scripts/data_building/build_main_dataset.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import csv
import hashlib
import json
import os
import math
import re
import shutil
import sys
from collections import Counter, defaultdict
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional, Tuple

ROOT = Path(os.environ.get("ENHANCED_REPLICA_DATA_ROOT", r"F:\codex开发\研究"))
ROUTE_ROOT = ROOT / "研究成果" / "实验路线" / "20260408_SOTA路线重构"
DATASET_ROOT = ROUTE_ROOT / "data" / "dataset"

SRC_HUMAN = ROOT / "data" / "human"
SRC_QUARANTINE = ROOT / "data" / "human_quarantine"
SRC_RAW_EXTERNAL = ROOT / "data" / "raw_external"

NLPCC_DIR = SRC_RAW_EXTERNAL / "NLPCC-2025-Task1-main" / "data"
HC3_PLUS_DIR = SRC_RAW_EXTERNAL / "human_longform_20260406" / "HC3_Plus" / "raw" / "files" / "zh"
HC3_RAW_FILE = SRC_RAW_EXTERNAL / "human_longform_20260406" / "HC3" / "raw" / "modelscope_zh" / "all.jsonl"

QUARANTINE_HC_DIRS = [
    SRC_QUARANTINE / "failed_platforms_20260406" / "weibo_quality_v2" / "high_confidence",
    SRC_QUARANTINE / "failed_platforms_20260406" / "weibo_quality_v3" / "exports" / "high_confidence",
]

P2_CLTS_DIR = SRC_RAW_EXTERNAL / "CLTS数据集"
P2_ZHIHU_DIR = SRC_RAW_EXTERNAL / "huggingface_zhihu"

REQUIRED_FIELDS = ["record_id", "text", "label", "source", "split", "length_char", "topic", "model_slug"]
SPLITS = ["train", "dev", "test"]
LABEL_MAP = {"human": 0, "ai": 1}
MIN_CHAR_LEN = 30

LENGTH_BUCKETS = [
    (0, 29, "000_029"),
    (30, 99, "030_099"),
    (100, 299, "100_299"),
    (300, 599, "300_599"),
    (600, 999, "600_999"),
    (1000, 1499, "1000_1499"),
    (1500, 2199, "1500_2199"),
    (2200, 10**9, "2200_plus"),
]

EXCLUSION_LOG_ROWS: List[Dict[str, str]] = []
OVERLAP_AUDIT: Dict[str, Dict[str, int]] = {}


def now_str() -> str:
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")


def iso_now() -> str:
    return datetime.now().isoformat(timespec="seconds")


def log(msg: str) -> None:
    print(f"[{now_str()}] {msg}")


def ensure_dir(path: Path) -> None:
    path.mkdir(parents=True, exist_ok=True)


def write_json(path: Path, obj: dict) -> None:
    ensure_dir(path.parent)
    with path.open("w", encoding="utf-8") as f:
        json.dump(obj, f, ensure_ascii=False, indent=2)


def write_jsonl(path: Path, records: List[dict]) -> None:
    ensure_dir(path.parent)
    with path.open("w", encoding="utf-8", newline="\n") as f:
        for rec in records:
            out = {k: rec[k] for k in REQUIRED_FIELDS}
            f.write(json.dumps(out, ensure_ascii=False) + "\n")


def write_csv(path: Path, fieldnames: List[str], rows: List[dict]) -> None:
    ensure_dir(path.parent)
    with path.open("w", encoding="utf-8-sig", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        for r in rows:
            writer.writerow(r)


def read_text_with_fallback(path: Path) -> str:
    raw = path.read_bytes()
    for enc in ("utf-8-sig", "utf-8", "gb18030"):
        try:
            return raw.decode(enc)
        except UnicodeDecodeError:
            continue
    return raw.decode("utf-8", errors="replace")


def load_json_with_fallback(path: Path):
    return json.loads(read_text_with_fallback(path))


def iter_jsonl(path: Path):
    text = read_text_with_fallback(path)
    for lineno, line in enumerate(text.splitlines(), 1):
        line = line.strip()
        if not line:
            continue
        try:
            yield lineno, json.loads(line)
        except json.JSONDecodeError:
            EXCLUSION_LOG_ROWS.append(
                {
                    "dataset_id": "GLOBAL",
                    "stage": "read_jsonl",
                    "source_file": str(path),
                    "record_ref": f"line_{lineno}",
                    "label": "",
                    "drop_reason": "json_decode_error",
                    "raw_length": "",
                    "clean_length": "",
                    "text_hash": "",
                }
            )


def normalize_label(label) -> Optional[int]:
    if label is None:
        return None
    if isinstance(label, bool):
        return int(label)
    if isinstance(label, int):
        return 1 if label == 1 else 0 if label == 0 else None
    s = str(label).strip().lower()
    if s in {"0", "human", "h"}:
        return 0
    if s in {"1", "ai", "machine", "gpt"}:
        return 1
    return None


def text_hash(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def non_ws_len(text: str) -> int:
    return len(re.sub(r"\s+", "", text))


def length_bucket(length_char: int) -> str:
    for lo, hi, name in LENGTH_BUCKETS:
        if lo <= length_char <= hi:
            return name
    return "unknown"
MD_CODE_BLOCK_RE = re.compile(r"```[\s\S]*?```", flags=re.MULTILINE)
MD_INLINE_CODE_RE = re.compile(r"`([^`]+)`")
MD_IMG_RE = re.compile(r"!\[([^\]]*)\]\(([^)]+)\)")
MD_LINK_RE = re.compile(r"\[([^\]]+)\]\(([^)]+)\)")
MD_HEADING_RE = re.compile(r"^\s{0,3}#{1,6}\s*")
MD_QUOTE_RE = re.compile(r"^\s{0,3}>\s?")
MD_LIST_RE = re.compile(r"^\s{0,3}(?:[-*+]|\d+\.)\s+")


def clean_text(text: str) -> str:
    if text is None:
        return ""
    t = str(text)
    t = t.replace("\r\n", "\n").replace("\r", "\n")
    t = t.replace("\\n", "\n").replace("\\r", "\n")
    t = re.sub(r"(?<=\S)\s*/n\s*(?=\S)", "\n", t)
    t = re.sub(r"^\s*/n\s*$", "", t, flags=re.MULTILINE)
    t = t.replace("/n/n", "\n\n")
    t = re.sub(
        r"(?<=[\u4e00-\u9fffA-Za-z0-9，。！？；：,.!?;:])\s*/n\s*(?=[\u4e00-\u9fffA-Za-z0-9])",
        "\n",
        t,
    )

    t = MD_CODE_BLOCK_RE.sub(" ", t)
    t = MD_INLINE_CODE_RE.sub(r"\1", t)
    t = MD_IMG_RE.sub(r"\1", t)
    t = MD_LINK_RE.sub(r"\1", t)

    lines = []
    for line in t.split("\n"):
        line = MD_HEADING_RE.sub("", line)
        line = MD_QUOTE_RE.sub("", line)
        line = MD_LIST_RE.sub("", line)
        line = line.replace("**", "").replace("__", "")
        line = line.replace("`", "")
        lines.append(line.strip())

    t = "\n".join(lines)
    t = t.replace("\u3000", " ")
    t = re.sub(r"[ \t\f\v]+", " ", t)
    t = re.sub(r"\n{3,}", "\n\n", t)
    t = re.sub(r"[ ]+\n", "\n", t)
    t = re.sub(r"\n[ ]+", "\n", t)
    return t.strip()


def severe_artifact_reason(cleaned_text: str, length_char: int) -> Optional[str]:
    if not cleaned_text:
        return "empty_after_clean"
    if re.fullmatch(r"[\W_]+", cleaned_text):
        return "non_content_symbols_only"

    content_chars = re.findall(r"[\u4e00-\u9fffA-Za-z0-9]", cleaned_text)
    ratio = len(content_chars) / max(len(cleaned_text), 1)
    if length_char >= MIN_CHAR_LEN and ratio < 0.10:
        return "severe_artifact_low_content_ratio"

    compact = re.sub(r"\s+", "", cleaned_text)
    if compact:
        most_freq = Counter(compact).most_common(1)[0][1]
        if len(compact) >= 120 and most_freq / len(compact) > 0.60:
            return "severe_artifact_repetitive_chars"
    return None


def add_exclusion(
    dataset_id: str,
    stage: str,
    source_file: str,
    record_ref: str,
    label,
    drop_reason: str,
    raw_length,
    clean_length,
    text_hash_value: str = "",
) -> None:
    EXCLUSION_LOG_ROWS.append(
        {
            "dataset_id": dataset_id,
            "stage": stage,
            "source_file": source_file,
            "record_ref": record_ref,
            "label": "" if label is None else str(label),
            "drop_reason": drop_reason,
            "raw_length": "" if raw_length is None else str(raw_length),
            "clean_length": "" if clean_length is None else str(clean_length),
            "text_hash": text_hash_value,
        }
    )


def build_record(
    ds_id: str,
    stage: str,
    source_file: Path,
    record_ref: str,
    raw_text,
    label: int,
    source: str,
    split: str,
    topic: str = "",
    model_slug: str = "",
    seen_hashes: Optional[set] = None,
) -> Optional[dict]:
    raw_text_str = "" if raw_text is None else str(raw_text)
    raw_len = non_ws_len(raw_text_str)
    cleaned = clean_text(raw_text_str)
    clean_len = non_ws_len(cleaned)

    if not cleaned:
        add_exclusion(ds_id, stage, str(source_file), record_ref, label, "empty_text", raw_len, clean_len)
        return None
    if clean_len < MIN_CHAR_LEN:
        add_exclusion(ds_id, stage, str(source_file), record_ref, label, f"too_short_lt_{MIN_CHAR_LEN}", raw_len, clean_len)
        return None

    reason = severe_artifact_reason(cleaned, clean_len)
    if reason:
        add_exclusion(ds_id, stage, str(source_file), record_ref, label, reason, raw_len, clean_len)
        return None

    h = text_hash(cleaned)
    if seen_hashes is not None:
        if h in seen_hashes:
            add_exclusion(ds_id, stage, str(source_file), record_ref, label, "duplicate_text", raw_len, clean_len, h)
            return None
        seen_hashes.add(h)

    return {
        "record_id": "",
        "text": cleaned,
        "label": int(label),
        "source": source,
        "split": split,
        "length_char": clean_len,
        "topic": "" if topic is None else str(topic),
        "model_slug": "" if model_slug is None else str(model_slug),
        "_hash": h,
    }


def split_counts(records: List[dict]) -> Dict[str, int]:
    c = Counter(r["split"] for r in records)
    return {k: int(c.get(k, 0)) for k in SPLITS}


def label_counts(records: List[dict]) -> Dict[str, int]:
    c = Counter(r["label"] for r in records)
    return {"0": int(c.get(0, 0)), "1": int(c.get(1, 0))}


def write_distribution_files(ds_dir: Path, records: List[dict]) -> None:
    label_rows = []
    lc = Counter((r["split"], r["label"]) for r in records)
    for sp in SPLITS:
        for lb in (0, 1):
            label_rows.append({"split": sp, "label": lb, "count": lc.get((sp, lb), 0)})
    label_rows.append({"split": "all", "label": 0, "count": sum(r["label"] == 0 for r in records)})
    label_rows.append({"split": "all", "label": 1, "count": sum(r["label"] == 1 for r in records)})
    write_csv(ds_dir / "label_dist.csv", ["split", "label", "count"], label_rows)

    source_rows = []
    sc = Counter((r["split"], r["source"], r["label"]) for r in records)
    for (sp, src, lb), cnt in sorted(sc.items()):
        source_rows.append({"split": sp, "source": src, "label": lb, "count": cnt})
    write_csv(ds_dir / "source_dist.csv", ["split", "source", "label", "count"], source_rows)

    length_rows = []
    lbc = Counter((r["split"], length_bucket(r["length_char"])) for r in records)
    for sp in SPLITS:
        for _, _, b in LENGTH_BUCKETS:
            length_rows.append({"split": sp, "length_bucket": b, "count": lbc.get((sp, b), 0)})
    write_csv(ds_dir / "length_dist.csv", ["split", "length_bucket", "count"], length_rows)

def finalize_dataset(
    ds_id: str,
    ds_dir: Path,
    records_by_split: Dict[str, List[dict]],
    build_note: str,
    source_inputs: List[str],
) -> dict:
    ensure_dir(ds_dir)
    all_records: List[dict] = []

    for sp in SPLITS:
        recs = records_by_split.get(sp, [])
        for idx, rec in enumerate(recs, 1):
            rec["split"] = sp
            rec["record_id"] = f"{ds_id.lower()}_{sp}_{idx:07d}"
        recs.sort(key=lambda x: x["record_id"])
        write_jsonl(ds_dir / f"{sp}.jsonl", recs)
        all_records.extend(recs)

    # Distribution stats are now covered comprehensively in 审阅报告.md;
    # skip generating intermediate CSVs to keep dataset directories clean.
    # write_distribution_files(ds_dir, all_records)

    manifest = {
        "dataset_id": ds_id,
        "created_at": iso_now(),
        "schema_fields": REQUIRED_FIELDS,
        "label_mapping": LABEL_MAP,
        "build_note": build_note,
        "source_inputs": source_inputs,
        "record_count_total": len(all_records),
        "record_count_by_split": split_counts(all_records),
        "record_count_by_label": label_counts(all_records),
        "source_set": sorted({r["source"] for r in all_records}),
    }
    write_json(ds_dir / "manifest.json", manifest)
    return manifest


def stratified_split(records: List[dict], ratios: Tuple[float, float, float], strata_fn) -> Dict[str, List[dict]]:
    if not records:
        return {"train": [], "dev": [], "test": []}
    train_ratio, dev_ratio, test_ratio = ratios
    if not math.isclose(train_ratio + dev_ratio + test_ratio, 1.0, rel_tol=1e-9):
        raise ValueError("ratios must sum to 1.0")

    grouped = defaultdict(list)
    for rec in records:
        grouped[strata_fn(rec)].append(rec)

    out = {"train": [], "dev": [], "test": []}
    for _, group in grouped.items():
        group = sorted(group, key=lambda x: x["_hash"])
        n = len(group)
        n_train = int(n * train_ratio)
        n_dev = int(n * dev_ratio)
        n_test = n - n_train - n_dev
        out["train"].extend(group[:n_train])
        out["dev"].extend(group[n_train : n_train + n_dev])
        out["test"].extend(group[n_train + n_dev : n_train + n_dev + n_test])

    for sp in SPLITS:
        out[sp].sort(key=lambda x: x["_hash"])
    return out


def collect_source_inventory(source_roots: Dict[str, Path]) -> dict:
    inventory = {"generated_at": iso_now(), "roots": {}, "global_digest": ""}
    global_hasher = hashlib.sha256()

    for name, base in source_roots.items():
        if not base.exists():
            inventory["roots"][name] = {
                "path": str(base),
                "exists": False,
                "file_count": 0,
                "total_size": 0,
                "mtime_min_ns": None,
                "mtime_max_ns": None,
                "digest": "",
            }
            continue

        file_count = 0
        total_size = 0
        mtime_min = None
        mtime_max = None
        local_hasher = hashlib.sha256()

        for p in sorted(base.rglob("*")):
            if not p.is_file():
                continue
            st = p.stat()
            rel = str(p.relative_to(base)).replace("\\", "/")
            signature = f"{name}|{rel}|{st.st_size}|{st.st_mtime_ns}"
            local_hasher.update(signature.encode("utf-8"))
            global_hasher.update(signature.encode("utf-8"))

            file_count += 1
            total_size += int(st.st_size)
            mtime_min = st.st_mtime_ns if mtime_min is None else min(mtime_min, st.st_mtime_ns)
            mtime_max = st.st_mtime_ns if mtime_max is None else max(mtime_max, st.st_mtime_ns)

        inventory["roots"][name] = {
            "path": str(base),
            "exists": True,
            "file_count": file_count,
            "total_size": total_size,
            "mtime_min_ns": mtime_min,
            "mtime_max_ns": mtime_max,
            "digest": local_hasher.hexdigest(),
        }

    inventory["global_digest"] = global_hasher.hexdigest()
    return inventory


def extract_text_from_obj(obj: dict) -> str:
    if not isinstance(obj, dict):
        return ""
    for key in ("text", "content", "body", "answer", "article", "message"):
        v = obj.get(key)
        if isinstance(v, str) and v.strip():
            return v
    return ""


def iter_dataset_records(ds_dir: Path):
    for sp in SPLITS:
        fp = ds_dir / f"{sp}.jsonl"
        if not fp.exists():
            continue
        for _, rec in iter_jsonl(fp):
            rec["split"] = sp
            yield rec


def coerce_answer_text(ans) -> str:
    if isinstance(ans, str):
        return ans
    if isinstance(ans, dict):
        for k in ("text", "answer", "content", "body"):
            v = ans.get(k)
            if isinstance(v, str) and v.strip():
                return v
    if isinstance(ans, list):
        return "\n".join(str(x) for x in ans if str(x).strip())
    return str(ans) if ans is not None else ""

def build_ds01_nlpcc() -> Tuple[Path, dict, set]:
    ds_id = "DS01_NLPCC_core_v1"
    ds_dir = DATASET_ROOT / "10_P0_主线必用" / ds_id
    log(f"Building {ds_id}")

    files = {
        "train": NLPCC_DIR / "train.json",
        "dev": NLPCC_DIR / "dev.json",
        "test": NLPCC_DIR / "test_with_label.json",
    }
    seen_hashes = set()
    records_by_split = {"train": [], "dev": [], "test": []}

    for sp, fp in files.items():
        rows = load_json_with_fallback(fp)
        for idx, row in enumerate(rows):
            lb = normalize_label(row.get("label"))
            if lb is None:
                add_exclusion(ds_id, "parse_nlpcc", str(fp), f"{sp}_{idx}", None, "invalid_label", "", "", "")
                continue
            rec = build_record(
                ds_id=ds_id,
                stage="clean_nlpcc",
                source_file=fp,
                record_ref=f"{sp}_{idx}",
                raw_text=row.get("text", ""),
                label=lb,
                source="nlpcc",
                split=sp,
                topic=row.get("source", ""),
                model_slug=row.get("model", ""),
                seen_hashes=seen_hashes,
            )
            if rec is not None:
                records_by_split[sp].append(rec)

    manifest = finalize_dataset(
        ds_id,
        ds_dir,
        records_by_split,
        build_note="NLPCC official train/dev/test_with_label with unified cleaning + dedupe.",
        source_inputs=[str(files["train"]), str(files["dev"]), str(files["test"])],
    )
    return ds_dir, manifest, seen_hashes


def build_ds02_hc3plus() -> Tuple[Path, dict]:
    ds_id = "DS02_HC3Plus_clean_v1"
    ds_dir = DATASET_ROOT / "20_P1_混合标签集" / ds_id
    log(f"Building {ds_id}")

    file_map = {
        "train": [HC3_PLUS_DIR / "train.jsonl"],
        "dev": [HC3_PLUS_DIR / "val_hc3_QA.jsonl", HC3_PLUS_DIR / "val_hc3_si.jsonl"],
        "test": [HC3_PLUS_DIR / "test_hc3_QA.jsonl", HC3_PLUS_DIR / "test_hc3_si.jsonl"],
    }
    seen_hashes = set()
    records_by_split = {"train": [], "dev": [], "test": []}

    for sp, fplist in file_map.items():
        for fp in fplist:
            for lineno, row in iter_jsonl(fp):
                lb = normalize_label(row.get("label"))
                if lb is None:
                    add_exclusion(ds_id, "parse_hc3_plus", str(fp), f"{lineno}", None, "invalid_label", "", "", "")
                    continue
                rec = build_record(
                    ds_id=ds_id,
                    stage="clean_hc3_plus",
                    source_file=fp,
                    record_ref=f"line_{lineno}",
                    raw_text=row.get("text", ""),
                    label=lb,
                    source="hc3_plus",
                    split=sp,
                    topic=fp.stem,
                    model_slug="",
                    seen_hashes=seen_hashes,
                )
                if rec is not None:
                    records_by_split[sp].append(rec)

    manifest = finalize_dataset(
        ds_id,
        ds_dir,
        records_by_split,
        build_note="HC3_Plus train + merged val->dev + merged test with unified cleaning + dedupe.",
        source_inputs=[str(x) for arr in file_map.values() for x in arr],
    )
    return ds_dir, manifest


def build_ds03_hc3raw_expanded() -> Tuple[Path, dict]:
    ds_id = "DS03_HC3raw_expanded_v1"
    ds_dir = DATASET_ROOT / "20_P1_混合标签集" / ds_id
    log(f"Building {ds_id}")

    seen_hashes = set()
    staged_records: List[dict] = []

    for lineno, row in iter_jsonl(HC3_RAW_FILE):
        topic = row.get("source", "")
        human_answers = row.get("human_answers") or []
        ai_answers = row.get("chatgpt_answers") or []
        if not isinstance(human_answers, list):
            human_answers = [human_answers]
        if not isinstance(ai_answers, list):
            ai_answers = [ai_answers]

        for idx, ans in enumerate(human_answers):
            rec = build_record(
                ds_id=ds_id,
                stage="expand_hc3_raw_human",
                source_file=HC3_RAW_FILE,
                record_ref=f"line_{lineno}_h_{idx}",
                raw_text=coerce_answer_text(ans),
                label=0,
                source="hc3_raw",
                split="train",
                topic=topic,
                model_slug="",
                seen_hashes=seen_hashes,
            )
            if rec is not None:
                rec["_stratum"] = f"{topic}|0"
                staged_records.append(rec)

        for idx, ans in enumerate(ai_answers):
            rec = build_record(
                ds_id=ds_id,
                stage="expand_hc3_raw_ai",
                source_file=HC3_RAW_FILE,
                record_ref=f"line_{lineno}_a_{idx}",
                raw_text=coerce_answer_text(ans),
                label=1,
                source="hc3_raw",
                split="train",
                topic=topic,
                model_slug="chatgpt",
                seen_hashes=seen_hashes,
            )
            if rec is not None:
                rec["_stratum"] = f"{topic}|1"
                staged_records.append(rec)

    split_map = stratified_split(staged_records, (0.8, 0.1, 0.1), strata_fn=lambda r: r["_stratum"])
    for sp in SPLITS:
        for rec in split_map[sp]:
            rec["split"] = sp
            rec.pop("_stratum", None)

    manifest = finalize_dataset(
        ds_id,
        ds_dir,
        split_map,
        build_note="Expanded HC3 raw answers (human_answers/chatgpt_answers), then stratified hash split 80/10/10.",
        source_inputs=[str(HC3_RAW_FILE)],
    )
    return ds_dir, manifest

def build_ds04_human_pools_merged() -> Tuple[Path, dict]:
    ds_id = "DS04_Human_pools_merged_v1"
    ds_dir = DATASET_ROOT / "30_P2_单标签原料池" / ds_id
    log(f"Building {ds_id}")

    seen_hashes = set()
    staged_records: List[dict] = []

    for fp in sorted(SRC_HUMAN.rglob("*.json")):
        try:
            row = load_json_with_fallback(fp)
        except Exception:
            add_exclusion(ds_id, "read_human", str(fp), "", 0, "json_decode_error", "", "", "")
            continue
        text = extract_text_from_obj(row)
        rel = fp.relative_to(SRC_HUMAN)
        top = rel.parts[0] if rel.parts else "unknown"
        rec = build_record(
            ds_id=ds_id,
            stage="clean_human",
            source_file=fp,
            record_ref=str(rel).replace("\\", "/"),
            raw_text=text,
            label=0,
            source="human",
            split="train",
            topic=row.get("topic", top) if isinstance(row, dict) else top,
            model_slug="",
            seen_hashes=seen_hashes,
        )
        if rec is not None:
            rec["_stratum"] = f"{top}|0"
            staged_records.append(rec)

    for base in QUARANTINE_HC_DIRS:
        version = "v2" if "v2" in str(base).lower() else "v3"
        for fp in sorted(base.rglob("*.json")):
            try:
                row = load_json_with_fallback(fp)
            except Exception:
                add_exclusion(ds_id, "read_quarantine", str(fp), "", 0, "json_decode_error", "", "", "")
                continue
            text = extract_text_from_obj(row)
            rel = fp.relative_to(base)
            rec = build_record(
                ds_id=ds_id,
                stage="clean_quarantine_hc",
                source_file=fp,
                record_ref=f"{version}/{str(rel).replace('\\', '/')}",
                raw_text=text,
                label=0,
                source="quarantine_hc",
                split="train",
                topic=row.get("topic", version) if isinstance(row, dict) else version,
                model_slug="",
                seen_hashes=seen_hashes,
            )
            if rec is not None:
                rec["_stratum"] = f"{version}|0"
                staged_records.append(rec)

    split_map: Dict[str, List[dict]] = {"train": [], "dev": [], "test": []}
    for rec in staged_records:
        rec["split"] = "train"
        rec.pop("_stratum", None)
        split_map["train"].append(rec)

    manifest = finalize_dataset(
        ds_id,
        ds_dir,
        split_map,
        build_note="Merged human pool from data/human (core) + quarantine v2+v3 high_confidence. All records go to train (pure single-label pool).",
        source_inputs=[str(SRC_HUMAN)] + [str(x) for x in QUARANTINE_HC_DIRS],
    )
    return ds_dir, manifest


def balanced_round_robin_select(records: List[dict], n: int, key: str = "source") -> List[dict]:
    if n <= 0:
        return []
    grouped = defaultdict(list)
    for r in records:
        grouped[r.get(key, "unknown")].append(r)
    keys = sorted(grouped.keys())
    for k in keys:
        grouped[k].sort(key=lambda x: x["_hash"])

    selected = []
    i = 0
    while len(selected) < n:
        progressed = False
        for k in keys:
            if i < len(grouped[k]):
                selected.append(grouped[k][i])
                progressed = True
                if len(selected) >= n:
                    break
        if not progressed:
            break
        i += 1
    return selected[:n]


def collect_pool_from_datasets(ds_dirs: List[Path], desired_label: int) -> List[dict]:
    pool = []
    for ds_dir in ds_dirs:
        for rec in iter_dataset_records(ds_dir):
            if rec.get("label") == desired_label:
                rec["_hash"] = text_hash(rec["text"])
                pool.append(rec)
    return pool


def build_ds06_external_core_balanced(
    ds01_hashes: set,
    ds04_dir: Path,
    ds11_dir: Path,
) -> Tuple[Path, dict]:
    ds_id = "DS06_External_core_balanced_v1"
    max_len = 1500
    ds_dir = DATASET_ROOT / "20_P1_混合标签集" / ds_id
    log(f"Building {ds_id}")

    human_pool_raw = collect_pool_from_datasets([ds04_dir], desired_label=0)
    ai_pool_raw = collect_pool_from_datasets([ds11_dir], desired_label=1)

    def filter_pool(pool: List[dict], label: int) -> List[dict]:
        filtered = []
        local_seen = set()
        overlap_drop = 0
        dedupe_drop = 0
        range_drop = 0

        for rec in pool:
            h = rec["_hash"]
            lch = rec["length_char"]
            if lch < 100 or lch > max_len:
                range_drop += 1
                add_exclusion(ds_id, "filter_len", rec.get("source", ""), rec.get("record_id", ""), label, f"length_out_of_range_100_{max_len}", lch, lch, h)
                continue
            if h in ds01_hashes:
                overlap_drop += 1
                add_exclusion(ds_id, "overlap_filter", rec.get("source", ""), rec.get("record_id", ""), label, "overlap_with_ds01", lch, lch, h)
                continue
            if h in local_seen:
                dedupe_drop += 1
                add_exclusion(ds_id, "dedupe_pool", rec.get("source", ""), rec.get("record_id", ""), label, "duplicate_text_pool", lch, lch, h)
                continue
            local_seen.add(h)
            filtered.append(rec)

        OVERLAP_AUDIT.setdefault(ds_id, {})
        OVERLAP_AUDIT[ds_id][f"label_{label}_raw"] = len(pool)
        OVERLAP_AUDIT[ds_id][f"label_{label}_after_filter"] = len(filtered)
        OVERLAP_AUDIT[ds_id][f"label_{label}_drop_overlap_ds01"] = overlap_drop
        OVERLAP_AUDIT[ds_id][f"label_{label}_drop_duplicate_pool"] = dedupe_drop
        OVERLAP_AUDIT[ds_id][f"label_{label}_drop_length"] = range_drop
        return filtered

    human_pool = filter_pool(human_pool_raw, 0)
    ai_pool = filter_pool(ai_pool_raw, 1)

    n = min(len(human_pool), len(ai_pool))
    if n <= 0:
        raise RuntimeError(f"{ds_id}: no balanced data available after filtering")

    combined = balanced_round_robin_select(human_pool, n, "source") + balanced_round_robin_select(ai_pool, n, "source")

    final = []
    seen_final = {}
    conflict_drop = 0
    for rec in sorted(combined, key=lambda x: x["_hash"]):
        h = rec["_hash"]
        lb = rec["label"]
        if h in seen_final and seen_final[h] != lb:
            conflict_drop += 1
            add_exclusion(ds_id, "final_dedupe", rec.get("source", ""), rec.get("record_id", ""), lb, "cross_label_conflict_hash", rec["length_char"], rec["length_char"], h)
            continue
        if h in seen_final:
            add_exclusion(ds_id, "final_dedupe", rec.get("source", ""), rec.get("record_id", ""), lb, "duplicate_text_final", rec["length_char"], rec["length_char"], h)
            continue
        seen_final[h] = lb
        final.append(rec)

    OVERLAP_AUDIT[ds_id]["balanced_n_before_conflict"] = 2 * n
    OVERLAP_AUDIT[ds_id]["drop_cross_label_conflict"] = conflict_drop
    OVERLAP_AUDIT[ds_id]["final_n"] = len(final)

    split_map = stratified_split(final, (0.6, 0.2, 0.2), strata_fn=lambda r: f"{r['source']}|{r['label']}")
    for sp in SPLITS:
        for rec in split_map[sp]:
            rec["split"] = sp

    manifest = finalize_dataset(
        ds_id,
        ds_dir,
        split_map,
        build_note=f"External balanced dataset from DS04 (human) and DS11 (generated ai standard), length 100-{max_len}, leakage-filtered vs DS01, stratified split 60/20/20.",
        source_inputs=[str(ds04_dir), str(ds11_dir)],
    )
    return ds_dir, manifest


def build_ds07_external_long(
    ds01_hashes: set,
    ds04_dir: Path,
    ds12_dir: Path,
) -> Tuple[Path, dict]:
    ds_id = "DS07_External_long_v1"
    max_len = 2200
    ds_dir = DATASET_ROOT / "20_P1_混合标签集" / ds_id
    log(f"Building {ds_id}")

    human_pool_raw = collect_pool_from_datasets([ds04_dir], desired_label=0)
    ai_pool_raw = collect_pool_from_datasets([ds12_dir], desired_label=1)

    def filter_pool(pool: List[dict], label: int) -> List[dict]:
        filtered = []
        local_seen = set()
        overlap_drop = 0
        dedupe_drop = 0
        range_drop = 0

        for rec in pool:
            h = rec["_hash"]
            lch = rec["length_char"]
            if lch < 100 or lch > max_len:
                range_drop += 1
                add_exclusion(ds_id, "filter_len", rec.get("source", ""), rec.get("record_id", ""), label, f"length_out_of_range_100_{max_len}", lch, lch, h)
                continue
            if h in ds01_hashes:
                overlap_drop += 1
                add_exclusion(ds_id, "overlap_filter", rec.get("source", ""), rec.get("record_id", ""), label, "overlap_with_ds01", lch, lch, h)
                continue
            if h in local_seen:
                dedupe_drop += 1
                add_exclusion(ds_id, "dedupe_pool", rec.get("source", ""), rec.get("record_id", ""), label, "duplicate_text_pool", lch, lch, h)
                continue
            local_seen.add(h)
            filtered.append(rec)

        OVERLAP_AUDIT.setdefault(ds_id, {})
        OVERLAP_AUDIT[ds_id][f"label_{label}_raw"] = len(pool)
        OVERLAP_AUDIT[ds_id][f"label_{label}_after_filter"] = len(filtered)
        OVERLAP_AUDIT[ds_id][f"label_{label}_drop_overlap_ds01"] = overlap_drop
        OVERLAP_AUDIT[ds_id][f"label_{label}_drop_duplicate_pool"] = dedupe_drop
        OVERLAP_AUDIT[ds_id][f"label_{label}_drop_length"] = range_drop
        return filtered

    human_pool = filter_pool(human_pool_raw, 0)
    ai_pool = filter_pool(ai_pool_raw, 1)

    n = min(len(human_pool), len(ai_pool))
    if n <= 0:
        raise RuntimeError(f"{ds_id}: no balanced data available after filtering")

    combined = balanced_round_robin_select(human_pool, n, "source") + balanced_round_robin_select(ai_pool, n, "source")

    final = []
    seen_final = {}
    conflict_drop = 0
    for rec in sorted(combined, key=lambda x: x["_hash"]):
        h = rec["_hash"]
        lb = rec["label"]
        if h in seen_final and seen_final[h] != lb:
            conflict_drop += 1
            add_exclusion(ds_id, "final_dedupe", rec.get("source", ""), rec.get("record_id", ""), lb, "cross_label_conflict_hash", rec["length_char"], rec["length_char"], h)
            continue
        if h in seen_final:
            add_exclusion(ds_id, "final_dedupe", rec.get("source", ""), rec.get("record_id", ""), lb, "duplicate_text_final", rec["length_char"], rec["length_char"], h)
            continue
        seen_final[h] = lb
        final.append(rec)

    OVERLAP_AUDIT[ds_id]["balanced_n_before_conflict"] = 2 * n
    OVERLAP_AUDIT[ds_id]["drop_cross_label_conflict"] = conflict_drop
    OVERLAP_AUDIT[ds_id]["final_n"] = len(final)

    split_map = stratified_split(final, (0.6, 0.2, 0.2), strata_fn=lambda r: f"{r['source']}|{r['label']}")
    for sp in SPLITS:
        for rec in split_map[sp]:
            rec["split"] = sp

    manifest = finalize_dataset(
        ds_id,
        ds_dir,
        split_map,
        build_note=f"External balanced dataset from DS04 (human) and DS12 (generated ai natural), length 100-{max_len}, leakage-filtered vs DS01, stratified split 60/20/20.",
        source_inputs=[str(ds04_dir), str(ds12_dir)],
    )
    return ds_dir, manifest

def build_p2_candidate_audit() -> dict:
    p2_dir = DATASET_ROOT / "30_P2_候选待验证"
    ensure_dir(p2_dir)

    def scan_dir(path: Path) -> dict:
        by_ext = Counter()
        total_size = 0
        files = []
        if path.exists():
            for fp in sorted(path.rglob("*")):
                if fp.is_file():
                    ext = fp.suffix.lower() if fp.suffix else "<no_ext>"
                    by_ext[ext] += 1
                    st = fp.stat()
                    total_size += int(st.st_size)
                    files.append(str(fp))
        return {
            "path": str(path),
            "exists": path.exists(),
            "file_count": len(files),
            "total_size": total_size,
            "ext_dist": dict(sorted(by_ext.items())),
            "sample_files": files[:200],
        }

    clts_info = scan_dir(P2_CLTS_DIR)
    zhihu_info = scan_dir(P2_ZHIHU_DIR)

    audit = {
        "generated_at": iso_now(),
        "policy": "P2 candidate only; not included in DS01-DS07 trainable main schema this round.",
        "candidates": {
            "CLTS数据集": {**clts_info, "why_not_direct_train": "parallel src/tgt corpus; no direct human-vs-ai detection labels"},
            "huggingface_zhihu": {**zhihu_info, "why_not_direct_train": "metadata/inspection-focused files; supervised label semantics need redesign"},
        },
    }

    write_json(p2_dir / "candidate_audit.json", audit)
    write_csv(
        p2_dir / "candidate_summary.csv",
        ["candidate", "file_count", "total_size", "why_not_direct_train"],
        [
            {
                "candidate": "CLTS数据集",
                "file_count": clts_info["file_count"],
                "total_size": clts_info["total_size"],
                "why_not_direct_train": "parallel corpus; no direct detection labels",
            },
            {
                "candidate": "huggingface_zhihu",
                "file_count": zhihu_info["file_count"],
                "total_size": zhihu_info["total_size"],
                "why_not_direct_train": "metadata/inspection-first; label semantics pending",
            },
        ],
    )
    return audit


def write_contract_files() -> None:
    cdir = DATASET_ROOT / "00_contract"
    ensure_dir(cdir)

    schema = {
        "title": "Unified Main Schema for 20260408 SOTA Route",
        "type": "object",
        "required": REQUIRED_FIELDS,
        "properties": {
            "record_id": {"type": "string"},
            "text": {"type": "string"},
            "label": {"type": "integer", "enum": [0, 1]},
            "source": {"type": "string"},
            "split": {"type": "string", "enum": ["train", "dev", "test"]},
            "length_char": {"type": "integer", "minimum": 0},
            "topic": {"type": "string"},
            "model_slug": {"type": "string"},
        },
    }
    write_json(cdir / "main_schema.json", schema)
    write_json(cdir / "label_mapping.json", LABEL_MAP)

    md = """# 主 Schema 合同（统一训练格式）

## 字段
- `record_id`：字符串唯一ID
- `text`：清洗后的正文
- `label`：整数标签（human=0, ai=1）
- `source`：数据族来源（如 `nlpcc/hc3_plus/hc3_raw/human/quarantine_hc`）
- `split`：`train/dev/test`
- `length_char`：去空白字符后的长度
- `topic`：主题字段（可空字符串）
- `model_slug`：AI来源模型（可空字符串）

## 全局规则
- 编码回退：`utf-8-sig -> utf-8 -> gb18030`
- 清洗：换行统一、空白压缩、`\\n`与明显`/n`噪声修正、Markdown去痕
- 质量过滤：空文本、`length_char < 30`、严重伪装文本、重复文本剔除
- 标签映射固定：`human=0, ai=1`
"""
    (cdir / "schema_contract.md").write_text(md, encoding="utf-8")


def write_experiment_views(ds_paths: Dict[str, Path]) -> None:
    edir = DATASET_ROOT / "40_experiment_views"
    ensure_dir(edir)

    mapping = {
        "E00": {"use": ["DS01"], "detail": "Schema check + distribution"},
        "E01": {"use": ["DS01"], "detail": "BERT baseline train/dev"},
        "E02": {"use": ["DS01"], "detail": "RoBERTa baseline train/dev"},
        "E03": {"use": ["DS01"], "detail": "Rule module on dev/test"},
        "E04": {"use": ["DS01"], "detail": "Static weighted vote"},
        "E05": {"use": ["DS01"], "detail": "Determinative override"},
        "E06": {"use": ["DS01"], "detail": "Dynamic vote by buckets"},
        "E07": {"use": ["DS01"], "detail": "Joint features MLP"},
        "E08": {"use": ["DS01", "DS02", "DS03"], "detail": "Full chain rerun / optional aug"},
        "E09": {"use": ["DS01", "DS06"], "detail": "Train on DS01, transfer eval on DS06"},
        "E10": {"use": ["DS01", "DS06"], "detail": "Calibrate with DS01 dev + DS06 dev/test"},
        "E11": {"use": ["DS01", "DS06"], "detail": "Bucket threshold on DS06"},
        "E12": {"use": ["DS01", "DS06", "DS07"], "detail": "Zero-shot branch"},
        "E13": {"use": ["DS01", "DS06", "DS07"], "detail": "Zero-shot branch"},
        "E14": {"use": ["DS01", "DS06", "DS07"], "detail": "LLM LoRA branch"},
        "E15": {"use": ["DS01", "DS06", "DS07"], "detail": "Combo search"},
    }

    resolved = {}
    rows = []
    for exp, item in mapping.items():
        resolved[exp] = {"detail": item["detail"], "datasets": {}}
        for ds in item["use"]:
            p = ds_paths[ds]
            resolved[exp]["datasets"][ds] = {
                "dataset_dir": str(p),
                "train": str(p / "train.jsonl"),
                "dev": str(p / "dev.jsonl"),
                "test": str(p / "test.jsonl"),
            }
            rows.append(
                {
                    "experiment_id": exp,
                    "dataset_id": ds,
                    "dataset_dir": str(p),
                    "train_jsonl": str(p / "train.jsonl"),
                    "dev_jsonl": str(p / "dev.jsonl"),
                    "test_jsonl": str(p / "test.jsonl"),
                }
            )

    write_json(edir / "experiment_views.json", {"generated_at": iso_now(), "views": resolved})
    write_csv(edir / "E00_E15_dataset_mapping.csv", ["experiment_id", "dataset_id", "dataset_dir", "train_jsonl", "dev_jsonl", "test_jsonl"], rows)


def validate_schema_sample(ds_paths: Dict[str, Path]) -> dict:
    report = {"generated_at": iso_now(), "datasets": {}}
    for ds, p in ds_paths.items():
        checked = 0
        invalid = 0
        examples = []
        for sp in SPLITS:
            fp = p / f"{sp}.jsonl"
            if not fp.exists():
                continue
            for _, rec in iter_jsonl(fp):
                checked += 1
                ok = True
                for k in REQUIRED_FIELDS:
                    if k not in rec:
                        ok = False
                        if len(examples) < 10:
                            examples.append(f"{sp}:missing:{k}")
                if rec.get("label") not in (0, 1):
                    ok = False
                    if len(examples) < 10:
                        examples.append(f"{sp}:bad_label:{rec.get('label')}")
                if rec.get("split") not in ("train", "dev", "test"):
                    ok = False
                    if len(examples) < 10:
                        examples.append(f"{sp}:bad_split:{rec.get('split')}")
                if not isinstance(rec.get("length_char"), int):
                    ok = False
                    if len(examples) < 10:
                        examples.append(f"{sp}:bad_length_type:{type(rec.get('length_char')).__name__}")
                if not ok:
                    invalid += 1
                if checked >= 1000:
                    break
            if checked >= 1000:
                break
        report["datasets"][ds] = {
            "checked_records": checked,
            "invalid_records": invalid,
            "valid": invalid == 0,
            "example_errors": examples,
        }
    return report


def write_exclusion_log() -> None:
    mdir = DATASET_ROOT / "90_manifests"
    ensure_dir(mdir)
    fieldnames = ["dataset_id", "stage", "source_file", "record_ref", "label", "drop_reason", "raw_length", "clean_length", "text_hash"]
    write_csv(mdir / "exclusion_log.csv", fieldnames, EXCLUSION_LOG_ROWS)

    reason_counter = Counter(r["drop_reason"] for r in EXCLUSION_LOG_ROWS)
    rows = [{"drop_reason": k, "count": v} for k, v in sorted(reason_counter.items(), key=lambda x: (-x[1], x[0]))]
    write_csv(mdir / "exclusion_reason_stats.csv", ["drop_reason", "count"], rows)

def main() -> int:
    source_roots = {
        "human": SRC_HUMAN,
        "human_quarantine": SRC_QUARANTINE,
        "raw_external": SRC_RAW_EXTERNAL,
    }

    if DATASET_ROOT.exists():
        log(f"Removing old dataset directory: {DATASET_ROOT}")
        shutil.rmtree(DATASET_ROOT)
    ensure_dir(DATASET_ROOT)
    ensure_dir(DATASET_ROOT / "90_manifests")

    log("Collecting source inventory (before)...")
    inv_before = collect_source_inventory(source_roots)
    write_json(DATASET_ROOT / "90_manifests" / "source_inventory_before.json", inv_before)

    log("Writing dataset contract files...")
    write_contract_files()

    ds_paths: Dict[str, Path] = {}
    dataset_manifests = {}

    ds01_dir, ds01_manifest, ds01_hashes = build_ds01_nlpcc()
    ds_paths["DS01"] = ds01_dir
    dataset_manifests["DS01"] = ds01_manifest

    ds02_dir, ds02_manifest = build_ds02_hc3plus()
    ds_paths["DS02"] = ds02_dir
    dataset_manifests["DS02"] = ds02_manifest

    ds03_dir, ds03_manifest = build_ds03_hc3raw_expanded()
    ds_paths["DS03"] = ds03_dir
    dataset_manifests["DS03"] = ds03_manifest

    ds04_dir, ds04_manifest = build_ds04_human_pools_merged()
    ds_paths["DS04"] = ds04_dir
    dataset_manifests["DS04"] = ds04_manifest

    ds11_dir = DATASET_ROOT / "30_P2_单标签原料池" / "DS11_Generated_AI_v1"
    ds12_dir = DATASET_ROOT / "30_P2_单标签原料池" / "DS12_Generated_AI_natural_v1"

    ds06_dir, ds06_manifest = build_ds06_external_core_balanced(
        ds01_hashes=ds01_hashes,
        ds04_dir=ds04_dir,
        ds11_dir=ds11_dir,
    )
    ds_paths["DS06"] = ds06_dir
    dataset_manifests["DS06"] = ds06_manifest

    ds07_dir, ds07_manifest = build_ds07_external_long(
        ds01_hashes=ds01_hashes,
        ds04_dir=ds04_dir,
        ds12_dir=ds12_dir,
    )
    ds_paths["DS07"] = ds07_dir
    dataset_manifests["DS07"] = ds07_manifest

    log("Building P2 candidate audit...")
    p2_audit = build_p2_candidate_audit()

    log("Writing experiment views...")
    write_experiment_views(ds_paths)

    log("Writing manifests + validation reports...")
    write_json(DATASET_ROOT / "90_manifests" / "dataset_manifests.json", dataset_manifests)
    write_json(DATASET_ROOT / "90_manifests" / "overlap_audit.json", OVERLAP_AUDIT)
    write_json(DATASET_ROOT / "90_manifests" / "p2_candidate_audit_copy.json", p2_audit)

    schema_report = validate_schema_sample(ds_paths)
    write_json(DATASET_ROOT / "90_manifests" / "schema_validation_report.json", schema_report)

    write_exclusion_log()

    log("Collecting source inventory (after)...")
    inv_after = collect_source_inventory(source_roots)
    write_json(DATASET_ROOT / "90_manifests" / "source_inventory_after.json", inv_after)

    compare = {
        "generated_at": iso_now(),
        "before_digest": inv_before.get("global_digest"),
        "after_digest": inv_after.get("global_digest"),
        "same_digest": inv_before.get("global_digest") == inv_after.get("global_digest"),
        "before": inv_before,
        "after": inv_after,
    }
    write_json(DATASET_ROOT / "90_manifests" / "source_inventory_compare.json", compare)
    if not compare["same_digest"]:
        log("ERROR: source inventory changed during build. Stopping.")
        return 2

    summary = {
        "generated_at": iso_now(),
        "dataset_root": str(DATASET_ROOT),
        "datasets": {k: {"path": str(v), "manifest": dataset_manifests.get(k, {})} for k, v in ds_paths.items()},
        "overlap_audit_file": str(DATASET_ROOT / "90_manifests" / "overlap_audit.json"),
        "exclusion_log_file": str(DATASET_ROOT / "90_manifests" / "exclusion_log.csv"),
        "schema_validation_report_file": str(DATASET_ROOT / "90_manifests" / "schema_validation_report.json"),
    }
    write_json(DATASET_ROOT / "90_manifests" / "build_summary.json", summary)

    log("Dataset build completed successfully.")
    return 0


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