AiDOPWarehouse/doc/db/mdp/_apply_1_0_259.py

"""Apply UpdateScripts/1.0.259.sql to aidopdev (idempotent)."""
from __future__ import annotations

import os
import pathlib

import pymysql

ROOT = pathlib.Path(__file__).resolve().parents[3]
SQL_PATH = ROOT / "server" / "Admin.NET.Web.Entry" / "UpdateScripts" / "1.0.259.sql"


def main() -> None:
    sql = SQL_PATH.read_text(encoding="utf-8")
    lines = []
    for line in sql.splitlines():
        if line.strip().startswith("--"):
            continue
        lines.append(line)
    body = "\n".join(lines)

    stmts: list[str] = []
    buf: list[str] = []
    for line in body.splitlines():
        buf.append(line)
        if line.rstrip().endswith(";"):
            stmts.append("\n".join(buf))
            buf = []
    if buf:
        stmts.append("\n".join(buf))

    conn = pymysql.connect(
        host=os.environ.get("MDP_DB_HOST", "127.0.0.1"),
        user=os.environ.get("MDP_DB_USER", "root"),
        password=os.environ.get("MDP_DB_PASSWORD", ""),
        database=os.environ.get("MDP_DB_NAME", "aidopdev"),
        charset="utf8mb4",
        autocommit=True,
    )
    cur = conn.cursor()
    ok = 0
    for i, st in enumerate(stmts, 1):
        st = st.strip()
        if not st:
            continue
        try:
            cur.execute(st)
            ok += 1
        except Exception as e:
            print(f"FAIL#{i}: {e}")
            print(st[:300])
            raise
    print(f"OK statements={ok}")

    cur.execute(
        "SELECT COUNT(*) FROM information_schema.COLUMNS "
        "WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='mdp_entity' AND COLUMN_NAME='biz_key_expr'"
    )
    print("biz_key_expr", cur.fetchone())
    cur.execute(
        "SELECT COUNT(*) FROM information_schema.STATISTICS "
        "WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='mdp_stg_purchase_receipt' AND INDEX_NAME='uk_source_key'"
    )
    print("uk_purchase", cur.fetchone())
    cur.execute(
        "SELECT entity_code, biz_key_expr FROM mdp_entity "
        "WHERE entity_code IN ("
        "'S5_PURCHASE_RECEIPT_DETAIL','S5_PURCHASE_RECEIPT_MASTER',"
        "'S7_NBR_DETAIL','S6_IPQC_INSPECTION_HEAD','S7_NBR_MASTER')"
    )
    for row in cur.fetchall():
        print(row)
    cur.close()
    conn.close()


if __name__ == "__main__":
    main()