v3_scale: code banc v4/gen/v3, rapports validés, corpus_v3 (sha256 vérifié), README
519e2ef verified | """SPIRA Hashing - verification independante (lecture seule) d'un corpus: tailles, sha256, overlap 64 o. | |
| Usage: verify_v3.py CORPUS_DIR OUT_JSON""" | |
| import sys, os, json, hashlib, numpy as np | |
| sys.path.insert(0, "/workspace/spira_shared/checker") | |
| C, OUT = sys.argv[1:3]; K = 64; P = np.uint64(1099511628211) | |
| def hashes(b): | |
| a = np.frombuffer(b, np.uint8).astype(np.uint64); n = len(a) - K + 1 | |
| if n <= 0: return np.zeros(0, np.uint64) | |
| h = np.zeros(n, np.uint64) | |
| with np.errstate(over="ignore"): | |
| for j in range(K): h = h * P + a[j:j+n] + np.uint64(1) | |
| return h | |
| W = {k: open(f"/workspace/spira_shared/spira_m7/windows/{k}.bin","rb").read() for k in ("fresh_en","fresh_fr","neither_code","test20k","queries")} | |
| try: | |
| from windows_v2 import freshcode_v2; W["fc2"] = freshcode_v2() | |
| except Exception as e: W["fc2_ERROR"] = b""; print("fc2 indisponible", e) | |
| F = {f: open(f"{C}/{f}.bin","rb").read() for f in ("train","test","oneshot") if os.path.exists(f"{C}/{f}.bin")} | |
| for _k in ("test","oneshot"): | |
| if _k in F: W["corpus_"+_k] = F[_k] | |
| rep = {"corpus": C, "files": {f: {"bytes": len(b), "sha256": hashlib.sha256(b).hexdigest()} for f, b in F.items()}, "overlap64": {}} | |
| ht = hashes(F["train"]); hs = np.sort(ht) | |
| for k, b in W.items(): | |
| hw = hashes(b); u = np.unique(hw) | |
| i = np.searchsorted(hs, u); i = np.minimum(i, len(hs)-1); cand = u[hs[i] == u] | |
| # confirmation octet par octet (anti-collision) | |
| tset = set(cand.tolist()); pos_t = {} | |
| for p in np.nonzero(np.isin(ht, cand))[0]: pos_t.setdefault(int(ht[p]), []).append(int(p)) | |
| conf = 0 | |
| for p in np.nonzero(np.isin(hw, cand))[0]: | |
| seg = b[p:p+K] | |
| if any(F["train"][q:q+K] == seg for q in pos_t.get(int(hw[p]), ())): conf += 1 | |
| rep["overlap64"][k] = {"window_bytes": len(b), "n_64grams": int(len(hw)), "hash_hits": int(np.isin(hw, cand).sum()), "confirmed_positions": conf} | |
| print(k, rep["overlap64"][k], flush=True) | |
| rep["valid_zero_overlap"] = all(v["confirmed_positions"] == 0 for v in rep["overlap64"].values()) and "fc2_ERROR" not in W | |
| m = f"{C}/corpus_manifest.json" if os.path.exists(f"{C}/corpus_manifest.json") else f"{C}/manifest.json" | |
| if os.path.exists(m): rep["manifest_checker"] = json.load(open(m)) | |
| json.dump(rep, open(OUT, "w"), indent=1); print("VALID" if rep["valid_zero_overlap"] else "INVALID") | |