mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-09-06 07:56:52 +08:00
refactor(application): filter_rules.py 更名 rules.py 并合并 RuleHelper
规则域收敛为单一事实来源 app/application/rules.py: - filter_rules.py 更名为 rules.py(内置规则集 + RuleParser) - filter.py 仅含 66 行 RuleHelper(用户规则组配置访问),并入 rules.py - 同步更新 5 处引用方导入、nettest stub 目标与架构文档 - 兼容映射 app.helper.rule 指向新路径,legacy 导入仍可用
This commit is contained in:
@@ -7,9 +7,9 @@ from typing import Any, Dict, Iterable, Optional
|
||||
from app.runtime.events import eventmanager
|
||||
from app.db.oper.subscribe import SubscribeOper
|
||||
from app.db.oper.systemconfig import SystemConfigOper
|
||||
from app.application.filter import RuleHelper
|
||||
from app.application.filter_rules import RuleParser
|
||||
from app.application.filter_rules import BUILTIN_RULE_SET
|
||||
from app.application.rules import RuleHelper
|
||||
from app.application.rules import RuleParser
|
||||
from app.application.rules import BUILTIN_RULE_SET
|
||||
from app.schemas import CustomRule, FilterRuleGroup
|
||||
from app.schemas.event import ConfigChangeEventData
|
||||
from app.schemas.types import EventType, SystemConfigKey
|
||||
|
||||
@@ -41,7 +41,7 @@ from app.adapters.external.market import (
|
||||
)
|
||||
from app.application.messaging.message import MessageHelper
|
||||
from app.runtime.progress import ProgressHelper
|
||||
from app.application.filter import RuleHelper
|
||||
from app.application.rules import RuleHelper
|
||||
from app.adapters.external.server import MoviePilotServerHelper
|
||||
from app.runtime.state import SystemHelper
|
||||
from app.runtime.log import logger
|
||||
|
||||
@@ -1,66 +0,0 @@
|
||||
from typing import List, Optional
|
||||
|
||||
from app.db.oper.systemconfig import SystemConfigOper
|
||||
from app.domain.context import MediaInfo
|
||||
from app.schemas import CustomRule, FilterRuleGroup
|
||||
from app.schemas.types import SystemConfigKey
|
||||
|
||||
|
||||
class RuleHelper:
|
||||
"""读取用户过滤规则配置,并按媒体上下文选择适用规则组。"""
|
||||
|
||||
@staticmethod
|
||||
def get_rule_groups() -> List[FilterRuleGroup]:
|
||||
"""返回用户配置的全部过滤规则组。"""
|
||||
rule_groups: List[dict] = SystemConfigOper().get(
|
||||
SystemConfigKey.UserFilterRuleGroups
|
||||
)
|
||||
if not rule_groups:
|
||||
return []
|
||||
return [FilterRuleGroup(**group) for group in rule_groups]
|
||||
|
||||
def get_rule_group(self, group_name: str) -> Optional[FilterRuleGroup]:
|
||||
"""按名称返回过滤规则组。"""
|
||||
return next(
|
||||
(group for group in self.get_rule_groups() if group.name == group_name),
|
||||
None,
|
||||
)
|
||||
|
||||
def get_rule_group_by_media(
|
||||
self,
|
||||
media: Optional[MediaInfo] = None,
|
||||
group_names: Optional[list] = None,
|
||||
) -> List[FilterRuleGroup]:
|
||||
"""按媒体类型、分类和候选名称筛选适用规则组。"""
|
||||
rule_groups = self.get_rule_groups()
|
||||
if group_names:
|
||||
rule_groups = [
|
||||
group for group in rule_groups if group.name in group_names
|
||||
]
|
||||
return [
|
||||
group
|
||||
for group in rule_groups
|
||||
if not group.media_type
|
||||
or (
|
||||
media
|
||||
and (
|
||||
(not group.category and group.media_type == media.type.value)
|
||||
or group.category == media.category
|
||||
)
|
||||
)
|
||||
]
|
||||
|
||||
@staticmethod
|
||||
def get_custom_rules() -> List[CustomRule]:
|
||||
"""返回用户配置的全部自定义过滤规则。"""
|
||||
rules: List[dict] = SystemConfigOper().get(SystemConfigKey.CustomFilterRules)
|
||||
if not rules:
|
||||
return []
|
||||
return [CustomRule(**rule) for rule in rules]
|
||||
|
||||
def get_custom_rule(self, rule_id: str) -> Optional[CustomRule]:
|
||||
"""按 ID 返回一条自定义过滤规则。"""
|
||||
return next(
|
||||
(rule for rule in self.get_custom_rules() if rule.id == rule_id),
|
||||
None,
|
||||
)
|
||||
@@ -1,11 +1,79 @@
|
||||
"""过滤规则解析器与内置规则定义,过滤模块与 Agent 工具共享同一事实来源。"""
|
||||
"""
|
||||
规则域:用户规则组配置访问、内置规则定义与规则解析器,
|
||||
过滤模块与 Agent 工具共享同一事实来源。
|
||||
"""
|
||||
|
||||
import threading
|
||||
from typing import Dict
|
||||
from typing import Dict, List, Optional
|
||||
|
||||
from pyparsing import Forward, Literal, Word, alphas, infix_notation, opAssoc, alphanums, Combine, nums, ParseResults
|
||||
|
||||
from app.adapters.system import rust as rust_accel
|
||||
from app.db.oper.systemconfig import SystemConfigOper
|
||||
from app.domain.context import MediaInfo
|
||||
from app.schemas import CustomRule, FilterRuleGroup
|
||||
from app.schemas.types import SystemConfigKey
|
||||
|
||||
|
||||
class RuleHelper:
|
||||
"""读取用户过滤规则配置,并按媒体上下文选择适用规则组。"""
|
||||
|
||||
@staticmethod
|
||||
def get_rule_groups() -> List[FilterRuleGroup]:
|
||||
"""返回用户配置的全部过滤规则组。"""
|
||||
rule_groups: List[dict] = SystemConfigOper().get(
|
||||
SystemConfigKey.UserFilterRuleGroups
|
||||
)
|
||||
if not rule_groups:
|
||||
return []
|
||||
return [FilterRuleGroup(**group) for group in rule_groups]
|
||||
|
||||
def get_rule_group(self, group_name: str) -> Optional[FilterRuleGroup]:
|
||||
"""按名称返回过滤规则组。"""
|
||||
return next(
|
||||
(group for group in self.get_rule_groups() if group.name == group_name),
|
||||
None,
|
||||
)
|
||||
|
||||
def get_rule_group_by_media(
|
||||
self,
|
||||
media: Optional[MediaInfo] = None,
|
||||
group_names: Optional[list] = None,
|
||||
) -> List[FilterRuleGroup]:
|
||||
"""按媒体类型、分类和候选名称筛选适用规则组。"""
|
||||
rule_groups = self.get_rule_groups()
|
||||
if group_names:
|
||||
rule_groups = [
|
||||
group for group in rule_groups if group.name in group_names
|
||||
]
|
||||
return [
|
||||
group
|
||||
for group in rule_groups
|
||||
if not group.media_type
|
||||
or (
|
||||
media
|
||||
and (
|
||||
(not group.category and group.media_type == media.type.value)
|
||||
or group.category == media.category
|
||||
)
|
||||
)
|
||||
]
|
||||
|
||||
@staticmethod
|
||||
def get_custom_rules() -> List[CustomRule]:
|
||||
"""返回用户配置的全部自定义过滤规则。"""
|
||||
rules: List[dict] = SystemConfigOper().get(SystemConfigKey.CustomFilterRules)
|
||||
if not rules:
|
||||
return []
|
||||
return [CustomRule(**rule) for rule in rules]
|
||||
|
||||
def get_custom_rule(self, rule_id: str) -> Optional[CustomRule]:
|
||||
"""按 ID 返回一条自定义过滤规则。"""
|
||||
return next(
|
||||
(rule for rule in self.get_custom_rules() if rule.id == rule_id),
|
||||
None,
|
||||
)
|
||||
|
||||
|
||||
# 内置规则只在这里维护一份,便于过滤模块和 Agent 工具共享同一套事实来源。
|
||||
BUILTIN_RULE_SET: Dict[str, dict] = {
|
||||
@@ -5,11 +5,11 @@ from typing import List, Tuple, Union, Dict, Optional
|
||||
|
||||
from app.domain.context import TorrentInfo, MediaInfo
|
||||
from app.domain.metainfo import MetaInfo, clear_rust_parse_options_cache, _rust_parse_options
|
||||
from app.application.filter import RuleHelper
|
||||
from app.application.rules import RuleHelper
|
||||
from app.runtime.log import logger
|
||||
from app.modules import _ModuleBase
|
||||
from app.application.filter_rules import RuleParser
|
||||
from app.application.filter_rules import BUILTIN_RULE_SET
|
||||
from app.application.rules import RuleParser
|
||||
from app.application.rules import BUILTIN_RULE_SET
|
||||
from app.schemas.types import ModuleType, OtherModulesType, SystemConfigKey
|
||||
from app.adapters.system import rust as rust_accel
|
||||
from app.foundation import size as size_tools
|
||||
|
||||
@@ -543,7 +543,7 @@ MODULE_ALIASES: Dict[str, ModuleAlias] = {
|
||||
introduced="v3.0.0", owner="application",
|
||||
),
|
||||
"app.helper.rule": ModuleAlias(
|
||||
target="app.application.filter", replacement="app.sdk.services",
|
||||
target="app.application.rules", replacement="app.sdk.services",
|
||||
introduced="v3.0.0", owner="application",
|
||||
),
|
||||
"app.helper.scraper": ModuleAlias(
|
||||
|
||||
+1
-1
@@ -3,7 +3,7 @@
|
||||
from app.runtime.extensions.service_registry import ServiceBaseHelper, ServiceConfigHelper
|
||||
from app.runtime.state import SystemHelper
|
||||
from app.application.downloader import DownloaderHelper
|
||||
from app.application.filter import RuleHelper
|
||||
from app.application.rules import RuleHelper
|
||||
from app.application.mediaserver import (
|
||||
MediaServerIdentityHelper,
|
||||
MediaServerHelper,
|
||||
|
||||
Reference in New Issue
Block a user