#!/usr/bin/env python3
"""Duplicate-answer filter for the Status Board queue panel (likely resubmissions).

After the 2026-10-24 genesis cut, unsettled Oct 23-24 answers can be resubmitted, so the same piece of work
can sit in more than one close's queue, or twice in one queue. This module flags likely resubmissions using
only fields the public, cached data actually has:
  - answer text (Cluster Lens's cache of public explorer bodies: answer, from, epoch, ts),
  - author and seal time (POAScan challenge index: from, epoch, ts),
  - pending-row ids (txids) from the public /api/epochs close record.
The close record's pending rows carry no author or text, and the chain stores no separate answer hash, so the
hash is computed here from the text.

Rule: an answer is a LIKELY RESUBMISSION when the SAME author sealed an EARLIER answer whose text is identical
after normalization (lowercase, letters only, the per-epoch seed words removed, since a resubmission to a new
epoch must swap them in). "exact" means the raw text (trimmed) is byte-identical too. Answers shorter than
MIN_TOKENS normalized words are never matched. Different authors are never merged here (that is Cluster Lens's
job). Answers without cached text can't be checked and are counted separately, never guessed.

Pure functions plus a CLI that freezes an archived segment's per-close counts from a cached /api/epochs file:
  python3 dedupe.py --freeze SEGMENT_ID --epochs /path/to/epochs.json [--force]
No network access; no key or token is read.
"""
import hashlib
import json
import os
import re
import sys
import time

HERE = os.path.dirname(os.path.abspath(__file__))
CLUSTERS_DATA = os.environ.get("DEDUPE_CLUSTERS_DATA", os.path.join(HERE, "..", "clusters", "data"))
OUT_DIR = os.path.join(HERE, "data", "dedupe_segments")
MIN_TOKENS = 20
# Same seed vocabulary Cluster Lens strips (clusters/lens.py SEED_WORDS); copied so this file stands alone.
SEED_WORDS = frozenset("""
anchor atlas basalt beacon bridge canyon cedar cipher comet copper crystal dawn delta dune
echo ember fable falcon ferrous forge gale garden glacier grove harbor haven hollow horizon
inlet island ivory jade jetty juniper karst kernel knoll lagoon lantern ledge lotus marble
meadow mesa mirror nadir nebula nectar oasis ocean onyx opal orchard pearl pigment plume
prism pulse quartz quill raven reef relic ridge ripple river sable saffron shale signal
sonnet sparrow spiral summit terrace thread tide timber trellis tundra umber vault velvet
vertex vine violet voyage walnut wander wharf whisper willow yarrow zenith zephyr
""".split())
# Segment starts (same boundary as clusters/fetch.py SEGMENTS); used only when a row has no `seg` tag.
SEGMENTS = [("poa-mainnet-1-segment1", 0), ("poa-mainnet-1-segment2", 1792872000)]
_WORD = re.compile(r"[a-z]+")


def segment_of(ts):
    seg = SEGMENTS[0][0]
    for sid, start in SEGMENTS:
        if isinstance(ts, (int, float)) and ts >= start:
            seg = sid
    return seg


def norm_tokens(text):
    return [t for t in _WORD.findall((text or "").lower()) if t not in SEED_WORDS]


def hashes(text):
    """(exact_hash, normalized_hash); normalized is None for answers too short to match safely."""
    exact = hashlib.sha256((text or "").strip().encode()).hexdigest()
    toks = norm_tokens(text)
    norm = hashlib.sha256(" ".join(toks).encode()).hexdigest() if len(toks) >= MIN_TOKENS else None
    return exact, norm


def build_corpus(bodies, index):
    """txid -> {from, ts, epoch, seg, exact, norm}; rows known only from the index have exact/norm None."""
    out = {}
    for t, v in (index or {}).items():
        if isinstance(v, dict) and v.get("from"):
            out[t] = {"from": v["from"], "ts": v.get("ts") or 0, "epoch": v.get("epoch"),
                      "seg": v.get("seg") or segment_of(v.get("ts")), "exact": None, "norm": None}
    for t, v in (bodies or {}).items():
        if not isinstance(v, dict) or not isinstance(v.get("answer"), str) or not v.get("from"):
            continue
        ex, nm = hashes(v["answer"])
        base = out.get(t, {})
        out[t] = {"from": v["from"], "ts": v.get("ts") or base.get("ts") or 0,
                  "epoch": v.get("epoch") if v.get("epoch") is not None else base.get("epoch"),
                  "seg": base.get("seg") or v.get("seg") or segment_of(v.get("ts")), "exact": ex, "norm": nm}
    return out


def mark_resubmissions(corpus):
    """txid -> {orig, kind, cross_segment, cross_epoch} for every answer that repeats an earlier answer by the
    same author (normalized text). The earliest answer of each (author, text) group is the original."""
    groups = {}
    for t, r in corpus.items():
        if r.get("norm"):
            groups.setdefault((r["from"], r["norm"]), []).append(t)
    flags = {}
    for ids in groups.values():
        if len(ids) < 2:
            continue
        ids.sort(key=lambda t: (corpus[t]["ts"] or 0, t))
        o = corpus[ids[0]]
        for t in ids[1:]:
            r = corpus[t]
            flags[t] = {"orig": ids[0], "kind": "exact" if r["exact"] == o["exact"] else "normalized",
                        "cross_segment": r["seg"] != o["seg"], "cross_epoch": (r["seg"], r["epoch"]) != (o["seg"], o["epoch"])}
    return flags


def count_close(pending_ids, corpus, flags, prior_authors=None):
    """Counts for one close's pending map. `prior_authors`: authors with unsettled rows in an earlier segment
    (for the author-only bound on rows that have no cached text)."""
    ids = set(pending_ids or [])
    dup = [t for t in ids if t in flags]
    row = {"pending": len(ids),
           "with_text": sum(1 for t in ids if (corpus.get(t) or {}).get("exact")),
           "author_known": sum(1 for t in ids if t in corpus),
           "likely_resubmissions": len(dup),
           "exact_copies": sum(1 for t in dup if flags[t]["kind"] == "exact"),
           "original_also_pending": sum(1 for t in dup if flags[t]["orig"] in ids),
           "cross_segment": sum(1 for t in dup if flags[t]["cross_segment"]),
           "cross_epoch": sum(1 for t in dup if flags[t]["cross_epoch"])}
    row["pending_without_duplicates"] = row["pending"] - row["likely_resubmissions"]
    if prior_authors is not None:
        pa = set(prior_authors)
        row["no_text_author_had_unsettled_prior"] = sum(
            1 for t in ids if t in corpus and not corpus[t].get("exact") and corpus[t]["from"] in pa)
    return row


def load_corpus(data_dir=CLUSTERS_DATA):
    def ld(name):
        try:
            return json.load(open(os.path.join(data_dir, name)))
        except (OSError, ValueError):
            return {}
    return build_corpus(ld("bodies.json"), ld("index.json"))


def unsettled_authors(epochs, corpus):
    """Authors of rows still pending (never settled) in the last two closes of a segment's record, i.e. the
    work the genesis-cut notice says may be resubmitted."""
    eps = sorted((e for e in epochs if isinstance(e, dict) and "epoch" in e), key=lambda e: int(e["epoch"]))[-2:]
    settled = {s.get("answer") for e in epochs if isinstance(e, dict) for s in ((e.get("basis") or {}).get("settled") or []) if isinstance(s, dict)}
    ids = {t for e in eps for t in ((e.get("basis") or {}).get("pending") or {})} - settled
    return sorted({corpus[t]["from"] for t in ids if t in corpus})


def freeze(seg, epochs_path, force=False):
    os.makedirs(OUT_DIR, exist_ok=True)
    path = os.path.join(OUT_DIR, seg + ".json")
    if os.path.exists(path) and not force:
        print("dedupe: %s already frozen (use --force to rebuild)" % seg)
        return 0
    epochs = json.load(open(epochs_path))
    corpus = load_corpus()
    flags = mark_resubmissions(corpus)
    closes = []
    for e in sorted((x for x in epochs if isinstance(x, dict) and "epoch" in x), key=lambda x: int(x["epoch"])):
        row = count_close(((e.get("basis") or {}).get("pending") or {}).keys(), corpus, flags)
        row.update({"epoch": int(e["epoch"]), "ts": e.get("ts")})
        closes.append(row)
    authors = unsettled_authors(epochs, corpus)
    json.dump({"segment": seg, "built_ts": int(time.time()), "closes": closes,
               "corpus_answers_with_text": sum(1 for r in corpus.values() if r.get("exact") and r["seg"] == seg),
               "unsettled_authors_last_two_closes": len(authors)}, open(path, "w"), indent=1)
    # addresses stay in data/ (used for the author-only bound), never copied to the site
    json.dump(authors, open(os.path.join(OUT_DIR, seg + "_unsettled_authors.json"), "w"))
    print("dedupe: froze %s (%d closes, %d unsettled authors)" % (seg, len(closes), len(authors)))
    return 0


def load_frozen():
    out = {}
    if os.path.isdir(OUT_DIR):
        for f in sorted(os.listdir(OUT_DIR)):
            if f.endswith(".json") and not f.endswith("_unsettled_authors.json"):
                try:
                    out[f[:-5]] = json.load(open(os.path.join(OUT_DIR, f)))
                except (OSError, ValueError):
                    pass
    return out


def load_prior_authors():
    out = set()
    if os.path.isdir(OUT_DIR):
        for f in os.listdir(OUT_DIR):
            if f.endswith("_unsettled_authors.json"):
                try:
                    out |= set(json.load(open(os.path.join(OUT_DIR, f))))
                except (OSError, ValueError):
                    pass
    return out


if __name__ == "__main__":
    a = sys.argv[1:]
    if "--freeze" in a and "--epochs" in a:
        sys.exit(freeze(a[a.index("--freeze") + 1], a[a.index("--epochs") + 1], "--force" in a))
    print(__doc__)
    sys.exit(2)