From 48e4ec94cb6e5eec076870cfb51f8488e5ef63b5 Mon Sep 17 00:00:00 2001 From: Estrella Pan Date: Tue, 20 Oct 2026 17:11:24 +0200 Subject: [PATCH] =?UTF-8?q?perf(rss):=20=E6=AF=8F=E8=BD=AE=E6=AF=8F?= =?UTF-8?q?=E4=B8=AA=E8=AE=A2=E9=98=85=E5=8F=AA=E6=8B=89=E5=8F=96=E4=B8=80?= =?UTF-8?q?=E6=AC=A1=EF=BC=8C=E8=81=9A=E5=90=88=E5=88=86=E6=9E=90=E6=8C=89?= =?UTF-8?q?=20(url,=20rss=5Fid)=20=E8=B7=B3=E8=BF=87=E5=B7=B2=E7=9F=A5?= =?UTF-8?q?=E7=A7=8D=E5=AD=90=EF=BC=8C=E5=81=8F=E5=A5=BD=E5=8E=BB=E9=87=8D?= =?UTF-8?q?=E6=8C=89=E5=80=99=E9=80=89=E5=8A=A0=E8=BD=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 聚合订阅分析改在 RSSEngine.refresh_rss 内进行，复用按主机排队拉取的结果（保留 2 s 间隔，#1026），不再二次请求 - 分析前按 (url, rss_id) 过滤本订阅已记录的种子；其他订阅已知的种子仍参与匹配，以追加 rss_link（#1053） - 每轮只加载一次番剧列表，生成新规则后才重新加载 - 偏好去重：批次中没有偏好候选时直接返回，只为有候选的番剧加载已下载版本 - 删除不再使用的 pull_rss 与 search_aggregate --- backend/src/module/core/loops.py | 25 +-- backend/src/module/database/bangumi.py | 11 +- backend/src/module/database/rss.py | 6 - backend/src/module/database/torrent.py | 7 +- backend/src/module/rss/analyser.py | 26 ++- backend/src/module/rss/engine.py | 106 +++++++--- backend/src/test/test_loops.py | 5 +- .../test/test_rss_analyser_typed_identity.py | 10 +- backend/src/test/test_rss_engine_new.py | 186 +++++++++++++----- 9 files changed, 256 insertions(+), 126 deletions(-) diff --git a/backend/src/module/core/loops.py b/backend/src/module/core/loops.py index da405e96..78659476 100644 --- a/backend/src/module/core/loops.py +++ b/backend/src/module/core/loops.py @@ -25,29 +25,8 @@ async def rss_tick(analyser: RSSAnalyser, notifier: NotificationManager) -> None """Analyse aggregate RSS feeds and refresh the RSS engine once.""" async with DownloaderPool() as downloaders: async with Database() as db: - engine = RSSEngine(db) - # Analyse RSS - rss_list = await db.rss.search_aggregate() - for rss in rss_list: - try: - await analyser.rss_to_data(rss, engine) - except Exception: - # 仅当该 RSS 确实已在遍历期间被 API 删除时才视为预期情况； - # 其余异常可能是真实 bug，需在 WARNING 级别带完整堆栈。 - if await db.rss.search_id(rss.id) is None: - logger.debug( - "Skipping RSS id=%s, deleted during iteration", - rss.id, - ) - else: - logger.warning( - "Error analysing RSS id=%s (%s)", - rss.id, - rss.url, - exc_info=True, - ) - # Run RSS Engine - events = await engine.refresh_rss(downloaders) + # 每个订阅本轮只拉取一次，聚合分析与下载匹配共用结果（#1026） + events = await RSSEngine(db).refresh_rss(downloaders, analyser=analyser) if events: # 站内通知中心要求事件总是送达 notifier（send_event 内部先落库， # 再按 settings.notification.enable 决定是否外发）。 diff --git a/backend/src/module/database/bangumi.py b/backend/src/module/database/bangumi.py index 97ec7472..c5a5be96 100644 --- a/backend/src/module/database/bangumi.py +++ b/backend/src/module/database/bangumi.py @@ -601,8 +601,15 @@ class BangumiDatabase: result = await self.session.execute(statement) return list(result.scalars().all()) - async def match_list(self, torrent_list: list, rss_link: str) -> list: - match_datas = await self.search_all() + async def match_list( + self, + torrent_list: list, + rss_link: str, + match_datas: list[Bangumi] | None = None, + ) -> list: + # 调用方可传入本轮已加载的番剧列表，省去每次 search_all + if match_datas is None: + match_datas = await self.search_all() if not match_datas: return torrent_list diff --git a/backend/src/module/database/rss.py b/backend/src/module/database/rss.py index 2316ab26..740131dd 100644 --- a/backend/src/module/database/rss.py +++ b/backend/src/module/database/rss.py @@ -116,12 +116,6 @@ class RSSDatabase: result = await self.session.execute(select(RSSItem).where(RSSItem.enabled)) return list(result.scalars().all()) - async def search_aggregate(self) -> list[RSSItem]: - result = await self.session.execute( - select(RSSItem).where(and_(RSSItem.aggregate, RSSItem.enabled)) - ) - return list(result.scalars().all()) - async def delete(self, _id: int) -> bool: try: # 先删除引用该 RSS 的 torrent，避免外键约束报错 diff --git a/backend/src/module/database/torrent.py b/backend/src/module/database/torrent.py index ed532653..3edce663 100644 --- a/backend/src/module/database/torrent.py +++ b/backend/src/module/database/torrent.py @@ -47,13 +47,18 @@ class TorrentDatabase: ) return list(result.scalars().all()) - async def check_new(self, torrents_list: list[Torrent]) -> list[Torrent]: + async def check_new( + self, torrents_list: list[Torrent], rss_id: int | None = None + ) -> list[Torrent]: + """返回库中没有记录的种子；给出 ``rss_id`` 时只看该订阅的记录。""" if not torrents_list: return [] urls = [t.url for t in torrents_list] # SQLModel 类属性在 mypy 看来是普通字段类型而非 InstrumentedAttribute， # 无法识别 .in_()/.is_() 等查询方法（无官方 mypy 插件支持）。 statement = select(Torrent.url).where(Torrent.url.in_(urls)) # type: ignore[attr-defined] + if rss_id is not None: + statement = statement.where(Torrent.rss_id == rss_id) result = await self.session.execute(statement) existing_urls = set(result.scalars().all()) return [t for t in torrents_list if t.url not in existing_urls] diff --git a/backend/src/module/rss/analyser.py b/backend/src/module/rss/analyser.py index 516805e6..75fbf67c 100644 --- a/backend/src/module/rss/analyser.py +++ b/backend/src/module/rss/analyser.py @@ -114,28 +114,36 @@ class RSSAnalyser: return None async def rss_to_data( - self, rss: RSSItem, engine: RSSEngine, full_parse: bool = True + self, + rss: RSSItem, + torrents: list[Torrent], + engine: RSSEngine, + bangumi_list: list[Bangumi], ) -> list[Bangumi]: + """用本轮已拉取的 ``torrents`` 为聚合订阅生成新规则。 + + ``bangumi_list`` 是本轮已加载的番剧列表，避免每个订阅各查一次。 + """ # One RSS workflow can cross several await boundaries and parse the # same resource during Movie match, Bangumi match, and persistence. # Keep those stages on the engine selected when the workflow started. with parser_engine_snapshot(): - rss_torrents = await self.get_rss_torrents(rss.url, full_parse) + # 只跳过本订阅已记录的种子：其他订阅已知的种子仍要匹配， + # 才能把本订阅追加到 bangumi.rss_link（#1053） + torrents = await engine.db.torrent.check_new(torrents, rss_id=rss.id) + if not torrents: + return [] # Filter out already-known movies first - torrents_after_movies = await engine.db.movie.match_list( - rss_torrents, rss.url - ) + torrents_after_movies = await engine.db.movie.match_list(torrents, rss.url) # Then filter out already-known bangumi torrents_to_add = await engine.db.bangumi.match_list( - torrents_after_movies, rss.url + torrents_after_movies, rss.url, bangumi_list ) if not torrents_to_add: logger.debug("No new title has been found.") return [] # Parse remaining torrents - new_bangumi, new_movies = await self.torrents_to_data( - torrents_to_add, rss, full_parse - ) + new_bangumi, new_movies = await self.torrents_to_data(torrents_to_add, rss) for movie in new_movies: await engine.db.movie.add(movie) if new_bangumi: diff --git a/backend/src/module/rss/engine.py b/backend/src/module/rss/engine.py index 5cb30579..2b8b6d3b 100644 --- a/backend/src/module/rss/engine.py +++ b/backend/src/module/rss/engine.py @@ -3,7 +3,7 @@ import logging import re from collections import defaultdict from datetime import datetime, timezone -from typing import Optional +from typing import TYPE_CHECKING, Optional from urllib.parse import urlparse from ab_sdk import points @@ -40,6 +40,9 @@ from module.parser.title_parser import _apply_title_hooks from module.plugin import host as plugin_host from module.plugin.views import bangumi_info, torrent_info +if TYPE_CHECKING: + from .analyser import RSSAnalyser + logger = logging.getLogger(__name__) # Delay between consecutive requests to the same host. Firing all feeds of one @@ -162,16 +165,12 @@ class RSSEngine: msg_zh="删除 RSS 成功。", ) - async def pull_rss(self, rss_item: RSSItem) -> list[Torrent]: - torrents = await self._get_torrents(rss_item) - new_torrents = await self.db.torrent.check_new(torrents) - return new_torrents - async def _pull_rss_with_status( self, rss_item: RSSItem ) -> tuple[list[Torrent], Optional[str]]: + """拉取订阅的全部条目（不去重），聚合分析与下载匹配共用这一份结果。""" try: - torrents = await self.pull_rss(rss_item) + torrents = await self._get_torrents(rss_item) return torrents, None except Exception as e: logger.warning(f"Failed to fetch RSS {rss_item.name}: {e}") @@ -279,12 +278,17 @@ class RSSEngine: ``refresh_rss`` 内部过滤，不做跨请求持久化。 """ skip_ids: set[int] = set() + candidate_ids = {b.id for _, b in matched if b.id in preference_bangumi} + if not candidate_ids: + # 本批次没有偏好番剧的候选：无需解析已下载版本 + return skip_ids # 已下载版本：按完整内容身份记录当前最高 (偏好得分, 修订号)。 + # 只解析本批次有候选的番剧。 existing_best: dict[_PreferenceKey, _PreferenceRank] = {} for bangumi_id, torrents in existing_downloaded.items(): bangumi = preference_bangumi.get(bangumi_id) - if not bangumi: + if not bangumi or bangumi_id not in candidate_ids: continue for torrent in torrents: release = parse_configured_release_title(torrent.name) @@ -335,17 +339,55 @@ class RSSEngine: return skip_ids async def refresh_rss( - self, downloaders: DownloaderPool, rss_id: Optional[int] = None + self, + downloaders: DownloaderPool, + rss_id: Optional[int] = None, + analyser: "RSSAnalyser | None" = None, ) -> list[SystemEvent]: """Refresh feeds with one parser engine for the complete workflow. 匹配到的种子投递到规则的下载器实例，规则未指定时用订阅的，再退回默认实例。 + 给出 ``analyser`` 时，先用同一次拉取的结果为聚合订阅生成新规则。 """ with parser_engine_snapshot(): - return await self._refresh_rss(downloaders, rss_id) + return await self._refresh_rss(downloaders, rss_id, analyser) + + async def _analyse_aggregate( + self, + analyser: "RSSAnalyser", + item_results: list[tuple[RSSItem, tuple[list[Torrent], Optional[str]]]], + bangumi_list: list[Bangumi], + ) -> list[Bangumi]: + """为聚合订阅生成新规则，返回之后用于匹配的番剧列表。""" + for rss_item, (torrents, error) in item_results: + if not rss_item.aggregate or error: + continue + try: + if await analyser.rss_to_data(rss_item, torrents, self, bangumi_list): + # 新规则要能被后续订阅与本轮下载匹配到 + bangumi_list = await self.db.bangumi.search_all() + except Exception: + # 仅当该 RSS 确实已在遍历期间被 API 删除时才视为预期情况； + # 其余异常可能是真实 bug，需在 WARNING 级别带完整堆栈。 + if await self.db.rss.search_id(rss_item.id) is None: + logger.debug( + "Skipping RSS id=%s, deleted during iteration", + rss_item.id, + ) + else: + logger.warning( + "Error analysing RSS id=%s (%s)", + rss_item.id, + rss_item.url, + exc_info=True, + ) + return bangumi_list async def _refresh_rss( - self, downloaders: DownloaderPool, rss_id: Optional[int] = None + self, + downloaders: DownloaderPool, + rss_id: Optional[int] = None, + analyser: "RSSAnalyser | None" = None, ) -> list[SystemEvent]: # Get All RSS Items if not rss_id: @@ -384,6 +426,10 @@ class RSSEngine: # torrent against it in memory (this is the job the old module-level # TTL cache existed to do). bangumi_list = await self.db.bangumi.search_all() + if analyser is not None: + bangumi_list = await self._analyse_aggregate( + analyser, item_results, bangumi_list + ) events: list[SystemEvent] = [] # Bangumi with a release-group/resolution preference need per-episode @@ -393,13 +439,6 @@ class RSSEngine: for b in bangumi_list if b.id is not None and (b.preferred_group or b.preferred_resolution) } - existing_downloaded: dict[int, list[Torrent]] = {} - if preference_bangumi: - existing_downloaded = ( - await self.db.torrent.search_downloaded_by_bangumi_ids( - list(preference_bangumi) - ) - ) # First pass: match every torrent against the bangumi list (this also # tags torrent.bangumi_id as a side effect) without downloading yet, @@ -408,20 +447,33 @@ class RSSEngine: item_matches: list[ tuple[RSSItem, list[Torrent], Optional[str], list[Optional[Bangumi]]] ] = [] - for rss_item, (new_torrents, error) in item_results: + for rss_item, (torrents, error) in item_results: + new_torrents = await self.db.torrent.check_new(torrents) matches = [self.match_torrent(t, bangumi_list) for t in new_torrents] matches = await self._apply_plugin_filters(new_torrents, matches) item_matches.append((rss_item, new_torrents, error, matches)) + matched_pairs = [ + (torrent, matched) + for _, new_torrents, _, matches in item_matches + for torrent, matched in zip(new_torrents, matches) + if matched is not None + ] + # 只为本批次有候选的偏好番剧加载已下载版本；没有候选时不查库 + candidate_ids = { + b.id + for _, b in matched_pairs + if b.id is not None and b.id in preference_bangumi + } + existing_downloaded: dict[int, list[Torrent]] = {} + if candidate_ids: + existing_downloaded = ( + await self.db.torrent.search_downloaded_by_bangumi_ids( + list(candidate_ids) + ) + ) skip_ids = self._select_preference_skips( - [ - (torrent, matched) - for _, new_torrents, _, matches in item_matches - for torrent, matched in zip(new_torrents, matches) - if matched is not None - ], - preference_bangumi, - existing_downloaded, + matched_pairs, preference_bangumi, existing_downloaded ) # Process results sequentially (DB operations) diff --git a/backend/src/test/test_loops.py b/backend/src/test/test_loops.py index 54cd1604..9f55b240 100644 --- a/backend/src/test/test_loops.py +++ b/backend/src/test/test_loops.py @@ -10,7 +10,7 @@ returned events into ``notifier.send_event`` calls, gated by import asyncio import time -from unittest.mock import AsyncMock, MagicMock, patch +from unittest.mock import ANY, AsyncMock, MagicMock, patch from module.core.loops import ( offset_scan_tick, @@ -93,7 +93,6 @@ class TestRssTick: analyser = AsyncMock() mock_db = AsyncMock() - mock_db.rss.search_aggregate = AsyncMock(return_value=[]) mock_engine = AsyncMock() mock_engine.refresh_rss = AsyncMock(return_value=events) @@ -110,6 +109,7 @@ class TestRssTick: mock_settings.bangumi_manage.eps_complete = False await rss_tick(analyser, notifier) + mock_engine.refresh_rss.assert_awaited_once_with(ANY, analyser=analyser) notifier.send_event.assert_awaited_once_with(events[0]) async def test_events_forwarded_even_when_external_push_disabled(self): @@ -119,7 +119,6 @@ class TestRssTick: analyser = AsyncMock() mock_db = AsyncMock() - mock_db.rss.search_aggregate = AsyncMock(return_value=[]) mock_engine = AsyncMock() mock_engine.refresh_rss = AsyncMock(return_value=events) diff --git a/backend/src/test/test_rss_analyser_typed_identity.py b/backend/src/test/test_rss_analyser_typed_identity.py index 7c6db5c8..d250b12a 100644 --- a/backend/src/test/test_rss_analyser_typed_identity.py +++ b/backend/src/test/test_rss_analyser_typed_identity.py @@ -97,8 +97,8 @@ async def test_rss_to_data_keeps_one_parser_engine_across_awaits( torrent = Torrent(name="resource", url="https://example.com/resource") analyser = RSSAnalyser() - async def get_rss_torrents(rss_link: str, full_parse: bool = True): - return [torrent] + async def check_new(torrents, rss_id: int | None = None): + return torrents async def match_movies(torrents, rss_link: str): assert parse_configured_release_title(torrent.name) is classic_release @@ -107,7 +107,7 @@ async def test_rss_to_data_keeps_one_parser_engine_across_awaits( assert parse_configured_release_title(torrent.name) is classic_release return torrents - async def match_bangumi(torrents, rss_link: str): + async def match_bangumi(torrents, rss_link: str, match_datas): assert parse_configured_release_title(torrent.name) is classic_release return torrents @@ -115,19 +115,19 @@ async def test_rss_to_data_keeps_one_parser_engine_across_awaits( assert parse_configured_release_title(torrent.name) is classic_release return [], [] - monkeypatch.setattr(analyser, "get_rss_torrents", get_rss_torrents) monkeypatch.setattr(analyser, "torrents_to_data", torrents_to_data) engine = cast( RSSEngine, SimpleNamespace( db=SimpleNamespace( + torrent=SimpleNamespace(check_new=check_new), movie=SimpleNamespace(match_list=match_movies), bangumi=SimpleNamespace(match_list=match_bangumi), ) ), ) - await analyser.rss_to_data(make_rss_item(parser="none"), engine) + await analyser.rss_to_data(make_rss_item(parser="none"), [torrent], engine, []) # The task-local scope is gone, so a later workflow sees the new setting. assert parse_configured_release_title(torrent.name) is preview_release diff --git a/backend/src/test/test_rss_engine_new.py b/backend/src/test/test_rss_engine_new.py index 1576a7f9..44b7f5ae 100644 --- a/backend/src/test/test_rss_engine_new.py +++ b/backend/src/test/test_rss_engine_new.py @@ -1,4 +1,4 @@ -"""Tests for RSS engine: pull_rss, match_torrent, refresh_rss, add_rss.""" +"""Tests for RSS engine: aggregate analysis, match_torrent, refresh_rss, add_rss.""" import asyncio from unittest.mock import AsyncMock, patch @@ -8,11 +8,15 @@ import pytest_asyncio from module.conf import settings from module.database import Database +from module.database.bangumi import BangumiDatabase +from module.database.movie import MovieDatabase from module.downloader import AddResult from module.models import Torrent +from module.network import RequestContent from module.notification.events import DownloadFailureEvent, RssFailureEvent from module.parser.analyser.selector import parse_configured_release_title from module.parser.analyser.tokenizer import ReleaseKind +from module.rss.analyser import RSSAnalyser from module.rss.engine import RSSEngine from test.factories import SingleClientPool, make_bangumi, make_rss_item, make_torrent @@ -24,68 +28,94 @@ async def rss_engine(db_engine): # --------------------------------------------------------------------------- -# pull_rss +# 聚合订阅分析（refresh_rss 带 analyser，即 rss_tick） # --------------------------------------------------------------------------- +FEED_A = "https://mikanani.me/RSS/MyBangumi?token=a" +FEED_B = "https://mikanani.me/RSS/MyBangumi?token=b" +KNOWN_URL = "https://example.com/known.torrent" -class TestPullRss: - async def test_returns_only_new_torrents(self, rss_engine): - """pull_rss filters out torrents already in the database.""" - rss_item = make_rss_item() - await rss_engine.db.rss.add(rss_item) - rss_item = await rss_engine.db.rss.search_id(1) - # Pre-insert one torrent into DB - existing = make_torrent(url="https://example.com/existing.torrent", rss_id=1) - await rss_engine.db.torrent.add(existing) - - # Mock _get_torrents to return 3 torrents (1 existing + 2 new) - all_torrents = [ - Torrent(name="existing", url="https://example.com/existing.torrent"), - Torrent(name="new1", url="https://example.com/new1.torrent"), - Torrent(name="new2", url="https://example.com/new2.torrent"), +async def _tick_with_feeds( + engine: RSSEngine, items: dict[str, list[Torrent]] +) -> AsyncMock: + """按订阅 URL 返回 items 跑一轮刷新，返回 HTTP 层 get_torrents 的替身。""" + client = AsyncMock() + client.add_torrent = AsyncMock(return_value=AddResult.ADDED) + get = AsyncMock( + side_effect=lambda url, *_a, **_k: [ + Torrent(name=t.name, url=t.url) for t in items[url] ] - with patch.object( - RSSEngine, "_get_torrents", new_callable=AsyncMock - ) as mock_get: - mock_get.return_value = all_torrents - result = await rss_engine.pull_rss(rss_item) + ) + with ( + patch.object(RequestContent, "get_torrents", get), + patch("module.rss.engine.RSS_PER_HOST_DELAY", 0), + patch("module.rss.analyser.enrich_bangumi", AsyncMock()), + patch("module.rss.analyser.enrich_movie", AsyncMock()), + ): + await engine.refresh_rss(SingleClientPool(client), analyser=RSSAnalyser()) + return get - assert len(result) == 2 - assert all(t.url != "https://example.com/existing.torrent" for t in result) - async def test_all_existing_returns_empty(self, rss_engine): - """When all torrents already exist, returns empty list.""" - rss_item = make_rss_item() - await rss_engine.db.rss.add(rss_item) - rss_item = await rss_engine.db.rss.search_id(1) +class TestAggregateAnalysis: + async def test_item_known_for_same_feed_skips_matching(self, rss_engine): + """本订阅已记录的种子不再匹配，也不会生成新规则。""" + await rss_engine.db.rss.add(make_rss_item(url=FEED_A)) + await rss_engine.db.torrent.add( + make_torrent( + name="[Sub] Dungeon Meshi - 01 [1080p].mkv", url=KNOWN_URL, rss_id=1 + ) + ) + known = Torrent(name="[Sub] Dungeon Meshi - 01 [1080p].mkv", url=KNOWN_URL) - existing = make_torrent(url="https://example.com/only.torrent", rss_id=1) - await rss_engine.db.torrent.add(existing) + with ( + patch.object( + BangumiDatabase, "match_list", new_callable=AsyncMock + ) as bangumi_match, + patch.object( + MovieDatabase, "match_list", new_callable=AsyncMock + ) as movie_match, + ): + await _tick_with_feeds(rss_engine, {FEED_A: [known]}) - with patch.object( - RSSEngine, "_get_torrents", new_callable=AsyncMock - ) as mock_get: - mock_get.return_value = [ - Torrent(name="only", url="https://example.com/only.torrent") - ] - result = await rss_engine.pull_rss(rss_item) + bangumi_match.assert_not_awaited() + movie_match.assert_not_awaited() + assert await rss_engine.db.bangumi.search_all() == [] - assert result == [] + async def test_known_torrent_in_second_feed_appends_rss_link(self, rss_engine): + """其他订阅已知的种子出现在新聚合订阅里：仍把新订阅追加到 rss_link（#1053）。""" + await rss_engine.db.rss.add(make_rss_item(url=FEED_A)) + await rss_engine.db.rss.add(make_rss_item(url=FEED_B)) + await rss_engine.db.bangumi.add( + make_bangumi(title_raw="Mushoku Tensei", filter="", rss_link=FEED_A) + ) + name = "[Sub] Mushoku Tensei - 12 [1080p].mkv" + await rss_engine.db.torrent.add( + make_torrent(name=name, url=KNOWN_URL, rss_id=1, bangumi_id=1) + ) + known = Torrent(name=name, url=KNOWN_URL) - async def test_empty_feed_returns_empty(self, rss_engine): - """When RSS feed has no torrents, returns empty list.""" - rss_item = make_rss_item() - await rss_engine.db.rss.add(rss_item) - rss_item = await rss_engine.db.rss.search_id(1) + await _tick_with_feeds(rss_engine, {FEED_A: [known], FEED_B: [known]}) - with patch.object( - RSSEngine, "_get_torrents", new_callable=AsyncMock - ) as mock_get: - mock_get.return_value = [] - result = await rss_engine.pull_rss(rss_item) + bangumi = await rss_engine.db.bangumi.search_id(1) + assert bangumi is not None + assert bangumi.rss_link.split(",") == [FEED_A, FEED_B] - assert result == [] + async def test_each_feed_fetched_once_per_tick(self, rss_engine): + """聚合订阅不再单独请求：每轮每个订阅只有一次 HTTP GET。""" + await rss_engine.db.rss.add(make_rss_item(url=FEED_A)) + await rss_engine.db.rss.add(make_rss_item(url=FEED_B, aggregate=False)) + await rss_engine.db.bangumi.add( + make_bangumi(title_raw="Mushoku Tensei", filter="", rss_link=FEED_A) + ) + items = { + FEED_A: [Torrent(name="[Sub] Mushoku Tensei - 12 [1080p].mkv", url="a")], + FEED_B: [Torrent(name="[Sub] Mushoku Tensei - 13 [1080p].mkv", url="b")], + } + + get = await _tick_with_feeds(rss_engine, items) + + assert sorted(c.args[0] for c in get.await_args_list) == [FEED_A, FEED_B] # --------------------------------------------------------------------------- @@ -1403,3 +1433,59 @@ class TestRefreshRssPreferenceDedup: await rss_engine.refresh_rss(SingleClientPool(client)) assert client.add_torrent.call_count == 2 + + +class TestPreferencePreload: + @pytest.fixture(autouse=True) + def use_preview_engine(self): + with patch.object(settings.rss_parser, "engine", "tokenizer"): + yield + + @pytest.mark.parametrize( + ("name", "expected_ids"), + [ + ("[GroupA] Dungeon Meshi - 01 [1080p].mkv", None), + ("[GroupA] Mushoku Tensei - 12 [1080p].mkv", [1]), + ], + ids=["no-preference-candidate", "one-preference-candidate"], + ) + async def test_downloaded_loaded_only_for_candidate_bangumi( + self, rss_engine, name, expected_ids + ): + """只为本批次有候选的偏好番剧加载已下载版本；没有候选时不查库。""" + await rss_engine.db.rss.add(make_rss_item(aggregate=False)) + for title_raw, preferred_group in ( + ("Mushoku Tensei", "GroupA"), + ("Dungeon Meshi", None), + ("Sousou no Frieren", "GroupA"), + ): + await rss_engine.db.bangumi.add( + make_bangumi( + official_title=title_raw, + title_raw=title_raw, + filter="", + preferred_group=preferred_group, + ) + ) + client = AsyncMock() + client.add_torrent = AsyncMock(return_value=AddResult.ADDED) + + with ( + patch.object( + RSSEngine, + "_get_torrents", + AsyncMock(return_value=[Torrent(name=name, url="https://x/1")]), + ), + patch.object( + rss_engine.db.torrent, + "search_downloaded_by_bangumi_ids", + AsyncMock(return_value={}), + ) as preload, + ): + await rss_engine.refresh_rss(SingleClientPool(client)) + + client.add_torrent.assert_awaited_once() + if expected_ids is None: + preload.assert_not_awaited() + else: + preload.assert_awaited_once_with(expected_ids)