zsjz-ai/docs/testing/fixtures/ai-llm-match-20261026/retest/retest_fixes.py

# -*- coding: utf-8 -*-
"""F-01/F-01B/F-03B/F-05/F-07 真实链路复测（F-06 由单测覆盖）。"""
import json
import os
import sys
import time

import requests

BASE = os.path.dirname(os.path.abspath(__file__))
FIXTURES = os.path.dirname(BASE)
sys.path.insert(0, FIXTURES)
import run_tests as rt  # noqa: E402

CASE_ID = 27
rt.CASE_ID = CASE_ID
sys.stdout.reconfigure(encoding="utf-8")


def log(msg):
    print(msg, flush=True)


FILES = [
    # (测试点, 文件名)
    ("F-01", "RT_F01_BILL-09_交行同义.xlsx"),
    ("F-01B", "RT_F01B_CALL-09_导出话单同义.xlsx"),
    ("F-01B", "RT_F01B_CALL-17_日期时间同义.xlsx"),
    ("F-01B", "RT_F01B_CALL-19_时长格式.xlsx"),
    ("F-01B", "RT_F01B_IND-03_开户同义.xlsx"),
    ("F-03B/F-05", "RT_F03B_F05_CALL-18_方向_新增列.xlsx"),
    ("F-03B/F-07", "RT_F03B_F07_BILL-方向_新增列.xlsx"),
]


def ai_data_page(c, tpl_id, size=30):
    r = c.request("GET", "/aiClean/data/page",
                  params={"aiTemplateId": tpl_id, "page": 1, "size": size}, timeout=60)
    j = r.json() or {}
    return (j.get("data") or {})


def main():
    c = rt.Client()
    c.open_case()
    batch_id = None
    roots = []
    recs = []
    log("== 上传 ==")
    for point, name in FILES:
        path = os.path.join(BASE, name)
        with open(path, "rb") as f:
            r = c.request("POST", "/dm/preFileUpload", params=({"batchId": batch_id} if batch_id else {}),
                          files={"file": (name, f)}, timeout=120)
        data = (r.json() or {}).get("data")
        if not data:
            log(f"  {name}: DATA_NULL {(r.json() or {}).get('message')}")
            recs.append(dict(point=point, file=name, upload="DATA_NULL"))
            continue
        batch_id = batch_id or data.get("batchId")
        roots.append(data)
        sheets = []
        for ch in data.get("children") or []:
            sheets.append(dict(sheet_id=ch.get("id"), sheet_name=ch.get("sheetName"),
                               status=ch.get("fileStatus"), template_id=ch.get("templateId"),
                               data_num=ch.get("dataNum")))
        log(f"  {name}: {[(s['sheet_name'], s['status'], s['template_id']) for s in sheets]}")
        recs.append(dict(point=point, file=name, file_id=data.get("id"),
                         batch_id=data.get("batchId"), sheets=sheets))

    log(f"batch={batch_id}; 发起智能清洗")
    sc = c.smart_clean(roots)
    log(f"smartClean: {sc.get('code')} {sc.get('message')} data={sc.get('data')}")
    if sc.get("code") != 200 or not sc.get("data"):
        json.dump(recs, open(os.path.join(BASE, "retest_result.json"), "w", encoding="utf-8"),
                  ensure_ascii=False, indent=1, default=str)
        return

    t0 = time.time()
    final = None
    while time.time() - t0 < 1800:
        st = c.status(batch_id)
        if st and st.get("stage") in ("DONE", "PARTIAL", "FAILED"):
            final = st
            break
        time.sleep(3)
    log(f"batch final={final}")

    sheet_ids = [s["sheet_id"] for r in recs for s in r.get("sheets") or []]
    jobs = rt.db_jobs(sheet_ids)
    for rec in recs:
        rec["jobs"] = []
        for s in rec.get("sheets") or []:
            job = jobs.get(int(s["sheet_id"])) if s.get("sheet_id") else None
            s["job"] = job
            if job:
                rec["jobs"].append(job)
        rec["ai_templates"] = []
        for j in rec["jobs"]:
            tid = j.get("ai_template_id")
            if tid:
                rec["ai_templates"].append(rt.db_ai_template(tid))
        log(f"\n[{rec['point']}] {rec['file']}")
        for j in rec["jobs"]:
            log(f"  job={j['job_id']} status={j['status']} exit={j['exit_code']} by={j['matched_by']} "
                f"tier={j['match_tier']} ref={j['ref_template_id']} ai={j['ai_template_id']} "
                f"llm={j['llm_calls']} conf={j['confidence']} rows={j['row_total']}/{j['row_loaded']} "
                f"fail={j['fail_reason']}")
            tpl_id = j.get("ai_template_id")
            tpl = next((t for t in rec["ai_templates"] if t and str(t["id"]) == str(tpl_id)), None)
            if tpl:
                log(f"  AI表名={tpl['table_name_cn']!r} headers={tpl['headers'][:80]!r}")
                # 方向字段核验
                dir_cols = [f for f in tpl.get("fields") or []
                            if f.get("field_type") in ("CALL_DIRECTION", "DEBIT_CREDIT_FLAG")]
                if dir_cols:
                    data = ai_data_page(c, tpl["id"], 30)
                    recs_data = (data.get("pages") or {}).get("records") or []
                    for f in dir_cols:
                        vals = sorted({str(r.get(f["column_name"])) for r in recs_data if r.get(f["column_name"]) is not None})
                        log(f"  direction field {f['field_name_cn']} ({f['field_type']}) distinct={vals[:10]}")
                        rec.setdefault("direction_values", {})[f["field_name_cn"]] = vals[:10]

    json.dump(recs, open(os.path.join(BASE, "retest_result.json"), "w", encoding="utf-8"),
              ensure_ascii=False, indent=1, default=str)
    log("\nresult -> retest_result.json")


if __name__ == "__main__":
    main()