144 lines
6.1 KiB
Python
144 lines
6.1 KiB
Python
#!/usr/bin/env python3
|
||
"""Common parsers over the coldrun-a paid artifacts (D39.58). $0 — no model calls.
|
||
Reads the durable book artifacts read-only (outside git, per CLAUDE.md). Override dir with $COLDRUN_A."""
|
||
import sqlite3, re, sys, os
|
||
|
||
SCRATCH = os.environ.get("COLDRUN_A", "/home/ubuntu/books/gu-zhenren/coldrun-a")
|
||
DB = os.path.join(SCRATCH, "guzhenren-coldrun-a.db")
|
||
|
||
def con():
|
||
c = sqlite3.connect(f"file:{DB}?mode=ro", uri=True)
|
||
return c
|
||
|
||
# ---------- Terminologist passes ----------
|
||
def pass_of(ts):
|
||
# ts like '2026-07-30 22:05:01'
|
||
if ts < "2026-07-31":
|
||
return 1
|
||
if ts < "2026-07-31 00:20":
|
||
return 2
|
||
return 3
|
||
|
||
def terminologist_rows():
|
||
"""Return list of dicts: pass, chunk_idx, ts, src, dst (one per output line)."""
|
||
c = con()
|
||
rows = []
|
||
for chunk_idx, ts, rt in c.execute(
|
||
"SELECT chunk_idx, created_at, response_text FROM checkpoints "
|
||
"WHERE role='terminologist' ORDER BY created_at"):
|
||
p = pass_of(ts)
|
||
for ln in rt.split("\n"):
|
||
ln = ln.rstrip("\n")
|
||
if not ln.strip():
|
||
continue
|
||
# format: src<TAB>dst (exactly two fields per prompt)
|
||
if "\t" in ln:
|
||
parts = ln.split("\t")
|
||
else:
|
||
# tolerant fallback: split on runs of spaces (report both counts)
|
||
parts = re.split(r"\s{2,}", ln.strip(), maxsplit=1)
|
||
if len(parts) < 2:
|
||
rows.append({"pass": p, "chunk_idx": chunk_idx, "ts": ts,
|
||
"src": parts[0].strip() if parts else "", "dst": "",
|
||
"malformed": True})
|
||
continue
|
||
rows.append({"pass": p, "chunk_idx": chunk_idx, "ts": ts,
|
||
"src": parts[0].strip(), "dst": parts[1].strip(),
|
||
"malformed": False})
|
||
c.close()
|
||
return rows
|
||
|
||
# ---------- bank-stop.txt ----------
|
||
def parse_bankstop():
|
||
"""Return dict src -> {dst, origin, type, freq, spread, evidence:[...], drafts:[(v,n)], ctx:[full multi-line window]}.
|
||
Captures the `evidence:` line and GROUPS multi-line KWIC windows (a `ctx:` line + its indented
|
||
continuation lines), matching terminology.go RenderBatch (fix after 01.08 author!=reviewer audit)."""
|
||
path = os.path.join(SCRATCH, "guzhenren-coldrun-a.db.bank-stop.txt")
|
||
lines = open(path, encoding="utf-8").read().split("\n")
|
||
recs = {}; cur = None; in_ctx = False
|
||
FIELD = ("origin=", "evidence:", "drafts:", "aliases:", "related:", "ctx:", "src ")
|
||
i = 0
|
||
while i < len(lines):
|
||
ln = lines[i]
|
||
if ln and not ln.startswith(" ") and "\t" in ln and i+1 < len(lines) and lines[i+1].lstrip().startswith("origin="):
|
||
src, dst = ln.split("\t", 1)
|
||
cur = {"src": src.strip(), "dst": dst.strip(), "evidence": [], "drafts": [], "ctx": []}
|
||
recs[src.strip()] = cur; in_ctx = False; i += 1; continue
|
||
if cur is not None:
|
||
s = ln.strip()
|
||
if s.startswith("origin="):
|
||
m = dict(re.findall(r"(\w+)=([^\s]+)", s))
|
||
cur.update(origin=m.get("origin",""), type=m.get("type",""),
|
||
freq=int(m.get("freq","0")), spread=int(m.get("spread","0"))); in_ctx=False
|
||
elif s.startswith("evidence:"):
|
||
cur["evidence"] = [e.strip() for e in s[len("evidence:"):].split(",") if e.strip()]; in_ctx=False
|
||
elif s.startswith("drafts:"):
|
||
for part in s[len("drafts:"):].strip().split(" | "):
|
||
mm = re.match(r"^(.*?)\s*×(\d+)$", part.strip())
|
||
if mm: cur["drafts"].append((mm.group(1).strip(), int(mm.group(2))))
|
||
elif part.strip(): cur["drafts"].append((part.strip(), 1))
|
||
in_ctx=False
|
||
elif s.startswith("ctx:"):
|
||
cur["ctx"].append(s[len("ctx:"):].strip()); in_ctx=True
|
||
elif s == "":
|
||
in_ctx=False
|
||
elif in_ctx and ln.startswith(" "):
|
||
cur["ctx"][-1] += "\n" + s # continuation line of the current KWIC window
|
||
i += 1
|
||
return recs
|
||
|
||
# ---------- mined-signature.yaml ----------
|
||
def parse_minedsig():
|
||
"""Return list of records with all scalar fields + parsed 'other proposals' from note."""
|
||
path = os.path.join(SCRATCH, "guzhenren-coldrun-a.db.mined-signature.yaml")
|
||
txt = open(path, encoding="utf-8").read()
|
||
recs = []
|
||
cur = None
|
||
for ln in txt.split("\n"):
|
||
if re.match(r"^ - src:", ln):
|
||
if cur: recs.append(cur)
|
||
cur = {}
|
||
cur["src"] = ln.split("src:",1)[1].strip().strip('"')
|
||
elif cur is not None:
|
||
m = re.match(r"^ (\w+): (.*)$", ln)
|
||
if m:
|
||
k, v = m.group(1), m.group(2)
|
||
cur[k] = v.strip()
|
||
if cur: recs.append(cur)
|
||
# parse note -> other proposals list of (variant,count)
|
||
for r in recs:
|
||
note = r.get("note", "")
|
||
props = []
|
||
mm = re.search(r"other proposals:\s*(.*?)(?:;|$)", note)
|
||
if mm:
|
||
for part in mm.group(1).split(","):
|
||
p = re.match(r'\s*"(.*?)"\s*×(\d+)', part)
|
||
if p:
|
||
props.append((p.group(1), int(p.group(2))))
|
||
r["other_proposals"] = props
|
||
# banknote chunk count
|
||
bn = re.search(r"banknote,\s*(\d+)\s*chunk", note)
|
||
r["banknote_chunks"] = int(bn.group(1)) if bn else None
|
||
return recs
|
||
|
||
# ---------- BANK-FULL.tsv ----------
|
||
def parse_bankfull():
|
||
path = os.path.join(SCRATCH, "BANK-FULL.tsv")
|
||
rows = []
|
||
with open(path, encoding="utf-8") as f:
|
||
header = f.readline().rstrip("\n").split("\t")
|
||
for ln in f:
|
||
cells = ln.rstrip("\n").split("\t")
|
||
d = dict(zip(header, cells))
|
||
rows.append(d)
|
||
return header, rows
|
||
|
||
if __name__ == "__main__":
|
||
tr = terminologist_rows()
|
||
print("terminologist output lines:", len(tr), "malformed:", sum(r["malformed"] for r in tr))
|
||
bs = parse_bankstop()
|
||
print("bank-stop terms:", len(bs))
|
||
ms = parse_minedsig()
|
||
print("mined-signature records:", len(ms))
|
||
h, bf = parse_bankfull()
|
||
print("BANK-FULL rows:", len(bf), "header:", h)
|