#!/usr/bin/env python3 """Record evidence-backed WP3 B00-B14 status without overstating completion.""" from __future__ import annotations import json import re from datetime import datetime from pathlib import Path import pymysql from pymysql.cursors import DictCursor ROOT = Path(__file__).resolve().parents[4] DATABASE_JSON = ( ROOT / "server" / "Admin.NET.Application" / "Configuration" / "Database.json" ) EVIDENCE = Path(__file__).with_name("WP3-B00-B14-evidence.json") TENANT_ID = 838257212780613 def connect() -> pymysql.Connection: raw = DATABASE_JSON.read_text(encoding="utf-8-sig") connection_string = next( value for value in re.findall( r'(?m)^\s*"ConnectionString"\s*:\s*"([^"]+)"', raw ) if "Database=aidopdev" in value ) parts = { item.split("=", 1)[0].strip().lower(): item.split("=", 1)[1].strip() for item in connection_string.split(";") if "=" in item } return pymysql.connect( host=parts["server"], port=int(parts["port"]), user=parts["uid"], password=parts["pwd"], database=parts["database"], charset="utf8mb4", autocommit=True, cursorclass=DictCursor, ) def scalar(cur: DictCursor, sql: str, args: tuple = ()) -> int: cur.execute(sql, args) return int(next(iter(cur.fetchone().values()))) def main() -> None: conn = connect() try: with conn.cursor() as cur: cur.execute( """ CREATE TABLE IF NOT EXISTS uat_step_execution ( id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY, tenant_id BIGINT NOT NULL, work_package VARCHAR(20) NOT NULL, step_code VARCHAR(20) NOT NULL, status VARCHAR(30) NOT NULL, evidence_json JSON NULL, verified_at DATETIME(3) NOT NULL, UNIQUE KEY uk_uat_step (tenant_id,work_package,step_code) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """ ) facts = { "users": scalar( cur, "SELECT COUNT(*) FROM SysUser WHERE TenantId=%s", (TENANT_ID,), ), "items": scalar( cur, "SELECT COUNT(*) FROM ItemMaster WHERE tenant_id=%s", (TENANT_ID,), ), "bom_headers": scalar( cur, "SELECT COUNT(*) FROM ProductStructureMaster WHERE tenant_id=%s", (TENANT_ID,), ), "routing_ops": scalar( cur, "SELECT COUNT(*) FROM RoutingOpDetail WHERE tenant_id=%s", (TENANT_ID,), ), "sales_orders": scalar( cur, "SELECT COUNT(*) FROM crm_seorder WHERE tenant_id=%s", (TENANT_ID,), ), "work_orders": scalar( cur, "SELECT COUNT(*) FROM WorkOrdMaster WHERE tenant_id=%s", (TENANT_ID,), ), "purchase_requests": scalar( cur, "SELECT COUNT(*) FROM srm_pr_main WHERE tenant_id=%s", (TENANT_ID,), ), "purchase_orders": scalar( cur, "SELECT COUNT(*) FROM PurOrdMaster WHERE tenant_id=%s", (TENANT_ID,), ), "supplier_shipments": scalar( cur, "SELECT COUNT(*) FROM scm_shd WHERE tenant_id=%s", (TENANT_ID,), ), "iqc_closed": scalar( cur, """ SELECT COUNT(*) FROM ApprovalFlowInstance WHERE BizId IN ( SELECT id FROM qms_qcp_inspbill WHERE tenant_id=%s AND FBILLNO='UATB-B11-IQC-001' ) AND Status=2 """, (TENANT_ID,), ), "ipqc_closed": scalar( cur, """ SELECT COUNT(*) FROM ApprovalFlowInstance WHERE BizId IN ( SELECT id FROM qms_gcjyd WHERE tenant_id=%s AND djbh='UATB-B12-IPQC-001' ) AND Status=2 """, (TENANT_ID,), ), "fqc_closed": scalar( cur, """ SELECT COUNT(*) FROM ApprovalFlowInstance WHERE BizId IN ( SELECT id FROM qms_qcpp_inspbill WHERE tenant_id=%s AND FBILLNO='UATB-B12-FQC-001' ) AND Status=2 """, (TENANT_ID,), ), "s8_exceptions": scalar( cur, "SELECT COUNT(*) FROM ado_s8_exception WHERE tenant_id=%s", (TENANT_ID,), ), "successful_modules": scalar( cur, """ SELECT COUNT(DISTINCT module_code) FROM ado_module_dashboard_rebuild_job WHERE tenant_id=%s AND status='SUCCESS' AND module_code IN ('S1','S2','S3','S4','S5','S6','S7') """, (TENANT_ID,), ), } steps = { "B00": ("PASSED" if facts["users"] >= 4 else "BLOCKED", {"users": facts["users"]}), "B01": ("PASSED" if facts["items"] >= 20 else "BLOCKED", {"items": facts["items"]}), "B02": ( "PASSED" if facts["bom_headers"] > 0 and facts["routing_ops"] > 0 else "BLOCKED", {"bom_headers": facts["bom_headers"], "routing_ops": facts["routing_ops"]}, ), "B03": ("CHECKPOINT_READY", {"sales_orders": facts["sales_orders"]}), "B04": ("CHECKPOINT_READY", {"sales_orders": facts["sales_orders"]}), "B05": ("CHECKPOINT_READY", {"work_orders": facts["work_orders"]}), "B06": ("CHECKPOINT_READY", {"work_orders": facts["work_orders"]}), "B07": ("CHECKPOINT_READY", {"purchase_requests": facts["purchase_requests"]}), "B08": ("CHECKPOINT_READY", {"purchase_orders": facts["purchase_orders"]}), "B09": ("CHECKPOINT_READY", {"supplier_shipments": facts["supplier_shipments"]}), "B10": ("CHECKPOINT_READY", {"supplier_shipments": facts["supplier_shipments"]}), "B11": ( "API_VERIFIED" if facts["iqc_closed"] else "PENDING_EXECUTION", {"iqc_closed": facts["iqc_closed"]}, ), "B12": ( "API_VERIFIED" if facts["ipqc_closed"] and facts["fqc_closed"] else "PENDING_EXECUTION", { "ipqc_closed": facts["ipqc_closed"], "fqc_closed": facts["fqc_closed"], }, ), "B13": ( "API_VERIFIED" if facts["s8_exceptions"] else "PENDING_EXECUTION", {"s8_exceptions": facts["s8_exceptions"]}, ), "B14": ( "PASSED" if facts["successful_modules"] == 7 else "BLOCKED", {"successful_modules": facts["successful_modules"]}, ), } for step_code, (status, evidence) in steps.items(): cur.execute( """ INSERT INTO uat_step_execution (tenant_id,work_package,step_code,status,evidence_json,verified_at) VALUES(%s,'WP3',%s,%s,%s,NOW(3)) ON DUPLICATE KEY UPDATE status=VALUES(status), evidence_json=VALUES(evidence_json), verified_at=VALUES(verified_at) """, (TENANT_ID, step_code, status, json.dumps(evidence)), ) finally: conn.close() payload = { "generated_at": datetime.now().isoformat(timespec="seconds"), "tenant_id": str(TENANT_ID), "facts": facts, "steps": { code: {"status": status, "evidence": evidence} for code, (status, evidence) in steps.items() }, } EVIDENCE.write_text( json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8", ) print(json.dumps(payload, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()