"""规则服务实现。"""

from __future__ import annotations

import hashlib
import logging
import time
from fastapi_common.fastapi_common_sqlalchemy.database import GetAsyncSession
from fastapi_common.fastapi_common_web.domain.responses import StatusCodeEnum
from fastapi_common.fastapi_common_web.exception.LeauditException import LeauditException
from fastapi_common.fastapi_common_storage.oss_path_utils import OssPathUtils
from sqlalchemy import text

from fastapi_modules.fastapi_leaudit.domian.vo.ruleVo import (
    RuleBindingVO,
    RuleContentVO,
    RuleSetVO,
    RuleValidationVO,
    RuleVersionVO,
)
from fastapi_modules.fastapi_leaudit.leaudit_bridge.ruleValidator import RuleValidator
from fastapi_modules.fastapi_leaudit.services import IOssService, IRuleService
from fastapi_modules.fastapi_leaudit.services.impl.ossServiceImpl import OssServiceImpl
from fastapi_modules.fastapi_leaudit.services.impl.ruleGroupSupport import sync_doc_type_bindings_from_group
from fastapi_modules.fastapi_leaudit.services.impl.ruleTenantScope import normalize_scoped_tenant_code, pick_effective_scoped_row
from fastapi_modules.fastapi_leaudit.services.impl.ssoUserCompat import SsoUserCompat
from fastapi_modules.fastapi_leaudit.services.impl.tenantResolver import TenantResolver


class RuleServiceImpl(IRuleService):
    """规则服务实现。"""

    _GLOBAL_LIST_SETS_CACHE: tuple[float, list[RuleSetVO]] | None = None
    _GLOBAL_RULE_COUNT_CACHE: dict[int, tuple[float, int]] = {}

    def __init__(
        self,
        OssService: IOssService | None = None,
        Validator: RuleValidator | None = None,
    ) -> None:
        self.OssService = OssService or OssServiceImpl()
        self.Validator = Validator or RuleValidator()
        self.TenantResolver = TenantResolver()
        self._list_sets_cache = self.__class__._GLOBAL_LIST_SETS_CACHE
        self._rule_count_cache = self.__class__._GLOBAL_RULE_COUNT_CACHE
        self._entry_module_tenant_table_exists_cache: bool | None = None
        self._column_exists_cache: dict[str, bool] = {}

    @classmethod
    def _set_list_sets_cache(cls, value: tuple[float, list[RuleSetVO]] | None) -> None:
        cls._GLOBAL_LIST_SETS_CACHE = value

    def InvalidateCaches(self, version_ids: list[int] | None = None) -> None:
        """清理规则集及规则数量缓存。"""
        self.__class__._set_list_sets_cache(None)
        if version_ids is None:
            self._rule_count_cache.clear()
            return
        for version_id in {int(item) for item in version_ids if item is not None}:
            self._rule_count_cache.pop(version_id, None)

    async def _resolve_unique_child_group_id(self, Session, DocTypeId: int) -> int | None:
        """仅当文档类型唯一对应一个二级分组时，返回该分组ID。"""
        row = (
            await Session.execute(
                text(
                    """
                    SELECT CASE WHEN COUNT(*) = 1 THEN MIN(id) END AS group_id
                    FROM leaudit_evaluation_point_groups
                    WHERE document_type_id = :doc_type_id
                      AND deleted_at IS NULL
                      AND COALESCE(pid, 0) <> 0
                    """
                ),
                {"doc_type_id": DocTypeId},
            )
        ).mappings().first()
        return int(row["group_id"]) if row and row.get("group_id") is not None else None

    async def _count_child_groups(self, Session, DocTypeId: int) -> int:
        """统计文档类型下已启用的二级分组数量，用于明确报错。"""
        row = (
            await Session.execute(
                text(
                    """
                    SELECT COUNT(*) AS total
                    FROM leaudit_evaluation_point_groups
                    WHERE document_type_id = :doc_type_id
                      AND deleted_at IS NULL
                      AND COALESCE(pid, 0) <> 0
                    """
                ),
                {"doc_type_id": DocTypeId},
            )
        ).mappings().first()
        return int(row["total"] or 0) if row else 0

    async def ListSets(self, CurrentUserId: int | None = None) -> list[RuleSetVO]:
        """列出所有规则集。"""
        now = time.monotonic()
        cached = self.__class__._GLOBAL_LIST_SETS_CACHE
        if CurrentUserId is None and cached and now - cached[0] <= 30:
            return cached[1]

        async with GetAsyncSession() as Session:
            current_user = await self._get_current_user_context(Session, CurrentUserId)
            if await self._column_exists(Session, "leaudit_rule_sets", "tenant_code"):
                rows = await self._load_scoped_rule_set_rows(Session, current_user=current_user)
            else:
                Result = await Session.execute(
                    text(
                        """
                        SELECT
                            rs.id,
                            rs.rule_type,
                            rs.rule_name,
                            rs.domain_type,
                            rs.current_version_id,
                            current_rv.id AS usable_current_version_id,
                            fallback_rv.id AS fallback_version_id,
                            CASE
                                WHEN current_rv.id IS NOT NULL OR fallback_rv.id IS NOT NULL THEN TRUE
                                ELSE FALSE
                            END AS has_usable_version,
                            rs.status
                        FROM leaudit_rule_sets rs
                        LEFT JOIN leaudit_rule_versions current_rv
                          ON current_rv.id = rs.current_version_id
                         AND current_rv.status = 'published'
                        LEFT JOIN LATERAL (
                            SELECT rv.id
                            FROM leaudit_rule_versions rv
                            WHERE rv.rule_set_id = rs.id
                              AND rv.status = 'published'
                              AND (rs.current_version_id IS NULL OR rv.id <> rs.current_version_id)
                            ORDER BY rv.version_seq DESC, rv.id DESC
                            LIMIT 1
                        ) fallback_rv ON TRUE
                        WHERE rs.deleted_at IS NULL
                        ORDER BY rs.id DESC
                        """
                    )
                )
                rows = Result.mappings().all()

        usable_counts: dict[int, int] = {}
        for row in rows:
            usable_version_id = row["usable_current_version_id"] or row["fallback_version_id"]
            if usable_version_id is not None and int(usable_version_id) not in usable_counts:
                usable_counts[int(usable_version_id)] = await self._GetRuleCountByVersionId(int(usable_version_id))

        items = [
            RuleSetVO(
                id=int(Row["id"]),
                ruleType=Row["rule_type"],
                ruleName=Row["rule_name"],
                effectiveTenantCode=str(Row.get("tenant_code") or "").strip() or None,
                effectiveScopeType="PUBLIC"
                if str(Row.get("tenant_code") or "").strip().upper() == "PUBLIC"
                else "PROVINCIAL"
                if str(Row.get("tenant_code") or "").strip().upper() in {"", "PROVINCIAL"}
                else "TENANT",
                isInherited=bool(current_user)
                and not bool(current_user.get("is_global"))
                and bool(str(current_user.get("tenant_code") or "").strip())
                and str(Row.get("tenant_code") or "").strip().upper()
                not in {"", str(current_user.get("tenant_code") or "").strip().upper()},
                sourceRuleSetId=int(Row["source_rule_set_id"]) if Row.get("source_rule_set_id") is not None else None,
                domainType=Row["domain_type"],
                currentVersionId=Row["current_version_id"],
                fallbackVersionId=Row["fallback_version_id"],
                hasUsableVersion=bool(Row["has_usable_version"]),
                usableRuleCount=usable_counts.get(int(Row["usable_current_version_id"] or Row["fallback_version_id"]), 0)
                if (Row["usable_current_version_id"] or Row["fallback_version_id"]) is not None
                else 0,
                status=Row["status"],
            )
            for Row in rows
        ]
        cache_value = (time.monotonic(), items)
        if CurrentUserId is None:
            self.__class__._set_list_sets_cache(cache_value)
            self._list_sets_cache = cache_value
        return items

    async def _GetRuleCountByVersionId(self, VersionId: int) -> int:
        """读取指定可用规则版本的规则数。"""
        cached = self._rule_count_cache.get(VersionId)
        now = time.monotonic()
        if cached and now - cached[0] <= 120:
            return cached[1]

        async with GetAsyncSession() as Session:
            Result = await Session.execute(
                text(
                    """
                    SELECT oss_url
                    FROM leaudit_rule_versions
                    WHERE id = :version_id
                    LIMIT 1
                    """
                ),
                {"version_id": VersionId},
            )
            Row = Result.mappings().first()

        if not Row or not Row["oss_url"]:
            return 0

        try:
            yaml_text = (await self.OssService.DownloadBytes(Row["oss_url"])).decode("utf-8")
            validation = self.Validator.ValidateYaml(yaml_text)
            count = int(validation.ruleCount or 0)
            self._rule_count_cache[VersionId] = (time.monotonic(), count)
            return count
        except Exception:
            return 0

    async def GetVersions(self, RuleType: str, CurrentUserId: int | None = None) -> list[RuleVersionVO]:
        """获取规则集的所有版本。"""
        async with GetAsyncSession() as Session:
            current_user = await self._get_current_user_context(Session, CurrentUserId)
            if await self._column_exists(Session, "leaudit_rule_sets", "tenant_code"):
                rule_set_row = await self._get_effective_rule_set_row_by_type(Session, RuleType, current_user=current_user)
                if not rule_set_row:
                    return []
                Result = await Session.execute(
                    text(
                        """
                        SELECT
                            rv.id,
                            rv.rule_set_id,
                            rv.version_no,
                            rv.status,
                            rv.oss_url,
                            rv.change_note,
                            rv.published_at
                        FROM leaudit_rule_versions rv
                        WHERE rv.rule_set_id = :rule_set_id
                        ORDER BY rv.version_seq DESC, rv.id DESC
                        """
                    ),
                    {"rule_set_id": int(rule_set_row["id"])},
                )
            else:
                Result = await Session.execute(
                    text(
                        """
                        SELECT
                            rv.id,
                            rv.rule_set_id,
                            rv.version_no,
                            rv.status,
                            rv.oss_url,
                            rv.change_note,
                            rv.published_at
                        FROM leaudit_rule_versions rv
                        JOIN leaudit_rule_sets rs ON rs.id = rv.rule_set_id
                        WHERE rs.rule_type = :rule_type
                          AND rs.deleted_at IS NULL
                        ORDER BY rv.version_seq DESC, rv.id DESC
                        """
                    ),
                    {"rule_type": RuleType},
                )
            return [self._BuildRuleVersionVo(Row) for Row in Result.mappings().all()]

    async def GetContent(self, VersionId: int, CurrentUserId: int | None = None) -> RuleContentVO:
        """获取指定版本的规则正文。"""
        async with GetAsyncSession() as Session:
            current_user = await self._get_current_user_context(Session, CurrentUserId)
            Result = await Session.execute(
                text(
                    """
                    SELECT
                        rv.id,
                        rv.rule_set_id,
                        rv.version_no,
                        rv.oss_url,
                        rs.rule_type
                    FROM leaudit_rule_versions rv
                    JOIN leaudit_rule_sets rs ON rs.id = rv.rule_set_id
                    WHERE rv.id = :version_id
                    LIMIT 1
                    """
                ),
                {"version_id": VersionId},
            )
            Row = Result.mappings().first()
            if not Row:
                raise LeauditException(StatusCodeEnum.HTTP_404_NOT_FOUND, "规则版本不存在")
            if await self._column_exists(Session, "leaudit_rule_sets", "tenant_code"):
                effective_rule_set = await self._get_effective_rule_set_row_by_type(
                    Session,
                    str(Row["rule_type"] or ""),
                    current_user=current_user,
                )
                if effective_rule_set is None or int(effective_rule_set["id"]) != int(Row["rule_set_id"]):
                    raise LeauditException(StatusCodeEnum.HTTP_403_FORBIDDEN, "当前租户不可访问该规则版本")

        YamlText = (await self.OssService.DownloadBytes(Row["oss_url"])).decode("utf-8")
        return RuleContentVO(
            id=int(Row["id"]),
            ruleSetId=int(Row["rule_set_id"]),
            ruleType=Row["rule_type"],
            versionNo=Row["version_no"],
            yamlText=YamlText,
            ossUrl=Row["oss_url"],
        )

    async def Validate(self, RuleType: str, YamlText: str) -> RuleValidationVO:
        """校验规则 YAML。"""
        Validation = self.Validator.ValidateYaml(YamlText)
        if Validation.ruleType and Validation.ruleType != RuleType:
            Validation = RuleValidationVO(
                valid=False,
                ruleType=Validation.ruleType,
                ruleName=Validation.ruleName,
                versionNo=Validation.versionNo,
                ruleCount=Validation.ruleCount,
                extractCount=Validation.extractCount,
                errors=[f"路径规则类型 {RuleType} 与 YAML metadata.type_id {Validation.ruleType} 不一致"],
            )
            return Validation

        return RuleValidationVO(
            valid=Validation.valid,
            ruleType=Validation.ruleType,
            ruleName=Validation.ruleName,
            versionNo=Validation.versionNo,
            ruleCount=Validation.ruleCount,
            extractCount=Validation.extractCount,
            errors=Validation.errors or [],
        )

    def _pick_writable_rule_set_row(
        self,
        rows: list[dict[str, object]],
        *,
        current_user: dict[str, object] | None,
    ) -> dict[str, object] | None:
        if not rows:
            return None
        if current_user is None:
            return dict(rows[0])
        current_user_tenant = str(current_user.get("tenant_code") or "").strip().upper()
        if current_user.get("is_global") and not current_user_tenant:
            effective = pick_effective_scoped_row(rows, "PROVINCIAL")
            return dict(effective) if effective is not None else dict(rows[0])
        effective = pick_effective_scoped_row(rows, current_user_tenant or None)
        if effective is None:
            return None
        normalized = str(effective.get("tenant_code") or "").strip().upper() or "PROVINCIAL"
        if normalized == "PUBLIC":
            raise LeauditException(StatusCodeEnum.HTTP_403_FORBIDDEN, "当前租户不能直接修改公共规则资产")
        return dict(effective)

    def _assert_version_belongs_to_writable_rule_set(
        self,
        version_row: dict[str, object] | None,
        writable_rule_set: dict[str, object] | None,
    ) -> None:
        if version_row is None or writable_rule_set is None:
            raise LeauditException(StatusCodeEnum.HTTP_404_NOT_FOUND, "规则版本不存在或当前租户无可写规则集")
        if int(version_row["rule_set_id"]) != int(writable_rule_set["id"]):
            raise LeauditException(StatusCodeEnum.HTTP_403_FORBIDDEN, "当前租户不能发布或回滚其他租户的规则版本")

    def _assert_rollback_target(
        self,
        version_row: dict[str, object] | None,
        rule_set_row: dict[str, object] | None,
    ) -> None:
        if version_row is None or rule_set_row is None:
            raise LeauditException(StatusCodeEnum.HTTP_404_NOT_FOUND, "规则版本不存在或当前租户无可写规则集")
        current_version_id = rule_set_row.get("current_version_id")
        if current_version_id is not None and int(version_row["id"]) == int(current_version_id):
            raise LeauditException(StatusCodeEnum.HTTP_400_BAD_REQUEST, "当前版本已经生效，无需回滚")
        status = str(version_row.get("status") or "").strip().lower()
        if status in {"draft", "validated"}:
            raise LeauditException(StatusCodeEnum.HTTP_400_BAD_REQUEST, "草稿版本不能作为回滚目标，请先发布或选择历史版本")

    async def CreateVersion(
        self,
        RuleType: str,
        YamlText: str,
        ChangeNote: str | None = None,
        EditorUserId: int | None = None,
        CurrentUserId: int | None = None,
    ) -> RuleVersionVO:
        """创建规则版本。"""
        Validation = await self.Validate(RuleType, YamlText)
        if not Validation.valid:
            raise LeauditException(StatusCodeEnum.HTTP_400_BAD_REQUEST, "; ".join(Validation.errors))

        RulesFile = self.Validator.ParseValidated(YamlText)
        FileSha256 = hashlib.sha256(YamlText.encode("utf-8")).hexdigest()
        FileSize = len(YamlText.encode("utf-8"))
        DomainType = self._ResolveDomainType(RuleType)

        async with GetAsyncSession() as Session:
            current_user = await self._get_current_user_context(Session, CurrentUserId)
            use_tenant_scope = await self._column_exists(Session, "leaudit_rule_sets", "tenant_code")
            self._assert_rule_tenant_schema_ready_for_write(use_tenant_scope=use_tenant_scope, current_user=current_user)
            writable_tenant_code: str | None = None
            writable_scope_type = "PROVINCIAL"
            source_rule_set_id: int | None = None
            source_rule_set_binding_ids: list[int] = []

            if use_tenant_scope:
                candidate_rows = await self._load_rule_set_rows_by_type(Session, RuleType)
                writable_rule_set = self._pick_writable_rule_set_row(candidate_rows, current_user=current_user)
                RuleSet = writable_rule_set
                if current_user and not current_user.get("is_global"):
                    writable_tenant_code = str(current_user.get("tenant_code") or "").strip().upper() or None
                    writable_scope_type = "TENANT"
                elif RuleSet is not None:
                    writable_tenant_code = str(RuleSet.get("tenant_code") or "").strip().upper() or "PROVINCIAL"
                    writable_scope_type = "PROVINCIAL" if writable_tenant_code != "PUBLIC" else "PUBLIC"
                if RuleSet is not None:
                    normalized_rule_set_tenant = str(RuleSet.get("tenant_code") or "").strip().upper() or "PROVINCIAL"
                    if (
                        writable_scope_type == "TENANT"
                        and writable_tenant_code
                        and normalized_rule_set_tenant != writable_tenant_code
                    ):
                        source_rule_set_id = int(RuleSet["id"])
                        source_rule_set_binding_ids = await self._load_source_group_binding_ids(Session, source_rule_set_id)
                        RuleSet = None
            else:
                RuleSet = await self._GetRuleSetByType(Session, RuleType)
            if not RuleSet:
                if use_tenant_scope:
                    CreateResult = await Session.execute(
                        text(
                            """
                            INSERT INTO leaudit_rule_sets (
                                rule_type,
                                rule_name,
                                domain_type,
                                description,
                                region,
                                status,
                                is_builtin,
                                owner_user_id,
                                tenant_code,
                                scope_type,
                                source_rule_set_id,
                                tenant_name_snapshot
                            ) VALUES (
                                :rule_type,
                                :rule_name,
                                :domain_type,
                                :description,
                                :region,
                                'draft',
                                false,
                                :owner_user_id,
                                :tenant_code,
                                :scope_type,
                                :source_rule_set_id,
                                :tenant_name_snapshot
                            )
                            RETURNING id, rule_type, rule_name, domain_type, current_version_id, status, tenant_code, source_rule_set_id
                        """
                    ),
                        {
                            "rule_type": RuleType,
                            "rule_name": RulesFile.metadata.name,
                            "domain_type": DomainType,
                            "description": RulesFile.metadata.description,
                            "region": self._legacy_region_for_scope(writable_tenant_code, writable_scope_type),
                            "owner_user_id": EditorUserId,
                            "tenant_code": writable_tenant_code or "PROVINCIAL",
                            "scope_type": writable_scope_type,
                            "source_rule_set_id": source_rule_set_id,
                            "tenant_name_snapshot": str(current_user.get("tenant_name") or "") if current_user else None,
                        },
                    )
                else:
                    CreateResult = await Session.execute(
                        text(
                            """
                            INSERT INTO leaudit_rule_sets (
                                rule_type,
                                rule_name,
                                domain_type,
                                description,
                                status,
                                is_builtin,
                                owner_user_id
                            ) VALUES (
                                :rule_type,
                                :rule_name,
                                :domain_type,
                                :description,
                                'draft',
                                false,
                                :owner_user_id
                            )
                            RETURNING id, rule_type, rule_name, domain_type, current_version_id, status
                            """
                        ),
                        {
                            "rule_type": RuleType,
                            "rule_name": RulesFile.metadata.name,
                            "domain_type": DomainType,
                            "description": RulesFile.metadata.description,
                            "owner_user_id": EditorUserId,
                        },
                    )
                RuleSet = CreateResult.mappings().first()
                if use_tenant_scope and writable_scope_type == "TENANT" and source_rule_set_binding_ids:
                    await self._clone_source_group_bindings_for_tenant(
                        Session,
                        source_binding_ids=source_rule_set_binding_ids,
                        tenant_rule_set_id=int(RuleSet["id"]),
                        current_user=current_user,
                    )
            else:
                await Session.execute(
                    text(
                        """
                        UPDATE leaudit_rule_sets
                        SET
                            rule_name = :rule_name,
                            domain_type = :domain_type,
                            description = :description,
                            updated_at = now()
                        WHERE id = :rule_set_id
                        """
                    ),
                    {
                        "rule_name": RulesFile.metadata.name,
                        "domain_type": DomainType,
                        "description": RulesFile.metadata.description,
                        "rule_set_id": RuleSet["id"],
                    },
                )

            SeqResult = await Session.execute(
                text(
                    """
                    SELECT COALESCE(MAX(version_seq), 0) AS max_seq
                    FROM leaudit_rule_versions
                    WHERE rule_set_id = :rule_set_id
                    """
                ),
                {"rule_set_id": RuleSet["id"]},
            )
            NextSeq = int(SeqResult.mappings().first()["max_seq"]) + 1
            VersionNo = RulesFile.metadata.version or f"v{NextSeq}"
            source_version_id: int | None = None
            if source_rule_set_id is not None:
                source_version_id = await self._load_latest_version_id_by_rule_set(Session, source_rule_set_id)
            ExistingVersion = await Session.execute(
                text(
                    """
                    SELECT id
                    FROM leaudit_rule_versions
                    WHERE rule_set_id = :rule_set_id
                      AND version_no = :version_no
                    LIMIT 1
                    """
                ),
                {
                    "rule_set_id": RuleSet["id"],
                    "version_no": VersionNo,
                },
            )
            if ExistingVersion.mappings().first():
                raise LeauditException(StatusCodeEnum.HTTP_409_CONFLICT, f"版本号 {VersionNo} 已存在")

            ObjectKey = OssPathUtils.BuildRuleYamlKey(RuleType, VersionNo)
            OssUrl = await self.OssService.UploadText(
                ObjectKey=ObjectKey,
                Content=YamlText,
                ContentType="application/x-yaml; charset=utf-8",
            )

            VersionResult = await Session.execute(
                text(
                    """
                    INSERT INTO leaudit_rule_versions (
                        rule_set_id,
                        version_no,
                        version_seq,
                        status,
                        source_type,
                        dsl_format,
                        oss_url,
                        file_sha256,
                        file_size,
                        metadata_type_id,
                        metadata_name,
                        metadata_version,
                        change_note,
                        editor_user_id
                    ) VALUES (
                        :rule_set_id,
                        :version_no,
                        :version_seq,
                        'draft',
                        'oss_yaml',
                        'yaml',
                        :oss_url,
                        :file_sha256,
                        :file_size,
                        :metadata_type_id,
                        :metadata_name,
                        :metadata_version,
                        :change_note,
                        :editor_user_id
                    )
                    RETURNING id, rule_set_id, version_no, status, oss_url, change_note, published_at
                    """
                ),
                {
                    "rule_set_id": RuleSet["id"],
                    "version_no": VersionNo,
                    "version_seq": NextSeq,
                    "oss_url": OssUrl,
                    "file_sha256": FileSha256,
                    "file_size": FileSize,
                    "metadata_type_id": RulesFile.metadata.type_id,
                    "metadata_name": RulesFile.metadata.name,
                    "metadata_version": RulesFile.metadata.version,
                    "change_note": ChangeNote,
                    "editor_user_id": EditorUserId,
                },
            )
            version_row = VersionResult.mappings().first()
            if use_tenant_scope and await self._column_exists(Session, "leaudit_rule_versions", "tenant_code_snapshot"):
                snapshot = self._build_version_scope_snapshot(
                    writable_tenant_code=writable_tenant_code,
                    writable_scope_type=writable_scope_type,
                    rule_set_row=dict(RuleSet),
                    source_version_id=source_version_id,
                )
                await Session.execute(
                    text(
                        """
                        UPDATE leaudit_rule_versions
                        SET
                            tenant_code_snapshot = :tenant_code_snapshot,
                            scope_type_snapshot = :scope_type_snapshot,
                            source_version_id = :source_version_id
                        WHERE id = :version_id
                        """
                    ),
                    {
                        "tenant_code_snapshot": snapshot["tenant_code_snapshot"],
                        "scope_type_snapshot": snapshot["scope_type_snapshot"],
                        "source_version_id": snapshot["source_version_id"],
                        "version_id": int(version_row["id"]),
                    },
                )
            await Session.commit()
            return self._BuildRuleVersionVo(version_row)

    async def Publish(
        self,
        RuleType: str,
        VersionId: int,
        OperatorUserId: int | None = None,
        CurrentUserId: int | None = None,
    ) -> RuleVersionVO:
        """发布指定版本。"""
        return await self._SwitchVersion(
            RuleType=RuleType,
            VersionId=VersionId,
            TargetStatus="published",
            OperatorUserId=OperatorUserId,
            CurrentUserId=CurrentUserId,
        )

    async def Rollback(
        self,
        RuleType: str,
        VersionId: int,
        OperatorUserId: int | None = None,
        CurrentUserId: int | None = None,
    ) -> RuleVersionVO:
        """回滚到指定历史版本。"""
        return await self._SwitchVersion(
            RuleType=RuleType,
            VersionId=VersionId,
            TargetStatus="published",
            OperatorUserId=OperatorUserId,
            CurrentUserId=CurrentUserId,
            Rollback=True,
        )

    async def ListBindings(
        self,
        RuleType: str | None = None,
        Region: str | None = None,
        CurrentUserId: int | None = None,
    ) -> list[RuleBindingVO]:
        """列出规则类型绑定，只读取新分组绑定。"""
        async with GetAsyncSession() as Session:
            current_user = await self._get_current_user_context(Session, CurrentUserId)
            params: dict[str, object] = {}
            filters = ["rs.deleted_at IS NULL", "dt.deleted_at IS NULL", "child.deleted_at IS NULL", "rgb.deleted_at IS NULL"]
            filters.append(
                await self._entry_module_tenant_scope_sql(
                    session=Session,
                    entry_module_expr="COALESCE(child.entry_module_id, parent.entry_module_id, dt.entry_module_id)",
                    current_user=current_user,
                    params=params,
                    param_prefix="rule_binding_scope",
                )
            )
            if RuleType:
                filters.append("rs.rule_type = :rule_type")
                params["rule_type"] = RuleType
            where_clause = " AND ".join(filters)

            result = await Session.execute(
                text(
                    f"""
                    SELECT
                        rgb.id,
                        dt.id AS doc_type_id,
                        dt.code AS doc_type_code,
                        rgb.rule_set_id,
                        'explicit' AS binding_mode,
                        rgb.priority,
                        rgb.is_active,
                        rgb.note,
                        rs.rule_type,
                        rs.rule_name
                    FROM leaudit_rule_group_bindings rgb
                    JOIN leaudit_evaluation_point_groups child ON child.id = rgb.group_id
                    LEFT JOIN leaudit_evaluation_point_groups parent ON parent.id = child.pid AND parent.deleted_at IS NULL
                    JOIN leaudit_document_types dt ON dt.id = child.document_type_id
                    JOIN leaudit_rule_sets rs ON rs.id = rgb.rule_set_id
                    WHERE {where_clause}
                      AND COALESCE(child.pid, 0) <> 0
                    ORDER BY dt.id ASC, COALESCE(child.sort_order, 0) ASC, rgb.priority DESC, rgb.id DESC
                    """
                ),
                params,
            )
            rows = result.mappings().all()
            bindings: list[RuleBindingVO] = []
            seen_doc_type_rule_pairs: set[tuple[int, int]] = set()
            for row in rows:
                pair = (int(row["doc_type_id"]), int(row["rule_set_id"]))
                if pair in seen_doc_type_rule_pairs:
                    continue
                seen_doc_type_rule_pairs.add(pair)
                bindings.append(
                    RuleBindingVO(
                        id=int(row["id"]),
                        docTypeId=int(row["doc_type_id"]),
                        docTypeCode=row["doc_type_code"],
                        ruleSetId=int(row["rule_set_id"]),
                        ruleType=row["rule_type"],
                        ruleName=row["rule_name"],
                        bindingMode=row["binding_mode"],
                        priority=int(row["priority"]),
                        isActive=bool(row["is_active"]),
                        note=row["note"],
                    )
                )
            return bindings

    async def CreateBinding(
        self,
        DocTypeId: int,
        RuleSetId: int,
        Region: str = "default",
        BindingMode: str = "explicit",
        Priority: int = 0,
        DocTypeCode: str | None = None,
        Note: str | None = None,
        CurrentUserId: int | None = None,
    ) -> RuleBindingVO:
        """创建规则类型绑定；若文档类型唯一对应一个二级分组，则优先写入新分组绑定。"""
        async with GetAsyncSession() as Session:
            current_user = await self._get_current_user_context(Session, CurrentUserId)
            RuleSet = await Session.execute(
                text("SELECT id, rule_type, rule_name FROM leaudit_rule_sets WHERE id = :rid AND deleted_at IS NULL LIMIT 1"),
                {"rid": RuleSetId},
            )
            RsRow = RuleSet.mappings().first()
            if not RsRow:
                raise LeauditException(StatusCodeEnum.HTTP_404_NOT_FOUND, "规则集不存在")

            await self._assert_document_type_access(Session, DocTypeId, current_user)
            GroupId = await self._resolve_unique_accessible_child_group_id(Session, DocTypeId, current_user)
            if GroupId is not None:
                ExistingGroupBinding = await Session.execute(
                    text(
                        """
                        SELECT id
                        FROM leaudit_rule_group_bindings
                        WHERE group_id = :group_id
                          AND rule_set_id = :rule_set_id
                          AND deleted_at IS NULL
                        LIMIT 1
                        """
                    ),
                    {"group_id": GroupId, "rule_set_id": RuleSetId},
                )
                if ExistingGroupBinding.mappings().first():
                    raise LeauditException(StatusCodeEnum.HTTP_409_CONFLICT, "该文档类型对应子组已绑定此规则集")

                Result = await Session.execute(
                    text(
                        """
                        INSERT INTO leaudit_rule_group_bindings (
                            group_id,
                            rule_set_id,
                            priority,
                            is_active,
                            note,
                            created_at,
                            updated_at
                        ) VALUES (
                            :group_id,
                            :rule_set_id,
                            :priority,
                            true,
                            :note,
                            NOW(),
                            NOW()
                        )
                        RETURNING id, rule_set_id, priority, is_active, note
                        """
                    ),
                    {
                        "group_id": GroupId,
                        "rule_set_id": RuleSetId,
                        "priority": Priority,
                        "note": Note,
                    },
                )
                await sync_doc_type_bindings_from_group(Session, GroupId)
                await Session.commit()
                Row = Result.mappings().first()
                return RuleBindingVO(
                    id=int(Row["id"]),
                    docTypeId=DocTypeId,
                    docTypeCode=DocTypeCode,
                    ruleSetId=int(Row["rule_set_id"]),
                    ruleType=RsRow["rule_type"],
                    ruleName=RsRow["rule_name"],
                    bindingMode="explicit",
                    priority=int(Row["priority"]),
                    isActive=bool(Row["is_active"]),
                    note=Row["note"],
                )

            child_group_count = await self._count_accessible_child_groups(Session, DocTypeId, current_user)
            if child_group_count == 0:
                raise LeauditException(
                    StatusCodeEnum.HTTP_409_CONFLICT,
                    "当前文档类型尚未创建运行子类型，无法绑定规则集；请先在评查点分组管理中补齐二级分组",
                )
            raise LeauditException(
                StatusCodeEnum.HTTP_409_CONFLICT,
                "当前文档类型存在多个运行子类型，不能再直接按文档类型绑定规则集；请改为在对应二级分组下绑定规则集",
            )

    async def UpdateBinding(
        self,
        BindingId: int,
        IsActive: bool | None = None,
        Priority: int | None = None,
        BindingMode: str | None = None,
        Note: str | None = None,
        CurrentUserId: int | None = None,
    ) -> RuleBindingVO:
        """更新规则类型绑定；若绑定ID来自新分组绑定，则优先更新新表。"""
        async with GetAsyncSession() as Session:
            current_user = await self._get_current_user_context(Session, CurrentUserId)
            GroupRow = await self._get_binding_row(Session, BindingId, current_user)
            if GroupRow:
                await Session.execute(
                    text(
                        """
                        UPDATE leaudit_rule_group_bindings
                        SET priority = :priority,
                            is_active = :is_active,
                            note = :note,
                            updated_at = NOW()
                        WHERE id = :binding_id
                        """
                    ),
                    {
                        "binding_id": BindingId,
                        "priority": Priority if Priority is not None else int(GroupRow.get("priority") or 0),
                        "is_active": IsActive if IsActive is not None else bool(GroupRow.get("is_active", True)),
                        "note": Note if Note is not None else GroupRow.get("note"),
                    },
                )
                await sync_doc_type_bindings_from_group(Session, int(GroupRow["group_id"]))
                await Session.commit()
                refreshed = await self._get_binding_row(Session, BindingId, current_user)
                return RuleBindingVO(
                    id=int(refreshed["id"]),
                    docTypeId=int(refreshed["doc_type_id"]),
                    docTypeCode=refreshed["doc_type_code"],
                    ruleSetId=int(refreshed["rule_set_id"]),
                    ruleType=refreshed["rule_type"],
                    ruleName=refreshed["rule_name"],
                    bindingMode=BindingMode or "explicit",
                    priority=int(refreshed["priority"]),
                    isActive=bool(refreshed["is_active"]),
                    note=refreshed["note"],
                )

            raise LeauditException(
                StatusCodeEnum.HTTP_404_NOT_FOUND,
                "绑定记录不存在或已迁移到新分组绑定，请刷新列表后重试",
            )

    async def DeleteBinding(self, BindingId: int, CurrentUserId: int | None = None) -> None:
        """删除规则类型绑定；若绑定ID来自新分组绑定，则优先删除新表。"""
        async with GetAsyncSession() as Session:
            current_user = await self._get_current_user_context(Session, CurrentUserId)
            GroupRow = await self._get_binding_row(Session, BindingId, current_user)
            if GroupRow:
                await Session.execute(
                    text("UPDATE leaudit_rule_group_bindings SET deleted_at = NOW(), updated_at = NOW() WHERE id = :bid"),
                    {"bid": BindingId},
                )
                await sync_doc_type_bindings_from_group(Session, int(GroupRow["group_id"]))
                await Session.commit()
                return

            raise LeauditException(
                StatusCodeEnum.HTTP_404_NOT_FOUND,
                "绑定记录不存在或已迁移到新分组绑定，请刷新列表后重试",
            )

    async def _SwitchVersion(
        self,
        *,
        RuleType: str,
        VersionId: int,
        TargetStatus: str,
        OperatorUserId: int | None = None,
        CurrentUserId: int | None = None,
        Rollback: bool = False,
    ) -> RuleVersionVO:
        """切换当前生效版本。"""
        async with GetAsyncSession() as Session:
            current_user = await self._get_current_user_context(Session, CurrentUserId)
            use_tenant_scope = await self._column_exists(Session, "leaudit_rule_sets", "tenant_code")
            self._assert_rule_tenant_schema_ready_for_write(use_tenant_scope=use_tenant_scope, current_user=current_user)
            if use_tenant_scope:
                candidate_rows = await self._load_rule_set_rows_by_type(Session, RuleType)
                RuleSet = self._pick_writable_rule_set_row(candidate_rows, current_user=current_user)
            else:
                RuleSet = await self._GetRuleSetByType(Session, RuleType)
            if not RuleSet:
                raise LeauditException(StatusCodeEnum.HTTP_404_NOT_FOUND, "规则集不存在")

            VersionRow = await self._GetVersion(Session, VersionId)
            self._assert_version_belongs_to_writable_rule_set(
                dict(VersionRow) if VersionRow else None,
                dict(RuleSet) if RuleSet else None,
            )
            if Rollback:
                self._assert_rollback_target(
                    dict(VersionRow) if VersionRow else None,
                    dict(RuleSet) if RuleSet else None,
                )

            await Session.execute(
                text(
                    """
                    UPDATE leaudit_rule_versions
                    SET
                        status = CASE
                            WHEN id = :version_id THEN :target_status
                            WHEN status = 'published' THEN :previous_status
                            ELSE status
                        END,
                        publisher_user_id = CASE
                            WHEN id = :version_id THEN :operator_user_id
                            ELSE publisher_user_id
                        END,
                        published_at = CASE
                            WHEN id = :version_id THEN now()
                            ELSE published_at
                        END,
                        updated_at = now()
                    WHERE rule_set_id = :rule_set_id
                    """
                ),
                {
                    "version_id": VersionId,
                    "target_status": TargetStatus,
                    "previous_status": "rollback" if Rollback else "deprecated",
                    "operator_user_id": OperatorUserId,
                    "rule_set_id": RuleSet["id"],
                },
            )
            await Session.execute(
                text(
                    """
                    UPDATE leaudit_rule_sets
                    SET
                        current_version_id = :version_id,
                        status = 'active',
                        updated_at = now()
                    WHERE id = :rule_set_id
                    """
                ),
                {
                    "version_id": VersionId,
                    "rule_set_id": RuleSet["id"],
                },
            )
            await Session.commit()

            self.InvalidateCaches()
            try:
                from fastapi_modules.fastapi_leaudit.services.impl.ruleConfigServiceImpl import GetRuleConfigServiceSingleton

                RuleConfigService = GetRuleConfigServiceSingleton()
                RuleConfigService.InvalidateSummaryCaches()
                await RuleConfigService.WarmPackSummaries(force=False)
            except Exception as exc:
                logging.getLogger("RULE").warning("刷新规则配置摘要缓存失败: %s", exc)

            VersionRow = await self._GetVersion(Session, VersionId)
            return self._BuildRuleVersionVo(VersionRow)

    async def _GetRuleSetByType(self, Session, RuleType: str):
        """按规则类型查询规则集。"""
        Result = await Session.execute(
            text(
                """
                SELECT id, rule_type, rule_name, domain_type, current_version_id, status
                FROM leaudit_rule_sets
                WHERE rule_type = :rule_type
                  AND deleted_at IS NULL
                LIMIT 1
                """
            ),
            {"rule_type": RuleType},
        )
        return Result.mappings().first()

    async def _load_rule_set_rows_by_type(self, Session, RuleType: str) -> list[dict[str, object]]:
        if not await self._column_exists(Session, "leaudit_rule_sets", "tenant_code"):
            row = await self._GetRuleSetByType(Session, RuleType)
            return [dict(row)] if row else []
        Result = await Session.execute(
            text(
                """
                SELECT
                    id,
                    rule_type,
                    rule_name,
                    domain_type,
                    current_version_id,
                    status,
                    COALESCE(NULLIF(BTRIM(tenant_code), ''), 'PROVINCIAL') AS tenant_code,
                    source_rule_set_id
                FROM leaudit_rule_sets
                WHERE rule_type = :rule_type
                  AND deleted_at IS NULL
                ORDER BY id DESC
                """
            ),
            {"rule_type": RuleType},
        )
        return [dict(row) for row in Result.mappings().all()]

    async def _GetVersion(self, Session, VersionId: int):
        """按版本 ID 查询规则版本。"""
        Result = await Session.execute(
            text(
                """
                SELECT
                    id,
                    rule_set_id,
                    version_no,
                    status,
                    oss_url,
                    change_note,
                    published_at
                FROM leaudit_rule_versions
                WHERE id = :version_id
                LIMIT 1
                """
            ),
            {"version_id": VersionId},
        )
        return Result.mappings().first()

    async def _load_latest_version_id_by_rule_set(self, Session, RuleSetId: int) -> int | None:
        row = (
            await Session.execute(
                text(
                    """
                    SELECT id
                    FROM leaudit_rule_versions
                    WHERE rule_set_id = :rule_set_id
                    ORDER BY version_seq DESC, id DESC
                    LIMIT 1
                    """
                ),
                {"rule_set_id": RuleSetId},
            )
        ).mappings().first()
        return int(row["id"]) if row and row.get("id") is not None else None

    def _build_version_scope_snapshot(
        self,
        *,
        writable_tenant_code: str | None,
        writable_scope_type: str,
        rule_set_row: dict[str, object],
        source_version_id: int | None,
    ) -> dict[str, object | None]:
        return {
            "tenant_code_snapshot": writable_tenant_code or str(rule_set_row.get("tenant_code") or "") or "PROVINCIAL",
            "scope_type_snapshot": writable_scope_type,
            "source_version_id": source_version_id,
        }

    def _legacy_region_for_scope(self, tenant_code: str | None, scope_type: str | None) -> str:
        """兼容旧唯一约束 rule_type + region；租户私有规则集不能再共用 default。"""
        normalized_tenant = normalize_scoped_tenant_code(str(tenant_code or ""), default="")
        normalized_scope = str(scope_type or "").strip().upper()
        if normalized_scope == "TENANT" and normalized_tenant:
            return normalized_tenant
        if normalized_tenant == "PUBLIC" or normalized_scope == "PUBLIC":
            return "PUBLIC"
        return "default"

    def _assert_rule_tenant_schema_ready_for_write(
        self,
        *,
        use_tenant_scope: bool,
        current_user: dict[str, object] | None,
    ) -> None:
        """租户用户不能在缺少规则域租户列时写共享规则集。"""
        if use_tenant_scope:
            return
        if current_user is None or current_user.get("is_global"):
            return
        if str(current_user.get("tenant_code") or "").strip():
            raise LeauditException(
                StatusCodeEnum.HTTP_409_CONFLICT,
                "规则域租户隔离表结构未就绪，已拒绝写入共享规则集；请先执行 schema_rule_domain_tenant_phase1.sql",
            )

    def _build_tenant_binding_clone_payload(
        self,
        *,
        current_user: dict[str, object],
        source_binding: dict[str, object],
        tenant_rule_set_id: int,
    ) -> dict[str, object | None]:
        tenant_code = normalize_scoped_tenant_code(str(current_user.get("tenant_code") or ""), default="")
        if not tenant_code:
            raise LeauditException(StatusCodeEnum.HTTP_403_FORBIDDEN, "当前租户上下文缺失，不能派生规则绑定")
        return {
            "group_id": int(source_binding["group_id"]),
            "rule_set_id": int(tenant_rule_set_id),
            "tenant_code": tenant_code,
            "scope_type": "TENANT",
            "tenant_name_snapshot": str(current_user.get("tenant_name") or "").strip() or None,
            "priority": int(source_binding.get("priority") or 0),
            "note": "由租户规则集派生自动补绑",
        }

    async def _load_source_group_binding_ids(self, Session, source_rule_set_id: int) -> list[int]:
        if not await self._column_exists(Session, "leaudit_rule_group_bindings", "tenant_code"):
            return []
        rows = (
            await Session.execute(
                text(
                    """
                    SELECT id
                    FROM leaudit_rule_group_bindings
                    WHERE rule_set_id = :rule_set_id
                      AND deleted_at IS NULL
                      AND is_active = TRUE
                    ORDER BY priority DESC, id ASC
                    """
                ),
                {"rule_set_id": source_rule_set_id},
            )
        ).mappings().all()
        return [int(row["id"]) for row in rows]

    async def _clone_source_group_bindings_for_tenant(
        self,
        Session,
        *,
        source_binding_ids: list[int],
        tenant_rule_set_id: int,
        current_user: dict[str, object] | None,
    ) -> None:
        if not source_binding_ids or not current_user or current_user.get("is_global"):
            return
        rows = (
            await Session.execute(
                text(
                    """
                    SELECT id, group_id, priority, note
                    FROM leaudit_rule_group_bindings
                    WHERE id = ANY(:binding_ids)
                      AND deleted_at IS NULL
                      AND is_active = TRUE
                    ORDER BY priority DESC, id ASC
                    """
                ),
                {"binding_ids": source_binding_ids},
            )
        ).mappings().all()
        for row in rows:
            payload = self._build_tenant_binding_clone_payload(
                current_user=current_user,
                source_binding=dict(row),
                tenant_rule_set_id=tenant_rule_set_id,
            )
            await Session.execute(
                text(
                    """
                    INSERT INTO leaudit_rule_group_bindings (
                        group_id,
                        rule_set_id,
                        tenant_code,
                        scope_type,
                        tenant_name_snapshot,
                        priority,
                        is_active,
                        note,
                        created_at,
                        updated_at
                    )
                    SELECT
                        :group_id,
                        :rule_set_id,
                        CAST(:tenant_code AS varchar),
                        :scope_type,
                        :tenant_name_snapshot,
                        :priority,
                        TRUE,
                        :note,
                        NOW(),
                        NOW()
                    WHERE NOT EXISTS (
                        SELECT 1
                        FROM leaudit_rule_group_bindings
                        WHERE group_id = :group_id
                          AND tenant_code = CAST(:tenant_code AS varchar)
                          AND deleted_at IS NULL
                    )
                    """
                ),
                payload,
            )

    async def _column_exists(self, Session, table_name: str, column_name: str) -> bool:
        cache_key = f"{table_name}.{column_name}"
        cached = self._column_exists_cache.get(cache_key)
        if cached is not None:
            return cached
        exists = bool(
            (
                await Session.execute(
                    text(
                        """
                        SELECT EXISTS (
                            SELECT 1
                            FROM information_schema.columns
                            WHERE table_schema = current_schema()
                              AND table_name = :table_name
                              AND column_name = :column_name
                        )
                        """
                    ),
                    {"table_name": table_name, "column_name": column_name},
                )
            ).scalar_one()
        )
        if exists:
            self._column_exists_cache[cache_key] = exists
        return exists

    async def _load_scoped_rule_set_rows(
        self,
        Session,
        *,
        current_user: dict[str, object] | None,
    ) -> list[dict[str, object]]:
        Result = await Session.execute(
            text(
                """
                SELECT
                    rs.id,
                    rs.rule_type,
                    rs.rule_name,
                    rs.domain_type,
                    rs.current_version_id,
                    current_rv.id AS usable_current_version_id,
                    fallback_rv.id AS fallback_version_id,
                    CASE
                        WHEN current_rv.id IS NOT NULL OR fallback_rv.id IS NOT NULL THEN TRUE
                        ELSE FALSE
                    END AS has_usable_version,
                    rs.status,
                    COALESCE(NULLIF(BTRIM(rs.tenant_code), ''), 'PROVINCIAL') AS tenant_code,
                    rs.source_rule_set_id
                FROM leaudit_rule_sets rs
                LEFT JOIN leaudit_rule_versions current_rv
                  ON current_rv.id = rs.current_version_id
                 AND current_rv.status = 'published'
                LEFT JOIN LATERAL (
                    SELECT rv.id
                    FROM leaudit_rule_versions rv
                    WHERE rv.rule_set_id = rs.id
                      AND rv.status = 'published'
                      AND (rs.current_version_id IS NULL OR rv.id <> rs.current_version_id)
                    ORDER BY rv.version_seq DESC, rv.id DESC
                    LIMIT 1
                ) fallback_rv ON TRUE
                WHERE rs.deleted_at IS NULL
                ORDER BY rs.rule_type ASC, rs.id DESC
                """
            )
        )
        rows = [dict(row) for row in Result.mappings().all()]
        current_user_tenant = str((current_user or {}).get("tenant_code") or "").strip().upper()
        if current_user is None or (current_user.get("is_global") and not current_user_tenant):
            return rows

        grouped: dict[str, list[dict[str, object]]] = {}
        for row in rows:
            grouped.setdefault(str(row.get("rule_type") or ""), []).append(row)

        effective_rows: list[dict[str, object]] = []
        for group_rows in grouped.values():
            effective = pick_effective_scoped_row(group_rows, current_user_tenant or None)
            if effective is not None:
                effective_rows.append(dict(effective))
        effective_rows.sort(key=lambda item: int(item["id"]), reverse=True)
        return effective_rows

    async def _get_effective_rule_set_row_by_type(
        self,
        Session,
        RuleType: str,
        *,
        current_user: dict[str, object] | None,
    ) -> dict[str, object] | None:
        if not await self._column_exists(Session, "leaudit_rule_sets", "tenant_code"):
            return await self._GetRuleSetByType(Session, RuleType)

        Result = await Session.execute(
            text(
                """
                SELECT
                    rs.id,
                    rs.rule_type,
                    rs.rule_name,
                    rs.domain_type,
                    rs.current_version_id,
                    rs.status,
                    COALESCE(NULLIF(BTRIM(rs.tenant_code), ''), 'PROVINCIAL') AS tenant_code,
                    rs.source_rule_set_id
                FROM leaudit_rule_sets rs
                WHERE rs.rule_type = :rule_type
                  AND rs.deleted_at IS NULL
                ORDER BY rs.id DESC
                """
            ),
            {"rule_type": RuleType},
        )
        rows = [dict(row) for row in Result.mappings().all()]
        if not rows:
            return None
        current_user_tenant = str((current_user or {}).get("tenant_code") or "").strip().upper()
        if current_user is None or (current_user.get("is_global") and not current_user_tenant):
            return rows[0]
        effective = pick_effective_scoped_row(rows, current_user_tenant or None)
        return dict(effective) if effective is not None else None

    async def _get_current_user_context(self, Session, CurrentUserId: int | None) -> dict[str, object] | None:
        if CurrentUserId is None:
            return None
        sso_user_columns = await SsoUserCompat.get_columns(Session)
        tenant_code_expr = SsoUserCompat.raw_optional_column(
            sso_user_columns,
            alias="u",
            column="tenant_code",
        )
        tenant_name_expr = SsoUserCompat.raw_optional_column(
            sso_user_columns,
            alias="u",
            column="tenant_name",
        )
        row = (
            await Session.execute(
                text(
                    f"""
                    SELECT
                        u.id,
                        COALESCE(u.area, '') AS area,
                        COALESCE(MAX({tenant_code_expr}), '') AS tenant_code,
                        COALESCE(MAX({tenant_name_expr}), '') AS tenant_name,
                        COALESCE(bool_or(r.role_key IN ('super_admin', 'provincial_admin')), FALSE) AS is_global
                    FROM sso_users u
                    LEFT JOIN user_role ur ON ur.user_id = u.id
                    LEFT JOIN roles r ON r.id = ur.role_id
                    WHERE u.id = :user_id
                      AND u.deleted_at IS NULL
                      AND u.status = 0
                    GROUP BY u.id, u.area
                    """
                ),
                {"user_id": CurrentUserId},
            )
        ).mappings().first()
        if not row:
            raise LeauditException(StatusCodeEnum.HTTP_404_NOT_FOUND, "当前用户不存在或已停用")
        tenant = await self.TenantResolver.ResolveUserContext(
            Area=str(row["area"] or ""),
            TenantCode=str(row.get("tenant_code") or "") or None,
            TenantName=str(row.get("tenant_name") or "") or None,
            Source="rule_binding_user_context",
        )
        return {
            "id": int(row["id"]),
            "tenant_code": tenant.tenant_code or (str(row.get("tenant_code") or "") or None),
            "tenant_name": tenant.tenant_name or (str(row.get("tenant_name") or "") or None),
            "is_global": bool(row["is_global"]),
        }

    async def _entry_module_tenant_table_exists(self, session) -> bool:
        if self._entry_module_tenant_table_exists_cache is not None:
            return self._entry_module_tenant_table_exists_cache
        exists = bool(
            (
                await session.execute(
                    text(
                        """
                        SELECT EXISTS (
                            SELECT 1
                            FROM information_schema.tables
                            WHERE table_schema = current_schema()
                              AND table_name = 'leaudit_entry_module_tenants'
                        )
                        """
                    )
                )
            ).scalar_one()
        )
        self._entry_module_tenant_table_exists_cache = exists
        return exists

    async def _entry_module_tenant_scope_sql(
        self,
        *,
        session,
        entry_module_expr: str,
        current_user: dict[str, object] | None,
        params: dict[str, object],
        param_prefix: str,
    ) -> str:
        if current_user is None or current_user.get("is_global"):
            return "1=1"
        if not await self._entry_module_tenant_table_exists(session):
            return "1=0"
        params[f"{param_prefix}_tenant_code"] = str(current_user.get("tenant_code") or "").strip()
        return f"""
            EXISTS (
                SELECT 1
                FROM leaudit_entry_module_tenants emt
                WHERE emt.entry_module_id = {entry_module_expr}
                  AND emt.deleted_at IS NULL
                  AND emt.is_enabled = TRUE
                  AND (
                      emt.tenant_code = :{param_prefix}_tenant_code
                      OR emt.tenant_code = 'PUBLIC'
                  )
            )
        """

    async def _assert_entry_module_access(self, Session, EntryModuleId: int | None, CurrentUser: dict[str, object] | None) -> None:
        if CurrentUser is None or CurrentUser.get("is_global"):
            return
        if EntryModuleId is None:
            raise LeauditException(StatusCodeEnum.HTTP_403_FORBIDDEN, "当前资源未绑定入口模块，当前租户不可访问")
        params: dict[str, object] = {"entry_module_id": int(EntryModuleId)}
        access_filter = await self._entry_module_tenant_scope_sql(
            session=Session,
            entry_module_expr="em.id",
            current_user=CurrentUser,
            params=params,
            param_prefix="assert_rule_binding_scope",
        )
        exists = (
            await Session.execute(
                text(
                    f"""
                    SELECT em.id
                    FROM leaudit_entry_modules em
                    WHERE em.id = :entry_module_id
                      AND em.deleted_at IS NULL
                      AND {access_filter}
                    LIMIT 1
                    """
                ),
                params,
            )
        ).scalar_one_or_none()
        if exists is None:
            raise LeauditException(StatusCodeEnum.HTTP_403_FORBIDDEN, "当前租户不可访问该入口模块")

    async def _assert_document_type_access(self, Session, DocTypeId: int, CurrentUser: dict[str, object] | None) -> None:
        row = (
            await Session.execute(
                text(
                    """
                    SELECT id, entry_module_id
                    FROM leaudit_document_types
                    WHERE id = :doc_type_id
                      AND deleted_at IS NULL
                    LIMIT 1
                    """
                ),
                {"doc_type_id": DocTypeId},
            )
        ).mappings().first()
        if not row:
            raise LeauditException(StatusCodeEnum.HTTP_404_NOT_FOUND, "文档类型不存在")
        await self._assert_entry_module_access(
            Session,
            int(row["entry_module_id"]) if row.get("entry_module_id") is not None else None,
            CurrentUser,
        )

    async def _resolve_unique_accessible_child_group_id(self, Session, DocTypeId: int, CurrentUser: dict[str, object] | None) -> int | None:
        params: dict[str, object] = {"doc_type_id": DocTypeId}
        access_filter = await self._entry_module_tenant_scope_sql(
            session=Session,
            entry_module_expr="COALESCE(child.entry_module_id, parent.entry_module_id, dt.entry_module_id)",
            current_user=CurrentUser,
            params=params,
            param_prefix="unique_child_scope",
        )
        row = (
            await Session.execute(
                text(
                    f"""
                    SELECT CASE WHEN COUNT(*) = 1 THEN MIN(child.id) END AS group_id
                    FROM leaudit_evaluation_point_groups child
                    LEFT JOIN leaudit_evaluation_point_groups parent ON parent.id = child.pid AND parent.deleted_at IS NULL
                    LEFT JOIN leaudit_document_types dt ON dt.id = child.document_type_id
                    WHERE child.document_type_id = :doc_type_id
                      AND child.deleted_at IS NULL
                      AND COALESCE(child.pid, 0) <> 0
                      AND {access_filter}
                    """
                ),
                params,
            )
        ).mappings().first()
        return int(row["group_id"]) if row and row.get("group_id") is not None else None

    async def _count_accessible_child_groups(self, Session, DocTypeId: int, CurrentUser: dict[str, object] | None) -> int:
        params: dict[str, object] = {"doc_type_id": DocTypeId}
        access_filter = await self._entry_module_tenant_scope_sql(
            session=Session,
            entry_module_expr="COALESCE(child.entry_module_id, parent.entry_module_id, dt.entry_module_id)",
            current_user=CurrentUser,
            params=params,
            param_prefix="count_child_scope",
        )
        row = (
            await Session.execute(
                text(
                    f"""
                    SELECT COUNT(*) AS total
                    FROM leaudit_evaluation_point_groups child
                    LEFT JOIN leaudit_evaluation_point_groups parent ON parent.id = child.pid AND parent.deleted_at IS NULL
                    LEFT JOIN leaudit_document_types dt ON dt.id = child.document_type_id
                    WHERE child.document_type_id = :doc_type_id
                      AND child.deleted_at IS NULL
                      AND COALESCE(child.pid, 0) <> 0
                      AND {access_filter}
                    """
                ),
                params,
            )
        ).mappings().first()
        return int(row["total"] or 0) if row else 0

    async def _get_binding_row(self, Session, BindingId: int, CurrentUser: dict[str, object] | None):
        params: dict[str, object] = {"bid": BindingId}
        access_filter = await self._entry_module_tenant_scope_sql(
            session=Session,
            entry_module_expr="COALESCE(child.entry_module_id, parent.entry_module_id, dt.entry_module_id)",
            current_user=CurrentUser,
            params=params,
            param_prefix="binding_detail_scope",
        )
        row = (
            await Session.execute(
                text(
                    f"""
                    SELECT
                        rgb.id,
                        rgb.group_id,
                        child.document_type_id AS doc_type_id,
                        dt.code AS doc_type_code,
                        rgb.rule_set_id,
                        rgb.priority,
                        rgb.is_active,
                        rgb.note,
                        rs.rule_type,
                        rs.rule_name
                    FROM leaudit_rule_group_bindings rgb
                    JOIN leaudit_evaluation_point_groups child ON child.id = rgb.group_id
                    LEFT JOIN leaudit_evaluation_point_groups parent ON parent.id = child.pid AND parent.deleted_at IS NULL
                    LEFT JOIN leaudit_document_types dt ON dt.id = child.document_type_id
                    JOIN leaudit_rule_sets rs ON rs.id = rgb.rule_set_id
                    WHERE rgb.id = :bid
                      AND rgb.deleted_at IS NULL
                      AND {access_filter}
                    LIMIT 1
                    """
                ),
                params,
            )
        ).mappings().first()
        if not row:
            raise LeauditException(
                StatusCodeEnum.HTTP_404_NOT_FOUND,
                "绑定记录不存在或当前租户不可访问，请刷新列表后重试",
            )
        return row

    def _BuildRuleVersionVo(self, Row) -> RuleVersionVO:
        """构造规则版本响应。"""
        return RuleVersionVO(
            id=int(Row["id"]),
            ruleSetId=int(Row["rule_set_id"]),
            versionNo=Row["version_no"],
            status=Row["status"],
            ossUrl=Row["oss_url"],
            changeNote=Row["change_note"],
            publishedAt=Row["published_at"].isoformat() if Row["published_at"] else None,
        )

    def _ResolveDomainType(self, RuleType: str) -> str:
        """按规则类型推导领域类型。"""
        Prefix = RuleType.split(".", 1)[0]
        Mapping = {
            "contract": "contract",
            "admin_license": "admin_license",
            "legal_doc": "legal_doc",
        }
        return Mapping.get(Prefix, Prefix or "unknown")


_RULE_SERVICE_SINGLETON: RuleServiceImpl | None = None


def GetRuleServiceSingleton() -> RuleServiceImpl:
    """返回共享规则服务实例，供控制器与聚合服务复用缓存。"""
    global _RULE_SERVICE_SINGLETON
    if _RULE_SERVICE_SINGLETON is None:
        _RULE_SERVICE_SINGLETON = RuleServiceImpl()
    return _RULE_SERVICE_SINGLETON