"""CharacterBench: checks every setting of the platform's protocol against its source (the paper, else the authors' code).

Each check names the source (paper: arXiv:2412.11912v1 e-print, file and the quoted sentence; code: thu-coai/CharacterBench
at commit 66c113c, file), reads the value there, reads the platform's value, and compares. The list of settings and the
reasoning behind each is docs/复现规范/CharacterBench-核查矩阵.md; this script is its executable part. Exit status 1 when
any check fails.

    curl -L -o cb.tar.gz https://arxiv.org/e-print/2412.11912v1 && mkdir eprint && tar xzf cb.tar.gz -C eprint
    git clone https://github.com/thu-coai/CharacterBench repo && git -C repo checkout 66c113c
    python3 -I tools/experiments/characterbench_spec_check.py eprint repo [val_zh.jsonl val_en.jsonl]

The optional files are the raw judge outputs of characterbench_judge_validation.py (10 samples per annotated response);
with them the self-consistency check is run (the paper's text and its numbers disagree; see C07). `table11 eprint` prints
the paper's role-play prompt as the platform must hold it (JSON), for characterbench_prompt_check.py.
"""
import collections
import hashlib
import json
import math
import os
import re
import statistics
import subprocess
import sys
import urllib.request

ROOT = os.path.abspath(os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", ".."))
EPRINT_SHA256 = {
    "appendix.tex": "efb8b679d170ed587f6ee310e422f50dc60fe82c3819b873d7f04b192fdd7951",
    "aaai25-main.tex": "586c90f013d82faaab6b6a116cc7d6fba2704b3602ab1f815d7a9515d8bcf8c3",
}
FILES = ["memory_consistency", "fact_accuracy", "boundary_consistency", "attribute_consistency_bot",
         "attribute_consistency_human", "behavior_consistency_bot", "behavior_consistency_human", "emotion_self_regulation",
         "empathetic_responsiveness", "morality_stability", "morality_robustness", "human_likeness", "engagement"]
# Paper, Table 3: Pearson (%) of CharacterJudge with and without self-consistency, AVG. zh/en.
TABLE3 = {"sc": {"zh": 68, "en": 64}, "wo_sc": {"zh": 64, "en": 60}}
JUDGE_CONFIG_URL = "https://huggingface.co/thu-coai/CharacterJudge/raw/main/generation_config.json"

results = []


def check(cid, what, source, expected, actual, ok=None):
    ok = (expected == actual) if ok is None else ok
    results.append(ok)
    print(f"{'PASS' if ok else 'FAIL'} {cid} {what}\n     source: {source}\n     source value: {expected!r}"[:2000])
    print(f"     platform:     {actual!r}"[:2000])


def read(path):
    return open(path, encoding="utf-8").read()


def sha256(path):
    return hashlib.sha256(open(path, "rb").read()).hexdigest()


def table11(eprint):
    """The paper's role-play prompt (appendix Table 11) from the LaTeX source: one string per language."""
    src = read(os.path.join(eprint, "appendix.tex"))
    m = re.search(r"\\begin\{tabular\}\{\|l\|\}\s*\\hline(.*?)\\hline\s*\\end\{tabular\}\}\s*\\caption\{Prompt for general-purpose LLMs", src, re.S)
    if not m:
        raise SystemExit("Table 11 not found in appendix.tex")
    zh_part, en_part = m.group(1).split("\\hdashline")
    out = {}
    for lang, part in (("zh", zh_part), ("en", en_part)):
        lines = []
        for raw in part.split("\\\\"):
            t = raw.strip()
            if "\\end{tabular}" in t:
                continue
            t = re.sub(r"\\begin\{CJK\*\}\{UTF8\}\{gbsn\}|\\end\{CJK\*\}", "", t)
            t = t.replace("\\texttt{\\{character\\_profile\\}}", "{character_profile}").replace("\\texttt{\\{dialogue\\_context\\}}", "{dialogue_context}")
            # \lbrack eats the space after it (a control word); the space before \rbrack is the paper's.
            t = re.sub(r"\\lbrack\s*", "[", t).replace("\\rbrack", "]")
            wrapped = t.lstrip().startswith("\\quad")
            t = re.sub(r"^(\\quad\s*|\\\s)+", "", t.strip()).strip()
            if wrapped and lines:
                # A line the table wraps for width: Chinese continues without a space, English with one.
                lines[-1] = lines[-1] + ("" if lang == "zh" else " ") + t
            else:
                lines.append(t)
        while lines and not lines[0]:
            lines.pop(0)
        while lines and not lines[-1]:
            lines.pop()
        # The English table wraps its first sentence without \quad.
        if lang == "en" and lines[1] == "character and the user.":
            lines[0:2] = [lines[0] + " " + lines[1]]
        out[lang] = "\n".join(lines)
    return out


def platform():
    js = """
const b = await import('./server/src/benchmarks/characterbench/build.ts');
const i = await import('./server/src/benchmarks/characterbench/index.ts');
const fs = await import('node:fs');
const a = i.characterBench;
console.log(JSON.stringify({
  roleplay: b.ROLEPLAY, files: b.FILES, judge: a.protocol.judges[0], replyAttempts: a.replyAttempts,
  lastAttemptExcludes: !!a.lastAttemptExcludes, exactSessionScores: !!a.exactSessionScores,
  scoring: a.defaultScoring(), version: a.protocol.version,
  indexSource: fs.readFileSync('server/src/benchmarks/characterbench/index.ts', 'utf8'),
}));
"""
    r = subprocess.run(["node", "--disable-warning=ExperimentalWarning", "--input-type=module", "-e", js], cwd=ROOT, capture_output=True, text=True)
    if r.returncode != 0:
        raise SystemExit(r.stderr)
    return json.loads(r.stdout)


def paper_rows():
    js = "const m = await import('./server/src/benchmarks/characterbench/published.ts'); console.log(JSON.stringify(m.PAPER_ROWS));"
    r = subprocess.run(["node", "--disable-warning=ExperimentalWarning", "--input-type=module", "-e", js], cwd=ROOT, capture_output=True, text=True)
    if r.returncode != 0:
        raise SystemExit(r.stderr)
    return json.loads(r.stdout)


def pearson(a, b):
    ma, mb = statistics.mean(a), statistics.mean(b)
    num = sum((x - ma) * (y - mb) for x, y in zip(a, b))
    den = math.sqrt(sum((x - ma) ** 2 for x in a) * sum((y - mb) ** 2 for y in b))
    return num / den if den else None


def read_official(text):
    """all_eval.py check(): the first line when there is a line break, else the whole output; 3.0 when not a number."""
    t = text.split("\n")[0] if "\n" in text else text
    try:
        return float(t)
    except ValueError:
        return 3.0


def main(eprint, repo, val):
    p = platform()
    main_tex = read(os.path.join(eprint, "aaai25-main.tex"))
    flat = re.sub(r"\s+", " ", main_tex)

    for f, h in EPRINT_SHA256.items():
        check("C00", f"paper source {f} is the audited one", "arXiv e-print 2412.11912v1", h, sha256(os.path.join(eprint, f)))

    t11 = table11(eprint)
    for lang in ("zh", "en"):
        check("C01" if lang == "zh" else "C02", f"role-play prompt ({lang}) is the paper's Table 11", "appendix.tex, Table 11 (label roleplay_prompt)", t11[lang], p["roleplay"][lang])
    check("C03", "length instruction", "appendix.tex, Table 11",
          ["注意：回复字数要控制在15字以内。", "Note: The length of your response should be limited to 15 words."],
          [l for lang in ("zh", "en") for l in p["roleplay"][lang].split("\n") if l.startswith(("注意", "Note"))])
    check("C04", "prompt placeholders (no user profile, no output format)", "appendix.tex, Table 11 caption: “{character_profile} and {dialogue_context} are placeholders”",
          ["character_profile", "dialogue_context"], sorted(set(re.findall(r"\{(\w+)\}", p["roleplay"]["zh"] + p["roleplay"]["en"]))))

    scale_parts = ["2-point scale for Morality Stability and Morality Robustness", "3-point scale for Boundary Consistency and Behavior Consistency (human query)",
                   "5-point scale for Human-likeness and Engagement", "4-point scale for other dimensions"]
    paper_scale = {f: 2 if f.startswith("morality") else 3 if f in ("boundary_consistency", "behavior_consistency_human") else 5 if f in ("human_likeness", "engagement") else 4 for f in FILES}
    check("C05a", "annotation scales stated in the paper", "aaai25-main.tex, Response Generation and Human Annotation", scale_parts, [x for x in scale_parts if x in flat])
    check("C05b", "column maxima = paper scales", "aaai25-main.tex (C05a)", paper_scale, {f["file"]: f["max"] for f in p["files"]})
    gold_max = {f: max(float(x["annotation_score"]) for x in json.load(open(os.path.join(repo, "eval_data/raw_data", f"{f}_test.json"), encoding="utf-8"))) for f in FILES}
    check("C05c", "column maxima = max human score per test file (all_eval.py divides by max(golds))", "code: all_eval.py, eval_data/raw_data",
          {f: float(v) for f, v in paper_scale.items()}, gold_max)

    check("C06", "judge samples per response", "aaai25-main.tex: “use self-consistency to generate 10 outcomes”", 10, p["judge"]["samples"],
          ok=("self-consistency to generate 10 outcomes" in flat and p["judge"]["samples"] == 10))
    normal = "means / mx * 5" in read(os.path.join(repo, "all_eval.py"))
    check("C08", "normalization to 5 points", "aaai25-main.tex: “We normalize the scores of all dimensions to a 5-point scale”; code: all_eval.py “means / mx * 5”",
          "score / column maximum × 5", p["scoring"]["normalization"], ok=normal and "normalize the scores of all dimensions to a 5-point scale" in flat and p["scoring"]["normalization"] == "score / column maximum × 5")

    rows, within, off = paper_rows(), 0, []
    cols = ["mc", "fa", "bck", "acb", "ach", "bcpb", "bcph", "es", "er", "ms", "mr", "hl", "eg"]
    for r in rows:
        for lang in ("zh", "en"):
            v = r.get(lang)
            if not v or v.get("avg") is None:
                continue
            m = sum(v[c] for c in cols) / 13
            if abs(m - v["avg"]) <= 0.0051:
                within += 1
            else:
                off.append(f"{r['name']} {lang}: {v['avg']} vs {m:.3f}")
    total = within + len(off)
    check("C09", f"AVG = mean of the 13 columns (code: all_eval.py average over files); paper rows within rounding: {within}/{total}; others: {off}",
          "code: all_eval.py; paper Table 4/10 printed values", True, within / total >= 0.75, ok=within / total >= 0.75)

    gen = read(os.path.join(repo, "model_generation_qwen_chat.py"))
    check("C10", "judge request: system message and max_tokens", "code: model_generation_qwen_chat.py create_payload",
          ["You are a helpful assistant.", 512],
          ["You are a helpful assistant." if "content: 'You are a helpful assistant.'" in p["indexSource"] else None, p["judge"]["maxTokens"]],
          ok=('"You are a helpful assistant."' in gen and '"max_tokens": 512' in gen and "content: 'You are a helpful assistant.'" in p["indexSource"] and p["judge"]["maxTokens"] == 512))
    try:
        cfg = json.loads(urllib.request.urlopen(urllib.request.Request(JUDGE_CONFIG_URL, headers={"User-Agent": "Mozilla/5.0"}), timeout=30).read())
        src = {"temperature": cfg["temperature"], "top_p": cfg["top_p"], "top_k": cfg["top_k"], "repetition_penalty": cfg["repetition_penalty"]}
        plat = {"temperature": p["judge"]["temperature"], **{k: p["judge"]["extra"][k] for k in ("top_p", "top_k", "repetition_penalty")}}
        check("C11", "judge sampling settings", "released checkpoint: generation_config.json (the paper gives none)", src, plat)
    except Exception as e:  # offline
        print(f"SKIP C11 judge sampling settings ({e})")
    check("C12", "attempts per sample", "code: process.py “max_retry = 5” (the paper gives none)", 5, p["replyAttempts"], ok="max_retry = 5" in read(os.path.join(repo, "process.py")) and p["replyAttempts"] == 5)
    proc = read(os.path.join(repo, "process.py"))
    check("C15", "a sample read only on its last attempt is dropped too", "code: process.py “if retry == max_retry-1: continue” after the retry loop (the paper gives none)",
          True, p["lastAttemptExcludes"], ok=("if retry == max_retry-1:" in proc) == p["lastAttemptExcludes"])
    ev = read(os.path.join(repo, "all_eval.py"))
    check("C16", "sample scores are averaged unrounded; rounding only for display", "code: all_eval.py “means = np.mean(preds)” … “means / mx * 5”",
          True, p["exactSessionScores"], ok="means = np.mean(preds)" in ev and p["exactSessionScores"])
    n = {f: len(json.load(open(os.path.join(repo, "eval_data/raw_data", f"{f}_test.json"), encoding="utf-8"))) for f in FILES}
    check("C13", "test set: 3,250 samples, 250 per column (125 in-domain + 125 out-of-domain)", "aaai25-main.tex Table 2 and Statistics", 3250, sum(n.values()), ok=sum(n.values()) == 3250 and set(n.values()) == {250})
    bch = json.load(open(os.path.join(repo, "eval_data/raw_data/behavior_consistency_human_test.json"), encoding="utf-8"))
    inside = sum(1 for x in bch if x["response_messages"]["character_style"].strip() in x["character_profile"])
    check("C14", "behavior consistency (human): the new behaviour P'' is in the profile the model receives (P')", "aaai25-main.tex, Persona: “P' and C_T are used to generate a response”", len(bch), inside)

    if val:
        out = {}
        for lang, path in zip(("zh", "en"), val):
            recs = [json.loads(l) for l in open(path, encoding="utf-8")]
            by = collections.defaultdict(list)
            for r in recs:
                by[r["file"]].append(r)
            agg = {"mean": [], "vote": []}
            for rs in by.values():
                h = [float(r["human"]) for r in rs]
                sc = [[read_official(t) for t in r["sc"]] for r in rs]
                agg["mean"].append(pearson([statistics.mean(v) for v in sc], h))
                # Majority vote; a tie takes the mean of the tied values (the most favourable tie rule measured).
                votes = []
                for v in sc:
                    c = collections.Counter(v)
                    top = max(c.values())
                    votes.append(statistics.mean([x for x in c if c[x] == top]))
                agg["vote"].append(pearson(votes, h))
            out[lang] = {k: round(statistics.mean(v) * 100, 1) for k, v in agg.items()}
        near = lambda k: all(abs(out[l][k] - TABLE3["sc"][l]) <= 1.5 for l in ("zh", "en"))
        check("C07", f"self-consistency aggregation: the paper says majority vote, its Table 3 (68/64 with SC, 64/60 without) is matched by: mean {out['zh']['mean']}/{out['en']['mean']}, vote {out['zh']['vote']}/{out['en']['vote']}",
              "aaai25-main.tex Table 3; platform: mean of the 10 samples", "mean", "mean" if near("mean") and not near("vote") else "undetermined")
    else:
        print("SKIP C07 self-consistency aggregation (no validation files given)")

    print(f"\n{sum(results)}/{len(results)} checks passed; protocol {p['version']}")
    sys.exit(0 if all(results) else 1)


if __name__ == "__main__":
    if sys.argv[1] == "table11":
        print(json.dumps(table11(sys.argv[2]), ensure_ascii=False))
    else:
        main(sys.argv[1], sys.argv[2], sys.argv[3:5] if len(sys.argv) >= 5 else None)
