perf(rss): 每轮每个订阅只拉取一次,聚合分析按 (url, rss_id) 跳过已知种子,偏好去重按候选加载

- 聚合订阅分析改在 RSSEngine.refresh_rss 内进行,复用按主机排队拉取的结果(保留 2 s 间隔,#1026),不再二次请求
- 分析前按 (url, rss_id) 过滤本订阅已记录的种子;其他订阅已知的种子仍参与匹配,以追加 rss_link(#1053)
- 每轮只加载一次番剧列表,生成新规则后才重新加载
- 偏好去重:批次中没有偏好候选时直接返回,只为有候选的番剧加载已下载版本
- 删除不再使用的 pull_rss 与 search_aggregate
This commit is contained in:
Estrella Pan
2026-10-20 17:11:24 +02:00
parent 843a014928
commit 48e4ec94cb
9 changed files with 256 additions and 126 deletions

View File

@@ -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 决定是否外发)。

View File

@@ -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

View File

@@ -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,避免外键约束报错

View File

@@ -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]

View File

@@ -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:

View File

@@ -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)

View File

@@ -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)

View File

@@ -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

View File

@@ -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)