shanxia2.0/backend/scripts/generate_dict_reimport_sql.py

#!/usr/bin/env python3
"""Merge zkxw-dict.sql + sx-dict.sql into shanxia2.0 sys_dict_* INSERT script."""

from __future__ import annotations

import re
from pathlib import Path

ROOT = Path(__file__).resolve().parents[2]
ZKXW_SQL = ROOT / "docs/sql/zkxw-dict.sql"
SX_SQL = ROOT / "docs/sql/sx-dict.sql"
SHANXIA_SQL = ROOT / "docs/sql/shanxia2.0.sql"
OUT_SQL = ROOT / "docs/sql/reimport-dict-from-zkxw-sx.sql"
TENANT = "shanxia2.0"
# sx 源库部分 CRM 字典无时间戳；目标表 create_time/update_time 为 NOT NULL
DEFAULT_DT = "2026-05-30 00:00:00"

# shanxia2.0 门户内置字典（不在 zkxw-dict / sx-dict 中，从基线 SQL 保留）
PORTAL_BASELINE_TYPES = frozenset(
    {
        "sys_status",
        "sys_sex",
        "sys_data_scope",
        "sys_visible",
        "menu_type",
        "leave_status",
        "wf_process_status",
        "wf_node_type",
        "wf_approve_type",
        "user_role",
    }
)

DICT_TYPE_RE = re.compile(r"^[a-z][a-z0-9_]*$")

TAG_TYPE_MAP = {
    "default": "",
    "primary": "primary",
    "success": "success",
    "danger": "danger",
    "warning": "warning",
    "info": "info",
}


def split_sql_values(values_part: str) -> list:
    parts: list[str] = []
    cur: list[str] = []
    in_str = False
    i = 0
    while i < len(values_part):
        c = values_part[i]
        if in_str:
            if c == "\\" and i + 1 < len(values_part):
                cur.append(c)
                cur.append(values_part[i + 1])
                i += 2
                continue
            if c == "'":
                if i + 1 < len(values_part) and values_part[i + 1] == "'":
                    cur.append("''")
                    i += 2
                    continue
                in_str = False
                cur.append(c)
            else:
                cur.append(c)
        else:
            if c == "'":
                in_str = True
                cur.append(c)
            elif c == ",":
                parts.append("".join(cur).strip())
                cur = []
            else:
                cur.append(c)
        i += 1
    if cur:
        parts.append("".join(cur).strip())
    return parts


def unquote(val: str):
    val = val.strip()
    if val.upper() == "NULL":
        return None
    if val.startswith("'") and val.endswith("'"):
        return val[1:-1].replace("''", "'")
    if re.fullmatch(r"-?\d+", val):
        return int(val)
    return val


def parse_insert(line: str) -> tuple[str, list[str], list] | None:
    m = re.match(
        r"INSERT INTO `(\w+)` \(([^)]+)\) VALUES \((.+)\);\s*$",
        line.strip(),
    )
    if not m:
        return None
    table = m.group(1)
    cols = [c.strip().strip("`") for c in m.group(2).split(",")]
    vals = [unquote(v) for v in split_sql_values(m.group(3))]
    return table, cols, vals


def sql_literal(val) -> str:
    if val is None:
        return "NULL"
    if isinstance(val, int):
        return str(val)
    s = str(val).replace("\\", "\\\\").replace("'", "''")
    return f"'{s}'"


def is_valid_dict_type(code: str | None) -> bool:
    return bool(code) and bool(DICT_TYPE_RE.match(code))


def map_tag_type(list_class: str | None) -> str:
    if not list_class:
        return ""
    key = str(list_class).strip().lower()
    return TAG_TYPE_MAP.get(key, key if key in TAG_TYPE_MAP.values() else "")


def sx_type_id(dict_id: int) -> str:
    return f"SXDT{dict_id:016X}"


def sx_data_id(dict_code: int) -> str:
    return f"SXDC{dict_code:016X}"


def normalize_row(row: dict) -> dict:
    out = dict(row)
    ct = out.get("create_time")
    ut = out.get("update_time")
    if not ct:
        out["create_time"] = ut or DEFAULT_DT
    if not out.get("update_time"):
        out["update_time"] = out["create_time"] or DEFAULT_DT
    return out


def row_to_insert(table: str, cols: list[str], row: dict) -> str:
    row = normalize_row(row)
    ordered = [row[c] for c in cols]
    vals = ", ".join(sql_literal(v) for v in ordered)
    col_sql = ", ".join(f"`{c}`" for c in cols)
    return f"INSERT INTO `{table}` ({col_sql}) VALUES ({vals});"


def load_portal_baseline():
    types: dict[str, dict] = {}
    data: list[dict] = []
    for line in SHANXIA_SQL.read_text(encoding="utf-8").splitlines():
        parsed = parse_insert(line)
        if not parsed:
            continue
        table, cols, vals = parsed
        row = dict(zip(cols, vals))
        if table == "sys_dict_type":
            code = row.get("dict_type")
            if code not in PORTAL_BASELINE_TYPES or row.get("del_flag", 0) != 0:
                continue
            types[code] = row
        elif table == "sys_dict_data":
            code = row.get("dict_type")
            if code not in PORTAL_BASELINE_TYPES or row.get("del_flag", 0) != 0:
                continue
            row.setdefault("tag_type", "")
            data.append(row)
    return types, data


def load_zkxw():
    types: dict[str, dict] = {}
    data: list[dict] = []
    for line in ZKXW_SQL.read_text(encoding="utf-8").splitlines():
        parsed = parse_insert(line)
        if not parsed:
            continue
        table, cols, vals = parsed
        row = dict(zip(cols, vals))
        if row.get("del_flag", 0) != 0:
            continue
        row["tenant_no"] = TENANT
        if table == "sys_dict_type":
            code = row.get("dict_type")
            if is_valid_dict_type(code):
                types[code] = row
        elif table == "sys_dict_data":
            code = row.get("dict_type")
            if not is_valid_dict_type(code):
                continue
            row["tenant_no"] = TENANT
            row.setdefault("tag_type", "")
            data.append(row)
    return types, data


def load_sx():
    types: dict[str, dict] = {}
    data: list[dict] = []
    for line in SX_SQL.read_text(encoding="utf-8").splitlines():
        parsed = parse_insert(line)
        if not parsed:
            continue
        table, cols, vals = parsed
        row = dict(zip(cols, vals))
        if table == "sys_dict_type":
            dict_id = row.get("dict_id")
            code = row.get("dict_type")
            name = row.get("dict_name")
            if not is_valid_dict_type(code):
                continue
            if str(name) in ("000000", TENANT, "zkxw", "0"):
                continue
            types[code] = {
                "id": sx_type_id(int(dict_id)),
                "dict_name": name,
                "dict_type": code,
                "tenant_no": TENANT,
                "order_num": 0,
                "remark": row.get("remark") or "",
                "create_by": None,
                "create_dept_id": "",
                "create_time": row.get("create_time"),
                "update_by": None,
                "update_time": row.get("update_time") or row.get("create_time"),
                "del_flag": 0,
            }
        elif table == "sys_dict_data":
            dict_code = row.get("dict_code")
            code = row.get("dict_type")
            label = row.get("dict_label")
            value = row.get("dict_value")
            if dict_code is None or not is_valid_dict_type(code):
                continue
            if not label or not str(label).strip():
                continue
            if str(label) in ("planned", "zsjy_visit_status") and str(value) in (
                "zsjy_visit_status",
                "planned",
            ):
                continue
            create_by = row.get("create_by")
            data.append(
                {
                    "id": sx_data_id(int(dict_code)),
                    "dict_label": label,
                    "dict_value": value,
                    "tag_type": map_tag_type(row.get("list_class")),
                    "dict_type": code,
                    "tenant_no": TENANT,
                    "order_num": row.get("dict_sort") or 0,
                    "remark": row.get("remark") or "",
                    "create_by": str(create_by) if create_by is not None else None,
                    "create_dept_id": "",
                    "create_time": row.get("create_time"),
                    "update_by": None,
                    "update_time": row.get("update_time") or row.get("create_time"),
                    "del_flag": 0,
                }
            )
    return types, data


def main():
    portal_types, portal_data = load_portal_baseline()
    zkxw_types, zkxw_data = load_zkxw()
    sx_types, sx_data = load_sx()

    merged_types = dict(portal_types)
    merged_types.update(zkxw_types)
    merged_types.update(sx_types)

    type_cols = [
        "id",
        "dict_name",
        "dict_type",
        "tenant_no",
        "order_num",
        "remark",
        "create_by",
        "create_dept_id",
        "create_time",
        "update_by",
        "update_time",
        "del_flag",
    ]
    data_cols = [
        "id",
        "dict_label",
        "dict_value",
        "tag_type",
        "dict_type",
        "tenant_no",
        "order_num",
        "remark",
        "create_by",
        "create_dept_id",
        "create_time",
        "update_by",
        "update_time",
        "del_flag",
    ]

    # portal baseline -> zkxw -> sx
    seen_data_ids: set[str] = set()
    merged_data: list[dict] = []
    for row in portal_data + zkxw_data + sx_data:
        rid = row["id"]
        if rid in seen_data_ids:
            continue
        seen_data_ids.add(rid)
        if row["dict_type"] not in merged_types:
            continue
        merged_data.append(row)

    lines = [
        "-- Reimport dictionary data for shanxia2.0",
        "-- Sources: shanxia2.0 portal baseline + zkxw-dict.sql + sx-dict.sql",
        "-- Generated by: backend/scripts/generate_dict_reimport_sql.py",
        "",
        "SET NAMES utf8mb4;",
        "SET FOREIGN_KEY_CHECKS = 0;",
        "",
        "DELETE FROM `sys_dict_data`;",
        "DELETE FROM `sys_dict_type`;",
        "",
        "BEGIN;",
        "",
        "-- sys_dict_type",
    ]
    for code in sorted(merged_types.keys()):
        lines.append(row_to_insert("sys_dict_type", type_cols, merged_types[code]))

    lines.append("")
    lines.append("-- sys_dict_data")
    for row in merged_data:
        lines.append(row_to_insert("sys_dict_data", data_cols, row))

    lines.extend(["", "COMMIT;", "", "SET FOREIGN_KEY_CHECKS = 1;", ""])

    OUT_SQL.write_text("\n".join(lines), encoding="utf-8")
    print(
        f"Wrote {OUT_SQL.relative_to(ROOT)}: "
        f"{len(merged_types)} types, {len(merged_data)} data rows "
        f"(portal={len(portal_types)}, zkxw={len(zkxw_types)}, sx={len(sx_types)})"
    )


if __name__ == "__main__":
    main()