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

#!/usr/bin/env python3
"""
Independently cross-verify a single candidate citation. This is the anti-
hallucination check: it never trusts a single source. If a check cannot be
run at all (e.g. no network), that check is reported as "skipped" -- never
silently counted as a pass.

CLI usage:
    python3 verify_citation.py --title "Compensation of magnetic ..." \\
        --arxiv-id 2401.12345 --doi 10.1109/TGRS.2024.1234567

Importable:
    from verify_citation import verify
"""
import sys
import os
import json
import argparse
import difflib
import urllib.parse
import urllib.error
import xml.etree.ElementTree as ET

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from http_utils import fetch

ATOM_NS = {"atom": "http://www.w3.org/2005/Atom"}


def title_similarity(a, b):
    if not a or not b:
        return 0.0
    return difflib.SequenceMatcher(None, a.lower().strip(), b.lower().strip()).ratio()


def check_arxiv_id(arxiv_id, timeout=20):
    """Confirm an arXiv ID actually resolves to a real paper."""
    try:
        url = f"http://export.arxiv.org/api/query?id_list={urllib.parse.quote(arxiv_id)}"
        data = fetch(url, timeout=timeout)
        root = ET.fromstring(data)
        entry = root.find("atom:entry", ATOM_NS)
        if entry is None:
            return {"status": "fail", "reason": "arXiv ID not found"}
        title_el = entry.find("atom:title", ATOM_NS)
        title = " ".join(title_el.text.split()) if title_el is not None else None
        return {"status": "pass", "canonical_title": title}
    except Exception as e:
        return {"status": "skipped", "reason": str(e)}


def check_doi(doi, timeout=20):
    """Confirm a DOI actually resolves via Crossref."""
    try:
        url = f"https://api.crossref.org/works/{urllib.parse.quote(doi)}"
        data = json.loads(fetch(url, headers={"User-Agent": "literature-search-verify-skill/1.0"}, timeout=timeout))
        titles = data.get("message", {}).get("title") or []
        return {"status": "pass", "canonical_title": titles[0] if titles else None}
    except urllib.error.HTTPError as e:
        if e.code == 404:
            return {"status": "fail", "reason": "DOI not found in Crossref"}
        return {"status": "skipped", "reason": f"HTTP {e.code}"}
    except Exception as e:
        return {"status": "skipped", "reason": str(e)}


def _best_title_match(candidates, title, title_field="title"):
    best = max(candidates, key=lambda p: title_similarity(title, p.get(title_field, "") or ""))
    sim = title_similarity(title, best.get(title_field, "") or "")
    return sim, best.get(title_field)


def check_title_cross_source_s2(title, timeout=20):
    """Independently re-search by title on Semantic Scholar and require a
    near-exact title match. This is what catches a plausible-sounding but
    entirely invented title/author combination."""
    try:
        params = {"query": title, "limit": 3, "fields": "title"}
        url = f"https://api.semanticscholar.org/graph/v1/paper/search?{urllib.parse.urlencode(params)}"
        data = json.loads(fetch(url, timeout=timeout))
        candidates = data.get("data", []) or []
        if not candidates:
            return {"status": "fail", "reason": "no matching title found on Semantic Scholar"}
        sim, matched = _best_title_match(candidates, title)
        status = "pass" if sim >= 0.9 else "fail"
        return {"status": status, "similarity": round(sim, 3), "matched_title": matched}
    except Exception as e:
        return {"status": "skipped", "reason": str(e)}


def check_title_cross_source_openalex(title, timeout=20):
    """Same idea as check_title_cross_source_s2 but against OpenAlex -- a
    second, independent source for the title cross-check so a single source
    (S2 in particular has real-world rate-limit issues) being unavailable
    doesn't drop this whole check to 'skipped' for a batch of candidates."""
    try:
        params = {"search": title, "per_page": 3, "select": "display_name"}
        url = f"https://api.openalex.org/works?{urllib.parse.urlencode(params)}"
        data = json.loads(fetch(url, timeout=timeout))
        candidates = data.get("results", []) or []
        if not candidates:
            return {"status": "fail", "reason": "no matching title found on OpenAlex"}
        sim, matched = _best_title_match(candidates, title, title_field="display_name")
        status = "pass" if sim >= 0.9 else "fail"
        return {"status": status, "similarity": round(sim, 3), "matched_title": matched}
    except Exception as e:
        return {"status": "skipped", "reason": str(e)}


def _combine_title_checks(s2_result, openalex_result):
    """Merge the two independent title cross-checks into one verdict-relevant
    check. Coverage differs between S2 and OpenAlex, so one source not
    indexing a (real) paper yet is common and must not, by itself, read as
    "actively contradicted" the way a single-source design would have taken
    it -- a pass from either source is enough; it only counts as an active
    fail once neither source could corroborate it and at least one of them
    genuinely searched (wasn't just skipped for network reasons)."""
    sources = {"semantic_scholar": s2_result, "openalex": openalex_result}
    passing = [r for r in sources.values() if r["status"] == "pass"]
    if passing:
        best = max(passing, key=lambda r: r.get("similarity", 0))
        return {"status": "pass", "sources": sources, "similarity": best.get("similarity"),
                "matched_title": best.get("matched_title")}
    non_skipped = [r for r in sources.values() if r["status"] != "skipped"]
    if non_skipped:
        return {"status": "fail", "sources": sources,
                "reason": "no title match on either Semantic Scholar or OpenAlex"}
    return {"status": "skipped", "sources": sources, "reason": "both sources unavailable"}


def verify(title=None, arxiv_id=None, doi=None):
    checks = {}
    if arxiv_id:
        checks["arxiv_id_check"] = check_arxiv_id(arxiv_id)
    if doi:
        checks["doi_check"] = check_doi(doi)
    if title:
        checks["title_cross_source_check"] = _combine_title_checks(
            check_title_cross_source_s2(title),
            check_title_cross_source_openalex(title),
        )

    passed = [c for c in checks.values() if c["status"] == "pass"]
    failed = [c for c in checks.values() if c["status"] == "fail"]

    if failed:
        verdict = "suspect"          # something actively contradicted it
    elif passed:
        verdict = "verified"         # at least one independent check passed
    else:
        verdict = "unverified"       # everything skipped (e.g. no network) -- NOT the same as verified

    return {"title": title, "arxiv_id": arxiv_id, "doi": doi, "verdict": verdict, "checks": checks}


if __name__ == "__main__":
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--title", default=None)
    ap.add_argument("--arxiv-id", default=None)
    ap.add_argument("--doi", default=None)
    args = ap.parse_args()
    if not any([args.title, args.arxiv_id, args.doi]):
        print(json.dumps({"error": "provide at least one of --title/--arxiv-id/--doi"}))
        sys.exit(1)
    result = verify(args.title, args.arxiv_id, args.doi)
    print(json.dumps(result, ensure_ascii=False, indent=2))