AiDOPWarehouse/doc/db/mdp/mock_api/seed_mock_db.py

"""登记模式 3 的三条 Mock 来源。不修改 WMS_API、TEST_INBOUND_ERP。

连接串从 server/Admin.NET.Application/Configuration/Database.json 的默认库读取，不打印密码。
只插入 aidopdev.mdp_source 中 system_code=MOCK_MODE3 的三条；库 aidop_mock_mode3 仅在有建库权限时创建。
"""
from __future__ import annotations

import sys
from datetime import datetime
from pathlib import Path

ROOT = Path(__file__).resolve().parents[4]
DB_JSON = ROOT / "server" / "Admin.NET.Application" / "Configuration" / "Database.json"


def connect(_database: str = "aidopdev"):
    import importlib.util

    helper = ROOT / "doc" / "plan" / "sql" / "uat-data" / "apply_sql_file.py"
    spec = importlib.util.spec_from_file_location("apply_sql_file", helper)
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod.connect()


def main() -> int:
    conn = connect("aidopdev")
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT COLUMN_NAME, IS_NULLABLE, COLUMN_DEFAULT, EXTRA, DATA_TYPE
                FROM information_schema.COLUMNS
                WHERE TABLE_SCHEMA='aidopdev' AND TABLE_NAME='mdp_source'
                """
            )
            meta = list(cur.fetchall())
            cols = {r["COLUMN_NAME"] for r in meta}
            needed = {"source_code", "system_code"}
            if not needed <= cols:
                print("mdp_source missing", needed - cols)
                return 1
            sources = [
                ("MOCK_MODE3_DB", "MOCK_MODE3", "aidop_mock_mode3"),
                ("MOCK_MODE3_API", "MOCK_MODE3", "http://127.0.0.1:8018"),
                ("MOCK_MODE3_INBOUND", "MOCK_MODE3", "API_INBOUND"),
            ]
            for code, system, hint in sources:
                cur.execute(
                    "SELECT COUNT(*) AS c FROM mdp_source WHERE source_code=%s",
                    (code,),
                )
                if cur.fetchone()["c"]:
                    print("exists", code)
                    continue
                values = {
                    "source_code": code,
                    "system_code": system,
                    "source_name": code,
                    "tenant_id": 0,
                    "status": 1,
                }
                if "conn_hint" in cols:
                    values["conn_hint"] = hint
                names = []
                vals = []
                for col in meta:
                    name = col["COLUMN_NAME"]
                    if "auto_increment" in (col["EXTRA"] or ""):
                        continue
                    if name in values:
                        names.append(name)
                        vals.append(values[name])
                        continue
                    if col["IS_NULLABLE"] == "NO" and col["COLUMN_DEFAULT"] is None:
                        kind = col["DATA_TYPE"]
                        names.append(name)
                        if kind in ("int", "bigint", "tinyint", "decimal"):
                            vals.append(0)
                        elif kind in ("datetime", "timestamp", "date"):
                            vals.append(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
                        else:
                            vals.append("")
                sql = "INSERT INTO mdp_source (" + ",".join(names) + ") VALUES (" + ",".join(["%s"] * len(names)) + ")"
                cur.execute(sql, vals)
                print("inserted", code, hint)
        try:
            with conn.cursor() as cur:
                cur.execute(
                    "CREATE DATABASE IF NOT EXISTS aidop_mock_mode3 DEFAULT CHARACTER SET utf8mb4"
                )
            print("database aidop_mock_mode3 ready")
        except Exception as ex:
            print("create database skipped:", type(ex).__name__)
    finally:
        conn.close()
    return 0


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