GithubBackup/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-09-10 18:06:48 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor(subscribe): separate match from reconciliation

Browse Source
This commit is contained in:
jxxghp
2026-09-01 17:08:53 +08:00
parent 04ecfbd7f6
commit 250b56c523
8 changed files with 155 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/application/subscription/candidates.py
+13 -2
View File
@@ -113,8 +113,14 @@ class CandidateIndex:
results.append(copied)
return results
def route_for_match(self, subscribe: SubscriptionSnapshot) -> CandidateGroups:
"""保守路由可能命中的候选，且只排除当前 canonical 逻辑必然拒绝的候选。"""
def route_for_match(
self,
subscribe: SubscriptionSnapshot,
*,
domains: Optional[set[str]] = None,
site_ids: Optional[set[int]] = None,
) -> CandidateGroups:
"""保守路由可能命中的候选，并在外部事实查询前应用站点范围。"""
if subscribe.custom_words:
positions = set(range(len(self._ordered)))
else:
@@ -133,6 +139,11 @@ class CandidateIndex:
for position, (domain, context) in enumerate(self._ordered):
if position not in positions:
continue
if domains and domain not in domains:
continue
torrent_info = getattr(context, "torrent_info", None)
if site_ids and getattr(torrent_info, "site", None) not in site_ids:
continue
if not subscribe.custom_words and (
not self.media_type_matches(context, subscribe)
or not self.season_matches(context, subscribe)
app/chain/subscribe/completion.py
+17
View File
@@ -6,6 +6,7 @@ from typing import Any, Dict, List, Optional, Tuple, Union
from app.application.subscription.contract import (
SubscriptionSnapshot,
subscribe_media_key,
subscribe_media_keys,
)
from app.application.subscription.mutation import SubscriptionActor
@@ -174,6 +175,22 @@ class SubscribeCompletionOwner(_SubscribeOwnerBase):
elif not downloads:
logger.info(f"{mediainfo.title_year} 继续洗版 ...")
def reconcile_subscription_completion(
self,
subscribe: SubscriptionSnapshot,
meta: MetaBase,
mediainfo: MediaInfo,
) -> bool:
"""使用已取得的新鲜媒体事实独立对账订阅完成状态。"""
mediakey = subscribe_media_key(subscribe)
completed, _no_exists = self.check_and_handle_existing_media(
subscribe=subscribe,
meta=meta,
mediainfo=mediainfo,
mediakey=mediakey,
)
return completed
def _SubscribeChain__update_subscribe_note(
self,
subscribe: SubscriptionSnapshot,
app/chain/subscribe/contract.py
+2
View File
@@ -79,6 +79,7 @@ if TYPE_CHECKING:
"""异步发送订阅通知。"""
raise NotImplementedError
check_and_handle_existing_media: Callable[..., Any]
check_and_reconcile: Callable[..., Any]
filter_torrents: Callable[..., Any]
finish_subscribe_or_not: Callable[..., Any]
get_params: Callable[..., Any]
@@ -96,6 +97,7 @@ if TYPE_CHECKING:
post_message: Callable[..., Any]
remote_list: Callable[..., Any]
resolve_subscribe_missing: Callable[..., Any]
reconcile_subscription_completion: Callable[..., Any]
_SubscribeOwnerBase = _SubscribeOwnerHost
else:
app/chain/subscribe/match.py
+9 -2
View File
@@ -181,6 +181,15 @@ class SubscribeMatchOwner(_SubscribeOwnerBase):
domains = []
if subscribe.sites:
domains = self.site_repository.get_domains_by_ids(subscribe.sites)
sub_sites = self.get_sub_sites(subscribe)
routed_torrents = candidate_index.route_for_match(
subscribe,
domains=set(domains) if domains else None,
site_ids=set(sub_sites) if sub_sites else None,
)
if not routed_torrents:
logger.info(f"订阅 {subscribe.name} 本轮没有可能相关的资源，跳过资源匹配准备")
continue
# 识别媒体信息
mediainfo: MediaInfo = MediaChain().recognize_media(
meta=meta,
@@ -220,7 +229,6 @@ class SubscribeMatchOwner(_SubscribeOwnerBase):
torrenthelper = TorrentHelper()
systemconfig = get_configured_system_config()
wordsmatcher = WordsMatcher()
routed_torrents = candidate_index.route_for_match(subscribe)
for domain, contexts in routed_torrents.items():
if runtime_stop_state.is_system_stopped:
break
@@ -237,7 +245,6 @@ class SubscribeMatchOwner(_SubscribeOwnerBase):
torrent_info = _context.torrent_info
# 不在订阅站点范围的不处理
sub_sites = self.get_sub_sites(subscribe)
if sub_sites and torrent_info.site not in sub_sites:
logger.debug(f"{torrent_info.site_name} - {torrent_info.title} 不符合订阅站点要求")
continue
app/chain/subscribe/refresh.py
+22 -1
View File
@@ -90,11 +90,16 @@ class SubscribeRefreshOwner(_SubscribeOwnerBase):
if progress_callback:
progress_callback(value=100, text="订阅刷新完成")
def check(self, progress_callback: Optional[Callable[..., None]] = None) -> None:
def check(
self,
progress_callback: Optional[Callable[..., None]] = None,
reconcile_completion: bool = False,
) -> None:
"""
定时检查订阅，更新订阅信息
:param progress_callback: 定时服务进度更新回调
:param reconcile_completion: 是否复用本次新鲜媒体事实执行独立完成对账
"""
# 查询所有订阅
repository = self.subscription_repository
@@ -219,6 +224,12 @@ class SubscribeRefreshOwner(_SubscribeOwnerBase):
update_data,
scene="metadata_refresh",
)
if reconcile_completion and subscribe.state in self.get_states_for_search("R"):
self.reconcile_subscription_completion(
subscribe=subscribe,
meta=meta,
mediainfo=mediainfo,
)
logger.info(f"{subscribe.name} 订阅元数据更新完成")
if progress_callback:
progress_callback(
@@ -229,6 +240,16 @@ class SubscribeRefreshOwner(_SubscribeOwnerBase):
if progress_callback:
progress_callback(value=100, text="订阅元数据更新完成")
def check_and_reconcile(
self,
progress_callback: Optional[Callable[..., None]] = None,
) -> None:
"""刷新订阅元数据，并复用同一轮新鲜事实执行完成对账。"""
return self.check(
progress_callback=progress_callback,
reconcile_completion=True,
)
async def cache_calendar(
self,
progress_callback: Optional[Callable[..., None]] = None,
app/startup/initializers/scheduler.py
+1 -1
View File
@@ -59,7 +59,7 @@ def configure_scheduler_services() -> None:
SchedulerServices(
sync_cookies=site_chain.sync_cookies,
sync_mediaserver=mediaserver_chain.sync,
check_subscribe=subscribe_chain.check,
check_subscribe=subscribe_chain.check_and_reconcile,
search_subscribe=subscribe_chain.search,
refresh_subscribe=subscribe_chain.refresh,
follow_subscribe=subscribe_chain.follow,
docs/refactor/moviepilot-subscription-governance-roadmap.md
+16 -4
View File
@@ -1,7 +1,7 @@
# MoviePilot 订阅执行治理
> 状态：`active（2026-09-01 已由 MoviePilot v3 接管）`
> 当前叶：`SUB-GOV-001C`
> 当前叶：`SUB-GOV-001D`
> 迁移来源：`V3-RDY-009A`、`V3-RDY-009A1`、`V3-RDY-009A1B`、
> `V3-RDY-009A1C`、`V3-RDY-009A1D`
> 适用范围：V3 订阅刷新、匹配、兜底搜索、下载提交与用户状态闭环
@@ -229,8 +229,8 @@ RSS/Spider 保持现有逐站点串行刷新，不通过并发化换取几秒收
| `SUB-GOV-000` | 固定历史行为和生产日志；保留 `V3-RDY-009A1A` 已交付的批量请求正确性 | 无 | `completed` |
| `SUB-GOV-001A` | 建立日常 Match 正确性和可重放夹具；撤销 `cache=True` 错误候选，只保留经验证不改变语义的失败识别状态回写 | 000 | `completed（2026-09-01）` |
| `SUB-GOV-001B` | 区分完整缓存与本轮 delta，建立无损候选索引；先证明命中集合与基线完全一致 | 001A | `completed（2026-09-01）` |
| `SUB-GOV-001C` | 将资源匹配与完成对账拆开；只有受候选影响的订阅进入匹配，完成对账独立保持新鲜语义 | 001B | `in_progress` |
| `SUB-GOV-001D` | 引入单轮新鲜事实租约，评估稳定元数据复用、轻量季集查询和单媒体服务器事实合并 | 001C | `pending` |
| `SUB-GOV-001C` | 将资源匹配与完成对账拆开；只有受候选影响的订阅进入匹配，完成对账独立保持新鲜语义 | 001B | `completed（2026-09-01）` |
| `SUB-GOV-001D` | 引入单轮新鲜事实租约，评估稳定元数据复用、轻量季集查询和单媒体服务器事实合并 | 001C | `in_progress` |
| `SUB-GOV-001E` | 若 001B–D 后仍有可重复等待热点，再通过隔离压测评估 2/4 worker 的有界准备；提交保持串行 | 001D | `conditional` |
| `SUB-GOV-002A` | 将 24 小时兜底搜索移出 Match/提交长锁，建立批次、订阅 single-flight、可取消等待和恢复游标 | 001C | `pending` |
| `SUB-GOV-002B` | 为兜底搜索建立每站点并发、间隔和错误冷却预算；RSS/Spider 保持基线压力 | 002A | `pending` |
@@ -239,7 +239,7 @@ RSS/Spider 保持现有逐站点串行刷新，不通过并发化换取几秒收
| `SUB-GOV-003B` | 后端提供订阅及批次业务状态，前端展示排队、匹配、搜索、提交、完成、失败和取消 | 002A, 003A | `pending` |
| `SUB-GOV-004` | 多条订阅记录指向同一媒体时的跨记录季集去重和产品规则 | 003A | `pending（最低优先级）` |
当前只激活 `SUB-GOV-001C`。001B–D 是日常 Match 主线；002A–C 是 24 小时兜底搜索主线；003A/B
当前只激活 `SUB-GOV-001D`。001B–D 是日常 Match 主线；002A–C 是 24 小时兜底搜索主线；003A/B
只有在前两条链路的身份和终态稳定后实施。每个叶子完成验收并更新本表后，才激活下一个满足依赖的叶子。
### 6.1 SUB-GOV-001A 验收证据
@@ -264,6 +264,18 @@ RSS/Spider 保持现有逐站点串行刷新，不通过并发化换取几秒收
- 验证：订阅、候选、音乐缓存和架构测试共 `228 passed`，错误级 Pylint 为 0，
`scripts/architecture/baseline.py --check-host` 与 `git diff --check` 通过。
### 6.3 SUB-GOV-001C 验收证据
- Match 在媒体识别和媒体库查询前应用候选身份、类型、季、站点域名和站点 ID 门禁；无可能候选时不读取
TMDB/媒体服务器事实，也不在资源路径隐式完成订阅；
- 新增 `reconcile_subscription_completion()`，复用 canonical 缺失查询与完成路径，不复制完成规则；
- 新增 `check_and_reconcile()`，现有每 6 小时 `subscribe_tmdb` 调度绑定该入口，在同一次 `cache=False`
新鲜识别中完成元数据更新和独立完成对账；普通 `check()` 兼容入口仍可只更新元数据；
- 固定重放验证明确冲突候选不触发外部事实查询，并验证无候选批次时独立巡检仍将 12→13 更新为 `13/1`
后进入完成对账；
- 验证：订阅专项 `131 passed`，API/调度器/架构组合 `183 passed`，错误级 Pylint 为 0，Host
架构基线和 `git diff --check` 通过。
## 7. 上线前验证与验收
### 7.1 场景
tests/test_subscription_governance_replay.py
+75
View File
@@ -176,3 +176,78 @@ def test_match_replays_fresh_episode_completion_contract(case, monkeypatch):
assert repository.current.total_episode == case["expected_total_episode"]
assert repository.current.lack_episode == case["expected_lack_episode"]
assert bool(completions) is case["expected_completed"]
def test_match_skips_external_facts_when_index_has_no_possible_candidates(monkeypatch):
"""本轮只有明确冲突候选时，不得为订阅调用 TMDB 或媒体服务器准备。"""
subscribe = _build_subscribe(_load_replay_cases()[0])
repository = _ReplaySubscriptionRepository(subscribe)
candidate = _build_unrelated_candidate()
candidate.meta_info.media_source = MediaSource.TMDB
candidate.meta_info.media_id = "999"
chain = SubscribeChain()
chain.subscription_repository = repository
chain.resolve_subscribe_missing = lambda **_kwargs: pytest.fail("不应查询媒体库缺失事实")
class _UnexpectedMediaChain:
"""任何媒体识别调用都表示候选门禁失效。"""
def recognize_media(self, **_kwargs):
"""阻止无候选订阅读取外部媒体事实。"""
pytest.fail("不应读取 TMDB 新鲜事实")
def recognize_by_meta(self, *_args, **_kwargs):
"""明确候选身份不应重新识别。"""
pytest.fail("不应重新识别明确候选")
monkeypatch.setattr("app.chain.subscribe.match.MediaChain", _UnexpectedMediaChain)
monkeypatch.setattr(
"app.chain.subscribe.query.get_configured_system_config",
lambda: SimpleNamespace(get=lambda _key: []),
)
chain.match({"replay.example": [candidate]})
def test_metadata_reconcile_reuses_fresh_fact_without_candidate_batch(monkeypatch):
"""独立元数据巡检应复用同一次新鲜识别执行完成对账。"""
subscribe = _build_subscribe(_load_replay_cases()[1])
repository = _ReplaySubscriptionRepository(subscribe)
recognition_calls = []
reconciled = []
fresh_media = MediaInfo(
media_source=MediaSource.TMDB,
media_id="100",
type=MediaType.TV,
title="增长中的剧集",
year="2026",
seasons={1: list(range(1, 14))},
)
class _ReplayMediaChain:
"""为独立完成对账返回固定新鲜媒体事实。"""
def recognize_media(self, **kwargs) -> MediaInfo:
"""记录识别参数并返回 13 集事实。"""
recognition_calls.append(kwargs)
return fresh_media
chain = SubscribeChain()
chain.subscription_repository = repository
chain._SubscribeChain__apply_subscribe_update = (
lambda _subscribe, update_data, **_kwargs: repository.update(dict(update_data))
)
chain.reconcile_subscription_completion = (
lambda **kwargs: reconciled.append(kwargs)
)
monkeypatch.setattr("app.chain.subscribe.refresh.MediaChain", _ReplayMediaChain)
chain.check_and_reconcile()
assert len(recognition_calls) == 1
assert recognition_calls[0]["cache"] is False
assert repository.current.total_episode == 13
assert repository.current.lack_episode == 1
assert len(reconciled) == 1
assert reconciled[0]["subscribe"] == repository.current
assert reconciled[0]["mediainfo"] is fresh_media