"""只读探查：`sys_user` 的认证字段现状 —— 判断"账号密码换令牌"这条路通不通。

只做 SELECT；`password_hash` **只统计前缀特征与长度，不输出完整值**（金融项目的基本习惯，
即使是测试库）。

    python tools/probe_auth_state.py
"""

from __future__ import annotations

import asyncio
import json
from collections import Counter
from pathlib import Path
from typing import Any

from sqlalchemy import text

from app.infrastructure.db import SessionFactory

OUTPUT = Path("docs/evidence/auth-state.json")

#: 常见密码哈希方案的特征前缀。任何一项命中，说明库里存在**真实可校验**的密码。
KNOWN_HASH_PREFIXES = (
    "$2a$", "$2b$", "$2y$",          # bcrypt
    "$argon2",                        # argon2
    "$pbkdf2", "$pbkdf2-sha256",     # passlib pbkdf2
    "$5$", "$6$", "$1$",             # crypt sha256/sha512/md5
    "$scrypt",
)


def _classify(values: list[str]) -> dict[str, Any]:
    prefixes: Counter[str] = Counter()
    lengths: Counter[int] = Counter()
    for value in values:
        lengths[len(value)] += 1
        matched = next(
            (prefix for prefix in KNOWN_HASH_PREFIXES if value.startswith(prefix)), None
        )
        prefixes[matched or f"<未知格式，首字符 {value[:1]!r}>"] += 1
    return {
        "length_distribution": dict(sorted(lengths.items())),
        "prefix_distribution": dict(prefixes.most_common()),
    }


async def collect() -> dict[str, Any]:
    report: dict[str, Any] = {}
    async with SessionFactory() as session:
        rows = (
            await session.execute(
                text(
                    "SELECT id, username, user_type, status, password_hash "
                    "FROM sys_user ORDER BY id"
                )
            )
        ).mappings().all()

        report["user_count"] = len(rows)
        report["by_status"] = dict(
            Counter(str(row["status"]) for row in rows).most_common()
        )
        report["by_user_type"] = dict(
            Counter(str(row["user_type"]) for row in rows).most_common()
        )
        report["password_hash_shape"] = _classify(
            [str(row["password_hash"] or "") for row in rows]
        )
        # 判断"能不能校验密码"的关键结论，直接给出来。
        hashes = [str(row["password_hash"] or "") for row in rows]
        report["has_real_password_hash"] = any(
            value.startswith(KNOWN_HASH_PREFIXES) for value in hashes
        )
        report["placeholder_examples"] = sorted(
            {
                value
                for value in hashes
                if not value.startswith(KNOWN_HASH_PREFIXES)
            }
        )[:5]
    return report


async def main() -> None:
    report = await collect()
    OUTPUT.parent.mkdir(parents=True, exist_ok=True)
    OUTPUT.write_text(
        json.dumps(report, ensure_ascii=False, indent=2, default=str),
        encoding="utf-8",
    )
    print(f"wrote {OUTPUT}")


if __name__ == "__main__":
    asyncio.run(main())