diff --git a/app/agent/tools/impl/_filter_rule_utils.py b/app/agent/tools/impl/_filter_rule_utils.py index b3eacab8c..ef4a00bfc 100644 --- a/app/agent/tools/impl/_filter_rule_utils.py +++ b/app/agent/tools/impl/_filter_rule_utils.py @@ -2,11 +2,16 @@ import copy import re -from typing import Any, Dict, Iterable, Optional +from typing import Any, Callable, Dict, Iterable, Optional from app.application.agentdata import get_agent_subscribe_port from app.application.configuration import get_configured_system_config -from app.application.rules import BUILTIN_RULE_SET, RuleHelper, RuleParser +from app.application.rules import ( + BUILTIN_RULE_SET, + RuleHelper, + RuleParser, + replace_group_name_in_list, +) from app.runtime.events import eventmanager from app.schemas.event import ConfigChangeEventData from app.schemas.rule import CustomRule @@ -430,17 +435,94 @@ async def save_system_config( success = await get_configured_system_config().async_set(key, normalized_value) if success: - await eventmanager.async_send_event( - etype=EventType.ConfigChanged, - data=ConfigChangeEventData( - key=key, - value=normalized_value, - change_type="update", - ), - ) + await _publish_rule_config_changed(key, normalized_value) return success +async def _publish_rule_config_changed( + key: SystemConfigKey, + value: Any, +) -> None: + """广播一项已经提交的规则配置变更。""" + await eventmanager.async_send_event( + etype=EventType.ConfigChanged, + data=ConfigChangeEventData( + key=key, + value=value, + change_type="update", + ), + ) + + +async def _rewrite_rule_group_references( + map_names: Callable[[Iterable[str]], list[str]], +) -> dict: + """按名称映射器更新全局、默认订阅配置和已有订阅引用。""" + changed = { + "global_settings": {}, + "subscribes": [], + } + system_config = get_configured_system_config() + for config_key in ( + SystemConfigKey.SearchFilterRuleGroups, + SystemConfigKey.SubscribeFilterRuleGroups, + SystemConfigKey.BestVersionFilterRuleGroups, + ): + original = system_config.get(config_key) or [] + updated = map_names(original) + if updated != original: + await save_system_config(config_key, updated) + changed["global_settings"][config_key.value] = updated + + for config_key in ( + SystemConfigKey.DefaultMovieSubscribeConfig, + SystemConfigKey.DefaultTvSubscribeConfig, + SystemConfigKey.DefaultMusicSubscribeConfig, + ): + original = system_config.get(config_key) or {} + original_groups = original.get("filter_groups") or [] + updated_groups = map_names(original_groups) + if updated_groups == original_groups: + continue + updated = copy.deepcopy(original) + updated["filter_groups"] = updated_groups + await save_system_config(config_key, updated) + changed["global_settings"][config_key.value] = updated + + subscribe_port = get_agent_subscribe_port() + subscribes = await subscribe_port.async_list() + for subscribe in subscribes: + original = subscribe.filter_groups or [] + updated = map_names(original) + if updated == original: + continue + await subscribe_port.async_update_filter_groups(subscribe.id, updated) + changed["subscribes"].append( + { + "subscribe_id": subscribe.id, + "name": subscribe.name, + "season": subscribe.season, + "filter_groups": updated, + } + ) + + return changed + + +async def rename_rule_group_references(old_name: str, new_name: str) -> dict: + """规则组改名后,联动更新全部配置和已有订阅引用。""" + return await _rewrite_rule_group_references( + lambda values: replace_group_name_in_list(values, old_name, new_name) + ) + + +async def remove_rule_group_references(group_name: str) -> dict: + """删除规则组后,清理全部配置和已有订阅中的悬空引用。""" + return await _rewrite_rule_group_references( + lambda values: [value for value in values or [] if value != group_name] + ) + + def replace_rule_id_in_rule_string( rule_string: str, old_rule_id: str, new_rule_id: str ) -> str: @@ -449,91 +531,3 @@ def replace_rule_id_in_rule_string( rf"(? list[str]: - """更新配置里的规则组名引用,并顺手去重。""" - result = [] - for value in values or []: - mapped = new_name if value == old_name else value - if mapped not in result: - result.append(mapped) - return result - - -async def rename_rule_group_references(old_name: str, new_name: str) -> dict: - """规则组改名后,联动更新全局设置和订阅引用。""" - changed = { - "global_settings": {}, - "subscribes": [], - } - - for config_key in ( - SystemConfigKey.SearchFilterRuleGroups, - SystemConfigKey.SubscribeFilterRuleGroups, - SystemConfigKey.BestVersionFilterRuleGroups, - ): - original = get_configured_system_config().get(config_key) or [] - updated = replace_group_name_in_list(original, old_name, new_name) - if updated != original: - await save_system_config(config_key, updated) - changed["global_settings"][config_key.value] = updated - - subscribe_oper = get_agent_subscribe_port() - subscribes = await subscribe_oper.async_list() - for subscribe in subscribes: - original = subscribe.filter_groups or [] - updated = replace_group_name_in_list(original, old_name, new_name) - if updated == original: - continue - await subscribe_oper.async_update_filter_groups(subscribe.id, updated) - changed["subscribes"].append( - { - "subscribe_id": subscribe.id, - "name": subscribe.name, - "season": subscribe.season, - "filter_groups": updated, - } - ) - - return changed - - -async def remove_rule_group_references(group_name: str) -> dict: - """删除规则组后,清理全局设置和订阅里的悬空引用。""" - changed = { - "global_settings": {}, - "subscribes": [], - } - - for config_key in ( - SystemConfigKey.SearchFilterRuleGroups, - SystemConfigKey.SubscribeFilterRuleGroups, - SystemConfigKey.BestVersionFilterRuleGroups, - ): - original = get_configured_system_config().get(config_key) or [] - updated = [value for value in original if value != group_name] - if updated != original: - await save_system_config(config_key, updated) - changed["global_settings"][config_key.value] = updated - - subscribe_oper = get_agent_subscribe_port() - subscribes = await subscribe_oper.async_list() - for subscribe in subscribes: - original = subscribe.filter_groups or [] - updated = [value for value in original if value != group_name] - if updated == original: - continue - await subscribe_oper.async_update_filter_groups(subscribe.id, updated) - changed["subscribes"].append( - { - "subscribe_id": subscribe.id, - "name": subscribe.name, - "season": subscribe.season, - "filter_groups": updated, - } - ) - - return changed diff --git a/app/api/endpoints/system.py b/app/api/endpoints/system.py index b6ebe63ca..4fcd33272 100644 --- a/app/api/endpoints/system.py +++ b/app/api/endpoints/system.py @@ -1192,7 +1192,10 @@ async def set_setting( try: with plugin_system_config_mutation(key): success = await get_configured_system_config().async_set(key, value) - if success: + if success or ( + success is None + and key == SystemConfigKey.UserFilterRuleGroups.value + ): # 发送配置变更事件 await eventmanager.async_send_event( etype=EventType.ConfigChanged, diff --git a/app/application/rules.py b/app/application/rules.py index 16b6656b9..46c1d4621 100644 --- a/app/application/rules.py +++ b/app/application/rules.py @@ -4,6 +4,7 @@ """ import threading +from collections.abc import Iterable from typing import Dict, List, Optional from pyparsing import ( @@ -85,6 +86,20 @@ class RuleHelper: ) +def replace_group_name_in_list( + values: Optional[Iterable[str]], + old_name: str, + new_name: str, +) -> list[str]: + """更新配置里的规则组名引用,并顺手去重。""" + result = [] + for value in values or []: + mapped = new_name if value == old_name else value + if mapped not in result: + result.append(mapped) + return result + + # 内置规则只在这里维护一份,便于过滤模块和 Agent 工具共享同一套事实来源。 BUILTIN_RULE_SET: Dict[str, dict] = { # 蓝光原盘 diff --git a/app/chain/subscribe.py b/app/chain/subscribe.py index 281e648cc..8dd138066 100644 --- a/app/chain/subscribe.py +++ b/app/chain/subscribe.py @@ -103,6 +103,24 @@ else: """返回快照字段,兼容整理链的响应转换。""" return dict(self.__dict__) + +def _rule_group_names(rule_groups: Optional[List[Any]]) -> set[str]: + """从规则组持久化字典中提取非空名称。""" + return { + str(group.get("name")).strip() + for group in rule_groups or [] + if isinstance(group, dict) and group.get("name") + } + + +def _retain_rule_group_names( + values: Optional[List[str]], + valid_names: set[str], +) -> List[str]: + """按当前规则组定义保留有效引用,并维持原有顺序。""" + return [value for value in values or [] if value in valid_names] + + @dataclass(frozen=True, slots=True) class _SubscribePostCommitContext: """订阅提交后副作用所需的不可变业务快照。""" @@ -1212,6 +1230,13 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): ] else: subscribes = subscribeoper.list(self.get_states_for_search(state)) + self._reconcile_rule_group_references( + valid_names=_rule_group_names( + _system_config().get(SystemConfigKey.UserFilterRuleGroups) + ), + subscribeoper=subscribeoper, + subscribes=subscribes, + ) total_num = len(subscribes) processed_subscribes = [] # 搜索链在整个订阅循环内复用,避免每轮订阅重复执行链初始化 @@ -3021,6 +3046,69 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): "sites": sites }) + @eventmanager.register(EventType.ConfigChanged) + def reconcile_rule_group_references(self, event: Event) -> None: + """规则组定义保存后清理默认配置和已有订阅中的悬空引用。""" + if not event: + return + event_data = event.event_data + if isinstance(event_data, dict): + changed_keys = event_data.get("key", set()) + value = event_data.get("value") + else: + changed_keys = getattr(event_data, "key", set()) + value = getattr(event_data, "value", None) + if isinstance(changed_keys, str): + changed_keys = {changed_keys} + normalized_keys = {str(key) for key in (changed_keys or set())} + if not normalized_keys.intersection({ + SystemConfigKey.UserFilterRuleGroups.value, + str(SystemConfigKey.UserFilterRuleGroups), + }): + return + + self._reconcile_rule_group_references(valid_names=_rule_group_names(value)) + + @staticmethod + def _reconcile_rule_group_references( + valid_names: set[str], + subscribeoper=None, + subscribes: Optional[List[Any]] = None, + ) -> None: + """持久化清理规则组引用,并同步当前搜索使用的订阅对象。""" + system_config = _system_config() + for config_key in ( + SystemConfigKey.SearchFilterRuleGroups, + SystemConfigKey.SubscribeFilterRuleGroups, + SystemConfigKey.BestVersionFilterRuleGroups, + ): + original = system_config.get(config_key) or [] + updated = _retain_rule_group_names(original, valid_names) + if updated != original: + system_config.set(config_key, updated) + + for config_key in ( + SystemConfigKey.DefaultMovieSubscribeConfig, + SystemConfigKey.DefaultTvSubscribeConfig, + SystemConfigKey.DefaultMusicSubscribeConfig, + ): + original = system_config.get(config_key) or {} + original_groups = original.get("filter_groups") or [] + updated_groups = _retain_rule_group_names(original_groups, valid_names) + if updated_groups == original_groups: + continue + updated = copy.deepcopy(original) + updated["filter_groups"] = updated_groups + system_config.set(config_key, updated) + + subscribeoper = subscribeoper or get_chain_subscribe_port() + for subscribe in subscribes if subscribes is not None else subscribeoper.list(): + original = getattr(subscribe, "filter_groups", None) or [] + updated = _retain_rule_group_names(original, valid_names) + if updated != original: + subscribeoper.update(subscribe.id, {"filter_groups": updated}) + subscribe.filter_groups = updated + @staticmethod def __get_default_subscribe_config(mtype: MediaType, default_config_key: str) -> Optional[str]: """ diff --git a/tests/fixtures/architecture/runtime-contract-baseline.json b/tests/fixtures/architecture/runtime-contract-baseline.json index 3ee0fec7d..dc31298f9 100644 --- a/tests/fixtures/architecture/runtime-contract-baseline.json +++ b/tests/fixtures/architecture/runtime-contract-baseline.json @@ -1965,7 +1965,7 @@ } }, "events": { - "consumer_count": 15, + "consumer_count": 16, "dynamic_consumers": [ { "caller": "app.adapters.system.fsproxy", @@ -2293,6 +2293,10 @@ }, "EventType.ConfigChanged": { "consumers": [ + { + "caller": "app.chain.subscribe", + "count": 1 + }, { "caller": "app.runtime.extensions.module_manager", "count": 1 diff --git a/tests/test_plugin_system_setting_admission.py b/tests/test_plugin_system_setting_admission.py index ada3d157f..9d0a84491 100644 --- a/tests/test_plugin_system_setting_admission.py +++ b/tests/test_plugin_system_setting_admission.py @@ -227,6 +227,35 @@ async def test_non_plugin_system_config_does_not_resolve_plugin_runtime( config.async_set.assert_awaited_once_with(SystemConfigKey.Directories.value, None) +@pytest.mark.asyncio +async def test_rule_group_setting_reconciles_stale_references_when_unchanged( + monkeypatch, +) -> None: + """重复保存规则组也应按有效名称对账,修复旧版本遗留的悬空订阅引用。""" + config = MagicMock() + config.async_set = AsyncMock(return_value=None) + monkeypatch.setattr( + system_endpoint, + "get_runtime_settings", + lambda: SimpleNamespace(contains=lambda _key: False), + ) + monkeypatch.setattr( + system_endpoint, + "get_configured_system_config", + lambda: config, + ) + monkeypatch.setattr(system_endpoint.eventmanager, "async_send_event", AsyncMock()) + + response = await system_endpoint.set_setting( + SystemConfigKey.UserFilterRuleGroups.value, + [{"name": "keep", "rule_string": "4K"}], + None, + ) + + assert response.success is True + system_endpoint.eventmanager.async_send_event.assert_awaited_once() + + @pytest.mark.asyncio async def test_agent_non_plugin_config_does_not_resolve_plugin_runtime( monkeypatch, diff --git a/tests/test_rule_group_media.py b/tests/test_rule_group_media.py index 69df0cfd4..fa4465eca 100644 --- a/tests/test_rule_group_media.py +++ b/tests/test_rule_group_media.py @@ -1,9 +1,15 @@ +from types import SimpleNamespace + +from app.chain import subscribe as subscribe_module +from app.chain.subscribe import SubscribeChain from app.agent.tools.impl._filter_rule_utils import normalize_media_type from app.application.rules import RuleHelper from app.domain.context import MediaInfo, MusicInfo, TorrentInfo from app.modules.filter import FilterModule +from app.runtime.events import Event +from app.schemas.event import ConfigChangeEventData from app.schemas.rule import FilterRuleGroup -from app.schemas.types import MediaType +from app.schemas.types import EventType, MediaType, SystemConfigKey def test_agent_rule_group_media_type_accepts_music_aliases(): @@ -66,3 +72,74 @@ def test_rule_group_category_cannot_cross_media_types(monkeypatch): media = MediaInfo(type=MediaType.TV, category="shared") assert helper.get_rule_group_by_media(media=media, group_names=["movie-category"]) == [] + + +def test_reconcile_rule_group_references_removes_all_dangling_bindings(monkeypatch): + """规则组设置保存后应清理全局默认项、订阅默认值和已有订阅中的悬空名称。""" + values = { + SystemConfigKey.SearchFilterRuleGroups: ["keep", "deleted"], + SystemConfigKey.SubscribeFilterRuleGroups: ["deleted"], + SystemConfigKey.BestVersionFilterRuleGroups: ["keep"], + SystemConfigKey.DefaultMovieSubscribeConfig: { + "quality": "WEB-DL", + "filter_groups": ["deleted", "keep"], + }, + SystemConfigKey.DefaultTvSubscribeConfig: {"filter_groups": ["deleted"]}, + SystemConfigKey.DefaultMusicSubscribeConfig: {}, + } + + class Config: + """记录规则引用对账产生的系统配置更新。""" + + def get(self, key): + """读取当前测试配置。""" + return values.get(key) + + def set(self, key, value): + """保存配置并更新测试快照。""" + values[key] = value + return True + + subscribes = [ + SimpleNamespace( + id=1, + name="Example", + season=1, + filter_groups=["deleted", "keep"], + ) + ] + updates = [] + subscribe_port = SimpleNamespace( + list=lambda: subscribes, + update=lambda subscribe_id, payload: updates.append((subscribe_id, payload)), + ) + monkeypatch.setattr(subscribe_module, "_system_config", lambda: Config()) + monkeypatch.setattr( + subscribe_module, + "get_chain_subscribe_port", + lambda: subscribe_port, + ) + + SubscribeChain.reconcile_rule_group_references( + object.__new__(SubscribeChain), + Event( + EventType.ConfigChanged, + ConfigChangeEventData( + key=SystemConfigKey.UserFilterRuleGroups, + value=[{"name": "keep", "rule_string": "4K"}], + ), + ), + ) + + assert values[SystemConfigKey.SearchFilterRuleGroups] == ["keep"] + assert values[SystemConfigKey.SubscribeFilterRuleGroups] == [] + assert values[SystemConfigKey.BestVersionFilterRuleGroups] == ["keep"] + assert values[SystemConfigKey.DefaultMovieSubscribeConfig] == { + "quality": "WEB-DL", + "filter_groups": ["keep"], + } + assert values[SystemConfigKey.DefaultTvSubscribeConfig] == { + "filter_groups": [], + } + assert updates == [(1, {"filter_groups": ["keep"]})] + assert subscribes[0].filter_groups == ["keep"]