admin
83cbfaf03f
- 修复Redis认证 (配置密码) - 启动Python管理后台 (端口9531, 15个功能开关) - 统一版本号 0.2.7 - 更新docker-compose.yml (镜像版本/Redis URL/Admin服务)
155 lines
4.7 KiB
Python
155 lines
4.7 KiB
Python
"""
|
||
百度网盘文件清理 — 删除文件 & 广告过滤
|
||
参考 cloud-auto-save 的 filter_ads + netdisk 的 delete
|
||
"""
|
||
|
||
import json
|
||
import logging
|
||
from typing import List
|
||
|
||
import requests
|
||
|
||
from ...errors import TransferError, TransferErrorCode
|
||
from .credential import BaiduCredentialManager, BAIDU_PAN_API
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 默认广告关键词
|
||
DEFAULT_AD_KEYWORDS = [
|
||
"公众号", "微信", "扫码", "加群", "QQ群", "广告",
|
||
"关注", "免费领取", "点击领取", "全网", "最全",
|
||
"防走丢", "防迷路", "备用", "务必下载", "必看",
|
||
"解压密码", "压缩密码",
|
||
]
|
||
|
||
|
||
class BaiduCleanup:
|
||
"""百度网盘文件清理 & 广告过滤"""
|
||
|
||
def __init__(self, session: requests.Session,
|
||
credential: BaiduCredentialManager,
|
||
ad_keywords: List[str] = None):
|
||
self.session = session
|
||
self.credential = credential
|
||
self.ad_keywords = ad_keywords or DEFAULT_AD_KEYWORDS
|
||
|
||
# ─── 删除文件 ────────────────────────────────────────────
|
||
|
||
def delete_files(self, paths: List[str]) -> bool:
|
||
"""批量删除文件（按网盘路径）
|
||
|
||
POST /api/filemanager?opera=delete&bdstoken={bdstoken}
|
||
Body: filelist=["/path/to/file1","/path/to/file2"]
|
||
|
||
Args:
|
||
paths: 文件在网盘中的完整路径列表，如 ["/dir/file.txt"]
|
||
|
||
Returns:
|
||
True 全部成功（包括文件不存在的 errno=2）
|
||
|
||
Raises:
|
||
TransferError: 删除失败
|
||
"""
|
||
if not paths:
|
||
logger.info("删除列表为空，跳过")
|
||
return True
|
||
|
||
bdstoken = self.credential.get_bdstoken()
|
||
url = f"{BAIDU_PAN_API}/api/filemanager"
|
||
params = {
|
||
"opera": "delete",
|
||
"bdstoken": bdstoken,
|
||
}
|
||
data = {
|
||
"filelist": json.dumps(paths, ensure_ascii=False),
|
||
}
|
||
headers = self.credential.get_headers()
|
||
headers["Content-Type"] = "application/x-www-form-urlencoded"
|
||
|
||
try:
|
||
resp = self.session.post(
|
||
url, params=params, data=data, headers=headers, timeout=30
|
||
)
|
||
resp.raise_for_status()
|
||
result = resp.json()
|
||
except Exception as e:
|
||
raise TransferError(
|
||
TransferErrorCode.NETWORK_ERROR,
|
||
message=f"百度删除请求失败: {e}",
|
||
platform="baidu",
|
||
)
|
||
|
||
errno = result.get("errno", -1)
|
||
|
||
# errno=0 成功; errno=2 文件不存在（视为成功）
|
||
if errno in (0, 2):
|
||
logger.info(f"百度删除完成: {len(paths)} 个路径 (errno={errno})")
|
||
return True
|
||
|
||
raise TransferError(
|
||
TransferErrorCode.NETWORK_ERROR,
|
||
message=f"百度删除失败 (errno={errno})",
|
||
platform="baidu",
|
||
details=result,
|
||
)
|
||
|
||
# ─── 广告过滤 ────────────────────────────────────────────
|
||
|
||
def filter_ads(self, files: List[dict]) -> List[dict]:
|
||
"""根据文件名过滤广告文件
|
||
|
||
Args:
|
||
files: [{"fs_id": "xxx", "name": "xxx"}, ...]
|
||
|
||
Returns:
|
||
过滤后的文件列表，仅保留非广告文件
|
||
"""
|
||
if not self.ad_keywords:
|
||
return files
|
||
|
||
retained = []
|
||
removed = []
|
||
for f in files:
|
||
name = f.get("name", "")
|
||
if self._is_ad(name):
|
||
removed.append(name)
|
||
else:
|
||
retained.append(f)
|
||
|
||
if removed:
|
||
logger.info(f"广告过滤: 移除 {len(removed)} 个文件: {removed}")
|
||
return retained
|
||
|
||
def filter_ad_ids(self, file_ids: List[str],
|
||
file_names: List[str]) -> List[str]:
|
||
"""根据文件名过滤广告，返回保留的 file_ids
|
||
|
||
Args:
|
||
file_ids: 文件 ID 列表
|
||
file_names: 对应的文件名列表（与 file_ids 一一对应）
|
||
|
||
Returns:
|
||
过滤后的 file_ids
|
||
"""
|
||
if not self.ad_keywords:
|
||
return file_ids
|
||
|
||
retained = []
|
||
for fid, name in zip(file_ids, file_names):
|
||
if not self._is_ad(name):
|
||
retained.append(fid)
|
||
else:
|
||
logger.info(f"广告过滤: 移除 {name}")
|
||
|
||
return retained
|
||
|
||
def _is_ad(self, filename: str) -> bool:
|
||
"""判断文件名是否为广告"""
|
||
if not filename:
|
||
return False
|
||
name_lower = filename.lower()
|
||
for kw in self.ad_keywords:
|
||
if kw.lower() in name_lower:
|
||
return True
|
||
return False
|