From 598d1663b1b60b24bed24ce01f2347cd5b45c2c0 Mon Sep 17 00:00:00 2001 From: Aqr-K <95741669+Aqr-K@users.noreply.github.com> Date: Fri, 23 Oct 2026 07:22:22 +0800 Subject: [PATCH] fix(download): preserve protocol failures and isolate site updates (#7143) * fix(download): preserve protocol failures and isolate site updates * test: cover tracker token lifecycle and record dependency facts --------- Co-authored-by: jxxghp --- app/api/endpoints/download.py | 3 +- app/application/rss.py | 16 +- app/application/site/mutation.py | 17 +- app/application/torrent/cache.py | 25 +- app/application/torrent/download.py | 76 +- app/chain/base.py | 10 +- app/chain/download/subtitle.py | 98 +-- app/chain/search/subtitle.py | 43 +- app/chain/site.py | 3 +- app/chain/torrents.py | 2 + app/db/oper/site.py | 118 ++- app/foundation/url.py | 19 + app/modules/indexer/parser/mtorrent.py | 4 + app/modules/indexer/parser/nexus_audiences.py | 22 +- app/modules/indexer/spider/mtorrent.py | 6 +- app/modules/indexer/spider/tnode.py | 27 +- app/modules/qbittorrent/__init__.py | 5 +- app/modules/qbittorrent/qbittorrent.py | 9 +- app/modules/rtorrent/__init__.py | 80 +- app/modules/rtorrent/rtorrent.py | 46 +- app/modules/transmission/transmission.py | 15 +- app/runtime/cache.py | 46 + app/schemas/exception.py | 6 + app/startup/composition/network.py | 59 +- app/startup/composition/site.py | 44 +- docs/architecture-overview.md | 2 +- .../architecture/complexity-baseline.json | 3 +- .../architecture/dependency-baseline.json | 32 +- tests/test_download_chain.py | 12 +- tests/test_indexer_category.py | 2 +- tests/test_indexer_spider_response_parity.py | 124 ++- .../test_review_download_site_regressions.py | 831 ++++++++++++++++++ tests/test_subtitle_signed_download.py | 2 +- 33 files changed, 1553 insertions(+), 254 deletions(-) create mode 100644 tests/test_review_download_site_regressions.py diff --git a/app/api/endpoints/download.py b/app/api/endpoints/download.py index 4f6b5bd90..eeca5b3c6 100644 --- a/app/api/endpoints/download.py +++ b/app/api/endpoints/download.py @@ -31,6 +31,7 @@ from app.domain.context import Context, MediaInfo, MusicInfo, SubtitleInfo, Torr from app.domain.media import is_music_media_source, normalize_music_type from app.domain.meta.metabase import MetaBase from app.domain.metainfo import MetaInfo +from app.foundation.url import UrlUtils from app.schemas.common import ServiceClientInfo as _SchemaServiceClientInfo from app.schemas.context import MediaInfo as _SchemaMediaInfo from app.schemas.context import SubtitleInfo as _SchemaSubtitleInfo @@ -85,7 +86,7 @@ def _prepare_subtitle_download( return False, "字幕站点信息不存在" subtitle.enclosure = clean_url - subtitle.site_cookie = site.cookie + subtitle.site_cookie = site.cookie if UrlUtils.same_origin(site.url, clean_url) else None subtitle.site_ua = site.ua subtitle.site_proxy = bool(site.proxy) return True, "" diff --git a/app/application/rss.py b/app/application/rss.py index 887e501db..f6487f363 100644 --- a/app/application/rss.py +++ b/app/application/rss.py @@ -454,6 +454,14 @@ class RssHelper: logger.warning(f"RSS文件过大: {len(raw_data) / 1024 / 1024:.1f}MB,跳过解析") return False + expired_messages = { + "RSS 链接已过期, 您需要获得一个新的!", + "RSS Link has expired, You need to get a new one!", + "RSS Link has expired, You need to get new!", + } + if ret.text.strip() in expired_messages: + return None + if raw_data: ret_xml = http_port.decode_xml( ret, @@ -601,14 +609,6 @@ class RssHelper: except Exception as e2: logger.error(f"解析RSS失败:{str(e2)} - {traceback.format_exc()}") - # RSS过期检查 - _rss_expired_msg = [ - "RSS 链接已过期, 您需要获得一个新的!", - "RSS Link has expired, You need to get a new one!", - "RSS Link has expired, You need to get new!" - ] - if ret_xml in _rss_expired_msg: - return None return False finally: if root is not None: diff --git a/app/application/site/mutation.py b/app/application/site/mutation.py index 15c53a364..1c17482f6 100644 --- a/app/application/site/mutation.py +++ b/app/application/site/mutation.py @@ -22,6 +22,7 @@ from app.application.subscription.contract import ( ) from app.application.subscription.delete import AsyncUnitOfWork from app.schemas.common import JsonData +from app.schemas.exception import SiteDomainConflictError from app.schemas.types import SystemConfigKey SiteMutationResult: TypeAlias = SiteWriteResult @@ -181,8 +182,12 @@ class SiteMutationCommand: "public": 1 if site_info.get("public") else 0, } ) - await self._repository.stage_create(SiteMutation(values)) - await self._commit() + try: + await self._repository.stage_create(SiteMutation(values)) + await self._commit() + except SiteDomainConflictError: + await self._unit_of_work.rollback() + return SiteMutationResult(False, f"{domain} 站点已存在") await self._publish_updated({"domain": domain}) return SiteMutationResult(True) @@ -204,8 +209,12 @@ class SiteMutationCommand: values["url"] = normalized_url values["domain"] = normalized_domain values.pop("id", None) - await self._repository.stage_update(site_id, SiteMutation(values)) - await self._commit() + try: + await self._repository.stage_update(site_id, SiteMutation(values)) + await self._commit() + except SiteDomainConflictError: + await self._unit_of_work.rollback() + return SiteMutationResult(False, f"{normalized_domain} 站点已存在") await self._publish_updated( { "site_id": site_id, diff --git a/app/application/torrent/cache.py b/app/application/torrent/cache.py index ee58edfad..5dd46111f 100644 --- a/app/application/torrent/cache.py +++ b/app/application/torrent/cache.py @@ -9,6 +9,7 @@ from app.domain.media import is_music_media_source, normalize_music_type from app.domain.meta.metamusic import MetaMusic from app.domain.metainfo import MetaInfo from app.foundation.crypto import HashUtils +from app.runtime.cache import async_cache_mutation from app.schemas.types import MUSIC_ENTITY_RECORDING, MediaSource, MediaType, MusicTargetEntityType @@ -113,12 +114,24 @@ class TorrentCacheRecognitionService: ) else: mediainfo.clear() - target_context.media_info = mediainfo - - video_cache, music_cache = self.torrents_chain.split_cache_contexts(cache_data) - video_file, music_file = self.torrents_chain.cache_files() - await self.torrents_chain.async_save_cache(video_cache, video_file) - await self.torrents_chain.async_save_cache(music_cache, music_file) + async with async_cache_mutation("torrent-snapshots"): + # 远端识别期间不占写锁;提交时重读当前快照,仅修改本次目标。 + latest = await self.torrents_chain.async_get_torrents() + current = next((context for context in latest.get(domain, []) + if HashUtils.md5(f"{context.torrent_info.title}{context.torrent_info.description}") + == torrent_hash), None) + if current is None: + return False, "种子缓存已刷新,目标条目不存在", None + current.media_info = mediainfo + video_cache, music_cache = self.torrents_chain.split_cache_contexts(latest) + video_file, music_file = self.torrents_chain.cache_files() + # 同一个资源不会跨分类移动;只写其所属缓存,避免无关文件失败或部分提交。 + is_music_cache = current.torrent_info.category in (MediaType.MUSIC, MediaType.MUSIC.value) + cache, filename = (music_cache, music_file) if is_music_cache else (video_cache, video_file) + try: + await self.torrents_chain.async_save_cache(cache, filename, strict=True) + except Exception: + return False, "种子缓存保存失败,请稍后重试", None return True, "重新识别完成", { "media_name": mediainfo.title if mediainfo else "", "media_year": mediainfo.year if mediainfo else "", diff --git a/app/application/torrent/download.py b/app/application/torrent/download.py index 0af68b35c..6874cb0a1 100644 --- a/app/application/torrent/download.py +++ b/app/application/torrent/download.py @@ -26,6 +26,7 @@ from app.domain.meta.metamusic import audio_quality_tier, normalize_audio_format from app.domain.metainfo import MetaInfo from app.foundation import text as text_tools from app.foundation.crypto import HashUtils +from app.foundation.url import UrlUtils from app.runtime.cache import FileCache, TTLCache from app.runtime.log import logger from app.schemas.media import resolve_media_identity @@ -327,6 +328,59 @@ class TorrentHelper: time.sleep(attempt + 1) return None + def _request_torrent_redirects( + self, + http_port: TorrentHttpPort, + *, + url: str, + cookie: Optional[str], + ua: Optional[str], + referer: Optional[str], + proxies: Optional[dict[str, Any]], + retry: bool, + ) -> tuple[Optional[TorrentResponsePort], str, Optional[str]]: + """ + 有界处理GET跳转,跨来源后永久移除原站点凭据和Referer。 + + :param http_port: 已装配HTTP端口 + :param url: 当前下载地址 + :param cookie: 仅供原始来源使用的站点Cookie + :param ua: 请求用户代理 + :param referer: 原始来源页地址 + :param proxies: 本次请求代理 + :param retry: 是否对普通GET执行既有瞬时失败重试 + :return: 最终响应、最终地址及允许在该地址使用的Cookie + :raises ValueError: 跳转缺少地址、目标非法、重复或超过上限 + """ + seen = set() + for _hop in range(9): + if url in seen: + raise ValueError("种子下载重定向出现环路") + seen.add(url) + response = self._request_torrent_get( + http_port, retry and not torrent_rules.is_expiring_download_url(url), + url=url, cookie=cookie, ua=ua, referer=referer, + proxies=proxies, allow_redirects=False, + ) + if response is None or response.status_code not in (301, 302, 303, 307, 308): + return response, url, cookie + location = response.headers.get("Location") or response.headers.get("location") + close = getattr(response, "close", None) + if callable(close): + close() + if not location: + raise ValueError("种子下载重定向缺少目标地址") + destination = urljoin(url, location) + if destination.startswith("magnet:"): + return None, destination, None + target = urlsplit(destination) + if target.scheme not in ("http", "https") or not target.hostname or target.username or target.password: + raise ValueError("种子下载重定向目标无效") + if not UrlUtils.same_origin(url, destination): + cookie, referer = None, None + url = destination + raise ValueError("种子下载重定向次数超过上限") + def _should_retry_torrent_response(self, response: Optional[TorrentResponsePort]) -> bool: """ 保留重定向、磁力和首次确认页,明确限流不得消耗瞬时重试预算。 @@ -380,19 +434,15 @@ class TorrentHelper: config = get_chain_runtime_config_snapshot() http_port = _require_torrent_port() retry = cache_invalid and not torrent_rules.is_expiring_download_url(url) - req = self._request_torrent_get( - http_port, retry, url=url, cookie=cookie, ua=ua, referer=referer, - proxies=config.proxy if proxy else None, allow_redirects=False, - ) - while req is not None and req.status_code in [301, 302]: - url = req.headers['Location'] - if url and url.startswith("magnet:"): - return None, url, "", [], "获取到磁力链接" - req = self._request_torrent_get( - http_port, retry and not torrent_rules.is_expiring_download_url(url), - url=url, cookie=cookie, ua=ua, referer=referer, - proxies=config.proxy if proxy else None, allow_redirects=False, + try: + req, url, cookie = self._request_torrent_redirects( + http_port, url=url, cookie=cookie, ua=ua, referer=referer, + proxies=config.proxy if proxy else None, retry=retry, ) + except ValueError as err: + return cache_path, None, "", [], str(err) + if url.startswith("magnet:"): + return None, url, "", [], "获取到磁力链接" if rate_limit_error := _torrent_rate_limit_error(req): return cache_path, None, "", [], rate_limit_error if req is not None and req.status_code == 200: @@ -416,7 +466,7 @@ class TorrentHelper: action, data = confirmation req = http_port.request( method="POST", url=action, cookie=cookie, ua=ua, - referer=url, proxies=config.proxy if proxy else None, data=data, + referer=url, proxies=config.proxy if proxy else None, data=data, allow_redirects=False, ) if rate_limit_error := _torrent_rate_limit_error(req): return cache_path, None, "", [], rate_limit_error diff --git a/app/chain/base.py b/app/chain/base.py index 9c1f20d49..17e439713 100644 --- a/app/chain/base.py +++ b/app/chain/base.py @@ -147,14 +147,20 @@ class ChainBase(RecognitionMixin, MessageProcessingMixin, NotificationMixin, met logger.error(f"异步加载缓存 {filename} 出错:{str(err)}") return None - async def async_save_cache(self, cache: Any, filename: str) -> None: + async def async_save_cache(self, cache: Any, filename: str, *, strict: bool = False) -> None: """ - 异步保存缓存 + 异步保存缓存,默认保留旧插件的失败日志合同。 + + :param cache: 待序列化的业务快照 + :param filename: 目标缓存键 + :param strict: 为True时向需要确认落盘的用例传播失败 """ try: await self.async_filecache.set(filename, pickle.dumps(cache)) except Exception as err: logger.error(f"异步保存缓存 {filename} 出错:{str(err)}") + if strict: + raise return def save_cache(self, cache: Any, filename: str) -> None: diff --git a/app/chain/download/subtitle.py b/app/chain/download/subtitle.py index fb1ca359c..edd8eba08 100644 --- a/app/chain/download/subtitle.py +++ b/app/chain/download/subtitle.py @@ -6,6 +6,7 @@ from __future__ import annotations import re import shutil +import tempfile import time from pathlib import Path from typing import TYPE_CHECKING, Any, Callable, List, Optional, Tuple, Union, cast @@ -33,6 +34,7 @@ from app.domain.context import ( SubtitleInfo, ) from app.domain.metainfo import MetaInfo +from app.foundation.url import UrlUtils from app.runtime.log import logger from app.schemas.file import FileItem as _SchemaFileItem from app.schemas.file import FileURI @@ -264,7 +266,9 @@ class DownloadSubtitleOwner(_DownloadOwnerBase): """ 生成安全的字幕文件名。 """ - file_name = Path(file_name or fallback_name).name + file_name = Path((file_name or fallback_name).replace("\\", "/")).name + if file_name in ("", ".", ".."): + file_name = Path(fallback_name).name if not Path(file_name).suffix and Path(fallback_name).suffix: file_name = f"{file_name}{Path(fallback_name).suffix}" return file_name @@ -471,10 +475,12 @@ class DownloadSubtitleOwner(_DownloadOwnerBase): return False, message, [] saved_files = [] - temp_file = self.runtime_config.temporary_path / file_name - temp_extract_dir = temp_file.with_name(temp_file.stem) + temp_root = None try: self.runtime_config.temporary_path.mkdir(parents=True, exist_ok=True) + temp_root = Path(tempfile.mkdtemp(prefix="subtitle-", dir=self.runtime_config.temporary_path)) + temp_file = temp_root / file_name + temp_extract_dir = temp_root / "extracted" temp_file.write_bytes(response.content) if self._is_subtitle_archive(file_name): try: @@ -492,6 +498,9 @@ class DownloadSubtitleOwner(_DownloadOwnerBase): temp_extract_dir, self.runtime_config.subtitle_extensions, ): + if not sub_file.resolve().is_relative_to(temp_extract_dir.resolve()): + logger.warning("字幕解压结果越过临时目录,拒绝上传") + continue uploaded_path, message = self._upload_subtitle_file( storage_chain=storage_chain, storage=storage, @@ -524,10 +533,8 @@ class DownloadSubtitleOwner(_DownloadOwnerBase): return False, message, [] finally: try: - if temp_extract_dir.exists(): - shutil.rmtree(temp_extract_dir) - if temp_file.exists(): - temp_file.unlink() + if temp_root is not None: + shutil.rmtree(temp_root) except Exception as err: logger.error(f"删除临时字幕文件失败:{str(err)}") @@ -634,62 +641,35 @@ class DownloadSubtitleOwner(_DownloadOwnerBase): if response.status_code != 200: logger.error(f"下载字幕文件失败:{sublink}") return - file_name = TorrentHelper.get_url_filename(response, sublink) - if not file_name: + raw_name = TorrentHelper.get_url_filename(response, sublink) + if not raw_name: logger.warn(f"链接不是字幕文件:{sublink}") return - archive_format = self._SUBTITLE_ARCHIVE_FORMATS.get( - Path(file_name).suffix.lower() - ) - if not archive_format: - if Path(file_name).suffix.lower() not in self.runtime_config.subtitle_extensions: - logger.warn(f"链接不是支持的字幕文件:{sublink} - {file_name}") - return - sub_file = self.runtime_config.temporary_path / file_name - sub_file.write_bytes(response.content) - if not working_dir_item.path: - logger.error("字幕工作目录路径为空") - return - target_sub_file = Path(working_dir_item.path) / sub_file.name - if storage_chain.get_file_item(storage, target_sub_file): - logger.info(f"字幕文件已存在:{target_sub_file}") - return - logger.info(f"转移字幕 {sub_file} 到 {target_sub_file} ...") - storage_chain.upload_file(working_dir_item, sub_file) + file_name = self._safe_subtitle_file_name(raw_name, "subtitle.srt") + archive_format = self._subtitle_archive_format(file_name) + if not archive_format and Path(file_name).suffix.lower() not in self.runtime_config.subtitle_extensions: + logger.warn(f"链接不是支持的字幕文件:{sublink} - {file_name}") return - - archive_file = self.runtime_config.temporary_path / file_name - archive_file.write_bytes(response.content) - archive_path = archive_file.with_name(archive_file.stem) - try: - archive_port.unpack( - archive_file, - archive_path, - archive_format=archive_format, - ) - for sub_file in archive_port.list_files( - archive_path, - self.runtime_config.subtitle_extensions, - ): - if not working_dir_item.path: - logger.error("字幕工作目录路径为空") - return - target_sub_file = Path(working_dir_item.path) / sub_file.name - if storage_chain.get_file_item(storage, target_sub_file): - logger.info(f"字幕文件已存在:{target_sub_file}") - continue - logger.info(f"转移字幕 {sub_file} 到 {target_sub_file} ...") - storage_chain.upload_file(working_dir_item, sub_file) - except Exception as err: - logger.error(f"字幕压缩包解压失败:{archive_file} - {str(err)}") - finally: + self.runtime_config.temporary_path.mkdir(parents=True, exist_ok=True) + # 临时目录只属于本次响应;清理范围不由远端文件名或压缩包名称决定。 + with tempfile.TemporaryDirectory(prefix="site-subtitle-", dir=self.runtime_config.temporary_path) as directory: + temporary = Path(directory) + subtitle_file = temporary / file_name + subtitle_file.write_bytes(response.content) try: - if archive_path.exists(): - shutil.rmtree(archive_path) - if archive_file.exists(): - archive_file.unlink() + if archive_format: + archive_path = temporary / "extracted" + archive_port.unpack(subtitle_file, archive_path, archive_format=archive_format) + files = archive_port.list_files(archive_path, self.runtime_config.subtitle_extensions) + else: + archive_path, files = temporary, [subtitle_file] + for sub_file in files: + if not sub_file.resolve().is_relative_to(archive_path.resolve()): + logger.warning("字幕解压结果越过临时目录,拒绝上传") + continue + self._upload_subtitle_file(storage_chain, storage, working_dir_item, sub_file) except Exception as err: - logger.error(f"删除临时文件失败:{str(err)}") + logger.error(f"保存站点字幕失败:{str(err)}") def download_site_subtitles( self, @@ -751,7 +731,7 @@ class DownloadSubtitleOwner(_DownloadOwnerBase): logger.info(f"找到字幕下载链接:{sublink},开始下载...") ret = http.get( sublink, - cookies=torrent.site_cookie, + cookies=torrent.site_cookie if UrlUtils.same_origin(torrent.page_url, sublink) else None, ua=torrent.site_ua, proxies=self.runtime_config.proxy if torrent.site_proxy else None, ) diff --git a/app/chain/search/subtitle.py b/app/chain/search/subtitle.py index a324f9034..2573a2e97 100644 --- a/app/chain/search/subtitle.py +++ b/app/chain/search/subtitle.py @@ -411,6 +411,7 @@ class SearchSubtitleOwner(_SearchOwnerBase): ) subtitles: List[SubtitleInfo] = [] + matches: List[SubtitleInfo] = [] search_count = 0 for search_word in keywords: if search_count > 0: @@ -424,22 +425,16 @@ class SearchSubtitleOwner(_SearchOwnerBase): or [] ) search_count += 1 - if not self.runtime_config.search_multiple_name and subtitles: - logger.info(f"共搜索到 {len(subtitles)} 个字幕,停止搜索") + matches = await run_in_threadpool( + self._parse_subtitle_result, subtitles=subtitles, mediainfo=mediainfo, + keyword=mediainfo.title, season_episodes=season_episodes, + episode=episode, custom_words=custom_words, + ) + if not self.runtime_config.search_multiple_name and matches: + logger.info(f"共匹配到 {len(matches)} 个字幕,停止别名搜索") break - return cast( - List[SubtitleInfo], - await run_in_threadpool( - self._parse_subtitle_result, - subtitles=subtitles, - mediainfo=mediainfo, - keyword=mediainfo.title, - season_episodes=season_episodes, - episode=episode, - custom_words=custom_words, - ), - ) + return matches async def _async_search_subtitles_for_media_stream( self, @@ -498,6 +493,7 @@ class SearchSubtitleOwner(_SearchOwnerBase): ) subtitles: List[SubtitleInfo] = [] + matches: List[SubtitleInfo] = [] search_count = 0 for search_word in keywords: if search_count > 0: @@ -516,8 +512,13 @@ class SearchSubtitleOwner(_SearchOwnerBase): } search_count += 1 - if not self.runtime_config.search_multiple_name and subtitles: - logger.info(f"共搜索到 {len(subtitles)} 个字幕,停止搜索") + matches = await run_in_threadpool( + self._parse_subtitle_result, subtitles=subtitles, mediainfo=mediainfo, + keyword=mediainfo.title, season_episodes=season_episodes, + episode=episode, custom_words=custom_words, + ) + if not self.runtime_config.search_multiple_name and matches: + logger.info(f"共匹配到 {len(matches)} 个字幕,停止别名搜索") break yield { @@ -527,15 +528,7 @@ class SearchSubtitleOwner(_SearchOwnerBase): "text": f"正在识别匹配 {len(subtitles)} 个候选字幕 ...", } - match_subtitles = await run_in_threadpool( - self._parse_subtitle_result, - subtitles=subtitles, - mediainfo=mediainfo, - keyword=mediainfo.title, - season_episodes=season_episodes, - episode=episode, - custom_words=custom_words, - ) + match_subtitles = matches final_items = [subtitle.to_dict() for subtitle in match_subtitles] yield { "type": "replace", diff --git a/app/chain/site.py b/app/chain/site.py index bcd3b70a0..1ce38b71f 100644 --- a/app/chain/site.py +++ b/app/chain/site.py @@ -749,7 +749,8 @@ class SiteChain(InteractionChainMixin, ChainBase): favicon_url = urljoin(url, fav_link[0]) with _open_site_response( - method="GET", url=favicon_url, cookie=cookie, timeout=15, ua=ua + method="GET", url=favicon_url, + cookie=cookie if url_tools.UrlUtils.same_origin(url, favicon_url) else None, timeout=15, ua=ua ) as res: if res: return favicon_url, base64.b64encode(res.content).decode() diff --git a/app/chain/torrents.py b/app/chain/torrents.py index ace713830..6828118b9 100644 --- a/app/chain/torrents.py +++ b/app/chain/torrents.py @@ -23,6 +23,7 @@ from app.domain.meta.metabase import MetaBase from app.domain.meta.metamusic import MetaMusic from app.domain.metainfo import MetaInfo from app.foundation.url import UrlUtils +from app.runtime.cache import cache_mutation from app.runtime.log import logger from app.runtime.stop import runtime_stop_state from app.schemas.media import resolve_media_identity @@ -606,6 +607,7 @@ class TorrentsChain(ChainBase): context.media_recognize_fail_count = 1 return context + @cache_mutation("torrent-snapshots") def refresh( self, stype: Optional[str] = None, diff --git a/app/db/oper/site.py b/app/db/oper/site.py index 47cc685ca..4401e0315 100644 --- a/app/db/oper/site.py +++ b/app/db/oper/site.py @@ -6,9 +6,11 @@ from datetime import datetime from typing import Any, List, Mapping, Optional, Tuple from sqlalchemy import delete as sqlalchemy_delete -from sqlalchemy import false, func, or_, select +from sqlalchemy import false, func, or_, select, text, update from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import Session +from sqlalchemy.sql.dml import Update +from sqlalchemy.sql.elements import TextClause from app.db.base import DbOper from app.db.models.site import Site @@ -16,6 +18,7 @@ from app.db.models.siteicon import SiteIcon from app.db.models.sitestatistic import SiteStatistic from app.db.models.siteuserdata import SiteUserData from app.db.oper.query import literal_contains +from app.schemas.exception import SiteDomainConflictError async def _async_first(session: AsyncSession, statement: Any) -> Optional[Site]: @@ -118,6 +121,64 @@ def _latest_userdata_join(statement: Any, subquery: Any) -> Any: ) +def _site_domain_lock(dialect: str) -> TextClause | Update: + """ + 在数据库事务中锁住站点域名准入,兼容空表及已有重复历史记录。 + + SQLite的零行UPDATE仍获取数据库写锁;PostgreSQL表锁覆盖空表与并发插入。 + 不持有进程锁,不提交事务,锁由调用方UoW提交或回滚时释放。 + + :param dialect: 当前会话绑定的SQLAlchemy方言名称 + :return: 获取事务写入准入的语句 + """ + if dialect == "postgresql": + return text("LOCK TABLE site IN SHARE ROW EXCLUSIVE MODE") + if dialect == "sqlite": + return update(Site).where(Site.id == -1).values(id=Site.id) + raise RuntimeError("站点域名并发准入不支持当前数据库方言") + + +def _check_site_domain(session: Session, domain: Optional[str], site_id: Optional[int] = None) -> None: + """ + 同步取得事务准入后重新检查域名,防止先查后插竞争。 + + :param session: 调用方持有的真实数据库会话 + :param domain: 新配置的规范域名 + :param site_id: 更新时允许保留该站点自身域名 + :raises SiteDomainConflictError: 已有另一站点占用域名 + """ + if domain is None: + return + session.execute(_site_domain_lock(session.get_bind().dialect.name)) + statement = select(Site.id).where(Site.domain == domain) + if site_id is not None: + statement = statement.where(Site.id != site_id) + if session.execute(statement.limit(1)).scalar_one_or_none() is not None: + raise SiteDomainConflictError("站点已存在") + + +async def _async_check_site_domain( + session: AsyncSession, domain: Optional[str], site_id: Optional[int] = None, +) -> None: + """ + 异步取得数据库事务准入后重新检查域名。 + + :param session: 调用方持有的真实数据库会话 + :param domain: 新配置的规范域名 + :param site_id: 更新时允许保留该站点自身域名 + :raises SiteDomainConflictError: 已有另一站点占用域名 + """ + if domain is None: + return + await session.execute(_site_domain_lock(session.get_bind().dialect.name)) + statement = select(Site.id).where(Site.domain == domain) + if site_id is not None: + statement = statement.where(Site.id != site_id) + result = await session.execute(statement.limit(1)) + if result.scalar_one_or_none() is not None: + raise SiteDomainConflictError("站点已存在") + + class SiteOper(DbOper): """ 站点管理 @@ -127,11 +188,18 @@ class SiteOper(DbOper): """ 新增站点 """ - site = Site(**kwargs) - if not self.get_by_domain(kwargs.get("domain")): - self._stage_create(site) + def stage(session: Session) -> Tuple[bool, str]: + """ + 在同一调用方事务中完成锁定、查重及暂存。 + """ + try: + _check_site_domain(session, kwargs.get("domain")) + except SiteDomainConflictError: + return False, "站点已存在" + session.add(Site(**kwargs)) return True, "新增站点成功" - return False, "站点已存在" + + return self._execute_sync_write(stage) def get(self, sid: int) -> Optional[Site]: """ @@ -169,6 +237,7 @@ class SiteOper(DbOper): raise RuntimeError("站点事务新增需要调用方提供 AsyncSession") values = dict(payload) values.pop("id", None) + await _async_check_site_domain(self._db, values.get("domain")) self._db.add(Site(**values)) async def stage_update( @@ -179,6 +248,8 @@ class SiteOper(DbOper): """ 暂存站点字段更新,事务由调用方统一提交。 """ + if "domain" in payload: + await _async_check_site_domain(self._db, payload["domain"], site_id) site = await self.async_get(site_id) if not site: return False @@ -331,20 +402,37 @@ class SiteOper(DbOper): """ 更新站点 """ - site = self.get(sid) - if not site: - return None - self._stage_update(site, payload) - return site + def stage(session: Session) -> Optional[Site]: + """ + 在同一事务中校验域名并更新现有站点。 + """ + if "domain" in payload: + _check_site_domain(session, payload["domain"], sid) + site: Optional[Site] = session.get(Site, sid) + if site is not None: + for key, value in payload.items(): + setattr(site, key, value) + return site + + return self._execute_sync_write(stage) async def async_update(self, sid: int, payload: dict[str, Any]) -> Optional[Site]: """ - 异步更新站点。 + 在调用方异步事务内校验域名并更新站点。 """ - site = await self.async_get(sid) - if site: - await self._stage_async_update(site, payload) - return site + async def stage(session: AsyncSession) -> Optional[Site]: + """ + 共享与站点创建相同的数据库准入锁。 + """ + if "domain" in payload: + await _async_check_site_domain(session, payload["domain"], sid) + site: Optional[Site] = await session.get(Site, sid) + if site is not None: + for key, value in payload.items(): + setattr(site, key, value) + return site + + return await self._execute_async_write(stage) def get_by_domain(self, domain: Optional[str]) -> Optional[Site]: """ diff --git a/app/foundation/url.py b/app/foundation/url.py index e71839f5b..e584613c3 100644 --- a/app/foundation/url.py +++ b/app/foundation/url.py @@ -44,6 +44,25 @@ class UrlUtils: return None return normalized_url + @staticmethod + def same_origin(source: object, destination: object) -> bool: + """ + 比较HTTP来源的协议、主机及有效端口;无效地址不能接收来源凭据。 + + :param source: 认证凭据所属的绝对HTTP地址 + :param destination: 准备请求的绝对HTTP地址 + :return: 两地址来源完全一致且均有效时为True + """ + source_url, destination_url = UrlUtils.normalize_http_url(source), UrlUtils.normalize_http_url(destination) + if source_url is None or destination_url is None: + return False + first, second = urlparse(source_url), urlparse(destination_url) + if first.username or first.password or second.username or second.password: + return False + return (first.scheme, first.hostname, first.port if first.port is not None else (443 if first.scheme == "https" else 80)) == ( + second.scheme, second.hostname, second.port if second.port is not None else (443 if second.scheme == "https" else 80) + ) + @staticmethod def standardize_base_url(host: str) -> str: """ diff --git a/app/modules/indexer/parser/mtorrent.py b/app/modules/indexer/parser/mtorrent.py index 93318d72b..8a72c820e 100644 --- a/app/modules/indexer/parser/mtorrent.py +++ b/app/modules/indexer/parser/mtorrent.py @@ -157,6 +157,10 @@ class MTorrentSiteUserInfo(SiteParserBase): self.seeding_size += page_seeding_size self.seeding_info.extend(page_seeding_info) + # 空页或没有有效条目的页面不再推进,汇总可能滞后于实时做种明细。 + if not page_seeding: + return None + # 查询总做种数 seeder_count = 0 try: diff --git a/app/modules/indexer/parser/nexus_audiences.py b/app/modules/indexer/parser/nexus_audiences.py index e1033b318..c7ed3feb3 100644 --- a/app/modules/indexer/parser/nexus_audiences.py +++ b/app/modules/indexer/parser/nexus_audiences.py @@ -371,16 +371,26 @@ class NexusAudiencesSiteUserInfo(NexusPhpSiteUserInfo): if not DomUtils.has_child_elements(html): return - for user_node in html.xpath('//*[@data-uploader-url or @data-uploader-stats]'): + userbars = html.xpath( + '//*[contains(concat(" ", normalize-space(@class), " "), " site-userbar ")]' + ) + scope = userbars[0] if userbars else html + nodes = scope.xpath('.//*[@data-uploader-url or @data-uploader-stats]') + # 兼容只有单个用户卡的旧页面;多用户卡且无用户栏时不能猜测当前账号。 + if not userbars: + nodes = [node for node in nodes if self.userid and + re.search(rf"userdetails\.php\?id={re.escape(str(self.userid))}(?:&|$)", + node.get("data-uploader-url") or "")] or (nodes if len(nodes) == 1 else []) + for user_node in nodes: self.__parse_user_identity(user_node) self.__parse_uploader_stats(user_node.get("data-uploader-stats")) # data-uploader-stats 不包含分享率,需从 compact metric 的 class 中读取。 - self.__parse_compact_metric(html, "ratio", "ratio") - self.__parse_compact_metric(html, "uploaded", "upload") - self.__parse_compact_metric(html, "downloaded", "download") - self.__parse_compact_metric(html, "bonus", "bonus") - self.__parse_compact_metric(html, "active", "active") + self.__parse_compact_metric(scope, "ratio", "ratio") + self.__parse_compact_metric(scope, "uploaded", "upload") + self.__parse_compact_metric(scope, "downloaded", "download") + self.__parse_compact_metric(scope, "bonus", "bonus") + self.__parse_compact_metric(scope, "active", "active") finally: if html is not None: del html diff --git a/app/modules/indexer/spider/mtorrent.py b/app/modules/indexer/spider/mtorrent.py index e5b7734c4..57294e22b 100644 --- a/app/modules/indexer/spider/mtorrent.py +++ b/app/modules/indexer/spider/mtorrent.py @@ -221,7 +221,11 @@ class MTorrentSpider: 统一判定搜索响应状态并投影 M-Team 种子结果。 """ if res and res.status_code == 200: - data = res.json().get('data', {}) + payload = res.json() + if str(payload.get("code")) != "0" or not isinstance(payload.get("data"), dict): + logger.warning(f"{self._name} 搜索失败,站点返回业务错误") + return True, [] + data = payload["data"] results = data.get("data") or [] # 分页事实只供逐页搜索判断末页,空结果仍按末页处理。 page_number, total_pages = data.get("pageNumber"), data.get("totalPages") diff --git a/app/modules/indexer/spider/tnode.py b/app/modules/indexer/spider/tnode.py index d919d8f93..32eb8aa44 100644 --- a/app/modules/indexer/spider/tnode.py +++ b/app/modules/indexer/spider/tnode.py @@ -7,13 +7,13 @@ from typing import Any, List, Optional, Tuple from app.adapters.network.http import AsyncRequestUtils, RequestUtils from app.foundation import temporal as time_tools -from app.foundation.singleton import SingletonClass +from app.foundation.crypto import HashUtils from app.runtime.cache import cached from app.runtime.log import logger from app.runtime.settings import get_runtime_setting -class TNodeSpider(metaclass=SingletonClass): +class TNodeSpider: """ TNode 页面令牌与种子搜索 API 适配器。 """ @@ -47,13 +47,28 @@ class TNodeSpider(metaclass=SingletonClass): if indexer.get('proxy'): self._proxy = get_runtime_setting('PROXY') self._cookie = indexer.get('cookie') + self._token_key = HashUtils.md5(repr((self._domain, self._cookie, indexer.get('ua')))) self._ua = indexer.get('ua') self._timeout = indexer.get('timeout') or 15 - @cached(region="indexer_spider", maxsize=1, ttl=60 * 60 * 24, skip_empty=True, shared_key="get_token") def __get_token(self) -> Optional[str]: + """ + 以当前站点和凭据隔离的缓存键获取同步CSRF令牌。 + """ + return self.__cached_token(self._token_key) + + async def __async_get_token(self) -> Optional[str]: + """ + 以当前站点和凭据隔离的缓存键获取异步CSRF令牌。 + """ + return await self.__async_cached_token(self._token_key) + + @cached(region="indexer_spider", maxsize=64, ttl=60 * 60 * 24, skip_empty=True, shared_key="get_token") + def __cached_token(self, _identity: str) -> Optional[str]: """ 同步请求站点首页并提取 CSRF 令牌。 + + :param _identity: 站点和凭据摘要,用于隔离共享令牌缓存 """ if not self._domain: return None @@ -63,10 +78,12 @@ class TNodeSpider(metaclass=SingletonClass): timeout=self._timeout).get_res(url=self._domain) return self.__parse_token_response(res) - @cached(region="indexer_spider", maxsize=1, ttl=60 * 60 * 24, skip_empty=True, shared_key="get_token") - async def __async_get_token(self) -> Optional[str]: + @cached(region="indexer_spider", maxsize=64, ttl=60 * 60 * 24, skip_empty=True, shared_key="get_token") + async def __async_cached_token(self, _identity: str) -> Optional[str]: """ 异步请求站点首页并提取 CSRF 令牌。 + + :param _identity: 站点和凭据摘要,用于隔离共享令牌缓存 """ if not self._domain: return None diff --git a/app/modules/qbittorrent/__init__.py b/app/modules/qbittorrent/__init__.py index 9c68d1c36..264648081 100644 --- a/app/modules/qbittorrent/__init__.py +++ b/app/modules/qbittorrent/__init__.py @@ -463,13 +463,12 @@ class QbittorrentModule(_DownloaderModuleBase[Qbittorrent]): :param hashs: 种子哈希或哈希列表 :param tags: 需要追加的标签列表 :param downloader: 下载器配置名称,省略时使用默认实例 - :return: 找到实例并调用标签接口后返回 True,不表示写入已经确认;找不到实例时返回 None + :return: 客户端接受标签写入时返回 True,RPC失败时返回 False;找不到实例时返回 None """ server = self.get_instance(downloader) if not server: return None - server.set_torrents_tag(ids=hashs, tags=tags) - return True + return server.set_torrents_tag(ids=hashs, tags=tags) def update_torrent( self, diff --git a/app/modules/qbittorrent/qbittorrent.py b/app/modules/qbittorrent/qbittorrent.py index bc1f495ed..ce335bb06 100644 --- a/app/modules/qbittorrent/qbittorrent.py +++ b/app/modules/qbittorrent/qbittorrent.py @@ -451,21 +451,22 @@ class Qbittorrent: logger.error(f"移除种子Tag出错:{str(err)}") return False - def set_torrents_tag(self, ids: Union[str, list[str]], tags: list[str]) -> None: + def set_torrents_tag(self, ids: Union[str, list[str]], tags: list[str]) -> bool: """ 向指定种子追加标签,保留原有标签。 - 未连接时直接返回,请求失败只记录日志;本方法不返回写入确认。 - :param ids: 种子哈希或哈希列表 :param tags: 待追加的标签列表 + :return: RPC成功完成时为True,未连接或失败时为False """ if not self.qbc: - return + return False try: self.qbc.torrents_add_tags(tags=tags, torrent_hashes=ids) + return True except Exception as err: logger.error(f"设置种子Tag出错:{str(err)}") + return False def is_force_resume(self) -> Optional[bool]: """ diff --git a/app/modules/rtorrent/__init__.py b/app/modules/rtorrent/__init__.py index 567425a86..6c11ff8aa 100644 --- a/app/modules/rtorrent/__init__.py +++ b/app/modules/rtorrent/__init__.py @@ -5,6 +5,7 @@ rTorrent 下载模块,负责实例路由、任务投影与下载文件选择 from pathlib import Path from typing import Dict, List, Optional, Set, Tuple, Union, cast +from app.domain import torrent as torrent_rules from app.domain.metainfo import MetaInfo from app.foundation import size as size_tools from app.foundation import temporal as time_tools @@ -115,6 +116,7 @@ class RtorrentModule(_DownloaderModuleBase[Rtorrent]): if not torrent_from_file and not is_magnet: return None, None, None, "添加种子任务失败:无法读取种子文件" resolved_content = cast(Union[str, bytes], resolved_content) + expected_hashes = torrent_rules.download_hashes(resolved_content) # 失败提示使用种子名称(磁力链接保留原文),不拼接种子文件的二进制内容 torrent_desc = torrent_from_file.name if torrent_from_file else resolved_content @@ -157,12 +159,8 @@ class RtorrentModule(_DownloaderModuleBase[Rtorrent]): if torrents: try: for torrent in torrents: - # 名称与大小相等则认为是同一个种子 - if torrent.get("name") == getattr( - torrent_from_file, "name", "" - ) and torrent.get("total_size") == getattr( - torrent_from_file, "total_size", 0 - ): + candidate_hash = str(torrent.get("hash") or "").lower() + if candidate_hash in expected_hashes: torrent_hash = cast(str, torrent.get("hash")) torrent_tags = [ str(t).strip() @@ -208,45 +206,14 @@ class RtorrentModule(_DownloaderModuleBase[Rtorrent]): else: torrent_hash = added_torrent_hash if episodes: - torrent_files = server.get_files(torrent_hash) - if not torrent_files: - return ( - downloader or self.get_default_config_name(), - torrent_hash, - torrent_layout, - "获取种子文件失败,下载任务可能在暂停状态", - ) - - # 不需要的文件ID - file_ids = [] - # 需要的集清单 - sucess_epidised = set() - try: - for torrent_file in torrent_files: - file_id = torrent_file["id"] - file_name = torrent_file["name"] - meta_info = MetaInfo(file_name) - if not meta_info.episode_list or not set( - meta_info.episode_list - ).issubset(episodes): - file_ids.append(file_id) - else: - sucess_epidised.update(meta_info.episode_list) - finally: - torrent_files.clear() - del torrent_files - selected_episodes = list(sucess_epidised) - if selected_episodes and file_ids: - # 设置不需要的文件优先级为0(不下载) - server.set_files( - torrent_hash=torrent_hash, file_ids=file_ids, priority=0 - ) - server.start_torrents(torrent_hash) + selected, message = self._select_episode_files(server, torrent_hash, episodes) + if not selected: + return None, None, None, message return ( downloader or self.get_default_config_name(), torrent_hash, torrent_layout, - f"添加下载成功,已选择集数:{selected_episodes}", + message, ) else: return ( @@ -256,6 +223,36 @@ class RtorrentModule(_DownloaderModuleBase[Rtorrent]): "添加下载成功", ) + @staticmethod + def _select_episode_files(server: Rtorrent, torrent_hash: str, episodes: Set[int]) -> Tuple[bool, str]: + """ + 对暂停添加的任务完成选集,所有RPC均成功才允许下载链确认任务。 + + :param server: 当前下载器客户端 + :param torrent_hash: 已暂停添加的任务Hash + :param episodes: 本次允许下载的目标集数 + :return: 是否确认选集启动成功及结果说明 + """ + torrent_files = server.get_files(torrent_hash) + if not torrent_files: + return False, "获取种子文件失败,下载任务保留暂停状态,请检查下载器" + file_ids, selected_episodes = [], set() + for torrent_file in torrent_files: + meta = MetaInfo(torrent_file["name"]) + if meta.episode_list and set(meta.episode_list).issubset(episodes): + selected_episodes.update(meta.episode_list) + else: + file_ids.append(torrent_file["id"]) + if not selected_episodes: + return False, "未匹配到目标集数,下载任务保留暂停状态" + if file_ids and not server.set_files(torrent_hash=torrent_hash, file_ids=file_ids, priority=0): + return False, "设置选集优先级失败,下载任务保留暂停状态" + if not server.start_torrents(torrent_hash): + stopped = server.stop_torrents(torrent_hash) + message = "启动下载失败,已请求暂停,请检查下载器确认状态" if stopped else "启动下载状态未知,请检查下载器" + return False, message + return True, f"添加下载成功,已选择集数:{sorted(selected_episodes)}" + def list_torrents( self, status: Optional[TorrentStatus] = None, @@ -448,6 +445,7 @@ class RtorrentModule(_DownloaderModuleBase[Rtorrent]): return server.delete_torrents( delete_file=delete_file if delete_file is not None else True, ids=hashs, + path_mapper=lambda path: self.normalize_return_path(path, downloader), ) def set_torrents_tag( diff --git a/app/modules/rtorrent/rtorrent.py b/app/modules/rtorrent/rtorrent.py index d8b4f48eb..725ed9f0a 100644 --- a/app/modules/rtorrent/rtorrent.py +++ b/app/modules/rtorrent/rtorrent.py @@ -7,7 +7,7 @@ import io import socket import traceback import xmlrpc.client -from collections.abc import Buffer +from collections.abc import Buffer, Callable from pathlib import Path from typing import Any, Dict, List, Optional, Tuple, TypedDict, Union, cast from urllib.parse import urlparse @@ -504,16 +504,18 @@ class Rtorrent: logger.error(f"停止种子出错:{str(err)}") return False - def delete_torrents(self, delete_file: bool, ids: Union[str, list[str]]) -> bool: + def delete_torrents(self, delete_file: bool, ids: Union[str, list[str]], + *, path_mapper: Optional[Callable[[Path], str]] = None) -> bool: """ 删除指定种子任务,并按选项删除下载文件。 - rTorrent 返回的内容路径在 MoviePilot 所在文件系统中删除。文件清理失败会记录警告; - 返回 True 只说明任务删除流程完成,不保证文件全部移除。 + 先将rTorrent内容路径映射到MoviePilot本地文件系统,文件清理成功后才移除任务。 + 清理失败或映射为远程存储时保留任务,以便用户核对和重试。 :param delete_file: 是否同时删除下载文件 :param ids: 种子哈希或哈希列表 - :return: 请求完成时返回 True;未连接或请求失败时返回 False,不代表任务已完成该动作 + :param path_mapper: 将下载器内容路径映射为宿主本地路径;省略时保留直接客户端调用合同 + :return: 文件清理及任务删除均完成时返回 True,失败时返回 False """ if not self._proxy: return False @@ -524,23 +526,23 @@ class Rtorrent: ids = [ids] for tid in ids: if delete_file: - # 先获取base_path用于删除文件 - try: - base_path = self._proxy.d.base_path(tid) - self._proxy.d.erase(tid) - if base_path: - import shutil - - path = Path(_rpc_text(base_path)) - if path.is_dir(): - shutil.rmtree(str(path), ignore_errors=True) - elif path.is_file(): - path.unlink(missing_ok=True) - except Exception as e: - logger.warning(f"删除种子文件出错:{str(e)}") - self._proxy.d.erase(tid) - else: - self._proxy.d.erase(tid) + base_path = self._proxy.d.base_path(tid) + if not base_path: + logger.warning("下载器未返回文件路径,保留任务以便人工处理") + return False + path = Path(_rpc_text(base_path)) + mapped = path_mapper(path) if path_mapper else path.as_posix() + # 云存储URI不能交给本地Path删除;下载器任务也保留以便核对。 + if ":" in mapped and not (len(mapped) > 2 and mapped[1] == ":"): + logger.warning("下载器文件映射为远程存储,拒绝本地删除") + return False + path = Path(mapped) + if path.is_dir(): + import shutil + shutil.rmtree(path) + elif path.exists(): + path.unlink() + self._proxy.d.erase(tid) return True except Exception as err: logger.error(f"删除种子出错:{str(err)}") diff --git a/app/modules/transmission/transmission.py b/app/modules/transmission/transmission.py index 41709b35b..a09a7993a 100755 --- a/app/modules/transmission/transmission.py +++ b/app/modules/transmission/transmission.py @@ -4,6 +4,7 @@ Transmission 客户端连接与下载任务操作封装。 import time from typing import List, Optional, Sequence, Tuple, Union +from urllib.parse import urlsplit import transmission_rpc from transmission_rpc import Client, File, Torrent @@ -32,7 +33,8 @@ class Transmission: "trackerStats", "leftUntilDone", "rateDownload", "rateUpload", "recheckProgress", "rateDownload", "rateUpload", "peersGettingFromUs", "peersSendingToUs", "uploadRatio", "uploadedEver", "downloadedEver", "downloadDir", - "error", "errorString", "doneDate", "queuePosition", "activityDate", "trackers"] + "error", "errorString", "doneDate", "queuePosition", "activityDate", "trackers", + "downloadLimit", "uploadLimit", "seedRatioLimit", "seedIdleLimit"] def __init__(self, host: Optional[str] = None, port: Optional[int] = None, username: Optional[str] = None, password: Optional[str] = None, @@ -51,12 +53,16 @@ class Transmission: self._host: Optional[str] = None self._port: Optional[int] = None self._protocol: object + self._path = "/transmission/rpc" if host and port: self._protocol, self._host, self._port = kwargs.get("protocol", "http"), host, port elif host: result = UrlUtils.parse_url_params(url=host) if result: - self._protocol, self._host, self._port, path = result + self._protocol, self._host, self._port, _path = result + self._path = urlsplit(host if "://" in host else f"http://{host}").path or self._path + if self._path == "/": + self._path = "/transmission/rpc" else: logger.error("Transmission配置不正确!") return @@ -102,6 +108,7 @@ class Transmission: trt = transmission_rpc.Client(protocol=self._protocol, # noqa host=self._host, port=self._port, + path=self._path, username=self._username, password=self._password, # transmission-rpc 运行时接受 Requests 的连接与读取超时二元组。 @@ -584,14 +591,14 @@ class Transmission: https://github.com/trim21/transmission-rpc/blob/8eb82629492a0eeb0bb565f82c872bf9ccdcb313/transmission_rpc/client.py#L654 :param hash_string: 种子哈希 - :param tracker_list: 要写入的 Tracker 地址列表,原样交给客户端 + :param tracker_list: 要写入的 Tracker 地址列表,每个地址作为独立层级交给客户端 :return: 请求完成时返回 True;未连接或请求失败时返回 False,不代表任务已完成该动作 """ if not self.trc: return False try: self.trc.change_torrent(ids=hash_string, - tracker_list=tracker_list) + tracker_list=[[url] for url in tracker_list] if tracker_list is not None else None) return True except Exception as err: logger.error(f"修改tracker出错:{str(err)}") diff --git a/app/runtime/cache.py b/app/runtime/cache.py index 97039fea9..99b907319 100644 --- a/app/runtime/cache.py +++ b/app/runtime/cache.py @@ -2,6 +2,7 @@ 缓存契约、进程内后端与装饰器,具体持久化后端由启动层注入。 """ +import asyncio import contextvars import inspect import logging @@ -53,6 +54,51 @@ _file_ttl_provider: Callable[[], int] = lambda: DEFAULT_CACHE_TTL _redis_probe: Optional[Callable[[], bool]] = None +# 多文件业务缓存的读改写必须共享同一进程锁,锁名由有限的宿主缓存合同提供。 +_mutation_locks: dict[str, threading.Lock] = {} +_mutation_locks_guard = threading.Lock() + + +def _cache_mutation_lock(name: str) -> threading.Lock: + """ + 获取同步线程与异步任务共用的业务缓存写锁。 + + :param name: 宿主定义的有限业务缓存名称 + :return: 当前进程该名称唯一的锁 + """ + with _mutation_locks_guard: + return _mutation_locks.setdefault(name, threading.Lock()) + + +@contextmanager +def cache_mutation(name: str) -> Generator[None, None, None]: + """ + 在同步调用中串行化一个业务缓存的完整读改写。 + + :param name: 宿主定义的有限业务缓存名称 + :yields: 本次读改写持有锁的作用域 + """ + with _cache_mutation_lock(name): + yield + + +@asynccontextmanager +async def async_cache_mutation(name: str) -> AsyncGenerator[None, None]: + """ + 非阻塞等待共享写锁,取消后不会有后台线程迟到取得锁。 + + :param name: 与同步调用共用的业务缓存名称 + :yields: 本次异步读改写持有锁的作用域 + """ + lock = _cache_mutation_lock(name) + while not lock.acquire(blocking=False): + await asyncio.sleep(0.01) + try: + yield + finally: + lock.release() + + def configure_cache_factories( *, backend_type_provider: Callable[[], str], diff --git a/app/schemas/exception.py b/app/schemas/exception.py index 6e6bf4f89..07df0b603 100644 --- a/app/schemas/exception.py +++ b/app/schemas/exception.py @@ -56,6 +56,12 @@ class BusinessRejectedError(Exception): pass +class SiteDomainConflictError(BusinessRejectedError): + """ + 站点域名已被另一持久化配置占用,当前事务应回滚并报告业务冲突。 + """ + + class OperationInterrupted(KeyboardInterrupt): """ 用于表示操作被中断。 diff --git a/app/startup/composition/network.py b/app/startup/composition/network.py index eee3bd2c3..6cffd08a1 100644 --- a/app/startup/composition/network.py +++ b/app/startup/composition/network.py @@ -7,6 +7,7 @@ from collections.abc import Mapping from dataclasses import dataclass from pathlib import Path from typing import Any, Callable, Optional, Union, cast +from urllib.parse import urljoin, urlsplit from app.adapters.network.doh import DohHelper from app.adapters.network.http import AsyncRequestUtils, RequestUtils @@ -59,6 +60,7 @@ from app.chain.system import ( configure_system_ports, reset_system_ports, ) +from app.foundation.url import UrlUtils from app.runtime.log import logger @@ -406,12 +408,57 @@ class _DownloadHttpAdapter: """ 发送下载链 GET 请求并原样保留响应三态。 """ - request = self._request(cookies=cookies, ua=ua, headers=headers, proxies=proxies, timeout=timeout) - kwargs: dict[str, Any] = {"raise_exception": raise_exception} - if params is not None: - kwargs["params"] = params - response = request.get_res(url, **kwargs) - return cast(Optional[DownloadResponsePort], response) + try: + return self._get_redirects(url, cookies=cookies, ua=ua, headers=headers, proxies=proxies, + timeout=timeout, params=params, raise_exception=raise_exception) + except ValueError: + if raise_exception: + raise + return None + + def _get_redirects( + self, url: str, *, cookies: Optional[Union[str, dict[str, str]]], ua: Optional[str], + headers: Optional[dict[str, str]], proxies: Optional[dict[str, str]], timeout: Optional[int], + params: Optional[dict[str, Any]], raise_exception: bool, + ) -> Optional[DownloadResponsePort]: + """ + 跟随有界GET跳转;跨来源后不恢复Cookie、Authorization或Referer。 + + :param url: 当前下载地址 + :param cookies: 仅供原始来源使用的Cookie + :param ua: 请求用户代理 + :param headers: 原始来源额外请求头 + :param proxies: 本次请求代理 + :param timeout: 请求超时 + :param params: 仅首跳附加的查询参数 + :param raise_exception: 是否保留底层传输异常 + :return: 最终响应或无响应 + :raises ValueError: 跳转非法、重复或超过上限 + """ + seen, request_headers = set(), dict(headers or {}) + for _hop in range(9): + if url in seen: + raise ValueError("下载重定向出现环路") + seen.add(url) + request = self._request(cookies=cookies, ua=ua, headers=request_headers, proxies=proxies, timeout=timeout) + kwargs: dict[str, Any] = {"raise_exception": raise_exception, "allow_redirects": False} + if params is not None: + kwargs["params"] = params + response = cast(Optional[DownloadResponsePort], request.get_res(url, **kwargs)) + if response is None or response.status_code not in (301, 302, 303, 307, 308): + return response + location = response.headers.get("Location") or response.headers.get("location") + response.close() + destination = urljoin(url, location or "") + target = urlsplit(destination) + if not location or UrlUtils.normalize_http_url(destination) is None or target.username or target.password: + raise ValueError("下载重定向目标无效") + if not UrlUtils.same_origin(url, destination): + cookies = None + request_headers = {key: value for key, value in request_headers.items() + if key.lower() not in ("cookie", "authorization", "proxy-authorization", "referer")} + url, params = destination, None + raise ValueError("下载重定向次数超过上限") def post( self, diff --git a/app/startup/composition/site.py b/app/startup/composition/site.py index 1d6268526..87847edbd 100644 --- a/app/startup/composition/site.py +++ b/app/startup/composition/site.py @@ -4,6 +4,7 @@ from contextlib import contextmanager from typing import Any, Callable, Iterator, Mapping, Optional, cast +from urllib.parse import urljoin, urlsplit from app.adapters.external.cookiecloud import CookieCloudHelper from app.adapters.external.ocr import OcrHelper @@ -42,6 +43,7 @@ from app.chain.site import ( configure_site_ports, reset_site_ports, ) +from app.foundation.url import UrlUtils class _RssHttpAdapter: @@ -222,16 +224,38 @@ class _SiteHttpAdapter: :param json_body: JSON 请求体,None 时保留无请求体行为 :yields: 在上下文有效期内提供响应或 None """ - request = RequestUtils( - headers=dict(headers or {}), - cookies=cookie or "", - ua=ua or "", - proxies=dict(proxies or {}), - timeout=timeout, - ) - request_options = {} if json_body is None else {"json": dict(json_body)} - with request.response_manager(method=method, url=url, **request_options) as response: - yield cast(Optional[SiteResponsePort], response) + request_headers = dict(headers or {}) + seen = set() + for _hop in range(9): + if url in seen: + raise ValueError("站点请求重定向出现环路") + seen.add(url) + request = RequestUtils( + headers=request_headers, cookies=cookie or "", ua=ua or "", + proxies=dict(proxies or {}), timeout=timeout, + ) + request_options: dict[str, Any] = {} if json_body is None else {"json": dict(json_body)} + if method.upper() == "GET": + request_options["allow_redirects"] = False + with request.response_manager(method=method, url=url, **request_options) as response: + if method.upper() != "GET" or response is None or response.status_code not in (301, 302, 303, 307, 308): + yield cast(Optional[SiteResponsePort], response) + return + location = response.headers.get("Location") + if not location: + raise ValueError("站点请求重定向缺少目标地址") + destination = urljoin(url, location) + target = urlsplit(destination) + if target.scheme not in ("http", "https") or not target.hostname or target.username or target.password: + raise ValueError("站点请求重定向目标无效") + if not UrlUtils.same_origin(url, destination): + cookie = None + request_headers = { + key: value for key, value in request_headers.items() + if key.lower() not in ("cookie", "authorization", "proxy-authorization", "referer") + } + url = destination + raise ValueError("站点请求重定向次数超过上限") class _SiteBrowserAdapter: diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index 663b41021..385ab1e98 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -817,7 +817,7 @@ flowchart LR | 指标 | 当前值 | |---|---:| | Python 模块 | 1129 | -| 内部导入边 | 9,690 | +| 内部导入边 | 9,706 | | 非平凡 SCC | 1(29 个模块的 TMDB vendor 精确 containment) | | Application / Chain 具体 Adapter 直连 | 0 / 0 | | Direct egress | 51(精确 policy 准入,Application / Chain 均为 0) | diff --git a/tests/fixtures/architecture/complexity-baseline.json b/tests/fixtures/architecture/complexity-baseline.json index 3429993c1..29818bba4 100644 --- a/tests/fixtures/architecture/complexity-baseline.json +++ b/tests/fixtures/architecture/complexity-baseline.json @@ -101,7 +101,7 @@ "app/modules/qbittorrent/__init__.py:QbittorrentModule.download": 20, "app/modules/qbittorrent/__init__.py:QbittorrentModule.list_torrents": 20, "app/modules/qqbot/gateway.py:run_gateway": 42, - "app/modules/rtorrent/__init__.py:RtorrentModule.download": 19, + "app/modules/rtorrent/__init__.py:RtorrentModule.download": 16, "app/modules/rtorrent/__init__.py:RtorrentModule.list_torrents": 20, "app/modules/slack/__init__.py:SlackModule.message_parser": 16, "app/modules/slack/slack.py:Slack.send_medias_msg": 18, @@ -150,7 +150,6 @@ "app/modules/jellyfin/jellyfin.py:Jellyfin.get_movies": 6, "app/modules/qbittorrent/__init__.py:QbittorrentModule.download": 7, "app/modules/rtorrent/__init__.py:RtorrentModule.download": 6, - "app/modules/rtorrent/rtorrent.py:Rtorrent.delete_torrents": 6, "app/modules/slack/__init__.py:SlackModule.send_direct_message": 6, "app/modules/slack/slack.py:Slack.send_medias_msg": 6, "app/modules/transmission/__init__.py:TransmissionModule.download": 7, diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index e1464e378..b4d2d4905 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -641,9 +641,6 @@ "uses": [] }, { - "source": "app.modules.plex.plex", - "target": "plexapi", - "kind": "network_sdk", "bindings": [ "from:plexapi.audio.Album as Album", "from:plexapi.audio.Artist as Artist", @@ -659,6 +656,10 @@ "from:plexapi.video.Season as Season", "from:plexapi.video.Show as Show" ], + "fingerprint": "6109de36f963b5c6193ed60941a88856cbee9c37303dedf13da3a3be94478543", + "kind": "network_sdk", + "source": "app.modules.plex.plex", + "target": "plexapi", "uses": [ "Plex.__fetch_item|call:exceptions.NotFound", "Plex.__fetch_item|call:fetchItem", @@ -686,8 +687,7 @@ "Plex.refresh_library_by_items|call:library.update", "Plex.refresh_library_by_items|call:query", "Plex.refresh_root_library|call:library.update" - ], - "fingerprint": "6109de36f963b5c6193ed60941a88856cbee9c37303dedf13da3a3be94478543" + ] }, { "bindings": [ @@ -961,8 +961,8 @@ "runtime_only": true } }, - "edge_count": 9690, - "edge_sha256": "2acb67e3a6c9e7c10938a8dfde7571359a644b174e8ac2505f71fd1202976633", + "edge_count": 9706, + "edge_sha256": "17f7a368a31cf6d8ade3d18e6a5219249c8b82c436e290ac3173cbf3e5759b13", "edges": [ "app -> app.foundation", "app -> app.foundation.environment", @@ -2506,6 +2506,8 @@ "app.api.endpoints.download -> app.domain.meta", "app.api.endpoints.download -> app.domain.meta.metabase", "app.api.endpoints.download -> app.domain.metainfo", + "app.api.endpoints.download -> app.foundation", + "app.api.endpoints.download -> app.foundation.url", "app.api.endpoints.download -> app.schemas", "app.api.endpoints.download -> app.schemas.common", "app.api.endpoints.download -> app.schemas.context", @@ -4170,6 +4172,7 @@ "app.application.site.mutation -> app.application.subscription.delete", "app.application.site.mutation -> app.schemas", "app.application.site.mutation -> app.schemas.common", + "app.application.site.mutation -> app.schemas.exception", "app.application.site.mutation -> app.schemas.types", "app.application.site.query -> app.application", "app.application.site.query -> app.application.site", @@ -4339,6 +4342,8 @@ "app.application.torrent.cache -> app.domain.metainfo", "app.application.torrent.cache -> app.foundation", "app.application.torrent.cache -> app.foundation.crypto", + "app.application.torrent.cache -> app.runtime", + "app.application.torrent.cache -> app.runtime.cache", "app.application.torrent.cache -> app.schemas", "app.application.torrent.cache -> app.schemas.types", "app.application.torrent.download -> app.application", @@ -4355,6 +4360,7 @@ "app.application.torrent.download -> app.foundation", "app.application.torrent.download -> app.foundation.crypto", "app.application.torrent.download -> app.foundation.text", + "app.application.torrent.download -> app.foundation.url", "app.application.torrent.download -> app.runtime", "app.application.torrent.download -> app.runtime.cache", "app.application.torrent.download -> app.runtime.log", @@ -4762,6 +4768,8 @@ "app.chain.download.subtitle -> app.domain", "app.chain.download.subtitle -> app.domain.context", "app.chain.download.subtitle -> app.domain.metainfo", + "app.chain.download.subtitle -> app.foundation", + "app.chain.download.subtitle -> app.foundation.url", "app.chain.download.subtitle -> app.runtime", "app.chain.download.subtitle -> app.runtime.log", "app.chain.download.subtitle -> app.schemas", @@ -5774,6 +5782,7 @@ "app.chain.torrents -> app.foundation", "app.chain.torrents -> app.foundation.url", "app.chain.torrents -> app.runtime", + "app.chain.torrents -> app.runtime.cache", "app.chain.torrents -> app.runtime.log", "app.chain.torrents -> app.runtime.stop", "app.chain.torrents -> app.schemas", @@ -6693,6 +6702,8 @@ "app.db.oper.site -> app.db.models.siteuserdata", "app.db.oper.site -> app.db.oper", "app.db.oper.site -> app.db.oper.query", + "app.db.oper.site -> app.schemas", + "app.db.oper.site -> app.schemas.exception", "app.db.oper.subscribe -> app.application", "app.db.oper.subscribe -> app.application.subscription", "app.db.oper.subscribe -> app.application.subscription.contract", @@ -7869,7 +7880,7 @@ "app.modules.indexer.spider.tnode -> app.adapters.network", "app.modules.indexer.spider.tnode -> app.adapters.network.http", "app.modules.indexer.spider.tnode -> app.foundation", - "app.modules.indexer.spider.tnode -> app.foundation.singleton", + "app.modules.indexer.spider.tnode -> app.foundation.crypto", "app.modules.indexer.spider.tnode -> app.foundation.temporal", "app.modules.indexer.spider.tnode -> app.runtime", "app.modules.indexer.spider.tnode -> app.runtime.cache", @@ -8133,6 +8144,7 @@ "app.modules.redis -> app.schemas.types", "app.modules.rtorrent -> app.domain", "app.modules.rtorrent -> app.domain.metainfo", + "app.modules.rtorrent -> app.domain.torrent", "app.modules.rtorrent -> app.foundation", "app.modules.rtorrent -> app.foundation.size", "app.modules.rtorrent -> app.foundation.temporal", @@ -9938,6 +9950,8 @@ "app.startup.composition.network -> app.chain.message", "app.startup.composition.network -> app.chain.scraping", "app.startup.composition.network -> app.chain.system", + "app.startup.composition.network -> app.foundation", + "app.startup.composition.network -> app.foundation.url", "app.startup.composition.network -> app.runtime", "app.startup.composition.network -> app.runtime.log", "app.startup.composition.outbox -> app.adapters", @@ -10076,6 +10090,8 @@ "app.startup.composition.site -> app.application.torrent.download", "app.startup.composition.site -> app.chain", "app.startup.composition.site -> app.chain.site", + "app.startup.composition.site -> app.foundation", + "app.startup.composition.site -> app.foundation.url", "app.startup.composition.subscription -> app.adapters", "app.startup.composition.subscription -> app.adapters.external", "app.startup.composition.subscription -> app.adapters.external.server", diff --git a/tests/test_download_chain.py b/tests/test_download_chain.py index d98e7dbfd..b5c114148 100644 --- a/tests/test_download_chain.py +++ b/tests/test_download_chain.py @@ -841,14 +841,19 @@ def test_save_subtitle_response_accepts_rar_filename_from_header(monkeypatch, tm """ storage_chain = _FakeSubtitleStorageChain() temp_path = tmp_path / "temp" - extracted_dir = temp_path / "Hypnosis_AKA_Saimin_(1999)_480i_JAPANESE_NTSC_DVD_REMUX_MPEG-2_DD_2.0-MeeSta" - extracted_subtitle = extracted_dir / "Hypnosis_AKA_Saimin_(1999).srt" + extracted_files = [] def fake_unpack_archive(archive_file, extract_dir, archive_format=None): + """ + 按真实端口传入的独占目录构造RAR解压结果。 + """ assert archive_format == "rar" assert archive_file.suffix == ".rar" + assert extract_dir.is_relative_to(temp_path) extract_dir.mkdir(parents=True, exist_ok=True) + extracted_subtitle = extract_dir / "Hypnosis_AKA_Saimin_(1999).srt" extracted_subtitle.write_text("subtitle", encoding="utf-8") + extracted_files.append(extracted_subtitle) monkeypatch.setattr(download_subtitle, "StorageChain", lambda: storage_chain) monkeypatch.setattr( @@ -876,7 +881,8 @@ def test_save_subtitle_response_accepts_rar_filename_from_header(monkeypatch, tm assert success assert message == "字幕文件保存成功" assert saved_files == ["/downloads/Hypnosis_AKA_Saimin_(1999).srt"] - assert storage_chain.uploaded_files == [extracted_subtitle] + assert storage_chain.uploaded_files == extracted_files + assert not list(temp_path.iterdir()) def test_save_subtitle_response_rejects_unsupported_filename_from_header(monkeypatch, tmp_path): diff --git a/tests/test_indexer_category.py b/tests/test_indexer_category.py index 2c52b1753..f0caa1dad 100644 --- a/tests/test_indexer_category.py +++ b/tests/test_indexer_category.py @@ -37,7 +37,7 @@ def test_search_requests_use_explicit_categories_before_media_type( ids = selected_ids[:1] if selection == "single" else selected_ids cat = ",".join(str(category) for category in ids) if selection else selection expected_ids = ids if selection else default_ids - payload = {"success": True, "data": []} if module is not mtorrent else {"data": {"data": []}} + payload = {"success": True, "data": []} if module is not mtorrent else {"code": "0", "data": {"data": []}} response = SimpleNamespace(status_code=200, json=lambda: payload) request = AsyncMock(return_value=response) if asynchronous else Mock(return_value=response) transport = module.AsyncRequestUtils if asynchronous else module.RequestUtils diff --git a/tests/test_indexer_spider_response_parity.py b/tests/test_indexer_spider_response_parity.py index 16d876a45..21059c9ef 100644 --- a/tests/test_indexer_spider_response_parity.py +++ b/tests/test_indexer_spider_response_parity.py @@ -72,8 +72,8 @@ _PROCESSOR_CASES = { "mtorrent": { "processor": "_MTorrentSpider__process_response", "parser": "_MTorrentSpider__parse_result", - "success": {"data": {"data": [{"marker": "hit"}]}}, - "empty": {"data": {"data": []}}, + "success": {"code": "0", "data": {"data": [{"marker": "hit"}]}}, + "empty": {"code": "0", "data": {"data": []}}, }, "hddolby": { "processor": "_HddolbySpider__process_response", @@ -309,6 +309,126 @@ def test_tnode_token_projection_is_identical_for_sync_and_async_responses(monkey assert processor(None) is None +@pytest.mark.parametrize("asynchronous", [False, True]) +@pytest.mark.parametrize("changed", ["domain", "cookie", "ua"]) +def test_tnode_search_caches_tokens_per_current_connection_identity(tmp_path, monkeypatch, asynchronous, changed): + """ + 真实搜索入口在站点、Cookie 或UA变化后使用各自CSRF,重复搜索复用各自缓存。 + """ + first_config = {**_COMMON_INDEXER, "cookie": f"session={tmp_path}"} + second_config = {**first_config, changed: { + "domain": "https://other.tracker.example/", + "cookie": f"session=other-{tmp_path}", + "ua": "Other-MoviePilot-Test", + }[changed]} + identities = [(config["domain"], config["cookie"], config["ua"]) for config in (first_config, second_config)] + tokens = dict(zip(identities, ("first-token", "second-token"))) + pages, submitted = [], [] + + def http_client(*_args, **kwargs): + """ + 只替换HTTP边界,令牌缓存、首页解析和搜索请求组装均由真实实现执行。 + """ + def get_res(url): + """ + 按当前连接身份返回独立令牌,并记录实际首页请求。 + """ + identity = (url, kwargs["cookies"], kwargs["ua"]) + pages.append(identity) + return _FakeResponse(text=f'') + + def post_res(url, json): + """ + 记录搜索提交的令牌和页码,返回合法空结果。 + """ + submitted.append((url, {key.lower(): value for key, value in kwargs["headers"].items()}, json)) + return _FakeResponse({"data": {"torrents": []}}) + + async def async_get_res(url): + """ + 在异步搜索入口复用同一首页响应合同。 + """ + return get_res(url) + + async def async_post_res(url, json): + """ + 在异步搜索入口记录同一请求合同。 + """ + return post_res(url, json) + + from types import SimpleNamespace + + return SimpleNamespace(get_res=async_get_res if asynchronous else get_res, + post_res=async_post_res if asynchronous else post_res) + + monkeypatch.setattr("app.modules.indexer.spider.tnode.AsyncRequestUtils" if asynchronous else + "app.modules.indexer.spider.tnode.RequestUtils", http_client) + first, second = TNodeSpider(first_config), TNodeSpider(second_config) + for spider in (first, second, first, second): + result = asyncio.run(spider.async_search("Movie", page=2)) if asynchronous else spider.search("Movie", page=2) + assert result == (False, []) + + assert pages == identities + assert [headers["x-csrf-token"] for _url, headers, _params in submitted] == ["first-token", "second-token"] * 2 + assert [params["page"] for _url, _headers, params in submitted] == [3] * 4 + + +@pytest.mark.parametrize("asynchronous", [False, True]) +def test_tnode_invalid_homepage_is_retried_before_search(tmp_path, monkeypatch, asynchronous): + """ + 首页拒绝不能缓存为成功或提交搜索,重试取得有效令牌后才执行搜索。 + """ + from types import SimpleNamespace + + responses = [_FakeResponse(status_code=403), _FakeResponse(text='')] + pages, submitted = [], [] + + def http_client(*_args, **kwargs): + """ + 模拟首页先拒绝后成功的传输边界,保留真实令牌重试逻辑。 + """ + def get_res(url): + """ + 依次返回拒绝和恢复响应,并记录重试次数。 + """ + pages.append(url) + return responses.pop(0) + + def post_res(url, json): + """ + 记录取得有效令牌后才允许发起的搜索。 + """ + del url, json + submitted.append(kwargs["headers"]) + return _FakeResponse({"data": {"torrents": []}}) + + async def async_get_res(url): + """ + 在异步入口复用可恢复的首页响应序列。 + """ + return get_res(url) + + async def async_post_res(url, json): + """ + 在异步入口记录取得有效令牌后的搜索。 + """ + return post_res(url, json) + + return SimpleNamespace(get_res=async_get_res if asynchronous else get_res, + post_res=async_post_res if asynchronous else post_res) + + monkeypatch.setattr("app.modules.indexer.spider.tnode.AsyncRequestUtils" if asynchronous else + "app.modules.indexer.spider.tnode.RequestUtils", http_client) + spider = TNodeSpider({**_COMMON_INDEXER, "cookie": f"retry={tmp_path}"}) + for expected in [(True, []), (False, []), (False, [])]: + result = asyncio.run(spider.async_search("Movie")) if asynchronous else spider.search("Movie") + assert result == expected + assert pages == [_COMMON_INDEXER["domain"]] * 2 + assert len(submitted) == 2 + assert all({key.lower(): value for key, value in headers.items()}["x-csrf-token"] == "renewed" + for headers in submitted) + + @pytest.mark.parametrize( ("name", "credential", "search_kwargs"), [ diff --git a/tests/test_review_download_site_regressions.py b/tests/test_review_download_site_regressions.py new file mode 100644 index 000000000..6a4d8c4a2 --- /dev/null +++ b/tests/test_review_download_site_regressions.py @@ -0,0 +1,831 @@ +""" +下载、站点与字幕审查问题的隔离回归。 +""" + +import asyncio +import json +import pickle +from concurrent.futures import ThreadPoolExecutor +from contextlib import contextmanager +from pathlib import Path +from threading import Event +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock + +import pytest +from torrentool.bencode import Bencode +from torrentool.torrent import Torrent +from transmission_rpc import Client + +from app.application import rss +from app.application.torrent import download as torrent_download +from app.application.torrent.cache import TorrentCacheRecognitionService +from app.chain.download.subtitle import DownloadSubtitleOwner +from app.chain.search import SearchChain +from app.chain.site import SiteChain +from app.chain.torrents import TorrentsChain +from app.domain.context import Context, MediaInfo, SubtitleInfo, TorrentInfo +from app.domain.metainfo import MetaInfo +from app.foundation.crypto import HashUtils +from app.modules.qbittorrent import QbittorrentModule +from app.modules.rtorrent import RtorrentModule +from app.modules.rtorrent.rtorrent import Rtorrent +from app.modules.transmission.transmission import Transmission +from app.schemas.types import MediaSource, MediaType + + +def _torrent(piece=b'x'): + """ + 生成只有一个视频文件的真实种子内容。 + """ + return Bencode.encode({'info': {'name': 'Demo.S01', 'piece length': 16384, + 'pieces': piece * 20, 'files': [{'length': 100, 'path': ['Demo.S01E01.mkv']}]}}) + + +def test_f034_transmission_tracker_rpc_preserves_urls(): + """ + SDK编码后Tracker仍为完整URL,不能被拆成字符。 + """ + client = object.__new__(Client) + client._Client__protocol_version = 17 + calls = [] + client._request = lambda *args, **kwargs: calls.append((args, kwargs)) + server = object.__new__(Transmission) + server.trc = client + urls = ['https://tracker.invalid/announce', 'udp://backup.invalid:80/announce'] + assert server.update_tracker('a' * 40, urls) + assert calls[-1][0][1]['trackerList'].split() == urls + + +def test_f035_transmission_custom_rpc_path(monkeypatch): + """ + 配置完整URL时保留代理RPC路径且不发网络请求。 + """ + monkeypatch.setattr(Client, 'get_session', lambda self: SimpleNamespace(rename_partial_files=True)) + server = Transmission(host='https://fictional.invalid/proxy/transmission/rpc') + try: + assert server.trc._url == 'https://fictional.invalid:443/proxy/transmission/rpc' + finally: + server.trc._http_session.close() + + +def test_f036_transmission_requests_limit_fields(): + """ + 任务RPC字段包含后续统一投影需要的限制值。 + """ + assert {'downloadLimit', 'uploadLimit', 'seedRatioLimit', 'seedIdleLimit'} <= set(Transmission._trarg) + + +@pytest.mark.parametrize('files,set_ok,start_ok', [ + (None, True, True), + ([{'id': 0, 'name': 'Demo.S01E01.mkv'}], True, True), + ([{'id': 0, 'name': 'Demo.S01E01.mkv'}, {'id': 1, 'name': 'Demo.S01E02.mkv'}], False, True), + ([{'id': 0, 'name': 'Demo.S01E02.mkv'}], True, False), +]) +def test_f037_rtorrent_selection_failure_never_reports_hash(monkeypatch, files, set_ok, start_ok): + """ + 未确认选集及启动成功时不返回能被下载链结算的Hash。 + """ + content = _torrent() + server = SimpleNamespace(add_torrent=Mock(return_value=True), + get_torrent_id_by_tag=Mock(return_value=Torrent.from_string(content).info_hash), + get_files=Mock(return_value=files), set_files=Mock(return_value=set_ok), + start_torrents=Mock(return_value=start_ok), stop_torrents=Mock(return_value=True)) + module = object.__new__(RtorrentModule) + monkeypatch.setattr(module, 'get_instance', lambda name: server) + monkeypatch.setattr(module, 'normalize_path', lambda path, name: str(path)) + monkeypatch.setattr('app.modules.rtorrent.get_runtime_setting', lambda key: '') + can_start = bool(files and set_ok and any('E02' in f['name'] for f in files)) + result = module.download(content, Path('/downloads'), '', episodes={2}, downloader='rt') + assert result[1] is None + if not can_start: + server.start_torrents.assert_not_called() + + +def test_f038_rtorrent_duplicate_requires_infohash(monkeypatch): + """ + 同名称大小不同InfoHash的任务不能接管或修改标签。 + """ + other = Torrent.from_string(_torrent(b'y')) + server = SimpleNamespace(add_torrent=Mock(return_value=False), + get_torrents=Mock(return_value=([{'name': other.name, 'total_size': other.total_size, + 'hash': other.info_hash, 'tags': '已整理'}], False)), remove_torrents_tag=Mock()) + module = object.__new__(RtorrentModule) + monkeypatch.setattr(module, 'get_instance', lambda name: server) + monkeypatch.setattr(module, 'normalize_path', lambda path, name: str(path)) + monkeypatch.setattr('app.modules.rtorrent.get_runtime_setting', lambda key: '') + result = module.download(_torrent(), Path('/downloads'), '', downloader='rt') + assert result[1] is None + server.remove_torrents_tag.assert_not_called() + + +def test_f039_rtorrent_delete_uses_host_mapping(monkeypatch, tmp_path): + """ + 删除文件使用宿主映射路径,不能误删下载器命名空间同名文件。 + """ + host = tmp_path / 'host' / 'Demo.mkv' + remote = tmp_path / 'remote' / 'Demo.mkv' + for path in (host, remote): + path.parent.mkdir() + path.write_bytes(b'keep until mapped delete') + server = object.__new__(Rtorrent) + server._proxy = SimpleNamespace(d=SimpleNamespace(base_path=lambda tid: str(remote), erase=Mock())) + module = object.__new__(RtorrentModule) + monkeypatch.setattr(module, 'get_instance', lambda name: server) + monkeypatch.setattr(module, 'get_config', lambda name: SimpleNamespace(path_mapping=[(str(host.parent), str(remote.parent))])) + assert module.remove_torrents('a' * 40, delete_file=True, downloader='rt') + assert not host.exists() + assert remote.exists() + + +def test_f040_concurrent_subtitle_writes_are_isolated(monkeypatch, tmp_path): + """ + 两次同名字幕上传交错仍读到各自内容。 + """ + first_upload, second_written, first_read = Event(), Event(), Event() + uploaded = {} + + def upload(directory, path): + """ + 仅在存储上传边界控制顺序并读取真实临时内容。 + """ + if directory.path == '/target-A': + first_upload.set() + assert second_written.wait(5) + uploaded['A'] = path.read_bytes() + first_read.set() + else: + uploaded['B'] = path.read_bytes() + second_written.set() + assert first_read.wait(5) + return SimpleNamespace(path=f'{directory.path}/{path.name}') + + storage = SimpleNamespace(get_folder=lambda storage, path: SimpleNamespace(path=str(path)), + get_file_item=lambda *args: None, upload_file=upload) + owner = object.__new__(DownloadSubtitleOwner) + owner.runtime_config = SimpleNamespace(temporary_path=tmp_path, subtitle_extensions=('.srt',)) + monkeypatch.setattr('app.chain.download.subtitle.StorageChain', lambda: storage) + subtitle = SubtitleInfo(title='Demo', subtitle_id='1', file_name='subtitle.srt', enclosure='https://fictional.invalid/sub') + + def save(label): + """ + 保存一个独立响应。 + """ + return owner._save_subtitle_response(subtitle, SimpleNamespace(content=label.encode()), 'local', Path(f'/target-{label}')) + + with ThreadPoolExecutor(max_workers=2) as pool: + first = pool.submit(save, 'A') + assert first_upload.wait(5) + second = pool.submit(save, 'B') + assert first.result(10)[0] and second.result(10)[0] + assert uploaded == {'A': b'A', 'B': b'B'} + assert not list(tmp_path.iterdir()) + + +def test_f041_auto_subtitle_cannot_clean_external_directory(tmp_path): + """ + 恶意响应文件名及解压失败不触及临时根外已有目录。 + """ + from app.startup.composition.network import _DownloadArchiveAdapter + temporary = tmp_path / 'temporary' + temporary.mkdir() + outside = tmp_path / 'outside' + outside.mkdir() + victim = outside / 'keep.txt' + victim.write_text('keep') + owner = object.__new__(DownloadSubtitleOwner) + owner.runtime_config = SimpleNamespace(temporary_path=temporary, subtitle_extensions=('.srt',)) + response = SimpleNamespace(status_code=200, content=b'invalid zip', + headers={'content-disposition': f'attachment; filename="{outside}.zip"'}) + owner._save_site_subtitle_response(response=response, sublink='https://fictional.invalid/sub', + archive_port=_DownloadArchiveAdapter(), storage_chain=None, storage='local', + working_dir_item=SimpleNamespace(path='/target')) + assert victim.read_text() == 'keep' + assert not list(temporary.iterdir()) + + +@pytest.mark.parametrize('case', ['cross-origin', 'relative', 'loop']) +def test_f042_f043_f044_torrent_redirect_contract(monkeypatch, case): + """ + 跳转不泄露Cookie,支持相对地址并在环路有界终止。 + """ + content = _torrent() + calls = [] + original = 'https://site.invalid/download' + + def request(**kwargs): + """ + 模拟重定向与真实种子响应,限定最大调用防止旧代码挂起。 + """ + calls.append(kwargs) + assert len(calls) <= 12, 'redirect loop was not bounded' + if len(calls) == 1 or case == 'loop': + location = {'cross-origin': 'https://cdn.invalid/file', 'relative': '/file', 'loop': original}[case] + return SimpleNamespace(status_code=302, headers={'Location': location}, content=b'', text='') + assert kwargs['url'].startswith('https://') + return SimpleNamespace(status_code=200, headers={}, content=content, text='') + + monkeypatch.setattr(torrent_download, '_require_torrent_port', lambda: SimpleNamespace(request=request)) + monkeypatch.setattr(torrent_download, 'FileCache', lambda **kwargs: SimpleNamespace(get=lambda *a, **k: None, set=lambda *a, **k: None)) + result = torrent_download.TorrentHelper().download_torrent(original, cookie='private=secret') + if case == 'loop': + assert result[1] is None + else: + assert result[1] == content + assert calls[1]['cookie'] == (None if case == 'cross-origin' else 'private=secret') + if case == 'relative': + assert calls[1]['url'] == 'https://site.invalid/file' + + +def test_f045_qb_label_failure_propagates(monkeypatch): + """ + 标签RPC失败向模块调用者报告失败。 + """ + module = object.__new__(QbittorrentModule) + monkeypatch.setattr(module, 'get_instance', lambda name: SimpleNamespace(set_torrents_tag=lambda **kwargs: False)) + assert module.set_torrents_tag('a' * 40, ['tag']) is False + + +@pytest.mark.parametrize('message', ['RSS Link has expired, You need to get a new one!', 'RSS 链接已过期, 您需要获得一个新的!']) +def test_f047_expired_rss_preserves_renewal_signal(monkeypatch, message): + """ + 站点过期纯文本保留None合同以进入续期路径。 + """ + response = SimpleNamespace(status_code=200, content=message.encode(), text=message) + ports = (SimpleNamespace(get=lambda **kwargs: response, decode_xml=lambda response, **kwargs: response.text), None, None) + monkeypatch.setattr(rss, '_require_rss_ports', lambda: ports) + monkeypatch.setattr(rss.RssHelper, '_RssHelper__parse_with_rust', lambda self, value: None) + assert rss.RssHelper().parse('https://site.invalid/rss') is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize('fail_writes', [False, True]) +async def test_f049_f050_cache_reidentification_merges_and_reports_write_error(fail_writes): + """ + 并发识别不同条目保留两项修改,落盘失败不得返回成功。 + """ + original = {'site.invalid': [Context(meta_info=MetaInfo(f'Demo{index}.2026'), + media_info=MediaInfo(type=MediaType.MOVIE, title=f'Original {index}'), + torrent_info=TorrentInfo(title=f'Demo{index}.2026', description='', category=MediaType.MOVIE.value)) for index in (1, 2)]} + files = {'__torrents_cache__': pickle.dumps(original), '__torrents_music_cache__': pickle.dumps({})} + + async def get(filename): + """ + 每次解码真实快照,避免共享引用掩盖丢更新。 + """ + return files.get(filename) + + async def set_value(filename, value): + """ + 模拟持久化端口失败或实际保存快照。 + """ + if fail_writes: + raise OSError('fictional cache write failure') + await asyncio.sleep(0) + files[filename] = value + + chain = object.__new__(TorrentsChain) + chain.runtime_config = SimpleNamespace(subscribe_mode='spider') + chain.async_filecache = SimpleNamespace(get=get, set=set_value) + arrived, ready = [], asyncio.Event() + + async def recognize(**kwargs): + """ + 两次识别均读取旧快照后同时完成远端结果。 + """ + arrived.append(kwargs['media_id']) + if len(arrived) == 2 or fail_writes: + ready.set() + await ready.wait() + return MediaInfo(type=MediaType.MOVIE, title='New', media_source=MediaSource.TMDB, media_id=kwargs['media_id']) + + service = TorrentCacheRecognitionService(chain, SimpleNamespace(async_recognize_media=recognize)) + + async def execute(index): + """ + 从真实应用入口更新指定目标。 + """ + return await service.execute('site.invalid', HashUtils.md5(f'Demo{index}.2026'), media_source=MediaSource.TMDB, media_id=str(index)) + + results = [await execute(1)] if fail_writes else await asyncio.gather(execute(1), execute(2)) + if fail_writes: + assert results[0][0] is False + else: + assert all(result[0] for result in results) + assert [c.media_info.media_id for c in pickle.loads(files['__torrents_cache__'])['site.invalid']] == ['1', '2'] + + +def test_f067_favicon_cross_origin_drops_cookie(monkeypatch): + """ + HTML声明跨域图标时不转发站点登录Cookie。 + """ + calls = [] + + @contextmanager + def response(**kwargs): + """ + 提供无网络有界HTTP响应。 + """ + calls.append(kwargs) + yield SimpleNamespace(text='', content=b'image') + + monkeypatch.setattr('app.chain.site._open_site_response', response) + assert SiteChain._SiteChain__parse_favicon('https://site.invalid/', 'private=secret', '')[1] + assert calls[1]['cookie'] is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize('stream', [False, True]) +async def test_f069_subtitle_alias_stops_only_after_matching(monkeypatch, stream): + """ + 首个名称只有无关候选时继续查询别名,列表和渐进入口一致。 + """ + chain = object.__new__(SearchChain) + chain.runtime_config = SimpleNamespace(search_multiple_name=False) + media = MediaInfo(title='Review Show', names=['Review Show', 'Review Alias'], year='2024', + tmdb_id=999999, type=MediaType.TV, season=1, season_years={1: '2024'}) + wrong = SubtitleInfo(title='Unrelated Show 2024 S01E01', site=1, enclosure='https://site.invalid/wrong.zip') + correct = SubtitleInfo(title='Review Show 2024 S01E01', site=1, enclosure='https://site.invalid/correct.zip') + + async def search(**kwargs): + """ + 第一名称候选不匹配,别名返回可匹配字幕。 + """ + return [wrong] if kwargs['keyword'] == 'Review Show' else [correct] + + async def search_stream(**kwargs): + """ + 提供渐进查询相同远端结果。 + """ + yield {'type': 'append', 'items': await search(**kwargs)} + + async def no_sleep(_delay): + """ + 隔离搜索退避等待。 + """ + monkeypatch.setattr(chain, '_async_search_subtitles_all_sites', search) + monkeypatch.setattr(chain, '_async_search_subtitles_all_sites_stream', search_stream) + monkeypatch.setattr('app.chain.search.subtitle.asyncio.sleep', no_sleep) + monkeypatch.setattr('app.chain.search.plan.get_chain_runtime_config_snapshot', lambda: SimpleNamespace(max_search_name_limit=None)) + if stream: + events = [event async for event in chain._async_search_subtitles_for_media_stream(media, season=1, episode=1)] + result = events[-1]['subtitles'] + else: + result = await chain._async_search_subtitles_for_media(media, season=1, episode=1) + assert result == [correct] + + +def test_f070_tnode_instances_follow_current_configuration(): + """ + 相同类的新配置有独立客户端与Cookie。 + """ + from app.modules.indexer.spider.tnode import TNodeSpider + first = TNodeSpider({'domain': 'https://site.invalid/', 'cookie': 'old'}) + second = TNodeSpider({'domain': 'https://site.invalid/', 'cookie': 'new'}) + assert first is not second + assert first._cookie == 'old' and second._cookie == 'new' + + +def test_f071_mteam_business_error_is_not_empty_success(): + """ + HTTP成功但业务失败不得伪装为空末页。 + """ + from app.modules.indexer.spider.mtorrent import MTorrentSpider + spider = object.__new__(MTorrentSpider) + spider._name, spider._domain = 'review', 'site.invalid' + response = SimpleNamespace(status_code=200, json=lambda: {'code': '1', 'message': 'invalid token'}) + error, items = spider._MTorrentSpider__process_response(response) + assert error is True and items == [] + + +def test_f072_mteam_empty_seeding_page_stops(): + """ + 空明细页即使汇总仍有做种也终止翻页。 + """ + from app.modules.indexer.parser.mtorrent import MTorrentSiteUserInfo + parser = MTorrentSiteUserInfo('review', 'https://site.invalid/', '', None, None) + parser._torrent_seeding_params = {'pageNumber': 2} + parser._get_page_content = lambda **kwargs: json.dumps({'code': '0', 'data': {'seeder': 100}}) + assert parser._parse_user_torrent_seeding_info(json.dumps({'code': '0', 'data': {'data': []}})) is None + + +def test_f073_audiences_ignores_other_uploader(): + """ + 页面其他上传者不能覆盖顶部当前用户身份和流量。 + """ + from app.modules.indexer.parser.nexus_audiences import NexusAudiencesSiteUserInfo + parser = NexusAudiencesSiteUserInfo('review', 'https://audiences.me/', '', None, None) + content = '''
+ current
+
other
''' + parser._parse_user_traffic_info(content) + assert parser.userid == '101' and parser.username == 'current' + assert parser.upload == 1024 ** 4 + + +@pytest.mark.asyncio +async def test_f068_concurrent_site_creates_have_one_committed_owner(tmp_path): + """ + 两个独立事务都先读到不存在时,数据库串行准入仍只提交一个站点。 + """ + from sqlalchemy import select + from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + + from app.application.site.mutation import SiteMutationCommand + from app.db.adapters.site import SessionSiteRepository + from app.db.models.site import Site + from app.db.uow import SqlAlchemyAsyncUnitOfWork + engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path}/race.db') + async with engine.begin() as connection: + await connection.run_sync(Site.__table__.create) + sessions = async_sessionmaker(engine, expire_on_commit=False) + readers, both_read = [], asyncio.Event() + + async def create_one(): + """ + 每个请求使用独立真实仓储与UoW,仅首次查询加屏障。 + """ + async with sessions() as session: + repository = SessionSiteRepository(session) + read = repository.async_get_by_domain + + async def synchronized_read(domain): + """ + 保证两个请求都越过原先的先查后插检查。 + """ + result = await read(domain) + readers.append(domain) + if len(readers) == 2: + both_read.set() + await asyncio.wait_for(both_read.wait(), 5) + return result + + repository.async_get_by_domain = synchronized_read + command = SiteMutationCommand(repository=repository, unit_of_work=SqlAlchemyAsyncUnitOfWork(session), + auth_level_provider=lambda: 2, indexer_loader=AsyncMock(return_value={'name': 'Review'}), + domain_extractor=lambda url: 'site.invalid', url_normalizer=lambda url: 'https://site.invalid/', + publish_updated=AsyncMock(), publish_deleted=AsyncMock()) + return await command.create({'url': 'https://site.invalid/'}) + + try: + results = await asyncio.gather(create_one(), create_one()) + async with sessions() as session: + rows = list((await session.scalars(select(Site))).all()) + assert len(rows) == 1 + assert sorted(result.success for result in results) == [False, True] + finally: + await engine.dispose() + + +def test_f068_legacy_sync_add_is_transactional_and_preserves_old_duplicates(tmp_path): + """ + legacy同步入口也串行查重,既有重复记录和配置不被自动删除。 + """ + from sqlalchemy import create_engine, select + from sqlalchemy.orm import Session + + from app.db.models.site import Site + from app.db.oper.site import SiteOper + engine = create_engine(f'sqlite:///{tmp_path}/legacy.db') + Site.__table__.create(engine) + + def add_one(): + """ + 让两个工作线程各持有独立真实事务。 + """ + with Session(engine) as session, session.begin(): + return SiteOper(db=session).add(name='Review', domain='site.invalid', url='https://site.invalid/') + + try: + with ThreadPoolExecutor(max_workers=2) as pool: + results = list(pool.map(lambda _: add_one(), range(2))) + assert sorted(result[0] for result in results) == [False, True] + with Session(engine) as session, session.begin(): + session.add_all([Site(name='A', domain='legacy.invalid', url='https://legacy.invalid/'), + Site(name='B', domain='legacy.invalid', url='https://legacy.invalid/')]) + with Session(engine) as session, session.begin(): + assert SiteOper(db=session).add(name='C', domain='legacy.invalid', url='https://legacy.invalid/')[0] is False + assert [row.name for row in session.scalars(select(Site).where(Site.domain == 'legacy.invalid')).all()] == ['A', 'B'] + finally: + engine.dispose() + + +def test_f068_sync_domain_update_cannot_collide(tmp_path): + """ + 修改域名入口和新增共享准入,不绕过站点身份边界。 + """ + from sqlalchemy import create_engine + from sqlalchemy.orm import Session + + from app.db.models.site import Site + from app.db.oper.site import SiteOper, _site_domain_lock + from app.schemas.exception import SiteDomainConflictError + engine = create_engine(f'sqlite:///{tmp_path}/update.db') + Site.__table__.create(engine) + try: + with Session(engine) as session, session.begin(): + session.add_all([Site(id=1, name='A', domain='a.invalid', url='https://a.invalid/'), + Site(id=2, name='B', domain='b.invalid', url='https://b.invalid/')]) + with Session(engine) as session: + with pytest.raises(SiteDomainConflictError): + SiteOper(db=session).update(2, {'domain': 'a.invalid'}) + session.rollback() + assert session.get(Site, 2).domain == 'b.invalid' + assert str(_site_domain_lock('postgresql')) == 'LOCK TABLE site IN SHARE ROW EXCLUSIVE MODE' + finally: + engine.dispose() + + +@pytest.mark.asyncio +async def test_f068_async_domain_update_reports_conflict(tmp_path): + """ + 请求事务域名变更冲突以受控失败返回,原记录保持不变。 + """ + from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + + from app.application.site.mutation import SiteMutationCommand + from app.db.adapters.site import SessionSiteRepository + from app.db.models.site import Site + from app.db.uow import SqlAlchemyAsyncUnitOfWork + engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path}/update.db') + sessions = async_sessionmaker(engine, expire_on_commit=False) + try: + async with engine.begin() as connection: + await connection.run_sync(Site.__table__.create) + async with sessions() as session, session.begin(): + session.add_all([Site(id=1, name='A', domain='a.invalid', url='https://a.invalid/'), + Site(id=2, name='B', domain='b.invalid', url='https://b.invalid/')]) + async with sessions() as session: + command = SiteMutationCommand(repository=SessionSiteRepository(session), + unit_of_work=SqlAlchemyAsyncUnitOfWork(session), auth_level_provider=lambda: 2, + indexer_loader=AsyncMock(), domain_extractor=lambda url: 'a.invalid', + url_normalizer=lambda url: url, publish_updated=AsyncMock(), publish_deleted=AsyncMock()) + result = await command.update({'id': 2, 'url': 'https://a.invalid/'}) + assert result.success is False + assert (await session.get(Site, 2)).domain == 'b.invalid' + finally: + await engine.dispose() + + +@pytest.mark.parametrize('status', [301, 302, 303, 307, 308]) +def test_redirect_headers_do_not_reappear_after_returning_to_original_origin(monkeypatch, status): + """ + 跨域跳转再回原站时也不恢复已丢弃的敏感信息,并及时关闭中间响应。 + """ + calls, closes = [], [] + urls = ['https://site.invalid/start', 'https://cdn.invalid/ticket', 'https://site.invalid/final'] + content = _torrent() + + def request(**kwargs): + """ + 提供固定三跳链并记录请求凭据。 + """ + calls.append(kwargs) + index = len(calls) - 1 + return SimpleNamespace(status_code=status if index < 2 else 200, + headers={'Location': urls[index + 1]} if index < 2 else {}, + content=content if index == 2 else b'', text='', close=lambda: closes.append(index)) + + monkeypatch.setattr(torrent_download, '_require_torrent_port', lambda: SimpleNamespace(request=request)) + monkeypatch.setattr(torrent_download, 'FileCache', lambda **kwargs: SimpleNamespace(get=lambda *a, **k: None, set=lambda *a, **k: None)) + assert torrent_download.TorrentHelper().download_torrent(urls[0], cookie='secret', referer='https://site.invalid/private')[1] == content + assert [call['cookie'] for call in calls] == ['secret', None, None] + assert [call['referer'] for call in calls] == ['https://site.invalid/private', None, None] + assert closes == [0, 1] + + +def test_f067_real_site_adapter_strips_headers_on_redirect(monkeypatch): + """ + 真实组合根在同域图标随后跨域时也清除Cookie和认证头。 + """ + from app.startup.composition.site import _SiteHttpAdapter + requests = [] + + class Request: + """ + 仅替换HTTP传输,保留真实Adapter的逐跳判断。 + """ + + def __init__(self, **kwargs): + """ + 保存Adapter传给传输的认证参数。 + """ + self.options = kwargs + + @contextmanager + def response_manager(self, **kwargs): + """ + 模拟同来源入口指向其他来源。 + """ + requests.append((self.options, kwargs)) + yield SimpleNamespace(status_code=302 if len(requests) == 1 else 200, + headers={'Location': 'https://cdn.invalid/icon.png'} if len(requests) == 1 else {}, content=b'image') + + monkeypatch.setattr('app.startup.composition.site.RequestUtils', Request) + with _SiteHttpAdapter().open(method='GET', url='https://site.invalid/icon', cookie='secret', + headers={'Authorization': 'Bearer secret', 'Referer': 'private'}) as result: + assert result.content == b'image' + assert requests[1][0]['cookies'] == '' + assert requests[1][0]['headers'] == {} + assert all(options['allow_redirects'] is False for _, options in requests) + + +@pytest.mark.asyncio +async def test_cache_lock_cancelled_waiter_does_not_leak(): + """ + 异步等待共享缓存锁被取消后,不会迟到取得锁或阻断后继写入。 + """ + from app.runtime.cache import async_cache_mutation, cache_mutation + + async def acquire(): + """ + 在缓存锁内完成一次可观测写入。 + """ + async with async_cache_mutation('review-cancel-test'): + return True + + with cache_mutation('review-cancel-test'): + waiter = asyncio.create_task(acquire()) + await asyncio.sleep(0) + waiter.cancel() + with pytest.raises(asyncio.CancelledError): + await waiter + assert await asyncio.wait_for(acquire(), 1) + + +@pytest.mark.parametrize('fails', [False, True]) +def test_f045_real_qb_client_propagates_label_rpc_result(monkeypatch, fails): + """ + 从真实下载器客户端到模块均传播RPC成功或失败,不能只修模块返回值。 + """ + from app.modules.qbittorrent.qbittorrent import Qbittorrent + server = object.__new__(Qbittorrent) + server.qbc = SimpleNamespace(torrents_add_tags=Mock(side_effect=OSError('offline') if fails else None)) + module = object.__new__(QbittorrentModule) + monkeypatch.setattr(module, 'get_instance', lambda name: server) + assert module.set_torrents_tag('a' * 40, ['review']) is (not fails) + + +def test_f068_site_admission_serializes_separate_processes(tmp_path): + """ + 独立Python进程共享临时SQLite时,数据库事务锁仍只允许一个域名创建成功。 + """ + import os + import subprocess + import sys + + from sqlalchemy import create_engine, select + from sqlalchemy.orm import Session + + from app.db.models.site import Site + database = tmp_path / 'process-race.db' + engine = create_engine(f'sqlite:///{database}') + Site.__table__.create(engine) + script = ''' +import sys, time +from pathlib import Path +from sqlalchemy import create_engine +from sqlalchemy.orm import Session +from app.db.oper.site import SiteOper +folder, database, identity = Path(sys.argv[1]), sys.argv[2], sys.argv[3] +(folder / ("ready-" + identity)).touch() +deadline = time.monotonic() + 10 +while len(list(folder.glob("ready-*"))) < 2: + if time.monotonic() >= deadline: + raise RuntimeError("test barrier timed out") + time.sleep(0.01) +engine = create_engine("sqlite:///" + database) +with Session(engine) as session, session.begin(): + success, message = SiteOper(db=session).add(name="Review", domain="site.invalid", url="https://site.invalid/") + print("created=" + str(success), flush=True) +engine.dispose() +''' + environment = dict(os.environ, CONFIG_DIR=str(tmp_path / 'config')) + processes = [subprocess.Popen([sys.executable, '-c', script, str(tmp_path), str(database), str(i)], + env=environment, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) for i in (1, 2)] + try: + outputs = [process.communicate(timeout=20) for process in processes] + assert [process.returncode for process in processes] == [0, 0], outputs + assert sorted('created=True' in output for output, _ in outputs) == [False, True] + with Session(engine) as session: + assert len(session.scalars(select(Site)).all()) == 1 + finally: + for process in processes: + if process.poll() is None: + process.kill() + process.communicate() + engine.dispose() + + +def test_subtitle_temp_allocation_failure_is_reported(monkeypatch, tmp_path): + """ + 独占目录申请失败仍返回字幕保存失败,不清理非本次拥有的目录。 + """ + storage = SimpleNamespace(get_folder=lambda storage, path: SimpleNamespace(path=str(path))) + owner = object.__new__(DownloadSubtitleOwner) + owner.runtime_config = SimpleNamespace(temporary_path=tmp_path, subtitle_extensions=('.srt',)) + monkeypatch.setattr('app.chain.download.subtitle.StorageChain', lambda: storage) + monkeypatch.setattr('app.chain.download.subtitle.tempfile.mkdtemp', Mock(side_effect=OSError('offline disk full'))) + subtitle = SubtitleInfo(title='Demo', file_name='subtitle.srt', enclosure='https://site.invalid/sub') + result = owner._save_subtitle_response(subtitle, SimpleNamespace(content=b'subtitle'), 'local', Path('/target')) + assert result[0] is False and 'offline disk full' in result[1] + + +@pytest.mark.parametrize('loop', [False, True]) +def test_f042_subtitle_adapter_redirects_are_bounded_and_private(monkeypatch, loop): + """ + 字幕下载适配器也手动解析相对跳转,跨来源移除凭据并终止环路。 + """ + from app.startup.composition.network import _DownloadHttpAdapter + requests, closed = [], [] + + class Request: + """ + 仅替换HTTP端口,检查真实Adapter构造的请求字段。 + """ + + def __init__(self, **kwargs): + """ + 保存本次请求的凭据和HTTP头。 + """ + self.options = kwargs + + def get_res(self, url, **kwargs): + """ + 返回确定跳转序列,旧自动跟随参数也能被断言发现。 + """ + requests.append((url, self.options, kwargs)) + if len(requests) == 1 or loop: + target = 'https://site.invalid/start' if loop else '//cdn.invalid/sub.srt' + return SimpleNamespace(status_code=302, headers={'Location': target}, close=lambda: closed.append(url)) + return SimpleNamespace(status_code=200, headers={}, content=b'subtitle') + + monkeypatch.setattr('app.startup.composition.network.RequestUtils', Request) + response = _DownloadHttpAdapter().get('https://site.invalid/start', cookies='secret', + headers={'Authorization': 'Bearer secret', 'Cookie': 'secret', 'Referer': 'private'}) + if loop: + assert response is None and len(requests) <= 9 + else: + assert response.status_code == 200 + assert requests[1][0] == 'https://cdn.invalid/sub.srt' + assert requests[1][1]['cookies'] is None and requests[1][1]['headers'] == {} + assert closed == ['https://site.invalid/start'] + assert all(kwargs['allow_redirects'] is False for _, _, kwargs in requests) + + +def test_f042_signed_subtitle_cross_origin_has_no_site_cookie(): + """ + 有效签名授权下载跨域资源也不授权向该来源转发站点Cookie。 + """ + from app import schemas + from app.api.endpoints.download import _prepare_subtitle_download + from app.application.security.url import SecurityUtils + destination = 'https://cdn.invalid/sub.srt' + subtitle = schemas.SubtitleInfo(site=1, title='Review', enclosure=SecurityUtils.sign_url(destination, purpose='subtitle-download:1')) + query = SimpleNamespace(get_sync=lambda _: SimpleNamespace(url='https://site.invalid/', cookie='secret', ua='review', proxy=False)) + assert _prepare_subtitle_download(subtitle, query=query)[0] + assert subtitle.site_cookie is None + + +def test_f042_automatic_subtitle_external_link_drops_cookie(monkeypatch, tmp_path): + """ + 自动字幕详情页提供跨域下载链接时,不转发详情页站点登录凭据。 + """ + import app.chain.download.subtitle as subtitle_module + owner = object.__new__(DownloadSubtitleOwner) + owner.runtime_config = SimpleNamespace(download_subtitle=True, temporary_path=tmp_path, proxy=None) + destination = 'https://cdn.invalid/sub.srt' + owner._site_subtitle_links = lambda context: [destination] + owner._save_site_subtitle_response = Mock() + response = SimpleNamespace(status_code=200, content=b'subtitle', close=Mock()) + get = Mock(return_value=response) + monkeypatch.setattr(subtitle_module, 'StorageChain', lambda: object()) + monkeypatch.setattr(subtitle_module, '_resolve_site_subtitle_target_dir', lambda **kwargs: ('local', tmp_path)) + monkeypatch.setattr(subtitle_module, '_prepare_site_subtitle_directory', lambda **kwargs: (SimpleNamespace(path=str(tmp_path)), True)) + monkeypatch.setattr(subtitle_module, '_enqueue_completed_site_subtitles', lambda **kwargs: None) + monkeypatch.setattr(subtitle_module, '_download_ports_snapshot', lambda: (SimpleNamespace(get=get), object())) + context = Context(torrent_info=TorrentInfo(page_url='https://site.invalid/details?id=1', site_cookie='secret')) + owner.download_site_subtitles(context, tmp_path, torrent_content=_torrent()) + assert get.call_args.kwargs['cookies'] is None + response.close.assert_called_once() + + +@pytest.mark.parametrize('source,target,expected', [ + ('https://site.invalid/', 'https://site.invalid:443/sub', True), + ('https://site.invalid/', 'https://SITE.INVALID/sub', True), + ('https://site.invalid/', 'https://site.invalid:444/sub', False), + ('https://site.invalid/', 'https://site.invalid:0/sub', False), + ('https://site.invalid/', 'http://site.invalid/sub', False), + ('https://site.invalid/', 'https://other.invalid/sub', False), + ('https://site.invalid/', 'https://secret@site.invalid/sub', False), + ('invalid', 'https://site.invalid/', False), + ('https://site.invalid/', 'https://site.invalid:bad/', False), +]) +def test_request_origin_checks_scheme_host_and_effective_port(source, target, expected): + """ + 认证来源匹配考虑协议、有效端口和URL凭据,不能只比较主机名。 + """ + from app.foundation.url import UrlUtils + assert UrlUtils.same_origin(source, target) is expected diff --git a/tests/test_subtitle_signed_download.py b/tests/test_subtitle_signed_download.py index 4825b20a4..9d81911c3 100644 --- a/tests/test_subtitle_signed_download.py +++ b/tests/test_subtitle_signed_download.py @@ -271,7 +271,7 @@ def test_download_subtitle_cleans_url_and_uses_server_site_request_fields(monkey 按预期站点标识返回服务端凭据,隔离真实站点配置查询。 """ assert site_id == SUBTITLE_SITE_ID - return SimpleNamespace(cookie="server-cookie=1", ua="ServerUA", proxy=True) + return SimpleNamespace(url="https://example.test/", cookie="server-cookie=1", ua="ServerUA", proxy=True) def get_sync(self, site_id): """