shanxia2.0/backend/scripts/compare_dict_snapshots.py

#!/usr/bin/env python3
"""Compare sys_dict_type / sys_dict_data across SQL snapshots."""
from __future__ import annotations

import re
from collections import defaultdict
from pathlib import Path

ROOT = Path(__file__).resolve().parents[2]
SX_SQL = ROOT / "docs/sql/sx.sql"
PORTAL_SQL = ROOT / "docs/sql/shanxia2.0.sql"
DIBANG_SQL = ROOT / "docs/sql/dibang.sql"


def parse_insert_values(line: str) -> list[str] | None:
    m = re.search(r"VALUES\s*\((.*)\)\s*;?\s*$", line, re.I)
    if not m:
        return None
    inner = m.group(1)
    parts: list[str] = []
    cur: list[str] = []
    in_str = False
    esc = False
    for ch in inner:
        if esc:
            cur.append(ch)
            esc = False
            continue
        if ch == "\\":
            esc = True
            cur.append(ch)
            continue
        if ch == "'":
            in_str = not in_str
            cur.append(ch)
            continue
        if ch == "," and not in_str:
            parts.append("".join(cur).strip())
            cur = []
            continue
        cur.append(ch)
    if cur:
        parts.append("".join(cur).strip())
    return parts


def unquote(v: str) -> str:
    v = v.strip()
    if v.upper() == "NULL":
        return ""
    if v.startswith("'") and v.endswith("'"):
        return v[1:-1].replace("''", "'").replace("\\'", "'")
    return v


def load_sx_dicts(path: Path) -> tuple[dict[str, dict], dict[str, list[dict]]]:
    types: dict[str, dict] = {}
    data_by_type: dict[str, list[dict]] = defaultdict(list)
    for line in path.read_text(encoding="utf-8", errors="ignore").splitlines():
        if "INSERT INTO `sys_dict_type`" in line:
            vals = parse_insert_values(line)
            if not vals or len(vals) < 4:
                continue
            # dict_id, tenant_id, dict_name, dict_type, ...
            dict_type = unquote(vals[3])
            types[dict_type] = {
                "dict_name": unquote(vals[2]),
                "tenant_id": unquote(vals[1]),
                "remark": unquote(vals[9]) if len(vals) > 9 else "",
            }
        elif "INSERT INTO `sys_dict_data`" in line:
            vals = parse_insert_values(line)
            if not vals or len(vals) < 6:
                continue
            # dict_code, tenant_id, dict_sort, dict_label, dict_value, dict_type, ...
            dict_type = unquote(vals[5])
            data_by_type[dict_type].append(
                {
                    "label": unquote(vals[3]),
                    "value": unquote(vals[4]),
                    "sort": unquote(vals[2]),
                    "css_class": unquote(vals[6]) if len(vals) > 6 else "",
                    "list_class": unquote(vals[7]) if len(vals) > 7 else "",
                }
            )
    return types, data_by_type


def load_portal_dicts(path: Path) -> tuple[dict[str, dict], dict[str, list[dict]]]:
    types: dict[str, dict] = {}
    data_by_type: dict[str, list[dict]] = defaultdict(list)
    for line in path.read_text(encoding="utf-8", errors="ignore").splitlines():
        if "INSERT INTO `sys_dict_type`" in line:
            vals = parse_insert_values(line)
            if not vals or len(vals) < 3:
                continue
            # id, dict_name, dict_type, tenant_no, ...
            dict_type = unquote(vals[2])
            types[dict_type] = {
                "dict_name": unquote(vals[1]),
                "tenant_no": unquote(vals[3]),
                "remark": unquote(vals[5]) if len(vals) > 5 else "",
            }
        elif "INSERT INTO `sys_dict_data`" in line:
            vals = parse_insert_values(line)
            if not vals or len(vals) < 5:
                continue
            # id, dict_label, dict_value, tag_type, dict_type, tenant_no, ...
            dict_type = unquote(vals[4])
            data_by_type[dict_type].append(
                {
                    "label": unquote(vals[1]),
                    "value": unquote(vals[2]),
                    "tag_type": unquote(vals[3]),
                    "sort": unquote(vals[6]) if len(vals) > 6 else "",
                }
            )
    return types, data_by_type


def item_key(item: dict) -> str:
    return f"{item['value']}|{item['label']}"


def compare_data(sx_items: list[dict], portal_items: list[dict]) -> dict:
    sx_map = {item_key(i): i for i in sx_items}
    portal_map = {item_key(i): i for i in portal_items}
    sx_keys = set(sx_map)
    portal_keys = set(portal_map)
    return {
        "only_sx": sorted(sx_keys - portal_keys),
        "only_portal": sorted(portal_keys - sx_keys),
        "common": sorted(sx_keys & portal_keys),
        "sx_count": len(sx_items),
        "portal_count": len(portal_items),
    }


def main() -> None:
    sx_types, sx_data = load_sx_dicts(SX_SQL)
    portal_types, portal_data = load_portal_dicts(PORTAL_SQL)

    sx_type_keys = set(sx_types)
    portal_type_keys = set(portal_types)

    only_sx_types = sorted(sx_type_keys - portal_type_keys)
    only_portal_types = sorted(portal_type_keys - sx_type_keys)
    common_types = sorted(sx_type_keys & portal_type_keys)

    def count_by_prefix(types: set[str], prefix: str) -> int:
        return sum(1 for t in types if t.startswith(prefix))

    sx_data_total = sum(len(v) for v in sx_data.values())
    portal_data_total = sum(len(v) for v in portal_data.values())
    portal_active = sum(
        1 for items in portal_data.values() for i in items
    )  # placeholder

    print("=" * 72)
    print("字典对比总览（基于 docs/sql 快照）")
    print("=" * 72)
    print(f"{'':28} {'sx.sql':>12} {'shanxia2.0':>12}")
    print(f"{'字典类型数':28} {len(sx_type_keys):>12} {len(portal_type_keys):>12}")
    print(f"{'字典项数':28} {sx_data_total:>12} {portal_data_total:>12}")
    print(f"{'crm_* 类型数':28} {count_by_prefix(sx_type_keys,'crm_'):>12} {count_by_prefix(portal_type_keys,'crm_'):>12}")
    print(f"{'xg_* 类型数(学工)':28} {count_by_prefix(sx_type_keys,'xg_'):>12} {count_by_prefix(portal_type_keys,'xg_'):>12}")
    print(f"{'sys_* 类型数':28} {count_by_prefix(sx_type_keys,'sys_'):>12} {count_by_prefix(portal_type_keys,'sys_'):>12}")
    print(f"{'pl_* 类型数(运输)':28} {count_by_prefix(sx_type_keys,'pl_'):>12} {count_by_prefix(portal_type_keys,'pl_'):>12}")
    print()

    # semantic alias pairs
    alias_pairs = [
        ("sys_user_sex", "sys_sex"),
        ("sys_show_hide", "sys_visible"),
        ("sys_normal_disable", "sys_status"),
    ]
    print("--- 语义相近但类型码不同的「别名」---")
    for sx_t, portal_t in alias_pairs:
        in_sx = sx_t in sx_types or portal_t in sx_types
        in_portal = sx_t in portal_types or portal_t in portal_types
        if in_sx or in_portal:
            print(
                f"  sx: {sx_t} ({len(sx_data.get(sx_t,[]))}项) "
                f"| portal: {portal_t} ({len(portal_data.get(portal_t,[]))}项) "
                f"+ portal {sx_t} ({len(portal_data.get(sx_t,[]))}项)"
            )
    print()

    crm_prefixes = ("crm_", "xs_", "bx_")
    crm_only_sx = [t for t in only_sx_types if t.startswith(crm_prefixes) or "crm" in t or t.startswith("xs_")]
    other_only_sx = [t for t in only_sx_types if t not in crm_only_sx]

    print("=" * 72)
    print("字典类型 (sys_dict_type) 差异明细")
    print("=" * 72)
    print(f"类型码相同（两边都有）: {len(common_types)}")
    print(f"仅 sx 有（门户缺失）: {len(only_sx_types)}")
    print(f"仅门户有（sx 无，多为 2.0 新增）: {len(only_portal_types)}")
    print()

    print("--- 仅门户有的字典类型（shanxia2.0 新增能力）---")
    for t in only_portal_types:
        print(f"  {t:40} {portal_types[t]['dict_name']}")
    print()

    print("--- 仅 sx 有 · CRM/业务相关（建议评估是否迁入）---")
    for t in crm_only_sx:
        print(f"  {t:40} {sx_types[t]['dict_name']}  (data:{len(sx_data.get(t, []))})")
    print()

    print("--- 仅 sx 有 · 学工/校园 xg_*（门户一般不需要）---")
    xg_only = [t for t in only_sx_types if t.startswith("xg_")]
    for t in xg_only[:20]:
        print(f"  {t:40} {sx_types[t]['dict_name']}  (data:{len(sx_data.get(t, []))})")
    if len(xg_only) > 20:
        print(f"  ... 共 {len(xg_only)} 个 xg_* 类型")
    print()

    print("--- 仅 sx 有 · 若依框架 sys_*（可能与门户同名不同义）---")
    sys_overlap_missing = [t for t in only_sx_types if t.startswith("sys_")]
    for t in sys_overlap_missing:
        print(f"  {t:40} {sx_types[t]['dict_name']}  (data:{len(sx_data.get(t, []))})")
    print()

    print("--- 仅 sx 有 · 其他 ---")
    for t in other_only_sx:
        if not t.startswith("xg_"):
            print(f"  {t:40} {sx_types[t]['dict_name']}  (data:{len(sx_data.get(t, []))})")
    print()

    print("=" * 72)
    print("同名类型下字典项 (sys_dict_data) 差异（sx ∩ 门户）")
    print("=" * 72)
    diff_types = []
    for t in common_types:
        cmp = compare_data(sx_data.get(t, []), portal_data.get(t, []))
        if cmp["only_sx"] or cmp["only_portal"] or cmp["sx_count"] != cmp["portal_count"]:
            diff_types.append((t, cmp))
    print(f"类型码相同但字典项不完全一致: {len(diff_types)} / {len(common_types)}")
    print()
    for t, cmp in sorted(diff_types, key=lambda x: x[0]):
        sx_name = sx_types[t]["dict_name"]
        portal_name = portal_types[t]["dict_name"]
        print(f"[{t}] sx:{sx_name} | portal:{portal_name}")
        print(f"  sx项数={cmp['sx_count']} portal项数={cmp['portal_count']}")
        if cmp["only_sx"]:
            print(f"  仅sx有 ({len(cmp['only_sx'])}):")
            for k in cmp["only_sx"][:8]:
                print(f"    - {k}")
            if len(cmp["only_sx"]) > 8:
                print(f"    ... +{len(cmp['only_sx']) - 8}")
        if cmp["only_portal"]:
            print(f"  仅门户有 ({len(cmp['only_portal'])}):")
            for k in cmp["only_portal"][:8]:
                print(f"    - {k}")
            if len(cmp["only_portal"]) > 8:
                print(f"    ... +{len(cmp['only_portal']) - 8}")
        print()

    print("=" * 72)
    print("CRM 业务字典（sx 有、门户无）明细")
    print("=" * 72)
    crm_types = sorted(
        t
        for t in only_sx_types
        if t.startswith("crm_") or t.startswith("xs_") or t in ("contract_type", "contract_nature")
    )
    for t in crm_types:
        print(f"\n## {t} — {sx_types[t]['dict_name']}")
        for item in sorted(sx_data.get(t, []), key=lambda x: (x.get("sort") or "", x["value"])):
            print(f"   {item['value']:20} {item['label']}")

    print()
    print("=" * 72)
    print("地磅系统 dibang.sql")
    print("=" * 72)
    dibang_text = DIBANG_SQL.read_text(encoding="utf-8", errors="ignore")
    has_dict = "sys_dict" in dibang_text
    print("含 sys_dict 表/数据:", "是" if has_dict else "否（地磅无独立字典表，枚举多在业务字段或前端硬编码）")


if __name__ == "__main__":
    main()