"""对账：代码声明的权限码 vs 库里存在的 vs 各角色实际拥有的（只读）。

## 为什么需要它

权限缺失的表现一律是 `AGENT_PERMISSION_DENIED: 缺少操作权限`，但底下有**三种完全不同的原因**，
报错区分不出来，排查时极易往错的方向找：

1. **权限码在库里根本不存在**（代码写了、没人建）→ 任何角色都过不去，
   看起来却像"这个角色没授权"；
2. 权限码存在，但**没授给这个角色**（例如投顾缺 `investment-goal:write:self`）；
3. 角色本身没有对应的**接口范围或界面**（例如运营只有 `offsite:write`，
   而它要用的 `financial:nl2sql:read` 从没建过）。

这类问题**服务层测试测不出来** —— 那层直接构造 `RequestContext`，权限是测试自己塞的。
只有把"代码声明"和"库里数据"放一起比才看得见。2026-09-12 用它在主干上查出
6 个不存在的权限码（`promotion:*` 四个导致推广材料整条线 403、`financial:nl2sql:read`、
`probe:read`）。

只读：不连业务逻辑、不改任何数据。

用法：

    python tools/check_permission_coverage.py
"""

from __future__ import annotations

import re
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from app.core.config import get_settings  # noqa: E402

PROJECT_ROOT = Path(__file__).resolve().parents[1]

#: 代码里声明权限码的几种写法。
PATTERNS: tuple[str, ...] = (
    r'require\(\s*context\s*,\s*"([^"]+)"',
    r'^\s*permission\s*[:=]\s*"([^"]+)"',
    r'^[A-Z_]*PERMISSION[A-Z_]*\s*[:=]\s*"([^"]+)"',
    r'required_permission\s*=\s*"([^"]+)"',
)

CODE_SHAPE = re.compile(r"^[a-z][a-z0-9_-]*(?::[a-z0-9_-]+)+$")


def collect_required(root: Path = PROJECT_ROOT) -> dict[str, set[str]]:
    found: dict[str, set[str]] = {}
    for path in (root / "app").rglob("*.py"):
        text = path.read_text(encoding="utf-8", errors="replace")
        for pattern in PATTERNS:
            for match in re.finditer(pattern, text, re.MULTILINE):
                code = match.group(1)
                if CODE_SHAPE.match(code):
                    found.setdefault(code, set()).add(path.relative_to(root).as_posix())
    return found


def load_from_db() -> tuple[dict[str, int], dict[str, set[str]]]:
    """返回 `(权限码 → id, 角色 → 权限码集合)`。"""
    settings = get_settings()
    dsn = settings.mysql_dsn.split("://", 1)[1]
    credentials, location = dsn.split("@", 1)
    user, password = credentials.split(":", 1)
    host_port, database = location.split("/", 1)
    host, _, port = host_port.partition(":")
    return _query(host, int(port or 3306), user, password, database)


def _query(host: str, port: int, user: str, password: str, database: str):
    import asyncio

    import asyncmy

    async def run():
        connection = await asyncmy.connect(
            host=host, port=port, user=user, password=password, db=database
        )
        try:
            cursor = connection.cursor()
            await cursor.execute("SELECT id, permission_code FROM sys_permission")
            permissions = {str(code): int(pid) for pid, code in await cursor.fetchall()}

            await cursor.execute(
                """
                SELECT r.role_code, p.permission_code
                FROM sys_role r
                LEFT JOIN sys_role_permission rp ON rp.role_id = r.id
                LEFT JOIN sys_permission p ON p.id = rp.permission_id
                """
            )
            roles: dict[str, set[str]] = {}
            for role_code, permission_code in await cursor.fetchall():
                roles.setdefault(str(role_code), set())
                if permission_code:
                    roles[str(role_code)].add(str(permission_code))
            return permissions, roles
        finally:
            connection.close()

    return asyncio.run(run())


def main() -> int:
    required = collect_required()
    permissions, roles = load_from_db()

    print(f"代码里声明的权限码：{len(required)} 个")
    print(f"库里已有的权限码：{len(permissions)} 个")

    missing = sorted(code for code in required if code not in permissions)
    print()
    print(f"【一】代码要求、但库里不存在（任何角色都过不去）：{len(missing)} 个")
    for code in missing:
        where = sorted(required[code])[:3]
        print(f"  ✗ {code:<44} 用于 {', '.join(where)}")

    unused = sorted(code for code in permissions if code not in required)
    if unused:
        print()
        print(f"【二】库里有、代码里没直接搜到（可能是变量拼接或历史遗留）：{len(unused)} 个")
        for code in unused:
            holders = sorted(r for r, s in roles.items() if code in s)
            print(f"  · {code:<44} 持有角色 {holders or '无'}")

    print()
    print("【三】各角色的权限数，以及「代码要求却没拿到」的码：")
    for role, codes in sorted(roles.items()):
        lacks = sorted(c for c in required if c in permissions and c not in codes)
        print(f"  {role:<16} 共 {len(codes):>2} 项；代码要求但未授予 {len(lacks)} 个")
        for code in lacks[:12]:
            print(f"      - {code}")
        if len(lacks) > 12:
            print(f"      … 另有 {len(lacks) - 12} 个")

    if missing:
        print()
        print("处置：把这几个码并进 `tools/seed_test_rbac.py` 的 `PERMISSIONS`（那是唯一定义源），")
        print("     再按角色在对应的 `grant_*.py` 里授权。")
        return 1
    print()
    print("结论：代码要求的权限码在库里都存在。")
    return 0


if __name__ == "__main__":
    sys.exit(main())