pipelines/rank12_apply.py
# Rank 12 slice 1 — APPLY the 33 .md first-heading title recoveries to
# T0_Source_Material_Registry. Deterministic; evidence = each file's own first heading
# (the D-MINING identity-recovery precedent: accuracy repair, never invention).
# Preconditions per row: the live cell still fails real_title AND the file still yields the
# same heading. Writes an evidence CSV beside this script. BOM/CRLF preserved; tmp+os.replace.
import csv, glob, io, os, re
ROOT = r'C:\dev\humanity-forgotten'
SP = os.path.dirname(os.path.abspath(__file__))
EV = os.path.join(SP, 'rank12_slice1_evidence.csv')
WORD3 = re.compile(r'[A-Za-z]{3,}')
ARTIFACT = ('admin,+journal', 'journal+manager', 'journal manager')
def real_title(t):
t = (t or '').strip()
if not t: return False
low = t.lower()
if any(m in low for m in ARTIFACT): return False
return len(WORD3.findall(t)) >= 2
def md_title(path):
with open(path, encoding='utf-8', errors='replace') as f:
for i, line in enumerate(f):
if i > 30: break
s = line.strip()
if not s: continue
return re.sub(r'^#+\s*', '', s)[:160]
return ''
reg = glob.glob(os.path.join(ROOT, r'registries\T0_Source_Material_Registry\*.csv'))[0]
raw = open(reg, 'rb').read()
bom = raw.startswith(b'\xef\xbb\xbf')
crlf = b'\r\n' in raw
rows = list(csv.reader(io.StringIO(raw.decode('utf-8-sig'))))
hdr = rows[0]
ix = {c: hdr.index(c) for c in ('source_id', 'source_filepath', 'source_title',
'extraction_status', 'story_seed_tags', 'recent_changes')}
applied = []
for r in rows[1:]:
while len(r) < len(hdr): r.append('')
if not (r[ix['story_seed_tags']] or '').strip(): continue
if (r[ix['extraction_status']] or '').strip().startswith('deep_read'): continue
if real_title(r[ix['source_title']]): continue
fp = (r[ix['source_filepath']] or '').strip()
if not fp.lower().endswith('.md') or not os.path.exists(fp): continue
t = md_title(fp)
if not t or not real_title(t): continue
old = r[ix['source_title']]
r[ix['source_title']] = t
rc = (r[ix['recent_changes']] or '').strip()
note = '2026-07-29 source_title recovered from the file own first heading (rank-12 slice 1, weave-blind cleanup)'
if rc in ('', '[]'):
r[ix['recent_changes']] = f'["{note}"]'
else:
assert rc.endswith(']'), f'{r[ix["source_id"]]}: unexpected recent_changes shape'
r[ix['recent_changes']] = rc[:-1] + f', "{note}"]'
applied.append({'source_id': r[ix['source_id']], 'old_title': old, 'new_title': t, 'file': fp})
assert len(applied) == 33, f'expected exactly 33 recoveries, found {len(applied)} (re-scan before applying)'
w = len(hdr)
assert all(len(r) == w for r in rows), 'ragged rows after edit'
buf = io.StringIO()
csv.writer(buf, lineterminator='\r\n' if crlf else '\n').writerows(rows)
data = buf.getvalue().encode('utf-8')
if bom: data = b'\xef\xbb\xbf' + data
tmp = reg + '.tmp'
open(tmp, 'wb').write(data)
os.replace(tmp, reg)
with open(EV, 'w', encoding='utf-8', newline='') as f:
wr = csv.DictWriter(f, fieldnames=['source_id', 'old_title', 'new_title', 'file'])
wr.writeheader()
wr.writerows(applied)
print(f'applied {len(applied)} title recoveries; evidence -> {EV}')