zhenxun_bot/zhenxun/builtin_plugins/hooks/auth_activation.py

from __future__ import annotations

from collections.abc import Iterable
import contextlib
from dataclasses import dataclass, field
import html
import re
from typing import Any, Literal
import weakref

from loguru import logger
from nonebot.matcher import Matcher

ActivationDecision = Literal["match", "miss", "unknown"]
ActivationLane = Literal[
    "command_exact",
    "command_shortcut",
    "command_regex",
    "system",
    "passive_light",
    "passive_db",
    "passive_http",
    "passive_ai",
    "passive_render",
]

KNOWN_SAFE_RULE_NAMES = frozenset(
    {
        "CommandRule",
        "ShellCommandRule",
        "RegexRule",
        "StartswithRule",
        "EndswithRule",
        "FullmatchRule",
        "KeywordsRule",
        "IsTypeRule",
        "ToMeRule",
    }
)


@dataclass(frozen=True, slots=True)
class ActivationRuleDescriptor:
    kind: str
    value: object | None = None
    flags: int = 0
    ignorecase: bool = False
    deterministic_text: bool = False
    command_like: bool = False


@dataclass(slots=True)
class HandlerDescriptor:
    matcher: type[Matcher]
    module: str
    matcher_type: str
    priority: int
    lane: ActivationLane
    temp: bool = False
    block: bool = False
    command_like: bool = False
    deterministic_text: bool = False
    has_custom_rule: bool = False
    commands: tuple[str, ...] = ()
    shortcuts: tuple[str, ...] | None = None
    alconna: tuple[AlconnaDescriptor, ...] = ()
    rules: tuple[ActivationRuleDescriptor, ...] = ()


@dataclass(frozen=True, slots=True)
class AlconnaShortcutDescriptor:
    pattern: str
    fuzzy: bool = False
    flags: int = 0


@dataclass(frozen=True, slots=True)
class AlconnaDescriptor:
    command: str = ""
    aliases: tuple[str, ...] = ()
    prefixes: tuple[str, ...] = ()
    shortcuts: tuple[AlconnaShortcutDescriptor, ...] = ()
    compact: bool = False
    skip_for_unmatch: bool = True
    before_rule_count: int = 0
    after_rule_count: int = 0
    before_rule_known_safe: bool = True
    after_rule_known_safe: bool = True
    input_rewrite_extensions: tuple[str, ...] = ()

    @property
    def has_reply_merge_extension(self) -> bool:
        return "ReplyMergeExtension" in self.input_rewrite_extensions

    @property
    def regex_command(self) -> bool:
        return self.command.startswith("re:")


class AlconnaActivationIndex:
    """Safe, metadata-only Alconna prefilter.

    This index never executes Alconna.parse() or Matcher.check_rule(). It only
    skips matchers when the command head/shortcut is known to miss. Anything
    custom or ambiguous stays fail-open to preserve NoneBot compatibility.
    """

    def __init__(self) -> None:
        self._safe = 0
        self._unknown = 0

    @property
    def safe_count(self) -> int:
        return self._safe

    @property
    def unknown_count(self) -> int:
        return self._unknown

    def rebuild(self, descriptors: Iterable[HandlerDescriptor]) -> None:
        safe = 0
        unknown = 0
        for descriptor in descriptors:
            if descriptor.alconna:
                if any(_alconna_can_prefilter(item) for item in descriptor.alconna):
                    safe += 1
                else:
                    unknown += 1
        self._safe = safe
        self._unknown = unknown
        if logger.level("DEBUG"):
            logger.debug(
                "alconna activation index rebuilt: safe={}, unknown={}",
                safe,
                unknown,
            )

    def select(
        self,
        descriptor: HandlerDescriptor,
        context: ActivationContext,
        texts: tuple[str, ...],
    ) -> ActivationDecision:
        if not descriptor.alconna:
            return "unknown"
        saw_unknown = False
        for alconna in descriptor.alconna:
            decision = matcher_alconna_head_matches(alconna, texts, context)
            if decision == "match":
                return "match"
            if decision == "unknown":
                saw_unknown = True
        return "unknown" if saw_unknown else "miss"


@dataclass(slots=True)
class ActivationContext:
    event_type: str
    event: object | None = None
    plain_text: str = ""
    raw_text: str = ""
    to_me: bool = False
    has_url: bool = False
    has_image: bool = False
    is_command_like: bool = False
    route_modules: set[str] = field(default_factory=set)
    ai_route_modules: set[str] = field(default_factory=set)
    ai_route_heads: set[str] = field(default_factory=set)


@dataclass(slots=True)
class ActivationResult:
    selected: list[type[Matcher]]
    deterministic_selected: set[type[Matcher]] = field(default_factory=set)
    total_descriptors: int = 0
    candidate_count: int = 0


class HandlerActivationIndex:
    """In-memory matcher activation index.

    The index is intentionally fail-open: only known NoneBot rule misses are
    rejected before matcher task creation. Custom rules, incomplete command
    metadata, and Alconna shortcut misses stay selected so plugin compatibility
    wins over dispatch aggressiveness.
    """

    def __init__(self) -> None:
        self._by_priority: dict[int, list[HandlerDescriptor]] = {}
        self._matcher_map: dict[type[Matcher], HandlerDescriptor] = {}
        self._source_keys: set[tuple[int, tuple[int, ...]]] = set()
        self._alconna_index = AlconnaActivationIndex()
        self._compiled = False

    @property
    def compiled(self) -> bool:
        return self._compiled

    def rebuild(self, matchers: dict[int, list[type[Matcher]]]) -> None:
        self._by_priority.clear()
        self._matcher_map.clear()
        source_keys: set[tuple[int, tuple[int, ...]]] = set()
        for priority, priority_matchers in matchers.items():
            descriptors = [
                self._build_descriptor(matcher, priority)
                for matcher in priority_matchers
            ]
            self._by_priority[priority] = descriptors
            for descriptor in descriptors:
                self._matcher_map[descriptor.matcher] = descriptor
            source_keys.add(
                (int(priority), tuple(id(matcher) for matcher in priority_matchers))
            )
        self._source_keys = source_keys
        self._alconna_index.rebuild(self._matcher_map.values())
        self._compiled = True

    def ensure_fresh(self, matchers: dict[int, list[type[Matcher]]]) -> None:
        source_keys = {
            (int(priority), tuple(id(matcher) for matcher in items))
            for priority, items in matchers.items()
        }
        if not self._compiled or source_keys != self._source_keys:
            self.rebuild(matchers)

    def descriptors_for_priority(
        self,
        priority: int,
        priority_matchers: Iterable[type[Matcher]],
    ) -> list[HandlerDescriptor]:
        priority_matchers_list = list(priority_matchers)
        descriptors = self._by_priority.get(priority)
        if descriptors is not None and len(descriptors) == len(priority_matchers_list):
            return descriptors
        # Fallback for dynamic matcher list changes inside a priority bucket.
        return [
            self._matcher_map.get(matcher) or self._build_descriptor(matcher, priority)
            for matcher in priority_matchers_list
        ]

    def descriptor_for(self, matcher: type[Matcher]) -> HandlerDescriptor | None:
        return self._matcher_map.get(matcher)

    def select_priority(
        self,
        priority: int,
        priority_matchers: list[type[Matcher]],
        context: ActivationContext,
        budget: dict[str, int],
    ) -> ActivationResult:
        result = ActivationResult(selected=[], total_descriptors=len(priority_matchers))
        descriptors = [
            self._matcher_map.get(matcher) or self._build_descriptor(matcher, priority)
            for matcher in priority_matchers
        ]
        for descriptor in descriptors:
            decision = self._select_descriptor(descriptor, context)
            if decision == "miss":
                continue
            if decision == "deterministic":
                result.selected.append(descriptor.matcher)
                result.deterministic_selected.add(descriptor.matcher)
                continue
            if _is_throttleable_broad_passive(descriptor, context):
                if not _consume_broad_passive_budget(descriptor, budget):
                    continue
            result.selected.append(descriptor.matcher)
        result.candidate_count = len(result.selected)
        return result

    def _select_descriptor(
        self,
        descriptor: HandlerDescriptor,
        context: ActivationContext,
    ) -> Literal["select", "miss", "deterministic"]:
        if descriptor.temp:
            return "select"
        matcher_type = descriptor.matcher_type
        if matcher_type and matcher_type != context.event_type:
            return "miss"
        if context.event_type != "message":
            return "miss" if descriptor.command_like else "select"
        if descriptor.command_like:
            return self._select_command_descriptor(descriptor, context)
        rule_match = matcher_rule_matches_text(
            descriptor.rules,
            context.raw_text,
            context.plain_text,
            event=context.event,
            to_me=context.to_me,
        )
        if rule_match == "match":
            return "deterministic"
        if rule_match == "miss":
            return "miss"
        return "select"

    def _select_command_descriptor(
        self,
        descriptor: HandlerDescriptor,
        context: ActivationContext,
    ) -> Literal["select", "miss", "deterministic"]:
        texts = text_match_candidates(
            context.plain_text,
            context.raw_text,
            context.event,
        )
        if not texts:
            return "select"
        rule_match = matcher_rule_matches_text(
            descriptor.rules,
            context.raw_text,
            context.plain_text,
            event=context.event,
            to_me=context.to_me,
        )
        if rule_match == "miss":
            return "miss"
        command_matched = False
        if descriptor.alconna:
            alconna_match = self._alconna_index.select(descriptor, context, texts)
            if alconna_match == "match":
                command_matched = True
            elif alconna_match == "miss":
                return "miss"
            else:
                return "select"
        elif descriptor.commands:
            if any(
                matcher_command_matches(text, command)
                for text in texts
                for command in descriptor.commands
            ):
                command_matched = True
            else:
                shortcut_match = matcher_alconna_shortcut_matches_any(
                    descriptor.shortcuts,
                    texts,
                )
                if shortcut_match == "match":
                    command_matched = True
                else:
                    return "miss" if not descriptor.has_custom_rule else "select"
        else:
            shortcut_match = matcher_alconna_shortcut_matches_any(
                descriptor.shortcuts,
                texts,
            )
            if shortcut_match == "match":
                command_matched = True

        if (
            rule_match == "match"
            and not descriptor.has_custom_rule
            and descriptor.shortcuts is None
            and not descriptor.alconna
        ):
            command_matched = True
        if (
            not (descriptor.commands or descriptor.shortcuts is not None)
            and not command_matched
            and not descriptor.alconna
        ):
            return "select"

        if (
            context.ai_route_modules
            and descriptor.module not in context.ai_route_modules
        ):
            if not matcher_matches_ai_route_heads(descriptor, context.ai_route_heads):
                return "select"

        if command_matched:
            return "select" if descriptor.has_custom_rule else "deterministic"
        return "select"

    def _build_descriptor(
        self,
        matcher: type[Matcher],
        priority: int,
    ) -> HandlerDescriptor:
        rules = extract_matcher_rule_descriptors(matcher)
        command_like = any(rule.command_like for rule in rules)
        deterministic = any(rule.deterministic_text for rule in rules)
        if hasattr(matcher, "command"):
            command_like = True
        commands = extract_matcher_command_literals(matcher) or ()
        alconna_descriptors = extract_matcher_alconna_descriptors(matcher)
        shortcuts = extract_matcher_alconna_shortcuts(matcher)
        if alconna_descriptors:
            command_like = True
        if shortcuts is not None:
            command_like = True
        module = matcher_module_name(matcher)
        lane = classify_lane(
            matcher,
            module=module,
            command_like=command_like,
            deterministic_text=deterministic,
            shortcuts=shortcuts,
            commands=commands,
        )
        return HandlerDescriptor(
            matcher=matcher,
            module=module,
            matcher_type=getattr(matcher, "type", "") or "",
            priority=priority,
            lane=lane,
            temp=bool(getattr(matcher, "temp", False)),
            block=bool(getattr(matcher, "block", False)),
            command_like=command_like,
            deterministic_text=deterministic,
            has_custom_rule=matcher_has_custom_rule(matcher),
            commands=commands,
            shortcuts=shortcuts,
            alconna=alconna_descriptors,
            rules=rules,
        )


def matcher_module_name(matcher_cls: type[Matcher]) -> str:
    module = getattr(matcher_cls, "plugin_name", "") or ""
    if module:
        return module
    plugin = getattr(matcher_cls, "plugin", None)
    if not plugin:
        return ""
    return (getattr(plugin, "name", "") or "").strip()


def classify_lane(
    matcher_cls: type[Matcher],
    *,
    module: str,
    command_like: bool,
    deterministic_text: bool,
    shortcuts: tuple[str, ...] | None,
    commands: tuple[str, ...],
) -> ActivationLane:
    if getattr(matcher_cls, "temp", False):
        return "system"
    if command_like:
        if shortcuts:
            return "command_shortcut"
        has_regex_command = any(
            is_regex_like_command_literal(item) for item in commands
        )
        if deterministic_text or has_regex_command:
            return "command_regex"
        return "command_exact"
    module_l = (module or "").casefold()
    if any(hint in module_l for hint in PASSIVE_AI_HINTS):
        return "passive_ai"
    if any(hint in module_l for hint in PASSIVE_RENDER_HINTS):
        return "passive_render"
    if any(hint in module_l for hint in PASSIVE_HTTP_HINTS):
        return "passive_http"
    if matcher_has_custom_rule(matcher_cls):
        return "passive_light"
    if any(hint in module_l for hint in PASSIVE_DB_HINTS):
        return "passive_db"
    return "passive_light"


def matcher_has_custom_rule(matcher_cls: type[Matcher]) -> bool:
    rule = getattr(matcher_cls, "rule", None)
    checkers = getattr(rule, "checkers", ()) or ()
    for checker in checkers:
        call = getattr(checker, "call", None)
        if call is None:
            continue
        call_module = call.__class__.__module__
        if call_module.startswith("nonebot.rule") or call_module.startswith(
            "nonebot_plugin_alconna.rule"
        ):
            continue
        return True
    return False


def matcher_is_command_like(matcher_cls: type[Matcher]) -> bool:
    rules = extract_matcher_rule_descriptors(matcher_cls)
    if any(rule.command_like for rule in rules):
        return True
    if hasattr(matcher_cls, "command"):
        return True
    if extract_matcher_alconna_descriptors(matcher_cls):
        return True
    return extract_matcher_alconna_shortcuts(matcher_cls) is not None


def matcher_has_deterministic_text_rule(matcher_cls: type[Matcher]) -> bool:
    return any(
        rule.deterministic_text
        for rule in extract_matcher_rule_descriptors(matcher_cls)
    )


def classify_matcher_lane(
    matcher_cls: type[Matcher],
    *,
    ai_route_modules: set[str] | None = None,
) -> ActivationLane:
    module = matcher_module_name(matcher_cls)
    if ai_route_modules and any(
        module.casefold() == route_module.casefold()
        for route_module in ai_route_modules
    ):
        return "passive_ai"
    rules = extract_matcher_rule_descriptors(matcher_cls)
    command_like = any(rule.command_like for rule in rules)
    deterministic = any(rule.deterministic_text for rule in rules)
    if hasattr(matcher_cls, "command"):
        command_like = True
    commands = extract_matcher_command_literals(matcher_cls) or ()
    alconna_descriptors = extract_matcher_alconna_descriptors(matcher_cls)
    shortcuts = extract_matcher_alconna_shortcuts(matcher_cls)
    if alconna_descriptors:
        command_like = True
    if shortcuts is not None:
        command_like = True
    return classify_lane(
        matcher_cls,
        module=module,
        command_like=command_like,
        deterministic_text=deterministic,
        shortcuts=shortcuts,
        commands=commands,
    )


def extract_matcher_rule_descriptors(
    matcher_cls: type[Matcher],
) -> tuple[ActivationRuleDescriptor, ...]:
    descriptors: list[ActivationRuleDescriptor] = []
    rule = getattr(matcher_cls, "rule", None)
    checkers = getattr(rule, "checkers", ()) or ()
    for checker in checkers:
        call = getattr(checker, "call", None)
        if call is None:
            continue
        call_module = call.__class__.__module__
        call_name = call.__class__.__name__
        if call_module.startswith("nonebot.rule"):
            if call_name == "CommandRule":
                descriptors.append(
                    ActivationRuleDescriptor(
                        "command",
                        getattr(call, "cmds", ()),
                        command_like=True,
                    )
                )
            elif call_name == "ShellCommandRule":
                descriptors.append(
                    ActivationRuleDescriptor(
                        "shell_command",
                        getattr(call, "cmds", ()),
                        command_like=True,
                    )
                )
            elif call_name == "RegexRule":
                descriptors.append(
                    ActivationRuleDescriptor(
                        "regex",
                        getattr(call, "regex", ""),
                        flags=int(getattr(call, "flags", 0) or 0),
                        deterministic_text=True,
                        command_like=True,
                    )
                )
            elif call_name == "StartswithRule":
                descriptors.append(
                    ActivationRuleDescriptor(
                        "startswith",
                        normalize_rule_string_tuple(getattr(call, "msg", ())),
                        ignorecase=bool(getattr(call, "ignorecase", False)),
                        deterministic_text=True,
                        command_like=True,
                    )
                )
            elif call_name == "EndswithRule":
                descriptors.append(
                    ActivationRuleDescriptor(
                        "endswith",
                        normalize_rule_string_tuple(getattr(call, "msg", ())),
                        ignorecase=bool(getattr(call, "ignorecase", False)),
                        deterministic_text=True,
                        command_like=True,
                    )
                )
            elif call_name == "FullmatchRule":
                descriptors.append(
                    ActivationRuleDescriptor(
                        "fullmatch",
                        normalize_rule_string_tuple(getattr(call, "msg", ())),
                        ignorecase=bool(getattr(call, "ignorecase", False)),
                        deterministic_text=True,
                        command_like=True,
                    )
                )
            elif call_name == "KeywordsRule":
                descriptors.append(
                    ActivationRuleDescriptor(
                        "keywords",
                        normalize_rule_string_tuple(getattr(call, "keywords", ())),
                        deterministic_text=True,
                        command_like=True,
                    )
                )
            elif call_name == "IsTypeRule":
                descriptors.append(
                    ActivationRuleDescriptor("is_type", getattr(call, "types", ()))
                )
            elif call_name == "ToMeRule":
                descriptors.append(ActivationRuleDescriptor("to_me"))
            else:
                descriptors.append(ActivationRuleDescriptor("custom"))
        elif (
            call_module.startswith("nonebot_plugin_alconna.rule")
            and call_name == "AlconnaRule"
        ):
            descriptors.append(ActivationRuleDescriptor("alconna", command_like=True))
        else:
            descriptors.append(ActivationRuleDescriptor("custom"))
    return tuple(descriptors)


def normalize_rule_string_tuple(value: object) -> tuple[str, ...]:
    if isinstance(value, str):
        return (value,)
    if isinstance(value, list | tuple | set | frozenset):
        return tuple(str(item) for item in value if str(item))
    return ()


def text_match_candidates(
    plain_text: str,
    raw_text: str = "",
    event: object | None = None,
) -> tuple[str, ...]:
    """Return text variants visible to different matcher rule providers."""

    candidates: list[str] = []

    def add(text: object) -> None:
        if not isinstance(text, str):
            return
        normalized = text.strip()
        if normalized and normalized not in candidates:
            candidates.append(normalized)

        unescaped = _unescape_message_text(normalized)
        if unescaped and unescaped not in candidates:
            candidates.append(unescaped)

    add(plain_text)
    if event is not None:
        with contextlib.suppress(Exception):
            getter = getattr(event, "get_plaintext", None)
            if callable(getter):
                add(getter())
    add(raw_text)
    return tuple(candidates)


def _unescape_message_text(text: str) -> str:
    if not text:
        return ""
    unescaped = html.unescape(text)
    return unescaped.replace("\\/", "/").replace("\\u002F", "/").replace("\\u002f", "/")


def matcher_rule_matches_text(
    descriptors: tuple[ActivationRuleDescriptor, ...],
    raw_text: str,
    plain_text: str,
    *,
    event: object | None = None,
    to_me: bool = False,
) -> ActivationDecision:
    matched_any = False
    saw_deterministic = False
    saw_unknown = False
    message_text = raw_text or plain_text
    plain_candidates = text_match_candidates(plain_text, raw_text, event)

    for descriptor in descriptors:
        kind = descriptor.kind
        if kind in {"custom", "alconna"}:
            saw_unknown = True
            continue
        if kind in {"command", "shell_command"}:
            saw_deterministic = True
            commands: set[str] = set()
            collect_command_literals(descriptor.value, commands)
            normalized_commands = {
                normalized
                for item in commands
                if (normalized := normalize_command(item))
            }
            if any(
                matcher_command_matches(text, command)
                for text in plain_candidates
                for command in normalized_commands
            ):
                matched_any = True
            else:
                return "miss"
        elif kind in {"regex", "regex_fullmatch"}:
            saw_deterministic = True
            pattern = str(descriptor.value or "")
            if not pattern:
                continue
            try:
                matched = (
                    re.fullmatch(pattern, message_text, descriptor.flags)
                    if kind == "regex_fullmatch"
                    else re.search(pattern, message_text, descriptor.flags)
                )
                if matched:
                    matched_any = True
                else:
                    return "miss"
            except re.error:
                return "unknown"
        elif kind == "startswith":
            saw_deterministic = True
            values = descriptor.value if isinstance(descriptor.value, tuple) else ()
            candidates = (
                tuple(item.casefold() for item in values)
                if descriptor.ignorecase
                else values
            )
            texts = (
                tuple(item.casefold() for item in plain_candidates)
                if descriptor.ignorecase
                else plain_candidates
            )
            if any(
                text.startswith(item) for text in texts for item in candidates if item
            ):
                matched_any = True
            else:
                return "miss"
        elif kind == "endswith":
            saw_deterministic = True
            values = descriptor.value if isinstance(descriptor.value, tuple) else ()
            candidates = (
                tuple(item.casefold() for item in values)
                if descriptor.ignorecase
                else values
            )
            texts = (
                tuple(item.casefold() for item in plain_candidates)
                if descriptor.ignorecase
                else plain_candidates
            )
            if any(
                text.endswith(item) for text in texts for item in candidates if item
            ):
                matched_any = True
            else:
                return "miss"
        elif kind == "fullmatch":
            saw_deterministic = True
            values = descriptor.value if isinstance(descriptor.value, tuple) else ()
            candidates = (
                tuple(item.casefold() for item in values)
                if descriptor.ignorecase
                else values
            )
            texts = (
                tuple(item.casefold() for item in plain_candidates)
                if descriptor.ignorecase
                else plain_candidates
            )
            if any(text in candidates for text in texts):
                matched_any = True
            else:
                return "miss"
        elif kind == "keywords":
            saw_deterministic = True
            values = descriptor.value if isinstance(descriptor.value, tuple) else ()
            if any(
                item and item in text for text in plain_candidates for item in values
            ):
                matched_any = True
            else:
                return "miss"
        elif kind == "to_me":
            if not to_me:
                return "miss"
        elif kind == "is_type":
            if event is None:
                return "unknown"
            types = descriptor.value
            if isinstance(types, type):
                if not isinstance(event, types):
                    return "miss"
            elif isinstance(types, tuple) and types:
                if not isinstance(event, types):
                    return "miss"
    if matched_any:
        return "unknown" if saw_unknown else "match"
    if saw_unknown:
        return "unknown"
    if saw_deterministic:
        return "miss"
    return "unknown"


def extract_matcher_command_literals(
    matcher_cls: type[Matcher],
) -> tuple[str, ...] | None:
    commands: set[str] = set()
    collect_command_literals(getattr(matcher_cls, "command", None), commands)
    rule = getattr(matcher_cls, "rule", None)
    checkers = getattr(rule, "checkers", ()) or ()
    for checker in checkers:
        call = getattr(checker, "call", None)
        if call is None:
            continue
        for attr in ("cmds", "command", "commands", "cmd"):
            collect_command_literals(getattr(call, attr, None), commands)
    normalized_commands = {
        normalized for item in commands if (normalized := normalize_command(item))
    }
    normalized = tuple(sorted(normalized_commands))
    return normalized or None


def collect_command_literals(value: Any, target: set[str], depth: int = 0) -> None:
    if depth > 3 or value is None:
        return
    if isinstance(value, str):
        text = value.strip()
        if text:
            target.add(text)
        return
    if isinstance(value, weakref.ReferenceType):
        resolved = value()
        if resolved is not None and resolved is not value:
            collect_command_literals(resolved, target, depth + 1)
        return
    if isinstance(value, list | tuple | set | frozenset):
        if all(isinstance(item, str) for item in value):
            parts = tuple(str(item).strip() for item in value if str(item).strip())
            if parts:
                target.add(" ".join(parts))
                target.add("".join(parts))
            return
        for item in value:
            collect_command_literals(item, target, depth + 1)
        return
    if callable(value) and getattr(value, "__self__", None) is not None:
        with contextlib.suppress(TypeError, RuntimeError, ReferenceError):
            resolved = value()
            if resolved is not None and resolved is not value:
                collect_command_literals(resolved, target, depth + 1)
                return
    for attr in (
        "name",
        "path",
        "aliases",
        "header_display",
        "command",
        "commands",
        "cmd",
        "cmds",
    ):
        nested = getattr(value, attr, None)
        if nested is not None and nested is not value:
            collect_command_literals(nested, target, depth + 1)


def extract_matcher_alconna_shortcuts(
    matcher_cls: type[Matcher],
) -> tuple[str, ...] | None:
    shortcuts: set[str] = set()
    for attr in ("command", "_rule", "rule"):
        collect_alconna_shortcuts(getattr(matcher_cls, attr, None), shortcuts)
    rule = getattr(matcher_cls, "rule", None)
    checkers = getattr(rule, "checkers", ()) or ()
    for checker in checkers:
        call = getattr(checker, "call", None)
        if call is None:
            continue
        if call.__class__.__name__ != "AlconnaRule":
            continue
        command = resolve_maybe_weakref(
            getattr(call, "command", None) or getattr(call, "alconna", None)
        )
        collect_alconna_shortcuts(command, shortcuts)
    normalized_shortcuts = {
        normalized
        for item in shortcuts
        if item and (normalized := normalize_shortcut_pattern(item))
    }
    normalized = tuple(sorted(normalized_shortcuts))
    return normalized if normalized else None


def collect_alconna_shortcuts(value: Any, target: set[str], depth: int = 0) -> None:
    if depth > 4 or value is None:
        return
    if isinstance(value, weakref.ReferenceType):
        resolved = value()
        if resolved is not None and resolved is not value:
            collect_alconna_shortcuts(resolved, target, depth + 1)
        return
    get_shortcuts = getattr(value, "get_shortcuts", None)
    if callable(get_shortcuts):
        with contextlib.suppress(Exception):
            raw_shortcuts = get_shortcuts()
            if isinstance(raw_shortcuts, list | tuple | set | frozenset):
                for shortcut in raw_shortcuts:
                    if isinstance(shortcut, str) and shortcut.strip():
                        target.add(shortcut.strip())
    elif callable(value):
        with contextlib.suppress(Exception):
            resolved = value()
            if resolved is not None and resolved is not value:
                collect_alconna_shortcuts(resolved, target, depth + 1)
                return
    formatter = getattr(value, "formatter", None)
    data = getattr(formatter, "data", None)
    if isinstance(data, dict):
        for trace in data.values():
            trace_shortcuts = getattr(trace, "shortcuts", None)
            if not isinstance(trace_shortcuts, dict):
                continue
            for shortcut in trace_shortcuts:
                if isinstance(shortcut, str) and shortcut.strip():
                    target.add(shortcut.strip())
    for attr in ("shortcut", "shortcuts"):
        shortcuts = getattr(value, attr, None)
        if isinstance(shortcuts, dict):
            for key in shortcuts:
                if isinstance(key, str) and key.strip():
                    target.add(key.strip())
        elif isinstance(shortcuts, list | tuple | set | frozenset):
            for item in shortcuts:
                if isinstance(item, str) and item.strip():
                    target.add(item.strip())
    with contextlib.suppress(Exception):
        from arclet.alconna import command_manager

        for shortcut_map in command_manager.get_shortcut(value).values():  # type: ignore[arg-type]
            origin_key = getattr(shortcut_map, "origin_key", None)
            if isinstance(origin_key, str) and origin_key.strip():
                target.add(origin_key.strip())
    for attr in ("command", "commands", "base", "formatter", "source"):
        nested = getattr(value, attr, None)
        if nested is not None and nested is not value:
            collect_alconna_shortcuts(nested, target, depth + 1)


def extract_matcher_alconna_descriptors(
    matcher_cls: type[Matcher],
) -> tuple[AlconnaDescriptor, ...]:
    descriptors: list[AlconnaDescriptor] = []
    rule = getattr(matcher_cls, "rule", None)
    checkers = getattr(rule, "checkers", ()) or ()
    for checker in checkers:
        call = getattr(checker, "call", None)
        if call is None:
            continue
        if call.__class__.__name__ != "AlconnaRule":
            continue
        if not call.__class__.__module__.startswith("nonebot_plugin_alconna.rule"):
            continue
        command = resolve_maybe_weakref(
            getattr(call, "command", None) or getattr(call, "alconna", None)
        )
        descriptor = _build_alconna_descriptor(call, command)
        if descriptor is not None:
            descriptors.append(descriptor)
    return tuple(descriptors)


def _build_alconna_descriptor(call: Any, command: Any) -> AlconnaDescriptor | None:
    if command is None:
        return None
    command_text = str(getattr(command, "command", "") or "").strip()
    aliases = tuple(
        str(item).strip()
        for item in getattr(command, "aliases", ()) or ()
        if str(item).strip() and str(item).strip() != command_text
    )
    prefixes = tuple(
        str(item)
        for item in getattr(command, "prefixes", ()) or ()
        if isinstance(item, str)
    )
    meta = getattr(command, "meta", None)
    shortcuts = _extract_alconna_shortcut_descriptors(command)
    return AlconnaDescriptor(
        command=command_text,
        aliases=aliases,
        prefixes=prefixes,
        shortcuts=shortcuts,
        compact=bool(getattr(meta, "compact", False)),
        skip_for_unmatch=bool(getattr(call, "skip", True)),
        before_rule_count=_rule_checker_count(getattr(call, "before_rules", None)),
        after_rule_count=_rule_checker_count(getattr(call, "after_rules", None)),
        before_rule_known_safe=_alconna_rule_is_known_safe(
            getattr(call, "before_rules", None)
        ),
        after_rule_known_safe=_alconna_rule_is_known_safe(
            getattr(call, "after_rules", None)
        ),
        input_rewrite_extensions=_extract_alconna_input_rewrite_extensions(call),
    )


def _rule_checker_count(rule: Any) -> int:
    checkers = getattr(rule, "checkers", ()) or ()
    with contextlib.suppress(TypeError):
        return len(checkers)
    return 1


def _alconna_rule_is_known_safe(rule: Any) -> bool:
    """Whether Alconna before/after Rule can be reasoned about statically.

    We still do not execute these rules here. A rule is considered safe only
    when every checker is an official NoneBot rule whose negative result can be
    reproduced by the selector. Custom rules stay fail-open.
    """

    checkers = getattr(rule, "checkers", ()) or ()
    for checker in checkers:
        call = getattr(checker, "call", None)
        if call is None:
            return False
        call_module = call.__class__.__module__
        call_name = call.__class__.__name__
        if not call_module.startswith("nonebot.rule"):
            return False
        if call_name not in KNOWN_SAFE_RULE_NAMES:
            return False
    return True


def _extract_alconna_shortcut_descriptors(
    command: Any,
) -> tuple[AlconnaShortcutDescriptor, ...]:
    shortcuts: list[AlconnaShortcutDescriptor] = []
    with contextlib.suppress(Exception):
        from arclet.alconna import command_manager

        raw_shortcuts = command_manager.get_shortcut(command)  # type: ignore[arg-type]
        if isinstance(raw_shortcuts, dict):
            for key, args in raw_shortcuts.items():
                pattern = str(key or "").strip()
                if not pattern:
                    continue
                shortcuts.append(
                    AlconnaShortcutDescriptor(
                        pattern=pattern,
                        fuzzy=bool(getattr(args, "fuzzy", False)),
                        flags=int(getattr(args, "flags", 0) or 0),
                    )
                )
    if shortcuts:
        return tuple(shortcuts)
    fallback: set[str] = set()
    collect_alconna_shortcuts(command, fallback)
    return tuple(
        AlconnaShortcutDescriptor(pattern=item)
        for item in sorted(fallback)
        if item.strip()
    )


def _extract_alconna_input_rewrite_extensions(call: Any) -> tuple[str, ...]:
    executor = getattr(call, "executor", None)
    if executor is None:
        return ()
    result: list[str] = []
    for attr in ("extensions", "_extensions", "exts", "context"):
        extensions = getattr(executor, attr, None)
        if not isinstance(extensions, list | tuple | set | frozenset):
            continue
        for extension in extensions:
            overrides = getattr(extension.__class__, "_overrides", None)
            if not isinstance(overrides, dict):
                overrides = getattr(extension, "_overrides", None)
            if not isinstance(overrides, dict):
                continue
            if not (
                bool(overrides.get("message_provider"))
                or bool(overrides.get("receive_wrapper"))
            ):
                continue
            name = extension.__class__.__name__
            if name not in result:
                result.append(name)
    return tuple(result)


def _alconna_can_prefilter(alconna: AlconnaDescriptor) -> bool:
    if not (alconna.command or alconna.aliases or alconna.shortcuts):
        return False
    if not (alconna.before_rule_known_safe and alconna.after_rule_known_safe):
        return False
    if any(name != "ReplyMergeExtension" for name in alconna.input_rewrite_extensions):
        return False
    return True


def matcher_alconna_head_matches(
    alconna: AlconnaDescriptor,
    texts: Iterable[str],
    context: ActivationContext,
) -> ActivationDecision:
    if not _alconna_can_prefilter(alconna):
        return "unknown"
    if alconna.has_reply_merge_extension and _event_has_reply(context.event):
        return "unknown"

    candidates = tuple(text.strip() for text in texts if text and text.strip())
    if not candidates:
        return "unknown"

    saw_unknown = False
    command_heads = (alconna.command, *alconna.aliases)
    for text in candidates:
        for command in command_heads:
            if not command:
                continue
            decision = _alconna_command_head_matches(
                text,
                command,
                alconna.prefixes,
                compact=alconna.compact,
            )
            if decision == "match":
                return "match"
            if decision == "unknown":
                saw_unknown = True
        for shortcut in alconna.shortcuts:
            decision = _alconna_shortcut_matches(text, shortcut)
            if decision == "match":
                return "match"
            if decision == "unknown":
                saw_unknown = True
    return "unknown" if saw_unknown else "miss"


def _alconna_command_head_matches(
    text: str,
    command: str,
    prefixes: tuple[str, ...],
    *,
    compact: bool,
) -> ActivationDecision:
    normalized = command.strip()
    if not normalized:
        return "unknown"
    if normalized.startswith("re:"):
        pattern = normalized.removeprefix("re:").strip()
        if not pattern:
            return "unknown"
        for prefix in prefixes or ("",):
            try:
                if re.match(rf"^{re.escape(prefix)}(?:{pattern})", text):
                    return "match"
            except re.error:
                return "unknown"
        return "miss"
    for prefix in prefixes or ("",):
        head = f"{prefix}{normalized}"
        if _alconna_literal_head_matches(text, head, compact=compact):
            return "match"
    return "miss"


def _alconna_literal_head_matches(text: str, head: str, *, compact: bool) -> bool:
    if not text or not head:
        return False
    if text == head:
        return True
    if not text.startswith(head):
        return False
    if len(text) == len(head):
        return True
    rest = text[len(head) :]
    if rest and rest[0].isspace():
        return True
    return bool(compact or not head[-1].isascii())


def _alconna_shortcut_matches(
    text: str,
    shortcut: AlconnaShortcutDescriptor,
) -> ActivationDecision:
    pattern = shortcut.pattern.strip()
    if not pattern:
        return "unknown"
    normalized = normalize_shortcut_pattern(pattern)
    placeholder_match = _placeholder_shortcut_decision(text, normalized)
    if placeholder_match == "match":
        return "match"
    if placeholder_match == "unknown":
        return "unknown"
    if not is_regex_like_shortcut(normalized) and matcher_command_matches(
        text,
        normalized,
    ):
        return "match"
    try:
        if shortcut.fuzzy:
            return "match" if re.match(f"^{pattern}", text, shortcut.flags) else "miss"
        return "match" if re.fullmatch(pattern, text, shortcut.flags) else "miss"
    except re.error:
        return "unknown"


def _event_has_reply(event: object | None) -> bool:
    if event is None:
        return False
    with contextlib.suppress(Exception):
        message = event.get_message()  # type: ignore[attr-defined]
        for segment in message:
            segment_type = getattr(segment, "type", None)
            if segment_type == "reply":
                return True
            if isinstance(segment, dict) and segment.get("type") == "reply":
                return True
    for attr in ("reply", "reply_message", "source"):
        with contextlib.suppress(Exception):
            if getattr(event, attr, None) is not None:
                return True
    raw_text = ""
    with contextlib.suppress(Exception):
        raw_text = str(event.get_message())  # type: ignore[attr-defined]
    lowered = raw_text.casefold()
    return any(
        marker in lowered
        for marker in ("[cq:reply", "type=reply", '"reply"', "'reply'")
    )


def resolve_maybe_weakref(value: Any) -> Any:
    if isinstance(value, weakref.ReferenceType):
        resolved = value()
        return resolved if resolved is not None else value
    if callable(value) and getattr(value, "__self__", None) is not None:
        with contextlib.suppress(TypeError, RuntimeError, ReferenceError):
            resolved = value()
            if resolved is not None and resolved is not value:
                return resolved
    return value


def normalize_command(command: str) -> str:
    text = command.strip()
    if not text:
        return ""
    text = re.sub(r"^(?:\s*(?:\[[^\]]*]|\<[^>]*>))+\s*", "", text)
    cut_points = [idx for idx in (text.find("["), text.find("<")) if idx >= 0]
    if cut_points:
        text = text[: min(cut_points)]
    text = re.sub(r"\s+", " ", text).strip()
    text = re.sub(r"(?:\s+[?*]+|[?*]+)$", "", text).strip()
    return text


def matcher_command_matches(text: str, command: str) -> bool:
    normalized = command.strip()
    if not normalized:
        return False
    if normalized.startswith("re:"):
        pattern = normalized.removeprefix("re:").strip()
        if not pattern:
            return False
        try:
            return re.search(pattern, text) is not None
        except re.error:
            return False
    if command_matches(text, normalized):
        return True
    return text.startswith(normalized) and not normalized[-1].isascii()


def command_matches(text: str, command: str) -> bool:
    if not text or not command:
        return False
    if text == command:
        return True
    if text.startswith(command):
        if len(text) == len(command):
            return True
        return text[len(command)].isspace()
    return False


def is_regex_like_command_literal(command: str) -> bool:
    text = command.strip()
    if not text:
        return False
    if text.startswith("re:"):
        return True
    return any(token in text for token in ("\\", "(", ")", "[", "]", "|", "^", "$"))


def normalize_shortcut_pattern(pattern: str) -> str:
    text = str(pattern or "").strip()
    if not text:
        return ""
    text = re.sub(r"^\[(?:[^\]]*)\]\s*", "", text)
    text = re.sub(r"\s*\.\.\.args?$", "", text).strip()
    text = re.sub(r"\s+", " ", text)
    text = re.sub(r"\s*\.\.\.$", "", text).strip()
    text = re.sub(r"^\^", "", text)
    return text


def matcher_alconna_shortcut_matches(
    shortcuts: tuple[str, ...] | None,
    text: str,
) -> ActivationDecision:
    if shortcuts is None:
        return "unknown"
    for shortcut in shortcuts:
        if shortcut_matches_text(text, shortcut):
            return "match"
    return "unknown"


def matcher_alconna_shortcut_matches_any(
    shortcuts: tuple[str, ...] | None,
    texts: Iterable[str],
) -> ActivationDecision:
    if shortcuts is None:
        return "unknown"
    for text in texts:
        if matcher_alconna_shortcut_matches(shortcuts, text) == "match":
            return "match"
    return "unknown"


def shortcut_matches_text(text: str, shortcut: str) -> bool:
    pattern = normalize_shortcut_pattern(shortcut)
    if not pattern:
        return False
    if placeholder_shortcut_matches(text, pattern):
        return True
    if is_regex_like_shortcut(pattern):
        try:
            return re.match(pattern, text) is not None
        except re.error:
            return False
    return matcher_command_matches(text, pattern)


def placeholder_shortcut_matches(text: str, pattern: str) -> bool:
    return _placeholder_shortcut_decision(text, pattern) == "match"


def _placeholder_shortcut_decision(text: str, pattern: str) -> ActivationDecision:
    if "{" not in pattern or "}" not in pattern:
        return "miss"
    pieces: list[str] = []
    last = 0
    for match in re.finditer(r"\{[^{}]+\}", pattern):
        pieces.append(re.escape(pattern[last : match.start()]))
        pieces.append(r"\S+")
        last = match.end()
    if not pieces:
        return "miss"
    pieces.append(re.escape(pattern[last:]))
    try:
        return (
            "match"
            if re.match(rf"^{''.join(pieces)}(?:\s|$)", text) is not None
            else "miss"
        )
    except re.error:
        return "unknown"


def is_regex_like_shortcut(pattern: str) -> bool:
    return any(token in pattern for token in ("\\", "(", ")", "[", "]", "|", "^", "$"))


def matcher_matches_ai_route_heads(
    descriptor: HandlerDescriptor,
    ai_route_heads: set[str],
) -> bool:
    if not ai_route_heads:
        return False
    for command in descriptor.commands:
        normalized_command = command.strip().casefold()
        if not normalized_command:
            continue
        for head in ai_route_heads:
            if not head:
                continue
            if matcher_command_matches(head, normalized_command) or command_matches(
                normalized_command,
                head,
            ):
                return True
    for shortcut in descriptor.shortcuts or ():
        for head in ai_route_heads:
            if head and shortcut_matches_text(head, shortcut):
                return True
    return False


def _is_throttleable_broad_passive(
    descriptor: HandlerDescriptor,
    context: ActivationContext,
) -> bool:
    """Only broad, no-rule passive message matchers may be budget-throttled."""

    if context.event_type != "message":
        return False
    if descriptor.temp or descriptor.lane == "system":
        return False
    if not descriptor.lane.startswith("passive_"):
        return False
    if descriptor.command_like or descriptor.deterministic_text:
        return False
    if descriptor.has_custom_rule or descriptor.rules:
        return False
    if descriptor.lane == "passive_http" and (
        context.has_url or _looks_like_rich_message(context.raw_text)
    ):
        return False
    return True


def _looks_like_rich_message(text: str) -> bool:
    lowered = (text or "").casefold()
    return any(
        marker in lowered
        for marker in (
            "[cq:json",
            "[json:",
            "[cq:xml",
            "[xml:",
            "qqdocurl",
            "jumpurl",
            "miniapp",
            "com.tencent",
        )
    )


def _consume_broad_passive_budget(
    descriptor: HandlerDescriptor,
    budget: dict[str, int],
) -> bool:
    lane = descriptor.lane
    if lane not in budget:
        return True
    if budget[lane] <= 0:
        return False
    budget[lane] -= 1
    return True


PASSIVE_DB_HINTS = (
    "word_bank",
    "black_word",
    "history",
    "statistics",
    "sign",
    "gold",
    "redbag",
    "mute",
    "group",
    "user",
    "admin",
    "ban",
    "limit",
    "check",
)
PASSIVE_HTTP_HINTS = (
    "http",
    "translate",
    "bilibili",
    "music",
    "comment",
    "nbnhhsh",
    "quote",
    "search",
    "jitang",
    "poetry",
    "anime",
    "cover",
)
PASSIVE_AI_HINTS = (
    "chatinter",
    "dialogue",
    "ai",
    "llm",
    "fudu",
    "bym_ai",
)
PASSIVE_RENDER_HINTS = (
    "render",
    "image",
    "meme",
    "memes",
    "word_cloud",
    "wordcloud",
    "pic",
    "picture",
    "coser",
    "luxun",
)


__all__ = [
    "ActivationContext",
    "ActivationDecision",
    "ActivationResult",
    "ActivationRuleDescriptor",
    "HandlerActivationIndex",
    "HandlerDescriptor",
    "classify_matcher_lane",
    "command_matches",
    "extract_matcher_alconna_shortcuts",
    "extract_matcher_command_literals",
    "extract_matcher_rule_descriptors",
    "matcher_command_matches",
    "matcher_has_custom_rule",
    "matcher_has_deterministic_text_rule",
    "matcher_is_command_like",
    "matcher_rule_matches_text",
]