Daan MeijerandClaude Opus 5.5
2bbabc08de
Only the bytes on disk were checked: at 296.6 GB with 2,300 ranges complete it marked 100 more (~310 GB). The marked ranges not complete count at the stored ranges' mean size, and no more are marked than fit under 300 GB. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
222 lines
10 KiB
Python
222 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""The grouping sweeps' corpus (REBUILD.md, corpus): stored XOVER chunks replayed with each build's rules.
|
|
|
|
Run on homelab from the deployment root (/mnt/bulk/usenet-rebuild):
|
|
|
|
python3 src/deploy/corpus_sweep.py capture [N] mark N more ranges to store (default 100), fetched first
|
|
python3 src/deploy/corpus_sweep.py replay edge-bench over every stored range with the deployed edge image
|
|
python3 src/deploy/corpus_sweep.py report what the replay grouped: suspects, and the change since the last build
|
|
|
|
The prefetcher stores the chunks of ranges marked `capture` in corpus/<group>/<start>-<end>/ (range.json once
|
|
complete) and reads them from there afterwards, so a reset doesn't fetch them again. Replays go to
|
|
corpus-out/<image id>/<group>/<start>-<end>.jsonl, one per build; `report` compares the newest two.
|
|
"""
|
|
import collections
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
|
|
ROOT = os.getcwd()
|
|
CORPUS = os.path.join(ROOT, 'corpus')
|
|
OUT = os.path.join(ROOT, 'corpus-out')
|
|
IMAGE = 'usenet-rebuild-edge:latest'
|
|
# The corpus may take this much of the HDD (user, 2026-10-12: up to 300 GB).
|
|
MAX_BYTES = 300e9
|
|
# Ranks in groups.tsv the new ranges come from first (the top 100), then the rest of what's indexed.
|
|
TOP = 100
|
|
|
|
|
|
def env(name):
|
|
for line in open(os.path.join(ROOT, '.env')):
|
|
if line.startswith(name + '='):
|
|
return line.split('=', 1)[1].strip()
|
|
raise SystemExit(f'{name} not in .env')
|
|
|
|
|
|
def psql(sql):
|
|
"""Runs SQL in the catalog; rows as lists of strings."""
|
|
out = subprocess.run(
|
|
['docker', 'compose', 'exec', '-T', 'postgres', 'psql', '-U', env('DB_USER'), '-d', env('DB_DATABASE'), '-At', '-F', '\t', '-v', 'ON_ERROR_STOP=1'],
|
|
input=sql, capture_output=True, text=True, check=True,
|
|
).stdout
|
|
# Without the command tag a RETURNING prints after its rows ("UPDATE 100").
|
|
return [line.split('\t') for line in out.splitlines() if line and not line.startswith(('UPDATE ', 'INSERT ', 'DELETE '))]
|
|
|
|
|
|
def corpus_bytes():
|
|
total = 0
|
|
for dirpath, _, files in os.walk(CORPUS):
|
|
total += sum(os.path.getsize(os.path.join(dirpath, f)) for f in files)
|
|
return total
|
|
|
|
|
|
def capture(n):
|
|
used = corpus_bytes()
|
|
# Marked ranges not complete yet take room too: at the stored ranges' mean size, less what they hold already
|
|
# (2026-10-13: 2,300 complete at 296.6 GB and 100 more marked would have made ~310 GB).
|
|
complete = sum(1 for _ in complete_ranges())
|
|
marked = int(psql('SELECT count(*) FROM ranges WHERE capture;')[0][0])
|
|
mean = used / complete if complete else 0
|
|
pending = max(0, marked - complete) * mean
|
|
room = MAX_BYTES - max(used, complete * mean + pending)
|
|
n = min(n, int(room // mean)) if mean else n
|
|
if n <= 0:
|
|
print(f'corpus at {used / 1e9:.1f} GB of {MAX_BYTES / 1e9:.0f} (with {marked - complete} ranges still filling, ~{(complete * mean + pending) / 1e9:.0f} GB): nothing more marked')
|
|
return
|
|
top = [line.split('\t')[1] for line in open(os.path.join(ROOT, 'groups.tsv')).read().splitlines()[:TOP]]
|
|
names = ','.join("'" + g.replace("'", "''") + "'" for g in top)
|
|
# Spread over the groups and their history: per group, the ranges farthest from those stored already
|
|
# (at random in a group without any), the top-100 groups first. Only ranges not started (a range half
|
|
# spooled would miss its first chunks; a done one is stored at the next reset, which re-reads them all).
|
|
# Priority 0: fetched before the rest of the history (the tail stays first).
|
|
rows = psql(f"""
|
|
WITH stored AS (SELECT group_id, array_agg(start_id) AS starts FROM ranges WHERE capture GROUP BY 1),
|
|
pick AS (
|
|
SELECT r.id, g.name IN ({names}) AS top, row_number() OVER (PARTITION BY r.group_id ORDER BY
|
|
coalesce((SELECT min(abs(r.start_id - s)) FROM unnest(st.starts) s), 0) DESC, random()) AS n
|
|
FROM ranges r JOIN groups g ON g.id = r.group_id LEFT JOIN stored st ON st.group_id = r.group_id
|
|
WHERE r.priority >= 0 AND NOT r.capture AND r.status = 'planned' AND r.spooled_chunks = 0 AND r.next_id = r.start_id
|
|
),
|
|
chosen AS (SELECT id FROM pick ORDER BY top DESC, n, random() LIMIT {int(n)})
|
|
UPDATE ranges r SET capture = true, priority = 0 FROM chosen WHERE r.id = chosen.id RETURNING r.id;
|
|
""")
|
|
total = psql('SELECT count(*), count(*) FILTER (WHERE status = \'done\') FROM ranges WHERE capture;')[0]
|
|
print(f'marked {len(rows)} ranges; {total[0]} marked in all, {total[1]} indexed; corpus {used / 1e9:.1f} GB')
|
|
|
|
|
|
def complete_ranges():
|
|
for group in sorted(os.listdir(CORPUS)) if os.path.isdir(CORPUS) else []:
|
|
for rng in sorted(os.listdir(os.path.join(CORPUS, group))):
|
|
if os.path.exists(os.path.join(CORPUS, group, rng, 'range.json')):
|
|
yield group, rng
|
|
|
|
|
|
def image_id():
|
|
return subprocess.run(['docker', 'image', 'inspect', '-f', '{{.Id}}', IMAGE], capture_output=True, text=True, check=True).stdout.strip().split(':')[-1][:12]
|
|
|
|
|
|
def replay():
|
|
build = image_id()
|
|
out = os.path.join(OUT, build)
|
|
todo = [(g, r) for g, r in complete_ranges() if not os.path.exists(os.path.join(out, g, r + '.jsonl'))]
|
|
for g, _ in todo:
|
|
os.makedirs(os.path.join(out, g), exist_ok=True)
|
|
print(f'build {build}: {len(todo)} ranges to replay', flush=True)
|
|
if not todo:
|
|
return
|
|
# One container, 6 at a time (the edge worker keeps its threads).
|
|
script = 'xargs -P 6 -n 2 sh -c \'edge-bench "/corpus/$0/$1" --releases "/out/$0/$1.jsonl.part" >/dev/null 2>&1 && mv "/out/$0/$1.jsonl.part" "/out/$0/$1.jsonl"\''
|
|
started = datetime.now()
|
|
subprocess.run(
|
|
['docker', 'run', '--rm', '-i', '--user', '1000:1000', '-v', f'{CORPUS}:/corpus:ro', '-v', f'{out}:/out', '--entrypoint', 'sh', IMAGE, '-c', script],
|
|
input=''.join(f'{g} {r}\n' for g, r in todo), text=True, check=True,
|
|
)
|
|
print(f'replayed in {(datetime.now() - started).total_seconds():.0f} s')
|
|
# Older builds' replays beyond the last two go.
|
|
builds = sorted((os.path.getmtime(os.path.join(OUT, b)), b) for b in os.listdir(OUT))
|
|
for _, b in builds[:-2]:
|
|
shutil.rmtree(os.path.join(OUT, b))
|
|
|
|
|
|
def replayed(build):
|
|
"""The ranges a build replayed, as paths relative to its directory."""
|
|
root = os.path.join(OUT, build)
|
|
return {os.path.relpath(os.path.join(d, f), root) for d, _, files in os.walk(root) for f in files if f.endswith('.jsonl')}
|
|
|
|
|
|
def load(build, only=None):
|
|
"""Releases of one build's replay (of the ranges in `only`, when given), merged by (group, key) as the catalog does."""
|
|
releases = {}
|
|
for rel in sorted(replayed(build)):
|
|
if only is not None and rel not in only:
|
|
continue
|
|
dirpath, f = os.path.split(os.path.join(OUT, build, rel))
|
|
if True:
|
|
for line in open(os.path.join(dirpath, f), encoding='utf-8', errors='replace'):
|
|
r = json.loads(line)
|
|
k = (r['group'], r['key'])
|
|
have = releases.get(k)
|
|
if have is None:
|
|
releases[k] = r
|
|
else:
|
|
have['files'] += r['files']
|
|
have['segments'] += r['segments']
|
|
have['subjects'] = (have['subjects'] + r['subjects'])[:3]
|
|
have['fileTotal'] = have['fileTotal'] or r['fileTotal']
|
|
return releases
|
|
|
|
|
|
def hour(r):
|
|
return datetime.fromtimestamp((r['postedAt'] or 0) / 1000, timezone.utc).strftime('%Y-%m-%d %H')
|
|
|
|
|
|
def show(r, prefix=' '):
|
|
print(f"{prefix}{r['files']} files, total {r['fileTotal']}, {r['source']}: {r['name'][:80]!r} ({r['poster'][:30]}, {r['group']})")
|
|
for s in r['subjects'][:2]:
|
|
print(f'{prefix} {s[:150]}')
|
|
|
|
|
|
def report():
|
|
builds = sorted((os.path.getmtime(os.path.join(OUT, b)), b) for b in os.listdir(OUT)) if os.path.isdir(OUT) else []
|
|
if not builds:
|
|
raise SystemExit('nothing replayed yet')
|
|
build = builds[-1][1]
|
|
rel = load(build)
|
|
ranges = sum(1 for _ in complete_ranges())
|
|
print(f'build {build}: {len(rel):,} releases, {sum(r["files"] for r in rel.values()):,} files from {ranges} stored ranges ({corpus_bytes() / 1e9:.1f} GB)')
|
|
|
|
print('\n== same name, poster and hour: 4+ releases (under-grouping)')
|
|
clusters = collections.defaultdict(list)
|
|
for r in rel.values():
|
|
clusters[(r['group'], r['poster'], r['name'], hour(r))].append(r)
|
|
for (g, p, name, h), rs in sorted(clusters.items(), key=lambda kv: -len(kv[1]))[:25]:
|
|
if len(rs) < 4:
|
|
break
|
|
print(f" {len(rs)} releases, {sum(r['files'] for r in rs)} files: {name[:70]!r} ({p[:25]}, {g}, {h})")
|
|
for s in rs[0]['subjects'][:1] + rs[-1]['subjects'][:1]:
|
|
print(f' {s[:150]}')
|
|
|
|
print('\n== single files of a counter of 5+, per group and poster (under-grouping)')
|
|
singles = collections.defaultdict(list)
|
|
for r in rel.values():
|
|
if r['files'] == 1 and (r['fileTotal'] or 0) >= 5:
|
|
singles[(r['group'], r['poster'])].append(r)
|
|
for (g, p), rs in sorted(singles.items(), key=lambda kv: -len(kv[1]))[:15]:
|
|
if len(rs) < 20:
|
|
break
|
|
print(f' {len(rs)} single files: {p[:30]} ({g})')
|
|
for s in rs[0]['subjects'][:1] + rs[-1]['subjects'][:1]:
|
|
print(f' {s[:150]}')
|
|
|
|
print('\n== 2000+ files, or 3x the counter (over-grouping)')
|
|
big = [r for r in rel.values() if r['files'] > 2000 or (r['fileTotal'] and r['files'] > 3 * r['fileTotal'] and r['files'] > 50)]
|
|
for r in sorted(big, key=lambda r: -r['files'])[:20]:
|
|
show(r, ' ')
|
|
|
|
if len(builds) >= 2:
|
|
# Over the ranges both replayed: the corpus grows between builds.
|
|
both = replayed(builds[-2][1]) & replayed(build)
|
|
before, after = load(builds[-2][1], both), load(build, both)
|
|
print(f'\n== change since build {builds[-2][1]} on the {len(both)} ranges both replayed: {len(before):,} → {len(after):,} releases')
|
|
names = lambda rs: collections.Counter((r['group'], r['name']) for r in rs.values())
|
|
a, b = names(before), names(after)
|
|
moved = sorted(set(a) | set(b), key=lambda k: -abs(a[k] - b[k]))[:20]
|
|
for k in moved:
|
|
if a[k] != b[k]:
|
|
print(f' {a[k]:>6} → {b[k]:<6} {k[1][:80]!r} ({k[0]})')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
command = sys.argv[1] if len(sys.argv) > 1 else 'report'
|
|
if command == 'capture':
|
|
capture(int(sys.argv[2]) if len(sys.argv) > 2 else 100)
|
|
elif command == 'replay':
|
|
replay()
|
|
elif command == 'report':
|
|
report()
|
|
else:
|
|
raise SystemExit(__doc__)
|