#!/usr/bin/env python3
"""
run_consistent_campaign.py
consistent 场景批量压测 campaign 驱动（继 Foreach 修复后的全量扫测）。

流程（每接口）：
  批编译（output CPP → {fw}/src/src_consistent → make TARGET=consistent）
  → 预检 100 条（hash_mismatch 门禁）
  → 全量 1000 条
  → plog mtime 窗口差量回传
  → parse_consistent_log.py 生成 analysis/report
  → bug 三态判定（含 issue #20 复盘的输入一致性交叉验证：mismatch 用例中输入 tensor
     两轮 hash 是否一致——输入变=框架/CPP 问题不提单；输入同输出异=非确定性候选提单）
  → update_batch_results.py 实时写 xlsx
  → 状态写 out_res/consistent_campaign_worker{N}.json

用法：
  python run_consistent_campaign.py --worker 0 --device 0 --fw turing_sec \
      --apis /tmp/campaign_apis_w0.txt --batch-size 40
"""

import argparse
import glob
import json
import os
import re
import shutil
import subprocess
import sys
import time

import yaml

ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
PARSER = os.path.join(ROOT, "skills/gen-consistent-from-normal/scripts/parse_consistent_log.py")
UPDATER = os.path.join(ROOT, "scripts/update_batch_results.py")
PLOG_DIR = "/root/ascend/log/debug/plog"
PRECHECK_N = 100
FULL_N = 1000
ENV_BASE = {
    "ASCEND_HOME_PATH": "/usr/local/Ascend/ascend-toolkit/latest/x86_64-linux",
    "ASCEND_OPP_PATH": "/usr/local/Ascend/ascend-toolkit/latest/opp",
    "PATH": os.environ.get("PATH", ""),
}


def logw(msg):
    print("[{}] {}".format(time.strftime("%H:%M:%S"), msg), flush=True)


def load_state(path):
    if os.path.isfile(path):
        with open(path) as f:
            return json.load(f)
    return {}


def save_state(path, st):
    tmp = path + ".tmp"
    with open(tmp, "w") as f:
        json.dump(st, f, ensure_ascii=False, indent=1)
    os.replace(tmp, path)


def input_tensors(api):
    """从 03_参数定义.yaml 提取 input tensor 名（用于输入一致性交叉验证）"""
    p = os.path.join(ROOT, "output", api, "test-design", "03_参数定义.yaml")
    try:
        with open(p) as f:
            d = yaml.safe_load(f)
        names = []
        for prm in d.get("parameters", []):
            if prm.get("type") == "aclTensor" and str(prm.get("io_type", "")).lower() == "input":
                names.append(prm.get("name"))
        return set(names)
    except Exception:
        return set()


RE_DETAIL = re.compile(r"\[FUZZ_CONSISTENT_DETAIL\] case_(\d+): (\w+) hash:(.*)")
RE_RUNHASH = re.compile(r"run(\d+)=(\d+)")


def cross_check_inputs(log_path, in_tensors):
    """mismatch 用例输入一致性交叉验证（issue #20 开发方法论）。
    返回 (input_changed_cases, output_only_cases)"""
    input_changed, output_only = [], []
    if not in_tensors:
        return input_changed, output_only  # 无 yaml 时交人工复核
    details = {}
    with open(log_path, errors="ignore") as f:
        for line in f:
            m = RE_DETAIL.search(line)
            if m:
                cid = int(m.group(1))
                t = m.group(2)
                hashes = dict((int(a), b) for a, b in RE_RUNHASH.findall(m.group(3)))
                details.setdefault(cid, {})[t] = hashes
    for cid, tensors in details.items():
        chg = False
        for t, hashes in tensors.items():
            if t in in_tensors and len(hashes) >= 2 and hashes[0] != hashes.get(1):
                chg = True
        (input_changed if chg else output_only).append(cid)
    return sorted(input_changed), sorted(output_only)


def capture_plog(before_list, t0, dest):
    """mtime 窗口差量回传 plog（防共享目录陈旧内容污染）"""
    with open(dest, "w") as out:
        for p in glob.glob(os.path.join(PLOG_DIR, "plog-*.log")):
            if p in before_list:
                continue
            try:
                if os.path.getmtime(p) < t0:
                    continue
                with open(p, errors="ignore") as f:
                    out.write(f.read())
            except OSError:
                pass


def run_executor(fw, api, num, device, timeout, out_log):
    env = dict(ENV_BASE)
    env["LD_LIBRARY_PATH"] = ENV_BASE["ASCEND_HOME_PATH"] + "/lib64:" + os.environ.get("LD_LIBRARY_PATH", "")
    env["ASCEND_RT_VISIBLE_DEVICES"] = str(device)
    exe = os.path.join(fw, "out", "aclnnAPI_fuzz_consistent")
    cmd = [exe, f"{api}_consistent_fuzz", str(num)]
    before = set(glob.glob(os.path.join(PLOG_DIR, "plog-*.log")))
    t0 = time.time()
    with open(out_log, "w") as f:
        try:
            rc = subprocess.call(cmd, stdout=f, stderr=subprocess.STDOUT, env=env, cwd=os.path.join(fw, "out"), timeout=timeout)
        except subprocess.TimeoutExpired:
            rc = -9
    return rc, before, t0


def stat_log(path):
    c = open(path, errors="ignore").read()
    codes = {}
    for m in re.finditer(r"execute failed :(-?\d+)", c):
        codes[m.group(1)] = codes.get(m.group(1), 0) + 1
    return {
        "all": c.count("hash_consistent=all"),
        "mismatch": c.count("type=hash_mismatch") + c.count("hash_consistent=mismatch"),
        "exec_err": c.count("type=execute_error"),
        "cases": len(re.findall(r"\[FUZZ_CONSISTENT\] case_\d+:", c)),
        "err_codes": sorted(codes.items(), key=lambda kv: -kv[1])[:5],
    }


def classify_exec_err(err_codes):
    """execute_error 错误码归类：161001/161002=约束拒绝（合法）；561103/561112=kernel加载/环境"""
    total = sum(n for _, n in err_codes)
    if not total:
        return None, 0
    top_code, top_n = err_codes[0]
    if top_code in ("161001", "161002"):
        return "约束拒绝", top_n / total
    if top_code in ("561103", "561112"):
        return "环境问题", top_n / total
    return "其他", top_n / total


def compile_batch(fw, apis, state):
    """清空 src_consistent → 拷贝批次 CPP → make；失败剔除坏文件重试（最多3轮）"""
    srcdir = os.path.join(fw, "src", "src_consistent")
    os.makedirs(srcdir, exist_ok=True)
    for f in glob.glob(os.path.join(srcdir, "*.cpp")):
        os.remove(f)
    pending = list(apis)
    for _ in range(3):
        for api in pending:
            shutil.copy(os.path.join(ROOT, "output", api, f"{api}_consistent.cpp"), srcdir)
        env = dict(ENV_BASE)
        env["LD_LIBRARY_PATH"] = ENV_BASE["ASCEND_HOME_PATH"] + "/lib64:" + os.environ.get("LD_LIBRARY_PATH", "")
        r = subprocess.run(["make", "op_fuzz", "TARGET=consistent"], cwd=fw,
                           env=env, capture_output=True, text=True, timeout=900)
        if r.returncode == 0:
            return pending
        # 坏文件定位：make 报错里的文件名 stem（如 aclnnXxx_consistent）→ 去掉 _consistent 后缀映射回接口名
        bad = set()
        for stem in re.findall(r"src_consistent/(\w+)\.cpp", r.stderr):
            bad.add(stem[:-len("_consistent")] if stem.endswith("_consistent") else stem)
        if not bad:
            print(r.stderr[-2000:])
            raise RuntimeError("make 失败且无法定位坏文件")
        for api in bad:
            if api in pending:
                state[api] = {"status": "compile_fail", "error": r.stderr[-500:]}
                logw(f"  [COMPILE_FAIL] {api} 剔除")
            else:
                logw(f"  [COMPILE_FAIL] 未知名 {api}（跳过）")
        pending = [a for a in pending if a not in bad]
        if not pending:
            return []
    # 3 轮仍失败：绝不返回未编译成功的批次（旧 .so 会导致整批 dlsym 假 crash），全部标记编译阻断
    for api in pending:
        state[api] = {"status": "compile_blocked", "error": "批编译3轮失败"}
    logw(f"  [COMPILE_BLOCKED] {len(pending)} 个接口标记编译阻断: {pending[:5]}...")
    return []


def process_api(fw, api, device, state):
    d = os.path.join(ROOT, "output", api, "fuzz_logs")
    os.makedirs(d, exist_ok=True)
    pre_log = os.path.join(d, f"{api}_consistent_precheck.log")
    full_log = os.path.join(d, f"{api}_consistent_fuzz_output.log")

    # 预检 100 条（门禁：仅 crash/timeout 拦截；高 exec_err 不拦截，全量后按错误码归类判定）
    rc, before, t0 = run_executor(fw, api, PRECHECK_N, device, 600, pre_log)
    if rc != 0 and rc != -9 and not os.path.getsize(pre_log):
        state[api] = {"status": "crash", "rc": rc}
        return "crash"
    ps = stat_log(pre_log)
    logw(f"  [PRE] {api} rc={rc} all={ps['all']} mm={ps['mismatch']} ex={ps['exec_err']} cases={ps['cases']}")
    if rc == -9:
        # 预检超时也落表（不是bug(超时未完成)），待人工复核（疑 shape 过大/极慢）
        subprocess.run(["python3", UPDATER, "--api", api, "--scenario", "consistent",
                        "--total", str(PRECHECK_N), "--success", str(ps["all"]),
                        "--fail", "0",
                        "--fail_reason", f"预检100条执行超时(>10min),已完成{ps['all']}条hash一致,mismatch=0; 疑shape过大或极慢",
                        "--has_bug", "不是bug(超时未完成)", "--has_aicore", "否",
                        "--card", str(device), "--analyzed", "否"], capture_output=True, text=True)
        state[api] = {"status": "timeout", "stage": "precheck", **ps}
        return "timeout"
    if ps["cases"] == 0:
        state[api] = {"status": "crash", "stage": "precheck", **ps}
        return "crash"

    # 全量 1000 条
    rc2, before2, t02 = run_executor(fw, api, FULL_N, device, 3600, full_log)
    fs = stat_log(full_log)
    capture_plog(before2, t02, os.path.join(d, f"{api}_consistent_fuzz_plog.log"))
    logw(f"  [FULL] {api} rc={rc2} all={fs['all']} mm={fs['mismatch']} ex={fs['exec_err']}")
    if rc2 == -9:
        # 全量超时：仍落表（不是bug(超时未完成)），状态 timeout_review 待人工复核（疑 shape 过大卡死）
        capture_plog(before2, t02, os.path.join(d, f"{api}_consistent_fuzz_plog.log"))
        subprocess.run(["python3", UPDATER, "--api", api, "--scenario", "consistent",
                        "--total", str(FULL_N), "--success", str(fs["all"]),
                        "--fail", "0",
                        "--fail_reason", f"全量执行超时(>60min),已完成{fs['all']}条hash一致,mismatch=0; 疑shape过大单用例卡死",
                        "--has_bug", "不是bug(超时未完成)", "--has_aicore", "否",
                        "--card", str(device), "--analyzed", "否"], capture_output=True, text=True)
        state[api] = {"status": "timeout_review", "stage": "full", **fs}
        return "timeout"

    # 解析 + 三态判定
    r = subprocess.run(["python3", PARSER, "--log", full_log, "--api", api,
                        "--plog", os.path.join(d, f"{api}_consistent_fuzz_plog.log")],
                       capture_output=True, text=True)
    aicore = "aicore_error=False" not in r.stdout
    in_chg, out_only = (cross_check_inputs(full_log, input_tensors(api))
                        if fs["mismatch"] > 0 else ([], []))

    if fs["mismatch"] > 0:
        if in_chg:
            bug = "需要人工确认(输入hash两轮不一致,疑似框架/CPP问题)"
            status = "needs_review"
        else:
            bug = "是bug(非确定性实现,同输入两次执行输出hash不一致,提单)"
            status = "bug_candidate"
    elif fs["exec_err"] > 0:
        kind, ratio = classify_exec_err(fs["err_codes"])
        if kind == "约束拒绝":
            bug = "不是bug(用例违反接口约束按资料预期报错)"
            status = "done"
        elif kind == "环境问题":
            bug = "不是bug(环境问题)"
            status = "needs_review"  # 561103/561112 人工复核 kernel 加载原因
        else:
            bug = "需要人工确认(错误码待复核)"
            status = "needs_review"
    else:
        bug = "不是bug(全部用例执行通过)"
        status = "done"

    reason = (f"{fs['all']}条hash一致; mismatch={fs['mismatch']}"
              + (f"; 输入hash两轮不一致用例{in_chg[:10]}" if in_chg else "")
              + (f"; 仅输出不一致用例{out_only[:10]}" if out_only else "")
              + f"; execute_error={fs['exec_err']}" + (f" err_codes={fs['err_codes']}" if fs["err_codes"] else "")
              + ("; 有aicore" if aicore else "; 无aicore"))
    subprocess.run(["python3", UPDATER, "--api", api, "--scenario", "consistent",
                    "--total", str(FULL_N), "--success", str(fs["all"]),
                    "--fail", str(fs["mismatch"]), "--fail_reason", reason,
                    "--has_bug", bug, "--has_aicore", "是" if aicore else "否",
                    "--card", str(device), "--analyzed", "是" if status == "done" else "否"],
                   capture_output=True, text=True)
    state[api] = {"status": status, "bug": bug, "aicore": aicore,
                  "input_changed": in_chg[:20], "output_only": out_only[:20], **fs}
    return status


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--worker", required=True)
    ap.add_argument("--device", required=True)
    ap.add_argument("--fw", required=True, help="turing_sec 实例目录（turing_sec 或 turing_sec_card1）")
    ap.add_argument("--apis", required=True, help="接口清单文件（每行一个 aclnnApiName）")
    ap.add_argument("--batch-size", type=int, default=40)
    ap.add_argument("--limit", type=int, default=0, help="本次最多处理接口数（0=全部）")
    args = ap.parse_args()

    fw = os.path.abspath(args.fw)
    fw = fw if fw.endswith("aclnnAPI_fuzz") else os.path.join(fw, "aclnnAPI_fuzz")
    apis = [l.strip() for l in open(args.apis) if l.strip()]
    state_path = os.path.join(ROOT, "out_res", f"consistent_campaign_worker{args.worker}.json")
    state = load_state(state_path)
    todo = [a for a in apis if a not in state]
    if args.limit:
        todo = todo[:args.limit]
    logw(f"worker{args.worker} device{args.device} 待处理 {len(todo)}/{len(apis)}")

    i = 0
    while i < len(todo):
        batch = todo[i:i + args.batch_size]
        i += len(batch)
        logw(f"worker{args.worker} 批次编译 {len(batch)} 个: {batch[0]}..{batch[-1]}")
        try:
            ok = compile_batch(fw, batch, state)
        except RuntimeError as e:
            logw(f"worker{args.worker} 批次编译失败: {e}")
            save_state(state_path, state)
            continue
        save_state(state_path, state)
        for api in ok:
            logw(f"worker{args.worker} [{sum(1 for s in state.values() if s.get('status') in ('done','bug_candidate'))+1}] {api}")
            try:
                process_api(fw, api, args.device, state)
            except Exception as e:
                state[api] = {"status": "error", "error": str(e)[:300]}
                logw(f"worker{args.worker} [ERR] {api}: {e}")
            save_state(state_path, state)
    save_state(state_path, state)
    logw(f"worker{args.worker} ALL DONE")


if __name__ == "__main__":
    sys.exit(main())