autoPaper/.claude/skills/literature-search-verify/scripts/literature_search.py

#!/usr/bin/env python3
"""
End-to-end literature search: query arXiv + Semantic Scholar + Crossref +
OpenAlex, merge/dedupe candidates, independently verify each one, and emit
both a human-readable report and BibTeX for the entries that passed
verification.

This is the one script Claude should actually call for a normal literature
search -- the individual search_*.py / verify_citation.py scripts exist
mainly as building blocks it can reuse for one-off / follow-up lookups.

CLI usage:
    python3 literature_search.py "UAV magnetic compensation Tolles-Lawson" \\
        --max-per-source 8 --bib-out refs.bib

Output: prints a JSON report to stdout (one entry per merged candidate,
with its verdict), and if --bib-out is given, writes BibTeX for every
"verified" entry to that file (never for "suspect" or "unverified" ones).
"""
import sys
import os
import json
import argparse
import difflib
import re

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from search_arxiv import search_arxiv
from search_semantic_scholar import search_s2
from search_crossref import search_crossref
from search_openalex import search_openalex
from verify_citation import verify


def _similar(a, b, threshold=0.88):
    if not a or not b:
        return False
    return difflib.SequenceMatcher(None, a.lower().strip(), b.lower().strip()).ratio() >= threshold


def merge_candidates(all_results):
    """Dedupe candidates that are the same paper found via multiple sources,
    merging their metadata (preferring whichever source has an identifier)."""
    merged = []
    for item in all_results:
        placed = False
        for m in merged:
            if _similar(item.get("title"), m.get("title")):
                # merge: fill in any missing fields, keep track of all sources
                for key in ("doi", "arxiv_id", "abstract", "venue", "year", "citation_count", "pdf_url"):
                    if not m.get(key) and item.get(key):
                        m[key] = item[key]
                m["sources"] = sorted(set(m.get("sources", [m.get("source")]) + [item.get("source")]))
                placed = True
                break
        if not placed:
            item = dict(item)
            item["sources"] = [item.get("source")]
            merged.append(item)
    return merged


def make_bibtex_key(candidate, used_keys):
    authors = candidate.get("authors") or []
    surname = "unknown"
    if authors:
        first_author = authors[0]
        surname = first_author.strip().split()[-1].lower()
        surname = re.sub(r"[^a-z]", "", surname) or "unknown"
    year = str(candidate.get("year") or "nd")
    base = f"{surname}{year}"
    key = base
    suffix = ord("a")
    while key in used_keys:
        key = f"{base}{chr(suffix)}"
        suffix += 1
    used_keys.add(key)
    return key


def to_bibtex(candidate, key):
    authors = candidate.get("authors") or []
    author_str = " and ".join(authors) if authors else "Unknown"
    title = candidate.get("title") or ""
    year = candidate.get("year") or ""
    venue = candidate.get("venue") or ""
    doi = candidate.get("doi") or ""
    arxiv_id = candidate.get("arxiv_id") or ""

    if arxiv_id and not venue:
        entry_type = "misc"
        fields = [
            ("author", author_str),
            ("title", title),
            ("year", str(year)),
            ("eprint", arxiv_id),
            ("archivePrefix", "arXiv"),
        ]
    else:
        entry_type = "article"
        fields = [
            ("author", author_str),
            ("title", title),
            ("journal", venue),
            ("year", str(year)),
        ]
        if doi:
            fields.append(("doi", doi))

    lines = [f"@{entry_type}{{{key},"]
    for k, v in fields:
        if v:
            lines.append(f"  {k} = {{{v}}},")
    lines.append("}")
    return "\n".join(lines)


def run(query, max_per_source=8):
    all_results = []
    errors = {}
    sources = (
        ("arxiv", search_arxiv),
        ("semantic_scholar", search_s2),
        ("crossref", search_crossref),
        ("openalex", search_openalex),
    )
    for name, fn in sources:
        try:
            all_results.extend(fn(query, max_per_source))
        except Exception as e:
            errors[name] = str(e)

    merged = merge_candidates(all_results)

    used_keys = set()
    for cand in merged:
        result = verify(title=cand.get("title"), arxiv_id=cand.get("arxiv_id"), doi=cand.get("doi"))
        cand["verdict"] = result["verdict"]
        cand["verification_checks"] = result["checks"]
        if result["verdict"] == "verified":
            cand["bibtex_key"] = make_bibtex_key(cand, used_keys)

    return {"query": query, "search_errors": errors, "candidates": merged}


if __name__ == "__main__":
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("query")
    ap.add_argument("--max-per-source", type=int, default=8)
    ap.add_argument("--bib-out", default=None, help="path to write BibTeX for verified entries")
    args = ap.parse_args()

    try:
        report = run(args.query, args.max_per_source)
    except Exception as e:
        print(json.dumps({"error": str(e)}, ensure_ascii=False))
        sys.exit(1)

    print(json.dumps(report, ensure_ascii=False, indent=2))

    if args.bib_out:
        verified = [c for c in report["candidates"] if c["verdict"] == "verified"]
        with open(args.bib_out, "w", encoding="utf-8") as f:
            for cand in verified:
                f.write(to_bibtex(cand, cand["bibtex_key"]))
                f.write("\n\n")
        sys.stderr.write(f"Wrote {len(verified)} verified BibTeX entries to {args.bib_out}\n")