import datetime import re from pathlib import Path from typing import Tuple, Optional, List, Union, Dict, Any from urllib.parse import unquote from torrentool.api import Torrent from app.core.cache import FileCache from app.core.cache import TTLCache from app.core.config import settings from app.core.context import Context, TorrentInfo, MediaInfo from app.core.meta import MetaBase from app.core.metainfo import MetaInfo from app.db.site_oper import SiteOper from app.db.systemconfig_oper import SystemConfigOper from app.log import logger from app.schemas.types import MediaType, SystemConfigKey from app.utils.http import RequestUtils from app.utils.string import StringUtils class TorrentHelper: """ 种子帮助类 """ def __init__(self): self._invalid_torrents = TTLCache(maxsize=128, ttl=3600 * 24) def download_torrent(self, url: str, cookie: Optional[str] = None, ua: Optional[str] = None, referer: Optional[str] = None, proxy: Optional[bool] = False) \ -> Tuple[Optional[Path], Optional[Union[str, bytes]], Optional[str], Optional[list], Optional[str]]: """ 把种子下载到本地 :return: 种子缓存相对路径【用于索引缓存】, 种子内容、种子主目录、种子文件清单、错误信息 """ if url.startswith("magnet:"): return None, url, "", [], f"磁力链接" # 构建 torrent 种子文件的缓存路径 cache_path = Path(StringUtils.md5_hash(url)).with_suffix(".torrent") # 缓存处理器 cache_backend = FileCache() # 读取缓存的种子文件 torrent_content = cache_backend.get(cache_path.as_posix(), region="torrents") if torrent_content: # 缓存已存在 try: # 获取种子目录和文件清单 folder_name, file_list = self.get_fileinfo_from_torrent_content(torrent_content) # 无法获取信息，则认为缓存文件无效 if not folder_name and not file_list: raise ValueError("无效的缓存种子文件") # 成功拿到种子数据 return cache_path, torrent_content, folder_name, file_list, "" except Exception as err: logger.error(f"处理缓存的种子文件 {cache_path} 时出错: {err}，将重新下载") # 下载种子文件 req = RequestUtils( ua=ua, cookies=cookie, referer=referer, proxies=settings.PROXY if proxy else None ).get_res(url=url, allow_redirects=False) while req and req.status_code in [301, 302]: url = req.headers['Location'] if url and url.startswith("magnet:"): return None, url, "", [], f"获取到磁力链接" req = RequestUtils( ua=ua, cookies=cookie, referer=referer, proxies=settings.PROXY if proxy else None ).get_res(url=url, allow_redirects=False) if req and req.status_code == 200: if not req.content: return cache_path, None, "", [], "未下载到种子数据" # 解析内容格式 if req.content.startswith(b"magnet:"): # 磁力链接 return cache_path, req.text, "", [], f"获取到磁力链接" if "下载种子文件".encode("utf-8") in req.content: # 首次下载提示页面 skip_flag = False try: forms = re.findall(r'(.*?)', req.text, re.S) for form in forms: action = form[0] if action != "?": continue action = url inputs = re.findall(r'', form[1], re.S) if inputs: data = {} for item in inputs: data[item[0]] = item[1] # 改写req req = RequestUtils( ua=ua, cookies=cookie, referer=referer, proxies=settings.PROXY if proxy else None ).post_res(url=action, data=data) if req and req.status_code == 200: # 检查是不是种子文件，如果不是抛出异常 Torrent.from_string(req.content) # 跳过成功 logger.info(f"触发了站点首次种子下载，已自动跳过：{url}") skip_flag = True elif req is not None: logger.warn(f"触发了站点首次种子下载，且无法自动跳过，" f"返回码：{req.status_code}，错误原因：{req.reason}") else: logger.warn(f"触发了站点首次种子下载，且无法自动跳过：{url}") break except Exception as err: logger.warn(f"触发了站点首次种子下载，尝试自动跳过时出现错误：{str(err)}，链接：{url}") if not skip_flag: return cache_path, None, "", [], "种子数据有误，请确认链接是否正确，如为PT站点则需手工在站点下载一次种子" # 种子内容 if req.content: # 检查是不是种子文件，如果不是仍然抛出异常 try: # 获取种子目录和文件清单 folder_name, file_list = self.get_fileinfo_from_torrent_content(req.content) if file_list: # 保存到缓存 cache_backend.set(cache_path.as_posix(), req.content, region="torrents") # 成功拿到种子数据 return cache_path, req.content, folder_name, file_list, "" except Exception as err: logger.error(f"种子文件解析失败：{str(err)}") # 种子数据仍然错误 return cache_path, None, "", [], "种子数据有误，请确认链接是否正确" # 返回失败 return cache_path, None, "", [], "" elif req is None: return cache_path, None, "", [], "无法打开链接" elif req.status_code == 429: return cache_path, None, "", [], "触发站点流控，请稍后重试" else: # 把错误的种子记下来，避免重复使用 self.add_invalid(url) return cache_path, None, "", [], f"下载种子出错，状态码：{req.status_code}" def get_torrent_info(self, torrent_path: Path) -> Tuple[str, List[str]]: """ 获取种子文件的文件夹名和文件清单 :param torrent_path: 种子文件路径 :return: 文件夹名、文件清单，单文件种子返回空文件夹名 """ if not torrent_path or not torrent_path.exists(): return "", [] try: torrentinfo = Torrent.from_file(torrent_path) # 获取文件清单 return self.get_fileinfo_from_torrent(torrentinfo) except Exception as err: logger.error(f"种子文件解析失败：{str(err)}") return "", [] @staticmethod def get_fileinfo_from_torrent(torrent: Torrent) -> Tuple[str, List[str]]: """ 从种子文件中获取文件清单 :param torrent: 种子文件对象 :return: 文件夹名、文件清单，单文件种子返回空文件夹名 """ if not torrent or not torrent.files: return "", [] # 获取文件清单 if len(torrent.files) == 1 and torrent.files[0].name == torrent.name: # 单文件种子目录名返回空 folder_name = "" # 单文件种子 file_list = [torrent.name] else: # 目录名 folder_name = torrent.name # 文件清单，如果一级目录与种子名相同则去掉 file_list = [] for fileinfo in torrent.files: file_path = Path(fileinfo.name) # 根路径 root_path = file_path.parts[0] if root_path == folder_name: file_list.append(str(file_path.relative_to(root_path))) else: file_list.append(fileinfo.name) logger.debug(f"解析种子：{torrent.name} => 目录：{folder_name}，文件清单：{file_list}") return folder_name, file_list def get_fileinfo_from_torrent_content(self, torrent_content: Union[str, bytes]) -> Tuple[str, List[str]]: """ 从种子内容中获取文件夹名和文件清单 :param torrent_content: 种子内容 :return: 文件夹名、文件清单，单文件种子返回空文件夹名 """ if not torrent_content: return "", [] # 检查是否为磁力链接 if StringUtils.is_magnet_link(torrent_content): return "", [] try: # 解析种子内容 torrentinfo = Torrent.from_string(torrent_content) # 获取文件清单 return self.get_fileinfo_from_torrent(torrentinfo) except Exception as err: logger.error(f"种子内容解析失败：{str(err)}") return "", [] @staticmethod def get_url_filename(req: Any, url: str) -> str: """ 从下载请求中获取种子文件名 """ if not req: return "" disposition = req.headers.get('content-disposition') or "" file_name = re.findall(r"filename=\"?(.+)\"?", disposition) if file_name: file_name = unquote(str(file_name[0].encode('ISO-8859-1').decode()).split(";")[0].strip()) if file_name.endswith('"'): file_name = file_name[:-1] elif url and url.endswith(".torrent"): file_name = unquote(url.split("/")[-1]) else: file_name = str(datetime.datetime.now()) return file_name @staticmethod def sort_torrents(torrent_list: List[Context]) -> List[Context]: """ 对种子对行排序：torrent、site、upload、seeder """ if not torrent_list: return [] # 下载规则 priority_rule: List[str] = SystemConfigOper().get( SystemConfigKey.TorrentsPriority) or ["torrent", "upload", "seeder"] # 站点上传量 site_uploads = { site.name: site.upload for site in SiteOper().get_userdata_latest() } def get_sort_str(_context): """ 拼装排序字段 """ _meta = _context.meta_info _torrent = _context.torrent_info _media = _context.media_info # 标题 _title = str(_media.title).ljust(200, ' ') # 站点优先级 _site_order = str(999 - (_torrent.site_order or 0)).rjust(3, '0') # 站点上传量 _site_upload = str(site_uploads.get(_torrent.site_name) or 0).rjust(30, '0') # 资源优先级 _torrent_order = str(_torrent.pri_order or 0).rjust(3, '0') # 资源做种数 _torrent_seeders = str(_torrent.seeders or 0).rjust(10, '0') # 季集 if not _meta.episode_list: # 无集数的排最前面 _season_episode = "%s%s" % (str(len(_meta.season_list)).rjust(3, '0'), "9999") else: # 集数越多的排越前面 _season_episode = "%s%s" % (str(len(_meta.season_list)).rjust(3, '0'), str(len(_meta.episode_list)).rjust(4, '0')) # 根据下载规则的顺序拼装排序字符串 _sort_str = _title for rule in priority_rule: if rule == "torrent": _sort_str += _torrent_order elif rule == "site": _sort_str += _site_order elif rule == "upload": _sort_str += _site_upload elif rule == "seeder": _sort_str += _torrent_seeders _sort_str += _season_episode return _sort_str # 排序 return sorted(torrent_list, key=lambda x: get_sort_str(x), reverse=True) def sort_group_torrents(self, torrent_list: List[Context]) -> List[Context]: """ 对媒体信息进行排序、去重 """ if not torrent_list: return [] # 排序 torrent_list = self.sort_torrents(torrent_list) # 控重 result = [] _added = [] # 排序后重新加入数组，按真实名称控重，即只取每个名称的第一个 for context in torrent_list: # 控重的主链是名称、年份、季、集 meta = context.meta_info media = context.media_info if media.type == MediaType.TV: media_name = "%s%s" % (media.title_year, meta.season_episode) else: media_name = media.title_year if media_name not in _added: _added.append(media_name) result.append(context) return result @staticmethod def get_torrent_episodes(files: list) -> list: """ 从种子的文件清单中获取所有集数 """ episodes = [] for file in files: if not file: continue file_path = Path(file) if not file_path.suffix or file_path.suffix.lower() not in settings.RMT_MEDIAEXT: continue # 只使用文件名识别 meta = MetaInfo(file_path.name) if not meta.begin_episode: continue episodes = list(set(episodes).union(set(meta.episode_list))) return episodes def is_invalid(self, url: str) -> bool: """ 判断种子是否是无效种子 """ return url in self._invalid_torrents def add_invalid(self, url: str): """ 添加无效种子 """ if url not in self._invalid_torrents: self._invalid_torrents[url] = True @staticmethod def match_torrent(mediainfo: MediaInfo, torrent_meta: MetaBase, torrent: TorrentInfo) -> bool: """ 检查种子是否匹配媒体信息 :param mediainfo: 需要匹配的媒体信息 :param torrent_meta: 种子识别信息 :param torrent: 种子信息 """ # 比对词条指定的tmdbid if torrent_meta.tmdbid or torrent_meta.doubanid: if torrent_meta.tmdbid and torrent_meta.tmdbid == mediainfo.tmdb_id: logger.info( f'{mediainfo.title} 通过词表指定TMDBID匹配到资源：{torrent.site_name} - {torrent.title}') return True if torrent_meta.doubanid and torrent_meta.doubanid == mediainfo.douban_id: logger.info( f'{mediainfo.title} 通过词表指定豆瓣ID匹配到资源：{torrent.site_name} - {torrent.title}') return True # 要匹配的媒体标题、原标题 media_titles = { StringUtils.clear_upper(mediainfo.title), StringUtils.clear_upper(mediainfo.original_title) } - {""} # 要匹配的媒体别名、译名 media_names = {StringUtils.clear_upper(name) for name in mediainfo.names if name} # 识别的种子中英文名 meta_names = { StringUtils.clear_upper(torrent_meta.cn_name), StringUtils.clear_upper(torrent_meta.en_name) } - {""} # 比对种子识别类型 if torrent_meta.type == MediaType.TV and mediainfo.type != MediaType.TV: logger.debug(f'{torrent.site_name} - {torrent.title} 种子标题类型为 {torrent_meta.type.value}，' f'不匹配 {mediainfo.type.value}') return False # 比对种子在站点中的类型 if torrent.category == MediaType.TV.value and mediainfo.type != MediaType.TV: logger.debug(f'{torrent.site_name} - {torrent.title} 种子在站点中归类为 {torrent.category}，' f'不匹配 {mediainfo.type.value}') return False # 比对年份 if mediainfo.year: if mediainfo.type == MediaType.TV: # 剧集年份，每季的年份可能不同，没年份时不比较年份（很多剧集种子不带年份） if torrent_meta.year and torrent_meta.year not in [year for year in mediainfo.season_years.values()]: logger.debug(f'{torrent.site_name} - {torrent.title} 年份不匹配 {mediainfo.season_years}') return False else: # 电影年份，上下浮动1年，没年份时不通过 if not torrent_meta.year or torrent_meta.year not in [str(int(mediainfo.year) - 1), mediainfo.year, str(int(mediainfo.year) + 1)]: logger.debug(f'{torrent.site_name} - {torrent.title} 年份不匹配 {mediainfo.year}') return False # 比对标题和原语种标题 if meta_names.intersection(media_titles): logger.info(f'{mediainfo.title} 通过标题匹配到资源：{torrent.site_name} - {torrent.title}') return True # 比对别名和译名 if media_names: if meta_names.intersection(media_names): logger.info(f'{mediainfo.title} 通过别名或译名匹配到资源：{torrent.site_name} - {torrent.title}') return True # 标题拆分 if torrent_meta.org_string: # 只拆分出标题中的非英文单词进行匹配，英文单词容易误匹配（带空格的多个单词组合除外） titles = [StringUtils.clear_upper(t) for t in re.split( r'[\s/【】.\[\]\-]+', torrent_meta.org_string ) if not StringUtils.is_english_word(t)] # 在标题中判断是否存在标题、原语种标题 if media_titles.intersection(titles): logger.info(f'{mediainfo.title} 通过标题匹配到资源：{torrent.site_name} - {torrent.title}') return True # 在副标题中（非英文单词）判断是否存在标题、原语种标题、别名、译名 if torrent.description: subtitles = {StringUtils.clear_upper(t) for t in re.split( r'[\s/【】|]+', torrent.description) if not StringUtils.is_english_word(t)} if media_titles.intersection(subtitles) or media_names.intersection(subtitles): logger.info(f'{mediainfo.title} 通过副标题匹配到资源：{torrent.site_name} - {torrent.title}，' f'副标题：{torrent.description}') return True # 未匹配 logger.debug(f'{torrent.site_name} - {torrent.title} 标题不匹配，识别名称：{meta_names}') return False @staticmethod def filter_torrent(torrent_info: TorrentInfo, filter_params: Dict[str, str]) -> bool: """ 检查种子是否匹配订阅过滤规则 """ if not filter_params: return True # 匹配内容 content = (f"{torrent_info.title} " f"{torrent_info.description} " f"{' '.join(torrent_info.labels or [])} " f"{torrent_info.volume_factor}") # 包含 include = filter_params.get("include") if include: if not re.search(r"%s" % include, content, re.I): logger.info(f"{content} 不匹配包含规则 {include}") return False # 排除 exclude = filter_params.get("exclude") if exclude: if re.search(r"%s" % exclude, content, re.I): logger.info(f"{content} 匹配排除规则 {exclude}") return False # 质量 quality = filter_params.get("quality") if quality: if not re.search(r"%s" % quality, torrent_info.title, re.I): logger.info(f"{torrent_info.title} 不匹配质量规则 {quality}") return False # 分辨率 resolution = filter_params.get("resolution") if resolution: if not re.search(r"%s" % resolution, torrent_info.title, re.I): logger.info(f"{torrent_info.title} 不匹配分辨率规则 {resolution}") return False # 特效 effect = filter_params.get("effect") if effect: if not re.search(r"%s" % effect, torrent_info.title, re.I): logger.info(f"{torrent_info.title} 不匹配特效规则 {effect}") return False # 大小 size_range = filter_params.get("size") if size_range: if size_range.find("-") != -1: # 区间 size_min, size_max = size_range.split("-") size_min = float(size_min.strip()) * 1024 * 1024 size_max = float(size_max.strip()) * 1024 * 1024 if torrent_info.size < size_min or torrent_info.size > size_max: return False elif size_range.startswith(">"): # 大于 size_min = float(size_range[1:].strip()) * 1024 * 1024 if torrent_info.size < size_min: return False elif size_range.startswith("<"): # 小于 size_max = float(size_range[1:].strip()) * 1024 * 1024 if torrent_info.size > size_max: return False return True @staticmethod def match_season_episodes(torrent: TorrentInfo, meta: MetaBase, season_episodes: Dict[int, list]) -> bool: """ 判断种子是否匹配季集数 :param torrent: 种子信息 :param meta: 种子元数据 :param season_episodes: 季集数 {season:[episodes]} """ # 匹配季 seasons = season_episodes.keys() # 种子季 torrent_seasons = meta.season_list if not torrent_seasons: # 按第一季处理 torrent_seasons = [1] # 种子集 torrent_episodes = meta.episode_list if not set(torrent_seasons).issubset(set(seasons)): # 种子季不在过滤季中 logger.debug( f"种子 {torrent.site_name} - {torrent.title} 包含季 {torrent_seasons} 不是需要的季 {list(seasons)}") return False if not torrent_episodes: # 整季按匹配处理 return True if len(torrent_seasons) == 1: need_episodes = season_episodes.get(torrent_seasons[0]) if need_episodes \ and not set(torrent_episodes).intersection(set(need_episodes)): # 单季集没有交集的不要 logger.debug(f"种子 {torrent.site_name} - {torrent.title} " f"集 {torrent_episodes} 没有需要的集：{need_episodes}") return False return True