"""
过滤规则查询、校验、序列化和引用处理应用服务。
"""

import copy
import re
from collections.abc import Awaitable, Callable
from contextlib import AbstractAsyncContextManager
from typing import Any, Dict, Iterable, Optional

from app.application.configuration import get_configured_system_config
from app.application.rules import (
    BUILTIN_RULE_SET,
    AsyncRuleGroupMutationService,
    RuleHelper,
    RuleParser,
)
from app.application.subscription.contract import SubscriptionRepository
from app.schemas.rule import CustomRule, FilterRuleGroup
from app.schemas.types import SystemConfigKey

RULE_ID_PATTERN = re.compile(r"^[A-Za-z0-9]+$")
RULE_TOKEN_PATTERN = re.compile(r"[A-Za-z][A-Za-z0-9]*|[0-9][A-Za-z0-9]+")
NUMERIC_RANGE_PATTERN = re.compile(r"^\d+(?:\.\d+)?(?:\s*-\s*\d+(?:\.\d+)?)?$")
RuleConfigPublisher = Callable[[SystemConfigKey, Any], Awaitable[None]]

MEDIA_TYPE_ALIASES = {
    "movie": "电影",
    "film": "电影",
    "tv": "电视剧",
    "series": "电视剧",
    "show": "电视剧",
    "music": "音乐",
    "电影": "电影",
    "电视剧": "电视剧",
    "音乐": "音乐",
}

RULE_STRING_SYNTAX = {
    "level_separator": ">",
    "and_operator": "&",
    "not_operator": "!",
    "supported_grouping": "Parentheses are supported inside a single level.",
    "spacing_note": "Prefer spaces around '&', and '>' for readability; use '!RULE' for negation.",
    "match_order": "Levels are evaluated from left to right. The first matched level wins and stops further matching.",
    "match_result": "If no level matches, the torrent is filtered out. If a level matches, the torrent is kept.",
    "writing_workflow": [
        "First query built-in rules and custom rules to learn valid rule IDs.",
        "Compose one priority level with '&', '!' and optional parentheses.",
        "Join multiple priority levels with '>' from highest priority to lowest priority.",
        "Use spaces around '&', and '>' for readability.",
    ],
    "examples": [
        {
            "description": "Prefer torrents with special subtitles and Chinese dubbing at 4K, otherwise fall back to Chinese subtitles and Chinese dubbing at 4K.",
            "rule_string": "SPECSUB & CNVOI & 4K & !BLU & !REMUX & !WEBDL > CNSUB & CNVOI & 4K & !BLU & !REMUX & !WEBDL",
        },
        {
            "description": "Inside one level, require 4K and reject Blu-ray source.",
            "rule_string": "4K & !BLU",
        },
        {
            "description": "Inside one level, accept either special subtitles or Chinese subtitles, then also require 1080P.",
            "rule_string": "(SPECSUB | CNSUB) & 1080P",
        },
    ],
}


def normalize_optional_text(value: Optional[str]) -> Optional[str]:
    """
    把空白字符串折叠为 None，避免保存无意义的空值。
    """
    if value is None:
        return None
    value = str(value).strip()
    return value or None


def normalize_media_type(value: Optional[str]) -> Optional[str]:
    """
    兼容英中文媒体类型输入，最终统一为后端实际使用的中文值。
    """
    value = normalize_optional_text(value)
    if not value:
        return None
    normalized = MEDIA_TYPE_ALIASES.get(value.lower(), value)
    if normalized not in {"电影", "电视剧", "音乐"}:
        raise ValueError("media_type 仅支持 '电影'、'电视剧'、'音乐'、'movie'、'tv' 或 'music'")
    return normalized


def validate_numeric_range(field_name: str, value: Optional[str]) -> Optional[str]:
    """
    校验 size_range / publish_time 这类单值或区间值。
    """
    value = normalize_optional_text(value)
    if not value:
        return None
    if not NUMERIC_RANGE_PATTERN.match(value):
        raise ValueError(f"{field_name} 格式无效，支持 '1000' 或 '1000-5000' 这类数字区间格式")

    parts = [float(item.strip()) for item in value.split("-")]
    if len(parts) == 2 and parts[0] > parts[1]:
        raise ValueError(f"{field_name} 区间起始值不能大于结束值")
    return value


def validate_seeders(value: Optional[str]) -> Optional[str]:
    """
    做种人数最终会被 int() 解析，这里提前拦住非法值。
    """
    value = normalize_optional_text(value)
    if not value:
        return None
    if not value.isdigit():
        raise ValueError("seeders 必须是非负整数")
    return value


def get_builtin_rules() -> Dict[str, dict[str, Any]]:
    """
    返回内置规则的深拷贝，避免调用方误改共享常量。
    """
    return copy.deepcopy(BUILTIN_RULE_SET)


def get_custom_rules() -> list[CustomRule]:
    """
    读取当前配置中的自定义规则。
    """
    return RuleHelper().get_custom_rules()


def get_rule_groups() -> list[FilterRuleGroup]:
    """
    读取当前配置中的过滤规则组。
    """
    return RuleHelper().get_rule_groups()


def build_custom_rule_map(rules: Optional[Iterable[CustomRule]] = None) -> Dict[str, CustomRule]:
    """
    按非空规则标识构建自定义规则索引，未提供有效列表时读取当前配置。
    """
    return {rule.id: rule for rule in (rules or get_custom_rules()) if rule.id}


def build_rule_group_map(
    groups: Optional[Iterable[FilterRuleGroup]] = None,
) -> Dict[str, FilterRuleGroup]:
    """
    按非空组名构建规则组索引，未提供有效列表时读取当前配置。
    """
    return {group.name: group for group in (groups or get_rule_groups()) if group.name}


def extract_rule_tokens(rule_string: Optional[str]) -> list[str]:
    """
    从规则串里提取规则 ID，用于引用分析和未知规则校验。
    """
    if not rule_string:
        return []
    # dict.fromkeys 用来在保留顺序的同时去重，便于展示和报错。
    return list(dict.fromkeys(RULE_TOKEN_PATTERN.findall(rule_string)))


def parse_rule_string(rule_string: str) -> dict[str, Any]:
    """
    使用后端同款 RuleParser 解析规则串，并拆出每一层的元数据。
    """
    normalized = normalize_optional_text(rule_string)
    if not normalized:
        raise ValueError("rule_string 不能为空")

    parser = RuleParser()
    levels = [level.strip() for level in normalized.split(">")]
    if any(not level for level in levels):
        raise ValueError("rule_string 不能包含空层级，请检查 '>' 两侧内容")

    parsed_levels: list[dict[str, Any]] = []
    for index, level in enumerate(levels, start=1):
        try:
            parser.parse(level)
        # 依赖 pyparsing 的具体异常
        except Exception as exc:  # pragma: no cover
            raise ValueError(f"规则串第 {index} 层语法错误: {exc}") from exc

        parsed_levels.append(
            {
                "priority": index,
                "expression": level,
                "referenced_rules": extract_rule_tokens(level),
            }
        )

    return {
        "rule_string": " > ".join(levels),
        "levels": parsed_levels,
        "referenced_rules": extract_rule_tokens(normalized),
    }


def validate_rule_string(rule_string: str, available_rule_ids: Iterable[str]) -> dict[str, Any]:
    """
    校验规则串语法和引用规则是否都存在。
    """
    parsed = parse_rule_string(rule_string)
    available_ids = set(available_rule_ids)
    unknown_rules = sorted({rule_id for rule_id in parsed["referenced_rules"] if rule_id not in available_ids})
    if unknown_rules:
        raise ValueError(f"rule_string 引用了不存在的规则: {', '.join(unknown_rules)}")
    return parsed


def serialize_builtin_rule(rule_id: str, payload: dict[str, Any]) -> dict[str, Any]:
    """
    把内置规则整理成适合 Agent 阅读的结构。
    """
    data = copy.deepcopy(payload)
    data["id"] = rule_id
    data["source"] = "builtin"
    return data


def serialize_custom_rule(
    rule: CustomRule,
    group_refs: Optional[list[str]] = None,
) -> dict[str, Any]:
    """
    导出自定义规则并附带规则组引用，省略值为 None 的模型字段。
    """
    data = rule.model_dump(exclude_none=True)
    data["source"] = "custom"
    data["referenced_by_rule_groups"] = group_refs or []
    return data


def serialize_rule_group(
    group: FilterRuleGroup,
    usage: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    """
    查询时尽量附带解析结果，便于 Agent 理解优先级层级。
    """
    data = group.model_dump(exclude_none=True)
    if group.rule_string:
        try:
            parsed = parse_rule_string(group.rule_string)
            data["levels"] = parsed["levels"]
            data["referenced_rules"] = parsed["referenced_rules"]
            data["syntax_valid"] = True
        except ValueError as exc:
            data["syntax_valid"] = False
            data["syntax_error"] = str(exc)
            data["referenced_rules"] = extract_rule_tokens(group.rule_string)
    else:
        data["syntax_valid"] = False
        data["syntax_error"] = "rule_string 为空"
        data["referenced_rules"] = []
    data["usage"] = usage or default_rule_group_usage()
    return data


def default_rule_group_usage() -> dict[str, Any]:
    """
    创建尚无全局使用标记和订阅引用的独立规则组用量结构。
    """
    return {
        "used_in_global_search": False,
        "used_in_global_subscribe": False,
        "used_in_global_best_version": False,
        "subscribes": [],
    }


async def collect_rule_group_usages(
    repository: SubscriptionRepository,
    group_names: Optional[Iterable[str]] = None,
) -> Dict[str, dict[str, Any]]:
    """
    收集规则组在全局配置和订阅上的引用情况。
    """
    target_names = set(group_names or [])
    search_groups = set(get_configured_system_config().get(SystemConfigKey.SearchFilterRuleGroups) or [])
    subscribe_groups = set(get_configured_system_config().get(SystemConfigKey.SubscribeFilterRuleGroups) or [])
    best_version_groups = set(get_configured_system_config().get(SystemConfigKey.BestVersionFilterRuleGroups) or [])

    usage_map: dict[str, dict[str, Any]] = {
        name: default_rule_group_usage() for name in target_names
    }

    def ensure_usage(name: str) -> dict[str, Any]:
        """
        为规则组创建并复用本次统计中的使用量记录。
        """
        if name not in usage_map:
            usage_map[name] = default_rule_group_usage()
        return usage_map[name]

    for name in search_groups:
        if target_names and name not in target_names:
            continue
        ensure_usage(name)["used_in_global_search"] = True
    for name in subscribe_groups:
        if target_names and name not in target_names:
            continue
        ensure_usage(name)["used_in_global_subscribe"] = True
    for name in best_version_groups:
        if target_names and name not in target_names:
            continue
        ensure_usage(name)["used_in_global_best_version"] = True

    subscribes = await repository.async_list()
    for subscribe in subscribes:
        filter_groups = (
            [str(name) for name in subscribe.filter_groups] if isinstance(subscribe.filter_groups, list) else []
        )
        for name in filter_groups:
            if target_names and name not in target_names:
                continue
            ensure_usage(name)["subscribes"].append(
                {
                    "subscribe_id": subscribe.id,
                    "name": subscribe.name,
                    "season": subscribe.season,
                    "type": subscribe.type,
                    "username": subscribe.username,
                    "best_version": bool(subscribe.best_version),
                }
            )

    return usage_map


def collect_custom_rule_group_refs(
    rule_groups: Iterable[FilterRuleGroup],
    rule_ids: Optional[Iterable[str]] = None,
) -> Dict[str, list[str]]:
    """
    收集自定义规则被哪些规则组引用。
    """
    target_rule_ids = set(rule_ids or [])
    refs: Dict[str, list[str]] = {rule_id: [] for rule_id in target_rule_ids}

    for group in rule_groups:
        if not group.name or not group.rule_string:
            continue
        referenced = set(extract_rule_tokens(group.rule_string))
        for rule_id in referenced:
            if target_rule_ids and rule_id not in target_rule_ids:
                continue
            refs.setdefault(rule_id, []).append(group.name)

    for names in refs.values():
        names.sort()
    return refs


def normalize_custom_rule(
    rule_id: str,
    name: str,
    include: Optional[str],
    exclude: Optional[str],
    size_range: Optional[str],
    seeders: Optional[str],
    publish_time: Optional[str],
    existing_rules: Iterable[CustomRule],
    original_rule_id: Optional[str] = None,
) -> CustomRule:
    """
    新增/更新自定义规则时统一走这里，避免多处散落校验逻辑。
    """
    normalized_rule_id = normalize_optional_text(rule_id)
    normalized_name = normalize_optional_text(name)
    if not normalized_rule_id:
        raise ValueError("rule_id 不能为空")
    if not normalized_name:
        raise ValueError("name 不能为空")
    if not RULE_ID_PATTERN.match(normalized_rule_id):
        raise ValueError("rule_id 仅支持英文字母和数字")
    if normalized_rule_id in BUILTIN_RULE_SET and normalized_rule_id != original_rule_id:
        raise ValueError(f"rule_id '{normalized_rule_id}' 与内置规则冲突，不能覆盖内置规则")

    for existing_rule in existing_rules:
        if existing_rule.id == normalized_rule_id and existing_rule.id != original_rule_id:
            raise ValueError(f"rule_id '{normalized_rule_id}' 已存在")
        if existing_rule.name == normalized_name and existing_rule.id != original_rule_id:
            raise ValueError(f"规则名称 '{normalized_name}' 已存在")

    return CustomRule(
        id=normalized_rule_id,
        name=normalized_name,
        include=normalize_optional_text(include),
        exclude=normalize_optional_text(exclude),
        size_range=validate_numeric_range("size_range", size_range),
        seeders=validate_seeders(seeders),
        publish_time=validate_numeric_range("publish_time", publish_time),
    )


def normalize_rule_group(
    name: str,
    rule_string: str,
    media_type: Optional[str],
    category: Optional[str],
    existing_groups: Iterable[FilterRuleGroup],
    available_rule_ids: Iterable[str],
    original_name: Optional[str] = None,
) -> tuple[FilterRuleGroup, dict[str, Any]]:
    """
    新增/更新规则组时统一校验名字、适用范围和规则串。
    """
    normalized_name = normalize_optional_text(name)
    if not normalized_name:
        raise ValueError("规则组名称不能为空")

    for group in existing_groups:
        if group.name == normalized_name and group.name != original_name:
            raise ValueError(f"规则组名称 '{normalized_name}' 已存在")

    normalized_media_type = normalize_media_type(media_type)
    normalized_category = normalize_optional_text(category)
    if normalized_category and not normalized_media_type:
        raise ValueError("设置 category 时必须同时设置 media_type")

    parsed = validate_rule_string(rule_string, available_rule_ids)
    return (
        FilterRuleGroup(
            name=normalized_name,
            rule_string=parsed["rule_string"],
            media_type=normalized_media_type,
            category=normalized_category,
        ),
        parsed,
    )


async def save_system_config(
    key: SystemConfigKey,
    value: Any,
) -> Optional[bool]:
    """
    通过统一异步配置入口保存配置并等待运行时重载。
    """
    normalized_value = value
    if isinstance(normalized_value, list):
        normalized_value = [item for item in normalized_value if item is not None and item != ""]
        normalized_value = normalized_value or None

    return await get_configured_system_config().async_set(key, normalized_value)


def replace_rule_id_in_rule_string(rule_string: str, old_rule_id: str, new_rule_id: str) -> str:
    """
    只替换完整 token，避免误伤其他规则名。
    """
    pattern = re.compile(rf"(?<![A-Za-z0-9]){re.escape(old_rule_id)}(?![A-Za-z0-9])")
    return pattern.sub(new_rule_id, rule_string)


class FilterRuleService:
    """
    提供过滤规则和规则组的查询与事务化修改用例。
    """

    def __init__(
        self,
        subscriptions: SubscriptionRepository,
        mutation_scope: Callable[[], AbstractAsyncContextManager[AsyncRuleGroupMutationService]],
        publish_config_changed: RuleConfigPublisher,
    ) -> None:
        """
        注入订阅端口、规则组事务作用域和配置事件发布端口。
        """
        self._subscriptions = subscriptions
        self._mutation_scope = mutation_scope
        self._publish_config_changed = publish_config_changed

    @staticmethod
    def query_builtin(rule_ids: Optional[list[str]] = None) -> dict[str, Any]:
        """
        查询内置过滤规则及规则串语法。
        """
        rules = get_builtin_rules()
        if rule_ids:
            target_ids = set(rule_ids)
            rules = {key: value for key, value in rules.items() if key in target_ids}
        serialized = [serialize_builtin_rule(rule_id, payload) for rule_id, payload in rules.items()]
        return {
            "count": len(serialized),
            "rule_string_syntax": RULE_STRING_SYNTAX,
            "rules": serialized,
        }

    @staticmethod
    def query_custom(
        rule_ids: Optional[list[str]] = None,
        *,
        include_group_refs: bool = True,
    ) -> dict[str, Any]:
        """
        查询自定义过滤规则及规则组引用。
        """
        rules = get_custom_rules()
        if rule_ids:
            target_ids = set(rule_ids)
            rules = [rule for rule in rules if rule.id in target_ids]
        refs = (
            collect_custom_rule_group_refs(
                get_rule_groups(),
                [str(rule.id) for rule in rules if rule.id],
            )
            if include_group_refs
            else {}
        )
        serialized = [
            serialize_custom_rule(rule, refs.get(str(rule_id)))
            for rule in rules
            if (rule_id := rule.id)
        ]
        return {"count": len(serialized), "rules": serialized}

    async def query_groups(
        self,
        group_names: Optional[list[str]] = None,
        *,
        include_usage: bool = True,
    ) -> dict[str, Any]:
        """
        查询规则组、解析层级和可选引用位置。
        """
        groups = get_rule_groups()
        if group_names:
            target_names = set(group_names)
            groups = [group for group in groups if group.name in target_names]
        usage = (
            await collect_rule_group_usages(
                self._subscriptions,
                [str(group.name) for group in groups if group.name],
            )
            if include_usage
            else {}
        )
        serialized = [
            serialize_rule_group(group, usage.get(group.name or "")) for group in groups
        ]
        return {
            "count": len(serialized),
            "rule_string_syntax": RULE_STRING_SYNTAX,
            "rule_groups": serialized,
        }

    async def add_custom(self, **values: Any) -> dict[str, Any]:
        """
        校验并新增一条自定义过滤规则。
        """
        rules = get_custom_rules()
        new_rule = normalize_custom_rule(existing_rules=rules, **values)
        final_rules = [*rules, new_rule]
        await self._save_custom_rules(rules, final_rules)
        return {
            "message": f"已新增自定义过滤规则 {new_rule.id}",
            "custom_rule": serialize_custom_rule(new_rule),
            "count": len(final_rules),
        }

    async def _save_custom_rules(self, expected: list[CustomRule], final: list[CustomRule]) -> None:
        """
        通过现有原子作用域核对规则快照，拒绝并发新增、修改或删除覆盖。

        :param expected: 本次操作读取的旧规则集合
        :param final: 本次操作希望保存的完整规则集合
        :raises RuleGroupMutationConflictError: 配置已被另一请求修改，需要重新读取
        """
        groups = [group.model_dump(exclude_none=True) for group in get_rule_groups()]
        definitions = [rule.model_dump(exclude_none=True) for rule in final]
        async with self._mutation_scope() as mutation:
            await mutation.apply(groups, expected_rule_groups=groups, custom_rules=definitions,
                                 expected_custom_rules=[rule.model_dump(exclude_none=True) for rule in expected])
        await self._publish_config_changed(SystemConfigKey.CustomFilterRules, definitions)

    async def reorder_custom(
        self,
        rule_ids: list[str],
        *,
        expected_rule_ids: Optional[list[str]] = None,
    ) -> dict[str, Any]:
        """
        按完整 ID 列表重排规则，并用事务快照拒绝并发覆盖。
        """
        rules = get_custom_rules()
        current_ids = [str(rule.id) for rule in rules if rule.id]
        if len(current_ids) != len(rules):
            raise ValueError("自定义规则存在缺少 ID 的损坏项，不能调整顺序")
        self._validate_reorder(
            "自定义规则",
            current_ids,
            rule_ids,
            expected_rule_ids,
        )
        rules_by_id = {str(rule.id): rule for rule in rules if rule.id}
        ordered_rules = [rules_by_id[rule_id] for rule_id in rule_ids]
        expected_rules = [rule.model_dump(exclude_none=True) for rule in rules]
        rule_definitions = [rule.model_dump(exclude_none=True) for rule in ordered_rules]
        groups = get_rule_groups()
        group_definitions = [group.model_dump(exclude_none=True) for group in groups]
        async with self._mutation_scope() as mutation:
            await mutation.apply(
                group_definitions,
                expected_rule_groups=group_definitions,
                custom_rules=rule_definitions,
                expected_custom_rules=expected_rules,
            )
        await self._publish_config_changed(SystemConfigKey.CustomFilterRules, rule_definitions)
        return {
            "message": "已调整自定义过滤规则顺序",
            "count": len(rule_ids),
            "rule_ids": rule_ids,
        }

    async def update_custom(
        self,
        *,
        current_rule_id: str,
        new_rule_id: Optional[str] = None,
        name: Optional[str] = None,
        include: Optional[str] = None,
        exclude: Optional[str] = None,
        size_range: Optional[str] = None,
        seeders: Optional[str] = None,
        publish_time: Optional[str] = None,
    ) -> dict[str, Any]:
        """
        更新自定义规则，并在改名时原子重写规则组引用。
        """
        rules = get_custom_rules()
        current = next((rule for rule in rules if rule.id == current_rule_id), None)
        if current is None:
            raise ValueError(f"自定义过滤规则 '{current_rule_id}' 不存在")
        if current.id is None or current.name is None:
            raise ValueError("自定义过滤规则缺少 id 或 name")
        updated = normalize_custom_rule(
            rule_id=new_rule_id or current.id,
            name=name if name is not None else current.name,
            include=include if include is not None else current.include,
            exclude=exclude if exclude is not None else current.exclude,
            size_range=size_range if size_range is not None else current.size_range,
            seeders=seeders if seeders is not None else current.seeders,
            publish_time=(publish_time if publish_time is not None else current.publish_time),
            existing_rules=rules,
            original_rule_id=current.id,
        )
        expected_rules = [rule.model_dump(exclude_none=True) for rule in rules]
        final_rules = [updated if rule.id == current.id else rule for rule in rules]
        rule_definitions = [rule.model_dump(exclude_none=True) for rule in final_rules]
        groups = get_rule_groups()
        updated_groups = groups
        renamed_refs: list[str] = []
        if updated.id != current.id:
            updated_groups = []
            for group in groups:
                new_rule_string = replace_rule_id_in_rule_string(
                    group.rule_string or "",
                    current.id or "",
                    updated.id or "",
                )
                if new_rule_string != (group.rule_string or ""):
                    renamed_refs.append(group.name or "")
                    group = group.model_copy(update={"rule_string": new_rule_string})
                updated_groups.append(group)
            expected_groups = [group.model_dump(exclude_none=True) for group in groups]
            group_definitions = [group.model_dump(exclude_none=True) for group in updated_groups]
            async with self._mutation_scope() as mutation:
                await mutation.apply(
                    group_definitions,
                    expected_rule_groups=expected_groups,
                    custom_rules=rule_definitions,
                    expected_custom_rules=expected_rules,
                )
            await self._publish_config_changed(SystemConfigKey.CustomFilterRules, rule_definitions)
            await self._publish_config_changed(SystemConfigKey.UserFilterRuleGroups, group_definitions)
        else:
            await self._save_custom_rules(rules, final_rules)
        if updated.id is None:
            raise ValueError("更新后的自定义过滤规则缺少 id")
        refs = collect_custom_rule_group_refs(updated_groups, [updated.id]).get(updated.id, [])
        return {
            "message": f"已更新自定义过滤规则 {updated.id}",
            "custom_rule": serialize_custom_rule(updated, refs),
            "rule_groups_updated_for_rule_id_rename": renamed_refs,
        }

    async def delete_custom(self, rule_id: str) -> dict[str, Any]:
        """
        删除未被规则组引用的自定义过滤规则。
        """
        rules = get_custom_rules()
        if not any(rule.id == rule_id for rule in rules):
            raise ValueError(f"自定义过滤规则 '{rule_id}' 不存在")
        refs = collect_custom_rule_group_refs(get_rule_groups(), [rule_id]).get(rule_id, [])
        if refs:
            raise ValueError(f"自定义过滤规则 '{rule_id}' 仍被规则组引用: {', '.join(refs)}")
        remaining = [rule for rule in rules if rule.id != rule_id]
        await self._save_custom_rules(rules, remaining)
        return {
            "message": f"已删除自定义过滤规则 {rule_id}",
            "count": len(remaining),
        }

    async def add_group(
        self,
        *,
        name: str,
        rule_string: str,
        media_type: Optional[str] = None,
        category: Optional[str] = None,
    ) -> dict[str, Any]:
        """
        校验并新增规则组。
        """
        groups = get_rule_groups()
        available_ids = set(get_builtin_rules()) | set(build_custom_rule_map())
        new_group, parsed = normalize_rule_group(
            name,
            rule_string,
            media_type,
            category,
            groups,
            available_ids,
        )
        if new_group.name is None:
            raise ValueError("新规则组缺少名称")
        expected = [group.model_dump(exclude_none=True) for group in groups]
        definitions = [*expected, new_group.model_dump(exclude_none=True)]
        async with self._mutation_scope() as mutation:
            await mutation.apply(definitions, expected_rule_groups=expected)
        await self._publish_config_changed(SystemConfigKey.UserFilterRuleGroups, definitions)
        usage = await collect_rule_group_usages(self._subscriptions, [new_group.name])
        return {
            "message": f"已新增规则组 {new_group.name}",
            "rule_group": serialize_rule_group(new_group, usage.get(new_group.name)),
            "parsed": parsed,
            "count": len(definitions),
        }

    async def reorder_groups(
        self,
        group_names: list[str],
        *,
        expected_group_names: Optional[list[str]] = None,
    ) -> dict[str, Any]:
        """
        按完整名称列表重排规则组，并复用原子修改作用域。
        """
        groups = get_rule_groups()
        current_names = [str(group.name) for group in groups if group.name]
        if len(current_names) != len(groups):
            raise ValueError("规则组存在缺少名称的损坏项，不能调整顺序")
        self._validate_reorder(
            "规则组",
            current_names,
            group_names,
            expected_group_names,
        )
        groups_by_name = {str(group.name): group for group in groups if group.name}
        ordered_groups = [groups_by_name[name] for name in group_names]
        expected = [group.model_dump(exclude_none=True) for group in groups]
        definitions = [group.model_dump(exclude_none=True) for group in ordered_groups]
        async with self._mutation_scope() as mutation:
            await mutation.apply(definitions, expected_rule_groups=expected)
        await self._publish_config_changed(SystemConfigKey.UserFilterRuleGroups, definitions)
        return {
            "message": "已调整过滤规则组顺序",
            "count": len(group_names),
            "group_names": group_names,
        }

    async def update_group(
        self,
        *,
        current_name: str,
        new_name: Optional[str] = None,
        rule_string: Optional[str] = None,
        media_type: Optional[str] = None,
        category: Optional[str] = None,
    ) -> dict[str, Any]:
        """
        更新规则组并原子重写全局配置和订阅引用。
        """
        groups = get_rule_groups()
        current = next((group for group in groups if group.name == current_name), None)
        if current is None:
            raise ValueError(f"规则组 '{current_name}' 不存在")
        if current.name is None:
            raise ValueError("当前规则组缺少名称")
        available_ids = set(get_builtin_rules()) | set(build_custom_rule_map())
        updated, parsed = normalize_rule_group(
            new_name or current.name or "",
            rule_string if rule_string is not None else current.rule_string or "",
            media_type if media_type is not None else current.media_type,
            category if category is not None else current.category,
            groups,
            available_ids,
            original_name=current.name,
        )
        expected = [group.model_dump(exclude_none=True) for group in groups]
        final_groups = [updated if group.name == current.name else group for group in groups]
        definitions = [group.model_dump(exclude_none=True) for group in final_groups]
        async with self._mutation_scope() as mutation:
            result = await mutation.apply(
                definitions,
                expected_rule_groups=expected,
                previous_name=current.name,
                current_name=updated.name,
            )
        updated_name = updated.name
        if updated_name is None:
            raise ValueError("更新后的规则组缺少名称")
        await self._publish_config_changed(SystemConfigKey.UserFilterRuleGroups, definitions)
        usage = await collect_rule_group_usages(self._subscriptions, [updated_name])
        return {
            "message": f"已更新规则组 {updated_name}",
            "rule_group": serialize_rule_group(updated, usage.get(updated_name)),
            "parsed": parsed,
            "reference_updates": result.to_dict(),
        }

    async def delete_group(self, name: str) -> dict[str, Any]:
        """
        删除规则组并原子清理全部引用。
        """
        groups = get_rule_groups()
        if not any(group.name == name for group in groups):
            raise ValueError(f"规则组 '{name}' 不存在")
        expected = [group.model_dump(exclude_none=True) for group in groups]
        remaining = [group for group in groups if group.name != name]
        definitions = [group.model_dump(exclude_none=True) for group in remaining]
        async with self._mutation_scope() as mutation:
            result = await mutation.apply(
                definitions,
                expected_rule_groups=expected,
                previous_name=name,
            )
        await self._publish_config_changed(SystemConfigKey.UserFilterRuleGroups, definitions)
        return {
            "message": f"已删除规则组 {name}",
            "count": len(remaining),
            "reference_updates": result.to_dict(),
        }

    @staticmethod
    def _validate_reorder(
        item_label: str,
        current_names: list[str],
        requested_names: list[str],
        expected_names: Optional[list[str]],
    ) -> None:
        """
        验证重排列表完整、唯一且基于未过期的名称集合。
        """
        if any(not name.strip() for name in requested_names):
            raise ValueError(f"{item_label}顺序不能包含空名称")
        if len(set(requested_names)) != len(requested_names):
            raise ValueError(f"{item_label}顺序不能包含重复名称")
        if set(requested_names) != set(current_names):
            raise ValueError(f"{item_label}集合已变化，请重新读取后再试")
        if (
            expected_names is not None
            and current_names != expected_names
            and current_names != requested_names
        ):
            raise ValueError(f"{item_label}顺序已被其他请求修改，请重新读取后再试")