AiDOPWarehouse/doc/_check_close_gaps.py

#!/usr/bin/env python3
import json
import pymysql

T = 797403760988229
B = "MPO482024102300001"
CONN = dict(
    host="123.60.180.165",
    port=3306,
    user="aidopremote",
    password="1234567890aiDOP#",
    database="aidopdev",
    charset="utf8mb4",
    cursorclass=pymysql.cursors.DictCursor,
)

def main():
    with pymysql.connect(**CONN) as conn:
        with conn.cursor() as c:
            checks = {}
            c.execute(
                "SELECT COUNT(*) AS c FROM dwd_ship_trans WHERE tenant_id=%s AND order_no=%s",
                (T, B),
            )
            checks["dwd_ship_main"] = c.fetchone()["c"]
            c.execute(
                """
                SELECT job_code, status, batch_id, start_time
                FROM mdp_transform_run_log
                WHERE job_code='S1_MDP_SYNC_TRANSFORM'
                ORDER BY start_time DESC LIMIT 3
                """
            )
            checks["s1_mdp_runs"] = c.fetchall()
            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
                """,
                (T,),
            )
            checks["demand_published"] = c.fetchone()["c"]
            c.execute(
                "SELECT Id, itemnum, toschedqty, status, ishistoryversion FROM ic_demandschedule WHERE tenant_id=%s LIMIT 8",
                (T,),
            )
            checks["demand_rows"] = c.fetchall()
            c.execute(
                "SELECT entry_seq, progress FROM crm_seorderentry WHERE tenant_id=%s AND bill_no=%s",
                (T, B),
            )
            checks["order_progress"] = c.fetchall()
            c.execute("SELECT COUNT(*) AS c FROM srm_polist_ds WHERE tenant_id=%s", (T,))
            checks["ds_total"] = c.fetchone()["c"]
            c.execute(
                """
                SELECT sp.itemnum, sp.suppliercode, COUNT(*) AS po_lines
                FROM srm_purchase sp
                INNER JOIN PurOrdMaster m ON m.tenant_id=%s
                INNER JOIN PurOrdDetail d ON d.PurOrdRecID=m.RecID AND d.ItemNum=sp.itemnum
                WHERE sp.tenant_id=%s AND IFNULL(sp.IsDeleted,0)=0
                  AND (sp.is_active IN ('是','Y','1','y') OR sp.is_active=1)
                GROUP BY sp.itemnum, sp.suppliercode
                LIMIT 5
                """,
                (T, T),
            )
            checks["source_po_items"] = c.fetchall()
    print(json.dumps(checks, ensure_ascii=False, indent=2, default=str))

if __name__ == "__main__":
    main()