AiDOPWarehouse/doc/_prep_close_evidence.py

#!/usr/bin/env python3
"""CLOSE 证据准备：S1 MDP 刷新 + S3 净需求样例（幂等，可重复跑）。"""
import json
import subprocess
import sys
import time
import urllib.error
import urllib.request
from datetime import datetime, timedelta

import pymysql

BASE = "http://127.0.0.1:5005"
TENANT = 797403760988229
BILL_NO = "MPO482024102300001"
DEMAND_MARKER = "CLOSE_UAT_S3_DEMAND"
DEMAND_ID = 9106000999000001
CONN = dict(
    host="123.60.180.165",
    port=3306,
    user="aidopremote",
    password="1234567890aiDOP#",
    database="aidopdev",
    charset="utf8mb4",
    connect_timeout=20,
    cursorclass=pymysql.cursors.DictCursor,
)


def http_json(method, path, body=None, token=None, timeout=300):
    headers = {"Content-Type": "application/json", "Accept": "application/json"}
    if token:
        headers["Authorization"] = f"Bearer {token}"
    data = None if body is None else json.dumps(body).encode("utf-8")
    req = urllib.request.Request(f"{BASE}{path}", data=data, headers=headers, method=method)
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            raw = resp.read().decode("utf-8")
            try:
                return resp.status, json.loads(raw)
            except json.JSONDecodeError:
                return resp.status, {"raw": raw[:2000]}
    except urllib.error.HTTPError as e:
        raw = e.read().decode("utf-8", errors="replace")
        try:
            return e.code, json.loads(raw)
        except json.JSONDecodeError:
            return e.code, {"raw": raw[:2000]}


def login():
    node_script = r"""
const { sm2 } = require('sm-crypto-v2');
const PK = '0484C7466D950E120E5ECE5DD85D0C90EAA85081A3A2BD7C57AE6DC822EFCCBD66620C67B0103FC8DD280E36C3B282977B722AAEC3C56518EDCEBAFB72C5A05312';
console.log(JSON.stringify({ account: 'AIDOPDemo', password: sm2.doEncrypt('1234567890dop', PK, 1), tenantId: '797403760988229' }));
"""
    proc = subprocess.run(
        ["node", "-e", node_script],
        cwd=r"d:\Projects\Ai-DOP\SourceCode\ZZYDOP\Web",
        capture_output=True,
        text=True,
        timeout=30,
    )
    if proc.returncode != 0:
        raise RuntimeError(proc.stderr or proc.stdout)
    _, body = http_json("POST", "/api/sysAuth/login", json.loads(proc.stdout.strip()))
    token = (body.get("result") or {}).get("accessToken") or body.get("accessToken")
    if not token:
        raise RuntimeError(f"login failed: {body}")
    return token


def unwrap(body):
    if isinstance(body, dict) and body.get("code") == 200 and "result" in body:
        return body["result"]
    if isinstance(body, dict) and body.get("ok") is True:
        return body
    return body


def count_dwd(conn):
    with conn.cursor() as c:
        c.execute(
            "SELECT COUNT(*) AS c FROM dwd_ship_trans WHERE tenant_id=%s AND order_no=%s",
            (TENANT, BILL_NO),
        )
        return c.fetchone()["c"]


def count_demand(conn):
    with conn.cursor() as c:
        c.execute(
            """
            SELECT COUNT(*) AS c FROM ic_demandschedule
            WHERE tenant_id=%s AND IFNULL(status,'')='P' AND IFNULL(tosechedqty,0)>0
              AND (IFNULL(ishistoryversion,'')='' OR ishistoryversion='N') AND IFNULL(IsDeleted,0)=0
            """,
            (TENANT,),
        )
        return c.fetchone()["c"]


def find_seed_item(conn):
    with conn.cursor() as c:
        c.execute(
            """
            SELECT sp.number AS itemnum, sp.supplier_number, im.PurMfg
            FROM srm_purchase sp
            INNER JOIN ItemMaster im ON im.ItemNum = sp.number
            WHERE sp.tenant_id=%s AND IFNULL(sp.IsDeleted,0)=0
              AND IFNULL(sp.quota_rate,0)>0
              AND IFNULL(sp.supplier_number,'')<>''
              AND IFNULL(im.PurMfg,'')='P'
            ORDER BY sp.number
            LIMIT 1
            """,
            (TENANT,),
        )
        return c.fetchone()


def ensure_demand_seed(conn):
    existing = count_demand(conn)
    if existing > 0:
        return {"action": "skip", "reason": f"demand_published={existing}"}

    item = find_seed_item(conn)
    if not item:
        return {"action": "failed", "reason": "no item with source+PO open qty"}

    need_date = (datetime.now() + timedelta(days=7)).strftime("%Y-%m-%d %H:%M:%S")
    qty = 10.0
    with conn.cursor() as c:
        c.execute("SELECT Id FROM ic_demandschedule WHERE Id=%s", (DEMAND_ID,))
        if c.fetchone():
            c.execute(
                """
                UPDATE ic_demandschedule
                SET itemnum=%s, status='P', toschedqty=%s, sechedqty=%s,
                    requestdate=%s, arrivaldate=%s, ishistoryversion='N',
                    IsDeleted=0, remarks=%s, update_time=NOW()
                WHERE Id=%s AND tenant_id=%s
                """,
                (item["itemnum"], qty, qty, need_date, need_date, DEMAND_MARKER, DEMAND_ID, TENANT),
            )
        else:
            c.execute(
                """
                INSERT INTO ic_demandschedule
                (Id, itemnum, status, toschedqty, sechedqty, requestdate, arrivaldate,
                 ishistoryversion, remarks, tenant_id, IsDeleted, create_time, update_time)
                VALUES (%s,%s,'P',%s,%s,%s,%s,'N',%s,%s,0,NOW(),NOW())
                """,
                (DEMAND_ID, item["itemnum"], qty, qty, need_date, need_date, DEMAND_MARKER, TENANT),
            )
    conn.commit()
    return {"action": "seeded", "itemnum": item["itemnum"], "qty": qty, "demand_id": DEMAND_ID}


def refresh_s1_mdp(token):
    st, body = http_json("POST", "/api/AidopKanban/s1-mdp/refresh", None, token, timeout=600)
    return st, unwrap(body)


def wait_dwd(conn, target=2, attempts=12, sleep_s=5):
    for i in range(attempts):
        c = count_dwd(conn)
        if c >= target:
            return c, i + 1
        time.sleep(sleep_s)
    return count_dwd(conn), attempts


def main():
    report = {"tenant_id": str(TENANT), "bill_no": BILL_NO, "steps": []}
    token = login()
    report["login"] = "ok"

    with pymysql.connect(**CONN) as conn:
        report["before"] = {
            "dwd_ship_rows": count_dwd(conn),
            "demand_published": count_demand(conn),
        }

        st, mdp = refresh_s1_mdp(token)
        report["steps"].append({"step": "s1_mdp_refresh", "http": st, "body": mdp})

        dwd_after_mdp, tries = wait_dwd(conn, target=2)
        report["steps"].append({"step": "wait_dwd_ship", "rows": dwd_after_mdp, "attempts": tries})

        demand_result = ensure_demand_seed(conn)
        report["steps"].append({"step": "seed_s3_demand", **demand_result})

        report["after"] = {
            "dwd_ship_rows": count_dwd(conn),
            "demand_published": count_demand(conn),
        }

    report["checks"] = [
        {"name": "s1_mdp_http", "ok": report["steps"][0]["http"] == 200},
        {"name": "dwd_ship_main_ge2", "ok": report["after"]["dwd_ship_rows"] >= 2},
        {"name": "demand_published_gt0", "ok": report["after"]["demand_published"] > 0},
    ]
    report["passed"] = all(c["ok"] for c in report["checks"])

    out = r"d:\Projects\Ai-DOP\SourceCode\ZZYDOP\doc\_prep_close_evidence_result.json"
    with open(out, "w", encoding="utf-8") as f:
        json.dump(report, f, ensure_ascii=False, indent=2, default=str)

    print(json.dumps(report, ensure_ascii=False, indent=2, default=str))
    sys.exit(0 if report["passed"] else 7)


if __name__ == "__main__":
    main()