"""Auditable small-corpus pipeline. Quadratic near-dedup is intentionally lab-only."""
from collections import Counter
from dataclasses import dataclass, asdict
from hashlib import sha256
from pathlib import Path
import json
import re
import unicodedata
import numpy as np
from .tokenizer import ByteTokenizer

SECRET = re.compile(r"(?:hf_[A-Za-z0-9]{20,}|sk-[A-Za-z0-9_-]{20,}|-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----)")
EMAIL = re.compile(r"\b[\w.+-]+@[\w.-]+\.[A-Za-z]{2,}\b")
ALLOWED = {"MIT", "Apache-2.0", "BSD-3-Clause", "CC0-1.0", "owner-authored"}


@dataclass
class Record:
    id: str
    text: str
    source: str
    license: str
    domain: str
    split: str = "train"


def shingles(text, width=5):
    words = text.lower().split()
    return {tuple(words[i:i + width]) for i in range(max(1, len(words) - width + 1))}


def similarity(a, b):
    return len(a & b) / max(1, len(a | b))


def prepare(records, out_dir, *, holdouts=(), near_threshold=0.9):
    if not 0 <= near_threshold <= 1:
        raise ValueError("Invalid near duplicate threshold")
    out = Path(out_dir)
    out.mkdir(parents=True, exist_ok=True)
    seen, signatures, accepted, reasons = set(), [], [], Counter()
    holdout_sigs = [shingles(unicodedata.normalize("NFC", s)) for s in holdouts]
    for raw in records:
        r = Record(**raw) if isinstance(raw, dict) else raw
        if r.license not in ALLOWED or not r.source or r.split not in {"train", "validation", "test"}:
            reasons["license_or_provenance_or_split"] += 1
            continue
        text = unicodedata.normalize("NFC", r.text.replace("\r\n", "\n"))
        if SECRET.search(text):
            reasons["secret"] += 1
            continue
        text = EMAIL.sub("[EMAIL]", text)
        if len(text.strip()) < 24 or "\x00" in text:
            reasons["quality"] += 1
            continue
        digest = sha256(text.encode()).hexdigest()
        sig = shingles(text)
        if digest in seen:
            reasons["exact_duplicate"] += 1
            continue
        if any(similarity(sig, s) >= near_threshold for s in holdout_sigs):
            reasons["contamination"] += 1
            continue
        if any(similarity(sig, s) >= near_threshold for s in signatures):
            reasons["near_duplicate"] += 1
            continue
        seen.add(digest)
        signatures.append(sig)
        accepted.append({**asdict(r), "text": text, "sha256": digest,
                         "language_hint": "hi-or-mixed" if re.search("[\u0900-\u097f]", text) else "und",
                         "quality_score": min(1.0, len(set(text)) / 40)})
    tok, shards = ByteTokenizer(), {}
    for split in ("train", "validation", "test"):
        ids = [i for r in accepted if r["split"] == split for i in tok.encode(r["text"], special=True)]
        path = out / f"{split}.npy"
        np.save(path, np.asarray(ids, dtype=np.uint16), allow_pickle=False)
        shards[split] = {"file": path.name, "tokens": len(ids), "sha256": sha256(path.read_bytes()).hexdigest()}
    (out / "records.jsonl").write_text("".join(json.dumps(r, ensure_ascii=False) + "\n" for r in accepted), encoding="utf-8")
    manifest = {"version": 1, "tokenizer": "utf8-byte-v1", "accepted": len(accepted),
                "rejected": dict(reasons), "shards": shards, "limitations": ["PII regex is incomplete", "language hint is not language identification", "near dedup is quadratic"]}
    (out / "manifest.json").write_text(json.dumps(manifest, indent=2), encoding="utf-8")
    return manifest