#!/usr/bin/env python3
"""
run_scenario_campaign.py
长尾接口多场景压测 campaign（依次逐个：每接口按 sample→normal→non_contig→multithread→error→consistent
顺序执行已就绪场景，门禁与 AGENTS.md Step 3 一致）。

每场景流水线：目录切换(op→op_back) → 拷贝CPP → make → 预检/调试门禁 → 全量 →
plog 窗口差量归档 → 统计+错误码分类三态判定 → update_batch_results.py 落表 → 状态文件

用法：
  python run_scenario_campaign.py --worker 0 --device 0 --fw turing_sec --apis /tmp/scn_apis_w0.txt
"""

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

ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
PLOG_DIR = "/root/ascend/log/debug/plog"
UPDATER = os.path.join(ROOT, "scripts", "update_batch_results.py")
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", ""),
}
SCN_ORDER = ["sample", "normal", "non_contig", "multithread", "error", "consistent"]


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


def load_state(p):
    return json.load(open(p)) if os.path.isfile(p) else {}


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


def run_cmd(fw, args, timeout, out_log, device, remap_visible=True):
    env = dict(ENV_BASE)
    env["LD_LIBRARY_PATH"] = ENV_BASE["ASCEND_HOME_PATH"] + "/lib64:" + os.environ.get("LD_LIBRARY_PATH", "")
    if remap_visible:
        env["ASCEND_RT_VISIBLE_DEVICES"] = str(device)
    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(args, 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 capture_plog(before, t0, dest):
    with open(dest, "w") as out:
        for p in glob.glob(os.path.join(PLOG_DIR, "plog-*.log")):
            if p in before:
                continue
            try:
                if os.path.getmtime(p) < t0:
                    continue
                with open(p, errors="ignore") as f:
                    out.write(f.read())
            except OSError:
                pass


def stat_log(path, api):
    """统计执行结果 + 错误码分布 + aicore 初扫"""
    c = open(path, errors="ignore").read()
    ok = len(re.findall(r"\[FUZZ_RESULT\] case_\d+: success", c))
    fail = len(re.findall(r"\[FUZZ_RESULT\] case_\d+: faild", c))
    skip = len(re.findall(r"\[FUZZ_RESULT\] case_\d+: skip", c))
    codes = {}
    for m in re.finditer(r"execute failed :(-?\d+)", c):
        codes[m.group(1)] = codes.get(m.group(1), 0) + 1
    verify = re.findall(r"\[FUZZ_VERIFY\] case_\d+: .*match=(YES|NO)", c)
    scen = re.findall(r"== Scenario\s+\d+ \| ret=(-?\d+)\s*\| (\w+)\s*\| EZ:(PASS|FAIL)", c)
    base_ok = sum(1 for _, v, _ in scen if v == "BASE_CHECK_OK")
    unexp = sum(1 for _, v, _ in scen if v == "UNEXPECTED_SUCCESS")
    err_scen = [s for s in scen if s[1] not in ("BASE_CHECK_OK",)]
    ez_pass = sum(1 for _, _, z in err_scen if z == "PASS")
    ez_fail = sum(1 for _, _, z in err_scen if z == "FAIL")
    return {"ok": ok, "fail": fail, "skip": skip,
            "scen_total": len(scen), "base_ok": base_ok, "ez_pass": ez_pass, "ez_fail": ez_fail, "unexp_success": unexp,
            "err_codes": sorted(codes.items(), key=lambda kv: -kv[1])[:4],
            "verify_total": len(verify), "verify_no": sum(1 for v in verify if v == "NO")}


def scan_aicore(plog_path):
    if not plog_path or not os.path.isfile(plog_path):
        return False
    c = open(plog_path, errors="ignore").read().lower()
    for m in ("the error is aivec error", "aivec error", "aicore error", "aicore_error"):
        if m in c:
            return True
    for line in c.splitlines():
        if "kernel_name" in line and ("[error]" in line or " error " in line):
            return True
    return False


def classify(full, scenario):
    """三态判定（基础版：错误码分类；详细根因由后续分析）"""
    ok, fail = full.get("ok", 0), full.get("fail", 0)
    codes = full.get("err_codes", [])
    total = ok + fail
    if scenario == "error":
        if full.get("scen_total", 0) > 0:
            if full.get("unexp_success", 0) > 0:
                return "needs_review", "是bug候选(意外成功待复核)"
            if full.get("ez_fail", 0) == 0 and full.get("ez_pass", 0) > 0:
                return "done", "不是bug(全部用例按资料预期报错,维测信息完备)"
            if full.get("ez_fail", 0) > 0:
                return "needs_review", "需要人工确认(错误码不符或维测信息缺失待复核)"
            return "needs_review", "需要人工确认(error场景无可解析错误结果)"
        if total > 0:
            return "needs_review", "需要人工确认(error场景无可解析输出)"
        return "crash", "crash"
    if total == 0:
        return "crash", "crash"
    rate = ok * 100.0 / total
    if rate >= 95:
        return "done", "不是bug(全部用例执行通过)" if fail == 0 else "不是bug(执行通过率达标)"
    top = codes[0][0] if codes else ""
    if top in ("161001", "161002"):
        return "done", "不是bug(用例违反接口约束按资料预期报错)"
    if top in ("561103", "561112"):
        return "needs_review", "不是bug(环境问题)"
    return "needs_review", "需要人工确认(成功率未达标)"


def upd_xlsx(api, scenario, total, ok, fail, reason, bug, aicore, card):
    subprocess.run(["python3", UPDATER, "--api", api, "--scenario", scenario,
                    "--total", str(total), "--success", str(ok), "--fail", str(fail),
                    "--fail_reason", reason[:250], "--has_bug", bug[:60],
                    "--has_aicore", "是" if aicore else "否", "--card", str(card),
                    "--analyzed", "是" if bug.startswith("不是bug") else "否"],
                   capture_output=True, text=True)


def swap_dir(fw, sub, fname, rename=None):
    """清空目标场景目录并放入单个 CPP（op/op_multithread 机制）；
    multithread 场景文件必须重命名为 {api}.cpp（执行器按 目录stem+_fuzz 取符号）"""
    d = os.path.join(fw, "src", sub)
    if os.path.exists(d):
        shutil.rmtree(d)
    os.makedirs(d)
    if fname:
        dst = os.path.join(d, rename) if rename else os.path.join(d, os.path.basename(fname))
        shutil.copy(fname, dst)


def make_fw(fw, target):
    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=" + target] if target else []),
                       cwd=fw, env=env, capture_output=True, text=True, timeout=1200)
    return r.returncode == 0, r.stderr[-800:]


def run_scenario(fw, api, scn, device, state):
    d = os.path.join(ROOT, "output", api, "fuzz_logs")
    os.makedirs(d, exist_ok=True)
    cpp = os.path.join(ROOT, "output", api, f"{api}_{'non_contig' if scn == 'non_contig' else scn}.cpp") \
        if scn in ("normal", "non_contig", "error") else None
    if scn == "normal":
        cpp = os.path.join(ROOT, "output", api, f"{api}.cpp")
    if scn == "multithread":
        cpp = os.path.join(ROOT, "output", api, f"{api}_multithread.cpp")
    if scn == "consistent":
        cpp = os.path.join(ROOT, "output", api, f"{api}_consistent.cpp")

    # 编译
    if scn == "sample":
        okc, err = make_fw(fw, None)  # sample_fuzz 特殊：入口不同
        # sample 由独立函数处理
        return None
    if scn in ("normal", "non_contig", "error"):
        swap_dir(fw, "op", cpp)
        okc, err = make_fw(fw, None)
        exe_args = lambda n: [os.path.join(fw, "out", "aclnnAPI_fuzz"),
                              f"{api}{'_non_contig' if scn == 'non_contig' else ''}{'_error' if scn == 'error' else ''}_fuzz"] + ([str(n)] if scn != "error" else [])
    elif scn == "multithread":
        swap_dir(fw, "op_multithread", cpp, rename=f"{api}.cpp")
        okc, err = make_fw(fw, "multithread")
        exe_args = lambda n: [os.path.join(fw, "out", "aclnnAPI_fuzz_multithread"), "5", str(n), str(device), "5", "0"]
    else:  # consistent
        swap_dir(fw, "src_consistent", cpp)
        okc, err = make_fw(fw, "consistent")
        exe_args = lambda n: [os.path.join(fw, "out", "aclnnAPI_fuzz_consistent"), f"{api}_consistent_fuzz", str(n)]
    if not okc:
        state.setdefault(api, {})[scn] = {"status": "compile_fail", "error": err}
        logw(f"    [{scn}] COMPILE_FAIL")
        return "compile_fail"

    # 执行（normal: 预检1000→门禁→全量10000；mt: 调试1000→门禁→全量20000；其余直接跑）
    runs = []
    if scn in ("normal", "multithread"):
        pre_name = f"{api}_fuzz_precheck.log" if scn == "normal" else f"{api}_{scn}_precheck.log"
        full_name = f"{api}_fuzz_output.log" if scn == "normal" else f"{api}_{scn}_fuzz_output.log"
        runs = [(1000, pre_name, 2400), (10000 if scn == "normal" else 20000, full_name, 3600)]
    elif scn == "consistent":
        runs = [(100, f"{api}_consistent_precheck.log", 600), (1000, f"{api}_consistent_fuzz_output.log", 1800)]
    elif scn == "non_contig":
        runs = [(10000, f"{api}_non_contig_fuzz_output.log", 2400)]
    else:  # error：条数 1000（Scenario 轮转全覆盖），超时 1200s
        runs = [(0, f"{api}_error_fuzz_output.log", 1200)]

    result = {}
    gated = False
    for num, logname, tmo in runs:
        logpath = os.path.join(d, logname)
        args = exe_args(num) if (scn != "error" and num) else (
            [os.path.join(fw, "out", "aclnnAPI_fuzz"), f"{api}_error_fuzz"] if scn == "error" else exe_args(num))
        if scn == "error":
            args = [os.path.join(fw, "out", "aclnnAPI_fuzz"), f"{api}_error_fuzz", "1000"]
        rc, before, t0 = run_cmd(fw, args, tmo, logpath, device, remap_visible=(scn != "multithread"))
        plog_name = f"{api}_fuzz_plog.log" if scn == "normal" else f"{api}_{scn}_fuzz_plog.log"
        capture_plog(before, t0, os.path.join(d, plog_name) if not logname.endswith("precheck.log") else "/dev/null")
        st = stat_log(logpath, api)
        result["precheck" if "precheck" in logname else "full"] = {**st, "rc": rc}
        if "precheck" in logname or "调试" in logname:
            total = st["ok"] + st["fail"]  # skip（CPP自身护栏/约束拒绝）不计入执行分母，单独上报
            if rc == -9 or total == 0 or (scn != "consistent" and st["ok"] * 100.0 / max(total, 1) < 95):
                if scn == "consistent" and rc == 0 and total > 0:
                    pass  # consistent 门禁为 mismatch（此处不阻断，全量后判）
                else:
                    gated = True
                    break
    if gated:
        pre = result.get("precheck", {})
        if pre.get("ok", 0) == 0 and pre.get("fail", 0) == 0 and pre.get("skip", 0) > 0:
            reason = f"全部用例被约束拒绝(skip={pre.get('skip')}): CPP输入构造需修复, err={pre.get('err_codes')}"
            bug = "需要人工确认(CPP输入构造需修复)"
        else:
            reason = f"预检未达标: {pre.get('ok', 0)}/{pre.get('ok', 0) + pre.get('fail', 0) + pre.get('skip', 0)} err={pre.get('err_codes')}"
            bug = "需要人工确认(预检门禁未过)"
        status = "needs_review"
        upd_xlsx(api, scn, 1000, pre.get("ok", 0), pre.get("fail", 0) + pre.get("skip", 0), reason, bug, False, device)
        state.setdefault(api, {})[scn] = {"status": "gate_fail", **pre}
        logw(f"    [{scn}] GATE_FAIL {reason[:80]}")
        return "gate_fail"

    full = result.get("full") or result.get("precheck")
    if scn != "consistent" and full.get("ok", 0) == 0 and full.get("fail", 0) == 0 and full.get("skip", 0) > 0:
        num_allskip = 10000 if scn == "non_contig" else (20000 if scn == "multithread" else 1000)
        aicore_allskip = scan_aicore(os.path.join(d, f"{api}_{'fuzz' if scn=='normal' else scn}_fuzz_plog.log"))
        reason = f"全部用例被约束拒绝(skip={full['skip']}), err={full['err_codes']}"
        upd_xlsx(api, scn, num_allskip, 0, full["skip"], reason, "需要人工确认(CPP输入构造需修复)", aicore_allskip, device)
        state.setdefault(api, {})[scn] = {"status": "needs_review", **full}
        logw(f"    [{scn}] ALL_SKIP skip={full['skip']}")
        return "needs_review"
    num = 10000 if scn == "non_contig" else (20000 if scn == "multithread" else (0 if scn == "error" else 1000))
    total = full["ok"] + full["fail"] if full["ok"] + full["fail"] > 0 else (num or full["verify_total"])
    aicore = scan_aicore(os.path.join(d, f"{api}_{scn}_fuzz_plog.log"))
    produced = full.get("ok", 0) + full.get("fail", 0) + full.get("skip", 0)
    planned = num if num else produced
    if scn != "consistent" and produced > 0 and produced * 100 < planned * 50:
        reason = f"用例产出率过低: {produced}/{planned} (ok={full['ok']} fail={full['fail']} skip={full.get('skip',0)}), CPP生成有效用例不足"
        upd_xlsx(api, scn, planned, full["ok"], full["fail"] + full.get("skip", 0), reason,
                 "需要人工确认(用例产出率过低)", aicore, device)
        state.setdefault(api, {})[scn] = {"status": "needs_review", **full}
        logw(f"    [{scn}] LOW_COVERAGE {produced}/{planned}")
        return "needs_review"
    if scn == "consistent":
        c = open(os.path.join(d, f"{api}_consistent_fuzz_output.log"), errors="ignore").read() if os.path.isfile(
            os.path.join(d, f"{api}_consistent_fuzz_output.log")) else ""
        mm = c.count("type=hash_mismatch") + c.count("hash_consistent=mismatch")
        al = c.count("hash_consistent=all")
        status = "done" if mm == 0 else "needs_review"
        bug = "不是bug(hash一致)" if mm == 0 else "需要人工确认(mismatch待甄别)"
        reason = f"{al}条hash一致; mismatch={mm}; 无aicore" if not aicore else f"{al}条hash一致; mismatch={mm}; 有aicore"
        upd_xlsx(api, scn, 1000, al, mm, reason, bug, aicore, device)
    else:
        status, bug = classify(full, scn)
        if scn == "error":
            reason = (f"scenarios={full.get('scen_total',0)} base_ok={full.get('base_ok',0)} "
                      f"ez_pass={full.get('ez_pass',0)} ez_fail={full.get('ez_fail',0)} "
                      f"意外成功={full.get('unexp_success',0)}")
        else:
            reason = f"ok={full['ok']} fail={full['fail']}" + (f" err={full['err_codes']}" if full["err_codes"] else "")
        reason += "; 有aicore" if aicore else "; 无aicore"
        if scn == "error" and full.get("scen_total", 0) > 0:
            upd_xlsx(api, scn, full["scen_total"] - full.get("base_ok", 0), full["ez_pass"],
                     full["ez_fail"] + full.get("unexp_success", 0), reason, bug, aicore, device)
        else:
            upd_xlsx(api, scn, num or full["verify_total"], full["ok"], full["fail"], reason, bug, aicore, device)
    state.setdefault(api, {})[scn] = {"status": status, **full}
    logw(f"    [{scn}] {status} ok={full['ok']} fail={full['fail']} aicore={aicore}")
    return status


def run_sample(fw, api, device, state):
    d = os.path.join(ROOT, "output", api, "fuzz_logs")
    os.makedirs(d, exist_ok=True)
    scpp = os.path.join(fw, "src", "op_sample", f"{api}.cpp")
    if not os.path.isfile(scpp):
        state.setdefault(api, {})["sample"] = {"status": "no_sample"}
        return "no_sample"
    logp = os.path.join(d, f"{api}_sample_output.log")
    rc, before, t0 = run_cmd(fw, [os.path.join(fw, "out", "aclnnAPI_sample_fuzz"), api], 600, logp, device)
    capture_plog(before, t0, os.path.join(d, f"{api}_sample_plog.log"))
    st = "done" if rc == 0 else "needs_review"
    bug = "不是bug(样例通过)" if rc == 0 else "是bug(官方样例失败)" if rc and rc > 0 else "需要人工确认(样例超时)"
    upd_xlsx(api, "sample", 1, 1 if rc == 0 else 0, 0 if rc == 0 else 1,
             f"样例执行 rc={rc}", bug, scan_aicore(os.path.join(d, f"{api}_sample_plog.log")), device)
    state.setdefault(api, {})["sample"] = {"status": st, "rc": rc}
    logw(f"    [sample] rc={rc}")
    return st


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--worker", required=True)
    ap.add_argument("--device", required=True)
    ap.add_argument("--fw", required=True)
    ap.add_argument("--apis", required=True)
    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"scenario_campaign_worker{args.worker}.json")
    state = load_state(state_path)
    logw(f"worker{args.worker} device{args.device} 共{len(apis)} 接口")

    for api in apis:
        logw(f"worker{args.worker} >>> {api}")
        # sample：提取产物已就绪则执行
        try:
            run_sample(fw, api, args.device, state)
        except Exception as e:
            state.setdefault(api, {})["sample"] = {"status": "error", "e": str(e)[:200]}
        save_state(state_path, state)
        for scn in SCN_ORDER:
            if scn == "sample":
                continue
            # 场景完成判定：日志存在或状态存在
            done_logs = {"normal": f"{api}_fuzz_output.log", "non_contig": f"{api}_non_contig_fuzz_output.log",
                         "multithread": f"{api}_multithread_fuzz_output.log", "error": f"{api}_error_fuzz_output.log",
                         "consistent": f"{api}_consistent_fuzz_output.log"}
            dlog = os.path.join(ROOT, "output", api, "fuzz_logs", done_logs[scn])
            if os.path.isfile(dlog) and "FUZZ_RESULT" in open(dlog, errors="ignore").read(20 * 1024 * 1024):
                continue
            cppmap = {"normal": f"{api}.cpp", "non_contig": f"{api}_non_contig.cpp",
                      "multithread": f"{api}_multithread.cpp", "error": f"{api}_error.cpp",
                      "consistent": f"{api}_consistent.cpp"}
            if not os.path.isfile(os.path.join(ROOT, "output", api, cppmap[scn])):
                continue
            try:
                run_scenario(fw, api, scn, args.device, state)
            except Exception as e:
                state.setdefault(api, {})[scn] = {"status": "error", "e": str(e)[:200]}
                logw(f"    [{scn}] ERROR {e}")
            save_state(state_path, state)
    logw(f"worker{args.worker} ALL DONE")


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