Spaces:
Running
Running
siddhm11
Claude Opus 5
docs(backfill): measured estimate, and the sidecar caveat that reframes it
4f06b54 | """ | |
| Repair the truncated abstracts in the Turso corpus. | |
| THE PROBLEM | |
| ----------- | |
| Measured against live Turso on 2026-08-21: | |
| total papers 1,799,348 | |
| abstracts at the 500 cap 1,630,273 (90.6%) | |
| abstracts missing 0 | |
| titles missing 0 | |
| citation_count NULL 0 | |
| So the corpus is otherwise clean, and this is its one real defect — but it is a | |
| big one, because the truncation degrades four things at once: | |
| * the card and the paper page show a stump, often cut mid-word; | |
| * `groq_svc.explain_paper` summarises that stump, so the explanation is built | |
| from half a paper; | |
| * `readability` scores the stump, so the density badge measures a fragment; | |
| * the cross-encoder in the search path reranks on it (PHASE7 §3.2). | |
| WHY THIS IS A SCRIPT AND NOT A BACKGROUND JOB | |
| --------------------------------------------- | |
| Repairing all 1.63M rows is a bulk mutation of the production metadata store | |
| and a multi-hour run against a third party's rate-limited API. That is a | |
| decision for a person, not something a request handler should start. So this is | |
| explicit, resumable, and dry-run by default. | |
| It is also ordered by usefulness rather than by id: `--strategy served` repairs | |
| the papers the product actually shows first, so the first hour of running | |
| delivers most of the user-visible benefit. Paper popularity is Zipfian, so a | |
| few thousand rows cover a large share of what anyone ever reads. | |
| USAGE | |
| ----- | |
| # See the scale and the estimate, change nothing: | |
| .venv/bin/python scripts/backfill_abstracts.py --dry-run | |
| # Repair the 500 most-cited truncated papers: | |
| .venv/bin/python scripts/backfill_abstracts.py --limit 500 --apply | |
| # Repair everything (long; resumable — just re-run it): | |
| .venv/bin/python scripts/backfill_abstracts.py --all --apply | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import asyncio | |
| import sys | |
| import time | |
| sys.path.insert(0, ".") | |
| import xml.etree.ElementTree as ET # noqa: E402 | |
| from app import arxiv_svc, config, http_client # noqa: E402 | |
| # arXiv asks for no more than ONE request every three seconds for bulk use. | |
| # | |
| # The first run violated that badly and got 429'd from batch 11 onward, losing | |
| # 9 of 20 batches. The cause was reusing `arxiv_svc.fetch_metadata_batch`, which | |
| # is tuned for small interactive lookups: it splits any input into 20-id | |
| # sub-requests fired at ~3/s. So each "batch of 100" was really a burst of five | |
| # requests, and the 3s pause sat between bursts rather than between requests. | |
| # | |
| # This now issues exactly one HTTP request per batch and sleeps between them, so | |
| # the pause means what it says. A larger id_list per request is also what makes | |
| # the full sweep tractable: 200 ids per request is 7,183 requests, where 20 ids | |
| # would be 71,829. | |
| BATCH = 200 | |
| PAUSE_S = 3.0 | |
| MAX_RETRIES = 4 | |
| # Measured over real batches of 200 on 2026-08-24: 400 rows in 15s, | |
| # 200 rows in 8s. Used for the estimate only; nothing paces off it. | |
| SECONDS_PER_BATCH = 7.75 | |
| # Below this an abstract is not really longer than what we already hold, so | |
| # rewriting the row buys nothing. | |
| MIN_GAIN_CHARS = 40 | |
| # What "truncated" actually means: length EXACTLY at the cap. | |
| # | |
| # The first version of this used `>= 500`, which is wrong in both directions | |
| # that matter. It counted the 193,689 rows whose abstracts are legitimately | |
| # longer than 500 characters as damaged — overstating the job by 13% — and, | |
| # worse, a REPAIRED row still matches `>= 500`, so the script would have | |
| # re-selected and re-fetched its own completed work forever. It was not | |
| # resumable, despite a test asserting that it was. | |
| TRUNCATED = "length(abstract_preview) = 500" | |
| async def _pipeline(stmts: list[dict], timeout: int = 120) -> list: | |
| """Run several statements in ONE round trip. | |
| The write path originally issued one UPDATE per row. At ~150ms per round | |
| trip that is not a detail: a full sweep of 1.63M rows would spend ~68 hours | |
| on HTTP alone, dwarfing the 13.6 hours of arXiv rate limiting the estimate | |
| was based on. Turso's pipeline API takes a batch, so a batch is what it gets. | |
| """ | |
| url = config.TURSO_URL.replace("libsql://", "https://").rstrip("/") | |
| r = await http_client.get_client().post( | |
| f"{url}/v2/pipeline", | |
| json={"requests": [{"type": "execute", "stmt": s} for s in stmts] | |
| + [{"type": "close"}]}, | |
| headers={"Authorization": f"Bearer {config.TURSO_DB_TOKEN}", | |
| "Content-Type": "application/json"}, | |
| timeout=timeout) | |
| r.raise_for_status() | |
| out = [] | |
| for res in r.json()["results"]: | |
| if res.get("type") == "error": | |
| raise RuntimeError(str(res.get("error"))[:300]) | |
| resp = res.get("response", {}) | |
| if resp.get("type") == "execute": | |
| out.append(resp["result"]["rows"]) | |
| return out | |
| async def _turso(sql: str, args: list | None = None, timeout: int = 90): | |
| stmt = {"sql": sql} | |
| if args is not None: | |
| stmt["args"] = args | |
| rows = await _pipeline([stmt], timeout=timeout) | |
| return rows[0] if rows else [] | |
| def _cell(v): | |
| if v is None: | |
| return {"type": "null", "value": None} | |
| if isinstance(v, int): | |
| return {"type": "integer", "value": str(v)} | |
| return {"type": "text", "value": str(v)} | |
| _NS = {"atom": "http://www.w3.org/2005/Atom"} | |
| async def fetch_abstracts(ids: list[str]) -> dict[str, str]: | |
| """ONE arXiv request for the whole batch, with backoff on 429. | |
| Deliberately not `arxiv_svc.fetch_metadata_batch`: that is tuned for small | |
| interactive lookups and fans any input out into 20-id sub-requests at ~3/s, | |
| which is what got this script rate-limited out of 9 of its first 20 batches. | |
| Here one batch is one request, so the pause between them means what it says. | |
| """ | |
| params = {"id_list": ",".join(ids), "max_results": str(len(ids))} | |
| delay = PAUSE_S | |
| for attempt in range(MAX_RETRIES): | |
| try: | |
| r = await http_client.get_client().get( | |
| config.ARXIV_API_URL, params=params, timeout=60) | |
| if r.status_code == 429: | |
| # arXiv is explicitly asking us to slow down. Honour it rather | |
| # than burning the remaining retries at the same rate. | |
| delay *= 2 | |
| print(f" 429 — backing off {delay:.0f}s " | |
| f"(attempt {attempt + 1}/{MAX_RETRIES})") | |
| await asyncio.sleep(delay) | |
| continue | |
| r.raise_for_status() | |
| except Exception as e: | |
| if attempt == MAX_RETRIES - 1: | |
| raise | |
| delay *= 2 | |
| print(f" {type(e).__name__} — retrying in {delay:.0f}s") | |
| await asyncio.sleep(delay) | |
| continue | |
| out: dict[str, str] = {} | |
| root = ET.fromstring(r.text) | |
| for entry in root.findall("atom:entry", _NS): | |
| raw_id = (entry.findtext("atom:id", "", _NS) or "") | |
| summary = (entry.findtext("atom:summary", "", _NS) or "").strip() | |
| # arxiv_svc's own normaliser, not a hand-rolled one. Old-style ids | |
| # carry a category prefix (math/0309136v1) and a naive | |
| # rsplit("/")+split("v") mangles both halves — and CLAUDE.md §3.9 | |
| # makes id integrity a hard rule. | |
| aid = arxiv_svc._normalise_id(raw_id) | |
| if aid and summary: | |
| out[aid] = " ".join(summary.split()) | |
| return out | |
| return {} | |
| async def survey() -> dict: | |
| total = int((await _turso("SELECT COUNT(*) FROM papers"))[0][0]["value"]) | |
| capped = int((await _turso( | |
| f"SELECT COUNT(*) FROM papers WHERE {TRUNCATED}"))[0][0]["value"]) | |
| return {"total": total, "capped": capped} | |
| async def pick(limit: int, strategy: str) -> list[str]: | |
| """The truncated ids to repair, most useful first.""" | |
| order = { | |
| # What the product actually surfaces. Trending and the feed both lean on | |
| # citation_count, so this is the closest cheap proxy for "will be read". | |
| "served": "ORDER BY citation_count DESC", | |
| # Newest first — what a freshness-driven feed shows. | |
| "recent": "ORDER BY update_date DESC", | |
| # Whatever the index yields; fastest to scan, useful for a full sweep. | |
| "any": "", | |
| }[strategy] | |
| rows = await _turso( | |
| f"SELECT arxiv_id FROM papers WHERE {TRUNCATED} " | |
| f"{order} LIMIT {int(limit)}") | |
| return [r[0]["value"] for r in rows] | |
| async def repair(ids: list[str], apply: bool) -> dict: | |
| """Fetch full abstracts from arXiv and write the longer ones back.""" | |
| stats = {"fetched": 0, "longer": 0, "written": 0, "no_gain": 0, "missing": 0} | |
| for i in range(0, len(ids), BATCH): | |
| chunk = ids[i:i + BATCH] | |
| try: | |
| abstracts = await fetch_abstracts(chunk) | |
| except Exception as e: | |
| print(f" batch {i // BATCH}: fetch failed ({str(e)[:80]}) — skipping") | |
| continue | |
| updates = [] | |
| for aid in chunk: | |
| full = abstracts.get(aid) | |
| if not full: | |
| stats["missing"] += 1 | |
| continue | |
| stats["fetched"] += 1 | |
| if len(full) < 500 + MIN_GAIN_CHARS: | |
| stats["no_gain"] += 1 | |
| continue | |
| stats["longer"] += 1 | |
| updates.append((aid, full)) | |
| if apply and updates: | |
| # One round trip for the whole batch, not one per row. | |
| try: | |
| await _pipeline([ | |
| {"sql": "UPDATE papers SET abstract_preview = ? " | |
| "WHERE arxiv_id = ?", | |
| "args": [_cell(full), _cell(aid)]} | |
| for aid, full in updates | |
| ]) | |
| stats["written"] += len(updates) | |
| except Exception as e: | |
| print(f" batch write failed ({str(e)[:100]}) — rows unchanged") | |
| done = min(i + BATCH, len(ids)) | |
| print(f" {done}/{len(ids)} longer={stats['longer']} " | |
| f"written={stats['written']} no-gain={stats['no_gain']}") | |
| if done < len(ids): | |
| await asyncio.sleep(PAUSE_S) | |
| return stats | |
| async def main() -> None: | |
| ap = argparse.ArgumentParser(description=__doc__) | |
| ap.add_argument("--limit", type=int, default=200) | |
| ap.add_argument("--all", action="store_true", help="every truncated row") | |
| ap.add_argument("--strategy", choices=("served", "recent", "any"), | |
| default="served") | |
| ap.add_argument("--apply", action="store_true", | |
| help="actually write. Without this nothing is modified.") | |
| ap.add_argument("--dry-run", action="store_true") | |
| args = ap.parse_args() | |
| if not config.TURSO_URL or not config.TURSO_DB_TOKEN: | |
| print("TURSO_URL / TURSO_DB_TOKEN not configured") | |
| return | |
| s = await survey() | |
| pct = s["capped"] / s["total"] * 100 if s["total"] else 0 | |
| print(f"corpus: {s['total']:,} papers, {s['capped']:,} truncated ({pct:.1f}%)\n") | |
| batches = -(-s["capped"] // BATCH) | |
| # MEASURED, not modelled. Three estimates were wrong before this one: | |
| # 13.6h counted only the arXiv pause; 22.6h added a guessed write cost but | |
| # still used an inflated truncation count; 10h fixed the count but kept the | |
| # guess. Timing real batches gives ~7.75s each — arXiv pause, the fetch, and | |
| # one batched write round trip. | |
| hours = batches * SECONDS_PER_BATCH / 3600 | |
| print(f"full repair would be ~{batches:,} batches of {BATCH} " | |
| f"(~{hours:.1f} hours, arXiv rate limit + one write round trip each)\n") | |
| if args.dry_run or not args.apply: | |
| n = s["capped"] if args.all else args.limit | |
| print(f"DRY RUN — would repair {min(n, s['capped']):,} rows " | |
| f"(strategy: {args.strategy})") | |
| # Sample for real. The whole exercise rests on arXiv returning fuller | |
| # text than we hold, and that is an assumption worth testing before | |
| # committing thirteen hours to it. | |
| sample = await pick(min(20, n), args.strategy) | |
| if sample: | |
| abstracts = await fetch_abstracts(sample) | |
| gains = [len(v) for v in abstracts.values() if v] | |
| if gains: | |
| longer = sum(1 for g in gains if g >= 500 + MIN_GAIN_CHARS) | |
| print(f"\n sampled {len(gains)} from arXiv:") | |
| print(f" median full length {sorted(gains)[len(gains)//2]:,} chars " | |
| f"(stored: 500)") | |
| print(f" materially longer {longer}/{len(gains)}") | |
| print(f" mean gain " | |
| f"{sum(gains)/len(gains) - 500:+,.0f} chars per abstract") | |
| else: | |
| print("\n sample returned nothing from arXiv — check connectivity") | |
| print("\nre-run with --apply to write.") | |
| await http_client.aclose() | |
| return | |
| limit = s["capped"] if args.all else args.limit | |
| ids = await pick(limit, args.strategy) | |
| print(f"repairing {len(ids):,} rows, {args.strategy} first\n") | |
| t0 = time.time() | |
| stats = await repair(ids, apply=True) | |
| print(f"\ndone in {time.time() - t0:.0f}s: {stats}") | |
| print("re-run to continue; already-repaired rows drop out of the selection.") | |
| await http_client.aclose() | |
| if __name__ == "__main__": | |
| asyncio.run(main()) | |