diff --git a/backend/scripts/backfill_hf_author_claims.py b/backend/scripts/backfill_hf_author_claims.py new file mode 100644 index 00000000..736f35ee --- /dev/null +++ b/backend/scripts/backfill_hf_author_claims.py @@ -0,0 +1,118 @@ +"""Backfill Hugging Face claims on legacy author links and split off namesakes. + +Two steps (see backend/services/author_claim_backfill_service.py): + + # 1. Cache HF author lists of every paper with an unclaimed HF-linked link. + # Read-only on the database, paced for HF's rate limit; rerun to resume. + uv run --env-file backend/keys.env python -m backend.scripts.backfill_hf_author_claims \ + fetch --cache claims.jsonl + + # 2. Replay the cache, fix doubled links, split namesakes, merge provable + # duplicates, delete empty authors. Rolled back (a dry run) unless + # --apply; every change is written to --report as JSON lines either way. + uv run --env-file backend/keys.env python -m backend.scripts.backfill_hf_author_claims \ + run --cache claims.jsonl --report report.jsonl [--apply] +""" + +from __future__ import annotations + +import argparse +import json +import logging +from pathlib import Path + +from backend.database.connection import db_manager +from backend.services.author_claim_backfill_service import ( + DEFAULT_REQUESTS_PER_SECOND, + PHASES, + AuthorClaimCleanup, + HfPaperClient, + fetch_hf_papers, +) + +logger = logging.getLogger("backfill_hf_author_claims") + + +def _fetch(args: argparse.Namespace) -> None: + client = HfPaperClient(requests_per_second=args.rate) + with db_manager.session_local() as db: + fetched = fetch_hf_papers( + db, + client, + Path(args.cache), + limit=args.limit, + on_progress=lambda count, paper_id: logger.info( + "fetched=%d through paper %d", count, paper_id + ), + ) + logger.info("fetched %d papers into %s", fetched, args.cache) + + +def _run(args: argparse.Namespace) -> None: + phases = args.phases.split(",") + unknown = set(phases) - set(PHASES) + if unknown: + raise SystemExit(f"unknown phases: {', '.join(sorted(unknown))}") + with ( + db_manager.session_local() as db, + open(args.report, "w", encoding="utf-8") as report, + ): + cleanup = AuthorClaimCleanup( + db, lambda entry: report.write(json.dumps(entry) + "\n") + ) + steps = { + "claims": lambda: cleanup.replay_claims(Path(args.cache)), + "doubles": cleanup.fix_doubles, + "split": cleanup.split_namesakes, + "merge": cleanup.merge_namesakes, + "empty": cleanup.delete_empty_authors, + } + # Phases run in their fixed order; each builds on the previous one. + for phase in PHASES: + if phase not in phases: + continue + steps[phase]() + if args.apply: + db.commit() + logger.info("%s done: %s", phase, json.dumps(cleanup.stats.to_dict())) + if not args.apply: + db.rollback() + print( + json.dumps( + {"apply": args.apply, "phases": phases, **cleanup.stats.to_dict()}, + indent=2, + ) + ) + + +def main() -> int: + parser = argparse.ArgumentParser( + description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter + ) + commands = parser.add_subparsers(dest="command", required=True) + + fetch = commands.add_parser("fetch", help="Cache HF author lists (read-only)") + fetch.add_argument("--cache", required=True, help="JSON-lines cache to append to") + fetch.add_argument("--limit", type=int, help="Stop after this many requests") + fetch.add_argument( + "--rate", + type=float, + default=DEFAULT_REQUESTS_PER_SECOND, + help=f"HF requests per second (default {DEFAULT_REQUESTS_PER_SECOND})", + ) + + run = commands.add_parser("run", help="Replay claims and clean up authors") + run.add_argument("--cache", required=True, help="Cache written by fetch") + run.add_argument("--report", required=True, help="JSON-lines change report") + run.add_argument("--phases", default=",".join(PHASES)) + run.add_argument("--apply", action="store_true", help="Commit the changes") + + args = parser.parse_args() + logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s") + logging.getLogger("httpx").setLevel(logging.WARNING) + {"fetch": _fetch, "run": _run}[args.command](args) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/backend/services/author_claim_backfill_service.py b/backend/services/author_claim_backfill_service.py new file mode 100644 index 00000000..0afa4c2d --- /dev/null +++ b/backend/services/author_claim_backfill_service.py @@ -0,0 +1,671 @@ +"""Backfill Hugging Face claims on legacy author links and split off namesakes. + +Before PR #519 the importers matched arXiv bylines to the author row of +whoever first claimed that name on Hugging Face, so a common name's HF-linked +row (``/authors/``) lists many namesakes' papers. ``paper_authors.claimed`` +was only backfilled from HF's daily-papers list, so most legacy links are +unclaimed although HF records the account's claim on the paper itself. + +The cleanup runs in four phases, in this order: + +1. ``claims``: replay each paper's HF author list (fetched once into a JSON + lines cache by :func:`fetch_hf_papers`) through the daily sync, which marks + the account's links claimed and moves links another account claimed to that + account's row. +2. ``doubles``: a paper linked to two rows under one author name at the same + byline position keeps the claimed link, else the name-only one. Same-name + links at different positions may be two coauthors and are only reported. +3. ``split``: an unclaimed link of an HF-linked row stays only when the paper + shares a coauthor with one of the row's claimed papers; otherwise it moves + to the name-only author of that name. Rows without any claimed paper have + no evidence either way and are only reported. +4. ``merge``: a name-only link moves to the only HF account with that name + when the paper is provably the account's: it shares an official repository + with one of the account's anchor papers, or a coauthor whose identity is + certain on both papers (an HF-linked row with claimed or curated links), or + at least two distinctive name-only coauthors (no two accounts share the + name, and the row has at most ``MAX_COAUTHOR_PAPERS`` papers). Anchor papers are + the account's claimed papers, or all papers of a curated row (an HF username + set by hand, without an HF id). Papers without such evidence stay put. +5. ``empty``: name-only rows left without papers (and nothing else pointing at + them) are deleted; the row that took their papers, else a namesake row, + inherits their URL through ``author_aliases``. + +Each phase reports every change it makes, so a rolled-back run is the dry run. +""" + +from __future__ import annotations + +import json +import os +import time +from collections import Counter, defaultdict +from collections.abc import Callable, Iterator +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +import httpx +from sqlalchemy import func, or_, text +from sqlalchemy.orm import Session + +from backend.models.database import Author, AuthorAlias, PaperAuthor +from backend.services.author_hf_profile_sync_service import ( + AuthorHfProfileSyncService, + _name_key, +) +from backend.services.author_service import ( + get_or_create_author_by_name, + normalize_arxiv_author_name, +) + +HF_PAPER_URL = "https://huggingface.co/api/papers/{arxiv_id}" +USER_AGENT = "PapersWithCode-AuthorClaimBackfill/1.0 (https://paperswithcode.co)" +# HF allows 500 requests per 5 minutes per IP; stay below it. +DEFAULT_REQUESTS_PER_SECOND = 1.5 +MAX_ATTEMPTS = 6 +FIRST_BACKOFF_SECONDS = 30.0 + +PHASES = ("claims", "doubles", "split", "merge", "empty") +# Coauthors that tie a paper to an account when no stronger evidence exists. +MIN_SHARED_COAUTHORS = 2 +# A name-only row with more papers likely lumps several namesakes together. +MAX_COAUTHOR_PAPERS = 100 + +_MERGE_SETUP = ( + "DROP TABLE IF EXISTS merge_hf", + "DROP TABLE IF EXISTS merge_identity", + "DROP TABLE IF EXISTS merge_anchor", + "DROP TABLE IF EXISTS merge_candidate", + "DROP TABLE IF EXISTS merge_coauthor", + """ + CREATE TEMP TABLE merge_hf AS + SELECT id, lower(name) AS name_key, hf_id IS NULL AS curated + FROM authors + WHERE hf_id IS NOT NULL OR hf_username IS NOT NULL + """, + # Links whose author is certainly the person: claimed, or hand-curated. + """ + CREATE TEMP TABLE merge_identity AS + SELECT pa.author_id, pa.paper_id + FROM paper_authors pa + JOIN merge_hf h ON h.id = pa.author_id + WHERE pa.claimed OR h.curated + """, + "CREATE TEMP TABLE merge_anchor AS SELECT * FROM merge_identity", + """ + CREATE TEMP TABLE merge_candidate AS + SELECT pa.paper_id, pa.author_id AS name_only_id, h.id AS hf_author_id + FROM paper_authors pa + JOIN authors a ON a.id = pa.author_id + JOIN merge_hf h ON h.name_key = lower(a.name) + WHERE a.hf_id IS NULL AND a.hf_username IS NULL + AND h.name_key IN ( + SELECT name_key FROM merge_hf GROUP BY name_key HAVING count(*) = 1 + ) + AND EXISTS (SELECT 1 FROM merge_anchor an WHERE an.author_id = h.id) + """, + # Name-only coauthors distinctive enough to count as shared evidence. + f""" + CREATE TEMP TABLE merge_coauthor AS + SELECT a.id AS author_id + FROM authors a + WHERE a.hf_id IS NULL AND a.hf_username IS NULL + AND (SELECT count(*) FROM merge_hf h WHERE h.name_key = lower(a.name)) < 2 + AND ( + SELECT count(*) FROM paper_authors pa WHERE pa.author_id = a.id + ) <= {MAX_COAUTHOR_PAPERS} + AND EXISTS ( + SELECT 1 FROM paper_authors pa + JOIN merge_anchor an ON an.paper_id = pa.paper_id + WHERE pa.author_id = a.id + ) + """, +) + +_MERGE_EVIDENCE = """ +SELECT c.paper_id, c.name_only_id, c.hf_author_id, + EXISTS ( + SELECT 1 + FROM paper_repositories r1 + JOIN paper_repositories r2 + ON r2.repository_id = r1.repository_id AND r2.is_official + JOIN merge_anchor an + ON an.paper_id = r2.paper_id AND an.author_id = c.hf_author_id + WHERE r1.paper_id = c.paper_id AND r1.is_official + AND r2.paper_id <> c.paper_id + ) AS shares_repository, + EXISTS ( + SELECT 1 + FROM merge_identity co + JOIN merge_identity co2 ON co2.author_id = co.author_id + JOIN merge_anchor an + ON an.paper_id = co2.paper_id AND an.author_id = c.hf_author_id + WHERE co.paper_id = c.paper_id + AND co.author_id <> c.hf_author_id + AND co2.paper_id <> c.paper_id + ) AS shares_verified_coauthor, + ( + SELECT count(DISTINCT co.author_id) + FROM paper_authors co + JOIN merge_coauthor mc ON mc.author_id = co.author_id + JOIN paper_authors co2 ON co2.author_id = co.author_id + JOIN merge_anchor an + ON an.paper_id = co2.paper_id AND an.author_id = c.hf_author_id + WHERE co.paper_id = c.paper_id + AND co.author_id <> c.name_only_id + AND co2.paper_id <> c.paper_id + ) AS shared_coauthors +FROM merge_candidate c +ORDER BY c.hf_author_id, c.paper_id +""" + +Report = Callable[[dict[str, Any]], None] + + +def iter_claim_targets( + db: Session, *, after_id: int = 0, batch_size: int = 1000 +) -> Iterator[list[tuple[int, str]]]: + """Yield batches of (paper id, arXiv id) whose HF-linked links are unclaimed.""" + while True: + rows = db.execute( + text( + """ + SELECT DISTINCT p.id, p.arxiv_id + FROM papers p + JOIN paper_authors pa ON pa.paper_id = p.id + JOIN authors a ON a.id = pa.author_id + WHERE p.id > :after_id + AND p.is_latest_version + AND p.arxiv_id IS NOT NULL + AND NOT pa.claimed + AND (a.hf_id IS NOT NULL OR a.hf_username IS NOT NULL) + ORDER BY p.id + LIMIT :limit + """ + ), + {"after_id": after_id, "limit": batch_size}, + ).all() + if not rows: + return + yield [(row[0], row[1]) for row in rows] + after_id = rows[-1][0] + + +def cached_arxiv_ids(cache: Path) -> set[str]: + if not cache.exists(): + return set() + with cache.open(encoding="utf-8") as lines: + return {json.loads(line)["arxiv_id"] for line in lines if line.strip()} + + +class HfPaperClient: + """Paced Hugging Face papers API client that backs off on 429 and 5xx.""" + + def __init__( + self, + client: httpx.Client | None = None, + *, + requests_per_second: float = DEFAULT_REQUESTS_PER_SECOND, + sleep: Callable[[float], None] = time.sleep, + ): + headers = {"User-Agent": USER_AGENT} + if token := os.environ.get("HF_TOKEN"): + headers["Authorization"] = f"Bearer {token}" + self.client = client or httpx.Client( + timeout=httpx.Timeout(30, connect=10), follow_redirects=True + ) + self.headers = headers + self.interval = 1 / requests_per_second + self.sleep = sleep + self._last = 0.0 + + def authors(self, arxiv_id: str) -> list[dict[str, Any]] | None: + """Return the paper's HF author list, or None when HF has no such paper.""" + backoff = FIRST_BACKOFF_SECONDS + for attempt in range(MAX_ATTEMPTS): + wait = self._last + self.interval - time.monotonic() + if wait > 0: + self.sleep(wait) + self._last = time.monotonic() + response = self.client.get( + HF_PAPER_URL.format(arxiv_id=arxiv_id), headers=self.headers + ) + if response.status_code == 404: + return None + if response.status_code == 429 or response.status_code >= 500: + if attempt + 1 == MAX_ATTEMPTS: + response.raise_for_status() + retry_after = response.headers.get("Retry-After", "") + self.sleep(float(retry_after) if retry_after.isdigit() else backoff) + backoff *= 2 + continue + response.raise_for_status() + return [_claim_fields(a) for a in response.json().get("authors") or []] + raise AssertionError("unreachable") + + +def _claim_fields(author: dict[str, Any]) -> dict[str, Any]: + """Keep only what the claim sync reads, so the cache stays small.""" + kept = {k: author[k] for k in ("name", "hidden", "status") if k in author} + if user := author.get("user"): + kept["user"] = { + k: user[k] + for k in ("_id", "user", "fullname", "avatarUrl", "isPro", "type") + if k in user + } + return kept + + +def fetch_hf_papers( + db: Session, + client: HfPaperClient, + cache: Path, + *, + limit: int | None = None, + on_progress: Callable[[int, int], None] | None = None, +) -> int: + """Append HF author lists of uncached target papers to ``cache``. + + Read-only on the database; a rerun skips papers already in the cache. + """ + done = cached_arxiv_ids(cache) + fetched = 0 + with cache.open("a", encoding="utf-8") as out: + for batch in iter_claim_targets(db): + db.rollback() + for paper_id, arxiv_id in batch: + if arxiv_id in done: + continue + authors = client.authors(arxiv_id) + out.write(json.dumps({"arxiv_id": arxiv_id, "authors": authors}) + "\n") + done.add(arxiv_id) + fetched += 1 + if limit is not None and fetched >= limit: + return fetched + out.flush() + if on_progress: + on_progress(fetched, batch[-1][0]) + return fetched + + +def _is_hf(author: Author) -> bool: + return author.hf_id is not None or author.hf_username is not None + + +@dataclass +class CleanupStats: + counts: Counter = field(default_factory=Counter) + claim_sync: dict[str, int] = field(default_factory=dict) + + def to_dict(self) -> dict[str, Any]: + return {**dict(sorted(self.counts.items())), "claim_sync": self.claim_sync} + + +class AuthorClaimCleanup: + def __init__(self, db: Session, report: Report): + self.db = db + self.report = report + self.stats = CleanupStats() + # Name-only author id -> the account its papers merged into. + self.merged_into: dict[int, int] = {} + # Filled by replay_claims: HF byline names by arXiv id (name key -> + # name) and HF-linked author names before the replay renamed any. + self.bylines: dict[str, dict[str, str]] = {} + self.original_names: dict[int, str] = {} + + def replay_claims(self, cache: Path) -> None: + """Apply cached HF author lists through the daily claim sync.""" + # The sync renames an account's row to its HF full name; the split + # still needs the name the legacy links were matched by. + self.original_names = dict( + self.db.query(Author.id, Author.name).filter( + or_(Author.hf_id.isnot(None), Author.hf_username.isnot(None)) + ) + ) + sync = AuthorHfProfileSyncService(self.db) + with cache.open(encoding="utf-8") as lines: + for line in lines: + entry = json.loads(line) + if entry["authors"] is None: + self.stats.counts["claims_not_on_hf"] += 1 + continue + self.bylines[entry["arxiv_id"]] = { + _name_key(author["name"]): normalize_arxiv_author_name( + author["name"] + ) + for author in entry["authors"] + if author.get("name") + } + before = sync.stats.claimed + sync.sync_paper({"id": entry["arxiv_id"], "authors": entry["authors"]}) + if sync.stats.claimed > before: + self.report( + { + "phase": "claims", + "arxiv_id": entry["arxiv_id"], + "claimed": sync.stats.claimed - before, + } + ) + self.stats.claim_sync = sync.stats.to_dict() + + def fix_doubles(self) -> None: + """Keep one link per paper and author name: the claimed, else name-only.""" + groups: dict[tuple[int, str], list[tuple[PaperAuthor, Author]]] = defaultdict( + list + ) + doubled = ( + self.db.query(PaperAuthor, Author) + .join(Author, Author.id == PaperAuthor.author_id) + .filter( + PaperAuthor.paper_id.in_( + self.db.query(PaperAuthor.paper_id) + .join(Author, Author.id == PaperAuthor.author_id) + .group_by(PaperAuthor.paper_id, func.lower(Author.name)) + .having(text("count(*) > 1")) + ) + ) + ) + for link, author in doubled: + groups[(link.paper_id, _name_key(author.name))].append((link, author)) + for (paper_id, _), links in sorted(groups.items()): + if len(links) < 2: + continue + claimed = [pair for pair in links if pair[0].claimed] + name_only = [pair for pair in links if not _is_hf(pair[1])] + keep = claimed or name_only + # Two coauthors can share a name; only one byline slot is a double. + if len({link.author_order for link, _ in links}) > 1: + skip = "different_positions" + elif len(keep) != 1: + skip = "no_single_keeper" + else: + skip = None + if skip: + self.stats.counts[f"doubles_skipped_{skip}"] += 1 + self.report( + { + "phase": "doubles", + "action": "skip", + "reason": skip, + "paper_id": paper_id, + "author_ids": [author.id for _, author in links], + } + ) + continue + kept_link, kept = keep[0] + for link, author in links: + if link is kept_link: + continue + self.db.delete(link) + self.stats.counts["doubles_removed"] += 1 + self.report( + { + "phase": "doubles", + "action": "remove_link", + "paper_id": paper_id, + "author_id": author.id, + "hf_username": author.hf_username, + "kept_author_id": kept.id, + "name": author.name, + } + ) + self.db.flush() + + def split_namesakes(self) -> None: + """Move unclaimed links without coauthor evidence to name-only authors. + + Only bylines that two or more HF accounts publish under move: a second + account proves namesakes exist. A name with one account may just be + that person's unclaimed papers, so it stays. + """ + # An account renamed by a claim still counts under its former name. + accounts = Counter( + key + for author_id, name in self.db.query(Author.id, Author.name).filter( + Author.hf_id.isnot(None) + ) + for key in { + _name_key(name), + _name_key(self.original_names.get(author_id, name)), + } + ) + rows = self.db.execute( + text( + """ + WITH hf AS ( + SELECT id FROM authors + WHERE hf_id IS NOT NULL OR hf_username IS NOT NULL + ), + evidence AS ( + SELECT DISTINCT pa.author_id, co.author_id AS coauthor_id + FROM paper_authors pa + JOIN hf ON hf.id = pa.author_id + JOIN paper_authors co + ON co.paper_id = pa.paper_id AND co.author_id <> pa.author_id + WHERE pa.claimed + ) + SELECT pa.author_id, pa.paper_id, p.arxiv_id, + EXISTS ( + SELECT 1 FROM paper_authors pc WHERE pc.author_id = pa.author_id AND pc.claimed + ) AS has_claims, + EXISTS ( + SELECT 1 + FROM paper_authors co + JOIN evidence e + ON e.author_id = pa.author_id + AND e.coauthor_id = co.author_id + WHERE co.paper_id = pa.paper_id + AND co.author_id <> pa.author_id + ) AS shares_coauthor + FROM paper_authors pa + JOIN hf ON hf.id = pa.author_id + JOIN papers p ON p.id = pa.paper_id + WHERE NOT pa.claimed AND p.is_latest_version + ORDER BY pa.author_id, pa.paper_id + """ + ) + ).all() + namesakes: dict[str, Author] = {} + rows_moved_from: set[int] = set() + no_claims: Counter = Counter() + for author_id, paper_id, arxiv_id, has_claims, shares_coauthor in rows: + if not has_claims: + no_claims[author_id] += 1 + continue + if shares_coauthor: + self.stats.counts["split_kept_by_coauthor"] += 1 + continue + author = self.db.get(Author, author_id) + byline_name = self._byline_name(author, arxiv_id) + if byline_name is None: + self.stats.counts["split_skipped_name_not_on_byline"] += 1 + self.report( + { + "phase": "split", + "action": "skip_name_not_on_byline", + "paper_id": paper_id, + "author_id": author_id, + } + ) + continue + key = _name_key(byline_name) + if accounts[key] < 2: + self.stats.counts["split_skipped_unique_name"] += 1 + continue + namesake = namesakes.get(key) + if namesake is None: + namesake = get_or_create_author_by_name(self.db, byline_name) + namesakes[key] = namesake + rows_moved_from.add(author_id) + link = self.db.get(PaperAuthor, (paper_id, author_id)) + if self.db.get(PaperAuthor, (paper_id, namesake.id)) is not None: + self.db.delete(link) + action = "remove_link" + else: + order = link.author_order + self.db.delete(link) + self.db.flush() + self.db.add( + PaperAuthor( + paper_id=paper_id, author_id=namesake.id, author_order=order + ) + ) + action = "move_link" + self.db.flush() + self.stats.counts[f"split_{action}"] += 1 + self.report( + { + "phase": "split", + "action": action, + "paper_id": paper_id, + "author_id": author_id, + "hf_username": author.hf_username, + "to_author_id": namesake.id, + "name": byline_name, + } + ) + self.stats.counts["split_rows_moved_from"] = len(rows_moved_from) + self.stats.counts["split_rows_without_claims"] = len(no_claims) + self.stats.counts["split_links_on_rows_without_claims"] = sum( + no_claims.values() + ) + for author_id, links in no_claims.most_common(): + self.report( + { + "phase": "split", + "action": "skip_no_claims", + "author_id": author_id, + "unclaimed_links": links, + } + ) + + def _byline_name(self, author: Author, arxiv_id: str | None) -> str | None: + """Return the name the link was matched by, or None when unsure. + + It is the row's name, or its name before this run renamed it, as + listed on the paper's HF byline. Without a cached byline, only a row + this run did not rename has a known name. + """ + keys = { + _name_key(author.name), + _name_key(self.original_names.get(author.id, author.name)), + } + byline = self.bylines.get(arxiv_id) if arxiv_id else None + if byline is None: + return author.name if len(keys) == 1 else None + found = [byline[key] for key in keys if key in byline] + return found[0] if len(found) == 1 else None + + def merge_namesakes(self) -> None: + """Move name-only links to the account they provably belong to.""" + for statement in _MERGE_SETUP: + self.db.execute(text(statement)) + rows = self.db.execute(text(_MERGE_EVIDENCE)).all() + self.stats.counts["merge_candidates"] = len(rows) + for paper_id, name_only_id, hf_author_id, repo, coauthor, shared in rows: + evidence = [ + reason + for reason, found in ( + ("official_repository", repo), + ("verified_coauthor", coauthor), + ("shared_coauthors", shared >= MIN_SHARED_COAUTHORS), + ) + if found + ] + if not evidence: + continue + link = self.db.get(PaperAuthor, (paper_id, name_only_id)) + if self.db.get(PaperAuthor, (paper_id, hf_author_id)) is not None: + self.db.delete(link) + action = "remove_link" + else: + order = link.author_order + self.db.delete(link) + self.db.flush() + self.db.add( + PaperAuthor( + paper_id=paper_id, author_id=hf_author_id, author_order=order + ) + ) + action = "move_link" + self.db.flush() + self.merged_into[name_only_id] = hf_author_id + self.stats.counts[f"merge_{action}"] += 1 + for reason in evidence: + self.stats.counts[f"merge_by_{reason}"] += 1 + self.report( + { + "phase": "merge", + "action": action, + "paper_id": paper_id, + "author_id": name_only_id, + "to_author_id": hf_author_id, + "evidence": evidence, + "shared_coauthors": shared, + } + ) + self.stats.counts["merge_rows_into"] = len(set(self.merged_into.values())) + + def delete_empty_authors(self) -> None: + """Delete name-only authors without papers, aliasing them to a namesake.""" + empty = ( + self.db.execute( + text( + """ + SELECT a.id FROM authors a + WHERE a.hf_id IS NULL AND a.hf_username IS NULL + AND NOT EXISTS (SELECT 1 FROM paper_authors x WHERE x.author_id = a.id) + AND NOT EXISTS (SELECT 1 FROM author_google_scholar_profiles x WHERE x.author_id = a.id) + AND NOT EXISTS (SELECT 1 FROM author_google_scholar_candidates x WHERE x.author_id = a.id) + AND NOT EXISTS (SELECT 1 FROM author_photos x WHERE x.author_id = a.id) + AND NOT EXISTS ( + SELECT 1 FROM research_follows x + WHERE x.kind = 'author' AND x.entity_id = a.id + ) + ORDER BY a.id + """ + ) + ) + .scalars() + .all() + ) + for author_id in empty: + author = self.db.get(Author, author_id) + merged_into = self.merged_into.get(author_id) + target = self.db.get(Author, merged_into) if merged_into else None + target = target or ( + self.db.query(Author) + .filter( + Author.id != author_id, + Author.id.notin_(empty), + text("lower(authors.name) = lower(:name)"), + ) + .params(name=author.name) + .order_by( + Author.hf_id.isnot(None), Author.hf_username.isnot(None), Author.id + ) + .first() + ) + aliases = self.db.query(AuthorAlias).filter_by(author_id=author_id).all() + if target is None and aliases: + # Deleting would cascade away aliases that keep old URLs alive. + self.stats.counts["empty_kept_alias_target"] += 1 + continue + for alias in aliases: + alias.author_id = target.id + if target is not None: + self.db.add(AuthorAlias(alias_id=author_id, author_id=target.id)) + self.db.delete(author) + self.db.flush() + self.stats.counts["empty_deleted"] += 1 + self.stats.counts["empty_aliased"] += target is not None + self.report( + { + "phase": "empty", + "action": "delete_author", + "author_id": author_id, + "name": author.name, + "alias_to": target.id if target else None, + } + ) diff --git a/backend/tests/test_author_claim_backfill_service.py b/backend/tests/test_author_claim_backfill_service.py new file mode 100644 index 00000000..7517d6e5 --- /dev/null +++ b/backend/tests/test_author_claim_backfill_service.py @@ -0,0 +1,443 @@ +import json + +import httpx +import pytest +from sqlalchemy import BigInteger, create_engine, text +from sqlalchemy.dialects.postgresql import ARRAY, JSONB +from sqlalchemy.ext.compiler import compiles +from sqlalchemy.orm import Session + +from backend.models.database import ( + Author, + AuthorAlias, + AuthorGoogleScholarProfile, + AuthorPhoto, + Base, + Paper, + PaperAuthor, + PaperRepository, + Repository, + ResearchFollow, +) +from backend.services.author_claim_backfill_service import ( + AuthorClaimCleanup, + HfPaperClient, + fetch_hf_papers, +) + + +@compiles(BigInteger, "sqlite") +def _compile_bigint_sqlite(_element, _compiler, **_kw): + return "INTEGER" + + +@compiles(ARRAY, "sqlite") +@compiles(JSONB, "sqlite") +def _compile_postgres_container_sqlite(_element, _compiler, **_kw): + return "TEXT" + + +HAO = {"_id": "hf-hao", "user": "hao-li", "fullname": "Hao Li", "type": "user"} +OTHER = {"_id": "hf-other", "user": "other-hao", "type": "user"} + + +def _session() -> Session: + engine = create_engine("sqlite:///:memory:") + Base.metadata.create_all( + engine, + tables=[ + Paper.__table__, + Author.__table__, + PaperAuthor.__table__, + Repository.__table__, + PaperRepository.__table__, + AuthorAlias.__table__, + AuthorGoogleScholarProfile.__table__, + AuthorPhoto.__table__, + ResearchFollow.__table__, + ], + ) + session = Session(engine) + session.execute( + text("CREATE TABLE author_google_scholar_candidates (author_id INTEGER)") + ) + session.add_all( + Paper( + id=paper_id, + arxiv_id=f"2601.0000{paper_id}", + title=f"Paper {paper_id}", + version="v1", + is_latest_version=True, + ) + for paper_id in range(1, 6) + ) + session.add_all( + [ + # Legacy row: every "Hao Li" paper landed on the hao-li account. + Author(id=10, name="Hao Li", hf_username="hao-li", hf_id="hf-hao"), + Author(id=11, name="Ann Coauthor"), + Author(id=12, name="Bob Stranger"), + Author(id=13, name="Cat Stranger"), + ] + ) + links = {1: (10, 11), 2: (10, 11), 3: (10, 12), 4: (10, 13), 5: (10, 12)} + session.add_all( + PaperAuthor(paper_id=paper_id, author_id=author_id, author_order=order) + for paper_id, authors in links.items() + for order, author_id in enumerate(authors, start=1) + ) + session.commit() + return session + + +def _cache(tmp_path, papers): + path = tmp_path / "claims.jsonl" + path.write_text( + "".join( + json.dumps({"arxiv_id": arxiv_id, "authors": authors}) + "\n" + for arxiv_id, authors in papers.items() + ) + ) + return path + + +def _claim(name, user, status="claimed_verified"): + return {"name": name, "hidden": False, "status": status, "user": user} + + +def _authors_of(session, paper_id): + return { + (link.author_id, link.claimed) + for link in session.query(PaperAuthor).filter_by(paper_id=paper_id) + } + + +def _add_second_hao_account(session): + # A second HF account publishing as "Hao Li" proves namesakes exist. + session.add(Author(id=30, name="Hao Li", hf_username="hao-li-2", hf_id="hf-2")) + session.commit() + + +def _cleanup(session): + entries = [] + return AuthorClaimCleanup(session, entries.append), entries + + +def test_replayed_claims_mark_links_and_hand_other_claims_to_their_account( + tmp_path, +): + session = _session() + cleanup, entries = _cleanup(session) + cache = _cache( + tmp_path, + { + "2601.00001": [_claim("Hao Li", HAO)], + "2601.00004": [_claim("Hao Li", OTHER)], + "2601.00005": [_claim("Hao Li", HAO, status="extracted_pending")], + "2601.00003": None, + }, + ) + + cleanup.replay_claims(cache) + + assert _authors_of(session, 1) == {(10, True), (11, False)} + other = session.query(Author).filter_by(hf_username="other-hao").one() + assert _authors_of(session, 4) == {(other.id, True), (13, False)} + # An unconfirmed HF guess is not a claim. + assert (10, False) in _authors_of(session, 5) + assert cleanup.stats.counts["claims_not_on_hf"] == 1 + assert [e["arxiv_id"] for e in entries] == ["2601.00001", "2601.00004"] + + +def test_split_keeps_coauthor_backed_papers_and_moves_the_rest(tmp_path): + session = _session() + _add_second_hao_account(session) + cleanup, entries = _cleanup(session) + cleanup.replay_claims(_cache(tmp_path, {"2601.00001": [_claim("Hao Li", HAO)]})) + + cleanup.split_namesakes() + + # Paper 2 shares coauthor Ann with claimed paper 1, so it stays. + assert _authors_of(session, 2) == {(10, False), (11, False)} + namesake = session.query(Author).filter_by(name="Hao Li", hf_username=None).one() + for paper_id in (3, 4, 5): + assert (namesake.id, False) in _authors_of(session, paper_id) + assert (10, False) not in _authors_of(session, paper_id) + order = ( + session.query(PaperAuthor.author_order) + .filter_by(paper_id=3, author_id=namesake.id) + .scalar() + ) + assert order == 1 + assert cleanup.stats.counts["split_move_link"] == 3 + assert cleanup.stats.counts["split_kept_by_coauthor"] == 1 + assert {e["paper_id"] for e in entries if e["phase"] == "split"} == {3, 4, 5} + + +def test_split_uses_the_byline_name_of_a_row_renamed_by_its_claim(tmp_path): + session = _session() + _add_second_hao_account(session) + cleanup, entries = _cleanup(session) + # HF matched hao-li to a "Hao Li" byline; the account is really Zed Jia, + # who claims paper 1 as "Zed Jia" and so renames the row. + zed = {**HAO, "fullname": "Zed Jia"} + cleanup.replay_claims( + _cache( + tmp_path, + { + "2601.00001": [_claim("Zed Jia", zed)], + "2601.00003": [{"name": "Hao Li"}, {"name": "Bob Stranger"}], + "2601.00004": [{"name": "Someone Else"}], + }, + ) + ) + assert session.get(Author, 10).name == "Zed Jia" + + cleanup.split_namesakes() + + namesake = session.query(Author).filter_by(name="Hao Li", hf_username=None).one() + assert (namesake.id, False) in _authors_of(session, 3) + # Paper 4's byline lists neither name: the link stays and is reported. + assert (10, False) in _authors_of(session, 4) + # Paper 5 has no cached byline, and the row was renamed: unsure, stays. + assert (10, False) in _authors_of(session, 5) + assert cleanup.stats.counts["split_skipped_name_not_on_byline"] == 2 + + +def test_split_leaves_names_with_a_single_hf_account_alone(tmp_path): + session = _session() + cleanup, entries = _cleanup(session) + cleanup.replay_claims(_cache(tmp_path, {"2601.00001": [_claim("Hao Li", HAO)]})) + + cleanup.split_namesakes() + + for paper_id in (3, 4, 5): + assert (10, False) in _authors_of(session, paper_id) + assert cleanup.stats.counts["split_skipped_unique_name"] == 3 + assert not [e for e in entries if e["phase"] == "split"] + + +def test_split_leaves_rows_without_any_claim_alone(): + session = _session() + cleanup, entries = _cleanup(session) + + cleanup.split_namesakes() + + assert (10, False) in _authors_of(session, 3) + assert cleanup.stats.counts["split_rows_without_claims"] == 1 + assert cleanup.stats.counts["split_links_on_rows_without_claims"] == 5 + assert entries == [ + { + "phase": "split", + "action": "skip_no_claims", + "author_id": 10, + "unclaimed_links": 5, + } + ] + + +def test_doubles_keep_the_claimed_link_else_the_name_only_one(): + session = _session() + session.add(Author(id=20, name="Hao Li")) + session.add_all( + [ + PaperAuthor(paper_id=1, author_id=20, author_order=1), + PaperAuthor(paper_id=2, author_id=20, author_order=1), + # Two coauthors named Hao Li: both stay. + PaperAuthor(paper_id=3, author_id=20, author_order=3), + ] + ) + session.get(PaperAuthor, (1, 10)).claimed = True + session.commit() + cleanup, entries = _cleanup(session) + + cleanup.fix_doubles() + + assert _authors_of(session, 1) == {(10, True), (11, False)} + assert _authors_of(session, 2) == {(20, False), (11, False)} + assert _authors_of(session, 3) == {(10, False), (12, False), (20, False)} + assert cleanup.stats.counts["doubles_removed"] == 2 + assert cleanup.stats.counts["doubles_skipped_different_positions"] == 1 + assert [e["paper_id"] for e in entries if e["action"] == "skip"] == [3] + + +def test_empty_name_only_authors_are_deleted_and_aliased_to_a_namesake(): + session = _session() + session.add_all( + [ + Author(id=30, name="hao li"), + Author(id=31, name="Lonely Name"), + Author(id=32, name="Followed Name"), + Author(id=33, name="Pictured Name"), + AuthorAlias(alias_id=99, author_id=30), + ResearchFollow(hf_subject="u", kind="author", entity_id=32), + AuthorPhoto( + author_id=33, + image_data=b"x", + mime_type="image/jpeg", + sha256="0" * 64, + source_url="y", + credit="c", + license="z", + ), + ] + ) + session.commit() + cleanup, _ = _cleanup(session) + + cleanup.delete_empty_authors() + + assert session.get(Author, 30) is None + assert session.get(Author, 31) is None + assert session.get(Author, 32) is not None + assert session.get(Author, 33) is not None + aliases = {a.alias_id: a.author_id for a in session.query(AuthorAlias)} + assert aliases == {30: 10, 99: 10} + assert cleanup.stats.counts["empty_deleted"] == 2 + assert cleanup.stats.counts["empty_aliased"] == 1 + + +def test_fetch_caches_trimmed_author_lists_and_resumes(tmp_path): + session = _session() + calls = [] + + def handler(request): + arxiv_id = request.url.path.rsplit("/", 1)[1] + calls.append(arxiv_id) + if arxiv_id == "2601.00002": + return httpx.Response(404) + return httpx.Response( + 200, + json={ + "authors": [ + {**_claim("Hao Li", {**HAO, "extra": 1}), "_id": "x"}, + ] + }, + ) + + client = HfPaperClient( + httpx.Client(transport=httpx.MockTransport(handler)), + requests_per_second=1000, + sleep=lambda _: None, + ) + cache = tmp_path / "claims.jsonl" + + assert fetch_hf_papers(session, client, cache, limit=2) == 2 + assert fetch_hf_papers(session, client, cache) == 3 + + assert calls == [f"2601.0000{i}" for i in range(1, 6)] + lines = [json.loads(line) for line in cache.read_text().splitlines()] + assert lines[1] == {"arxiv_id": "2601.00002", "authors": None} + assert lines[0]["authors"] == [_claim("Hao Li", HAO)] + + +def _link(session, paper_id, author_id, order=9, claimed=False): + session.add( + PaperAuthor( + paper_id=paper_id, author_id=author_id, author_order=order, claimed=claimed + ) + ) + + +def test_merge_moves_a_name_only_paper_sharing_an_official_repo_with_a_curated_row(): + session = _session() + session.add_all( + [ + # A curator set the HF username by hand: no HF id, no claims. + Author(id=40, name="Vik Paruchuri", hf_username="vikp"), + Author(id=41, name="Vik Paruchuri"), + Repository(id=1, url="https://github.com/datalab-to/surya"), + ] + ) + _link(session, 1, 40) + _link(session, 2, 41, order=2) + session.add_all( + [ + PaperRepository(paper_id=1, repository_id=1, is_official=True), + PaperRepository(paper_id=2, repository_id=1, is_official=True), + ] + ) + session.commit() + cleanup, entries = _cleanup(session) + + cleanup.merge_namesakes() + cleanup.delete_empty_authors() + + assert (40, False) in _authors_of(session, 2) + assert session.get(PaperAuthor, (2, 40)).author_order == 2 + assert session.get(Author, 41) is None + assert session.get(AuthorAlias, 41).author_id == 40 + assert [e["evidence"] for e in entries if e["phase"] == "merge"] == [ + ["official_repository"] + ] + + +def _account_with_namesake(session): + session.add_all( + [ + Author(id=50, name="Zed Verified", hf_username="zed", hf_id="hf-zed"), + Author(id=51, name="Mia Account", hf_username="mia", hf_id="hf-mia"), + Author(id=52, name="Mia Account"), + ] + ) + # Paper 3 is Mia's claimed anchor, written with Zed (claimed), Bob and Ann. + _link(session, 3, 51, claimed=True) + _link(session, 3, 50, claimed=True) + _link(session, 3, 11) + + +def test_merge_needs_a_coauthor_whose_identity_is_certain_on_both_papers(): + session = _session() + _account_with_namesake(session) + _link(session, 4, 52) + _link(session, 4, 50, claimed=True) + _link(session, 5, 52) + _link(session, 5, 50) # Zed by name only on paper 5: could be a namesake. + session.commit() + cleanup, _ = _cleanup(session) + + cleanup.merge_namesakes() + + assert (51, False) in _authors_of(session, 4) + assert (52, False) in _authors_of(session, 5) + assert cleanup.stats.counts["merge_by_verified_coauthor"] == 1 + + +@pytest.mark.parametrize("cat_accounts", [0, 2]) +def test_merge_counts_distinctive_name_only_coauthors(cat_accounts): + session = _session() + _account_with_namesake(session) + # Mia's anchor (paper 3) lists Hao Li (an HF row, never counted), Bob (12) + # and Cat (13); paper 4 lists all three, paper 5 only Hao Li and Bob. + _link(session, 3, 13) + _link(session, 4, 52) + _link(session, 4, 12) + _link(session, 5, 52) + # Accounts named "Cat Stranger" mean the Cat row may lump namesakes. + session.add_all( + Author(id=70 + i, name="Cat Stranger", hf_username=f"cat{i}", hf_id=f"c{i}") + for i in range(cat_accounts) + ) + session.commit() + cleanup, entries = _cleanup(session) + + cleanup.merge_namesakes() + + merged = [(e["paper_id"], e["shared_coauthors"]) for e in entries] + assert merged == ([(4, 2)] if cat_accounts == 0 else []) + assert (52, False) in _authors_of(session, 5) + + +def test_merge_skips_names_shared_by_two_accounts(): + session = _session() + _account_with_namesake(session) + session.add(Author(id=53, name="Mia Account", hf_username="mia2", hf_id="hf-2")) + _link(session, 4, 52) + _link(session, 4, 50, claimed=True) + session.commit() + cleanup, _ = _cleanup(session) + + cleanup.merge_namesakes() + + assert (52, False) in _authors_of(session, 4) + assert cleanup.stats.counts["merge_candidates"] == 0