ai-research-survey

classify-paper-type.py (6692B)

---

 #!/usr/bin/env python3
 """
 Preliminary paper type classification using Haiku.
 
 Reads existing scan.json (title + key_findings + methodology_tags) and
 classifies into: empirical, benchmark-creation, survey, position, theoretical.
 
 Writes result to papers/{slug}/paper_type.json (separate file, non-destructive).
 
 Usage:
     python3 scripts/classify-paper-type.py                    # All unclassified
     python3 scripts/classify-paper-type.py --limit 50         # First N
     python3 scripts/classify-paper-type.py --parallel 8       # Concurrent (Haiku is fast+cheap)
     python3 scripts/classify-paper-type.py --id metr-rct-2025 # Specific paper
     python3 scripts/classify-paper-type.py --force            # Re-classify all
 """
 
 import json
 import subprocess
 import sys
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from pathlib import Path
 
 ROOT = Path(__file__).resolve().parent.parent
 PAPERS_DIR = ROOT / "papers"
 
 PROMPT = """Classify this research paper into exactly ONE category.
 
 Categories:
 1. **empirical** — runs experiments, reports quantitative results on benchmarks or datasets. The primary contribution is experimental findings.
 2. **benchmark-creation** — introduces a new benchmark, dataset, or evaluation framework. May run baselines, but the primary contribution is the benchmark itself.
 3. **survey** — reviews, surveys, or meta-analyzes existing work. Primary contribution is synthesis of the field.
 4. **position** — argues a viewpoint, proposes a conceptual framework, or makes prescriptive claims without experimental validation. Includes vision papers and opinion pieces.
 5. **theoretical** — proves something mathematically or analyzes properties formally. Primary contribution is theorems, proofs, or formal analysis.
 
 Paper information:
 Title: {title}
 Methodology tags: {tags}
 Key findings: {key_findings}
 
 Respond with ONLY a JSON object:
 {{"paper_type": "<one of: empirical, benchmark-creation, survey, position, theoretical>", "reason": "<one sentence>"}}"""
 
 
 def classify_one(paper_id, force=False):
     """Classify one paper. Returns (paper_id, type, reason) or (paper_id, None, error)."""
     scan_path = PAPERS_DIR / paper_id / "scan.json"
     type_path = PAPERS_DIR / paper_id / "paper_type.json"
 
     if type_path.exists() and not force:
         with open(type_path) as f:
             existing = json.load(f)
         return paper_id, existing.get("paper_type"), "already classified"
 
     if not scan_path.exists():
         return paper_id, None, "no scan.json"
 
     with open(scan_path) as f:
         scan = json.load(f)
 
     if scan.get("scan_version", 1) < 2:
         return paper_id, None, "v1 scan"
 
     paper = scan.get("paper", {})
     prompt = PROMPT.format(
         title=paper.get("title", ""),
         tags=", ".join(scan.get("methodology_tags", [])),
         key_findings=scan.get("key_findings", "")[:500],
     )
 
     try:
         result = subprocess.run(
             ["claude", "-p", "-", "--model", "haiku", "--max-turns", "1"],
             input=prompt,
             capture_output=True, text=True, timeout=30,
             cwd=str(ROOT),
         )
 
         if result.returncode != 0:
             return paper_id, None, f"claude exit {result.returncode}"
 
         output = result.stdout.strip()
         json_start = output.find("{")
         json_end = output.rfind("}") + 1
         if json_start == -1 or json_end == 0:
             return paper_id, None, "no JSON in output"
 
         parsed = json.loads(output[json_start:json_end])
         paper_type = parsed.get("paper_type", "")
         reason = parsed.get("reason", "")
 
         valid_types = ["empirical", "benchmark-creation", "survey", "position", "theoretical"]
         if paper_type not in valid_types:
             return paper_id, None, f"invalid type: {paper_type}"
 
         # Write separate file (non-destructive)
         with open(type_path, "w") as f:
             json.dump({"paper_type": paper_type, "reason": reason}, f, ensure_ascii=False, indent=2)
 
         return paper_id, paper_type, reason
 
     except json.JSONDecodeError as e:
         return paper_id, None, f"JSON parse error: {e}"
     except subprocess.TimeoutExpired:
         return paper_id, None, "timeout"
     except Exception as e:
         return paper_id, None, f"error: {e}"
 
 
 def main():
     args = sys.argv[1:]
     force = "--force" in args
     limit = None
     specific_id = None
     parallel = 1
 
     for i, arg in enumerate(args):
         if arg == "--limit" and i + 1 < len(args):
             limit = int(args[i + 1])
         if arg == "--id" and i + 1 < len(args):
             specific_id = args[i + 1]
         if arg == "--parallel" and i + 1 < len(args):
             parallel = int(args[i + 1])
 
     # Collect candidates
     candidates = []
     for scan_path in sorted(PAPERS_DIR.glob("*/scan.json")):
         pid = scan_path.parent.name
         if specific_id and pid != specific_id:
             continue
         with open(scan_path) as f:
             s = json.load(f)
         if s.get("scan_version", 1) < 2:
             continue
         type_path = scan_path.parent / "paper_type.json"
         if type_path.exists() and not force and not specific_id:
             continue
         candidates.append(pid)
 
     if limit:
         candidates = candidates[:limit]
 
     if not candidates:
         print("No papers to classify.")
         return
 
     print(f"Classifying {len(candidates)} papers"
           f"{f' (parallel={parallel})' if parallel > 1 else ''}:\n")
 
     from collections import Counter
     type_counts = Counter()
     failures = 0
 
     if parallel > 1:
         with ThreadPoolExecutor(max_workers=parallel) as executor:
             futures = {executor.submit(classify_one, pid, force): pid for pid in candidates}
             for future in as_completed(futures):
                 pid, ptype, reason = future.result()
                 if ptype:
                     type_counts[ptype] += 1
                 else:
                     failures += 1
                     print(f"  FAIL: {pid} — {reason}")
     else:
         for i, pid in enumerate(candidates):
             _, ptype, reason = classify_one(pid, force)
             if ptype:
                 type_counts[ptype] += 1
             else:
                 failures += 1
                 print(f"  FAIL: {pid} — {reason}")
             if (i + 1) % 50 == 0:
                 print(f"  ... {i+1}/{len(candidates)} done")
 
     total = sum(type_counts.values())
     print(f"\nDone. Classified: {total}, Failed: {failures}")
     print(f"Distribution:")
     for ptype, count in type_counts.most_common():
         print(f"  {ptype:20s} {count:>4d} ({count/total*100:.0f}%)")
 
 
 if __name__ == "__main__":
     main()