diff --git a/docs/architecture.md b/docs/architecture.md index 86fb62a..05cd5fd 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -62,6 +62,20 @@ A unified `research-toolkit` CLI (`scripts/cli.py`, `build-dashboard`, `freshness`, `export`, `resume-gather`, `compose-kg`, …) so the chain can be run by hand without `python scripts/.py` per stage. +**Auxiliary producer (off the linear chain):** + +| Producer | Kind | Consumes | Primary artifact | Verified by | +|---|---|---|---|---| +| `scripts/emit_bibtex.py` (`emit-bibtex`) | producer | `bib_ledger.yml` + cache | a biblatex `.bib` | `validators/bibtex_out.py` | + +`emit-bibtex` is a deterministic ledger→BibTeX transform for manuscript +authoring, not a gated pipeline stage. The ledger `authors` field is a display +string, so it reconstructs real author lists from the cached arXiv Highwire +`citation_author` tags (the same bytes the excerpt anchors point into), falling +back to a live arXiv Atom lookup (re-cached, so still anchored) and then to the +ledger display string (flagged on stderr). It escapes `& % # _ $` only — never +braces (that destroys BibTeX brace-protection). + ## What is agent-authored vs deterministic | Decision | Who makes it | Why | diff --git a/scripts/cli.py b/scripts/cli.py index 8dcfdb9..7aa50fb 100644 --- a/scripts/cli.py +++ b/scripts/cli.py @@ -21,6 +21,7 @@ import sys from collections.abc import Callable from pathlib import Path +from typing import TextIO if __package__ in (None, ""): sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) @@ -47,6 +48,7 @@ "backlog-stamp": ("scripts.backlog_stamp", "main", SHAPE_FULL), "resume-gather": ("scripts.resume_gather_from_cache", "main", SHAPE_SLICED), "compose-kg": ("scripts.compose_cross_project_kg", "main", SHAPE_SLICED), + "emit-bibtex": ("scripts.emit_bibtex", "main", SHAPE_SLICED), } # One-line help shown by ``research-toolkit --help`` (no module import needed). @@ -62,6 +64,7 @@ "backlog-stamp": "Stamp a topic_backlog.yml entry as handed off / done.", "resume-gather": "Rebuild a sources-JSON skeleton from the content-addressed cache.", "compose-kg": "Merge per-project claim graphs into a cross-project KG snapshot.", + "emit-bibtex": "Emit a biblatex .bib from bib_ledger.yml (authors from cached Highwire tags).", } @@ -104,7 +107,7 @@ def _dispatch(name: str, rest: list[str]) -> int: return code if isinstance(code, int) else 1 -def _print_top_help(stream: object) -> None: +def _print_top_help(stream: TextIO) -> None: print("usage: research-toolkit [args...]", file=stream) print("", file=stream) print("Unified entry point for the research_toolkit pipeline.", file=stream) diff --git a/scripts/emit_bibtex.py b/scripts/emit_bibtex.py new file mode 100644 index 0000000..f20c8fd --- /dev/null +++ b/scripts/emit_bibtex.py @@ -0,0 +1,435 @@ +#!/usr/bin/env python3 +"""Emit a biblatex ``.bib`` from dossier ``bib_ledger.yml`` files. + +The ledger ``authors`` field is a display string ("Surname et al. (Year)"), +unusable as a BibTeX author list, and the ledgers carry no ``doi``/``arxiv_id``/ +``year`` field. But for arXiv sources the cached ``/abs/`` blob -- the same bytes +the excerpt anchors point into -- retains the Highwire ``citation_author`` / +``citation_title`` / ``citation_date`` / ``citation_arxiv_id`` meta tags, from +which a real biblatex record is reconstructed. Resolution order per entry: + +1. Cache Highwire tags from the capture whose ``source_url == primary_url``. +2. arXiv Atom fallback (``export.arxiv.org``) if an arXiv entry's cache lacks the + tags -- re-cached through ``cache_source`` (write-back) so it stays anchored; + ``--no-live`` disables this. +3. Non-arXiv sources (GitHub/HF/blog) keep the ledger display string, and every + such entry is listed on stderr for later hand-fixing. + +Field VALUES escape ``& % # _ $`` only -- braces are BibTeX grouping and must +never be escaped (that destroys brace-protection, the bug in +``research-kb``'s ``bibtex_generator.py``). + +Usage: + emit_bibtex.py [more...] [--out FILE] [--seed FILE] + [--no-live] [--no-overwrite] [--reconcile BIB [BIB ...]] + +Exit codes: 0 ok; 1 data error (output failed self-validation); 2 usage. +""" +from __future__ import annotations + +import argparse +import re +import sys +from datetime import date +from html import unescape +from pathlib import Path +from typing import Any + +if __package__ in (None, ""): + sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import yaml + +from validators._common import ARXIV_ID_RE +from validators.bibtex_out import parse_entries, validate_text + +_DEFAULT_CACHE_ROOT = Path.home() / "Claude" / "research_cache" +# Escape BibTeX-special chars in field VALUES. Braces are NEVER escaped (that +# destroys brace-protection). Backslash/tilde/caret ARE escaped -- a raw `\`, `~` +# or `^` in a cached title would otherwise be a TeX command / active char. +_BIB_ESCAPE = { + "\\": r"\textbackslash{}", "&": r"\&", "%": r"\%", "#": r"\#", "_": r"\_", + "$": r"\$", "~": r"\textasciitilde{}", "^": r"\textasciicircum{}", +} +_ARXIV_ATOM_API = "http://export.arxiv.org/api/query?id_list=" + + +def bib_escape(value: str) -> str: + """Escape BibTeX-special chars in a field VALUE. Braces are NOT escaped.""" + return "".join(_BIB_ESCAPE.get(ch, ch) for ch in value) + + +def _highwire(text: str, field: str) -> list[str]: + """All ``content`` values of ```` (either attr order). + + The content capture backreferences the opening quote (``(["'])(.*?)\\1``), so + an apostrophe inside a double-quoted value (``content="O'Connor, Alice"``) is + kept, not truncated; a negative lookbehind on the attribute names rejects + ``data-name=`` / ``data-content=``.""" + out: list[str] = [] + for pat in ( + rf']*?(?]*?(?]*?(?]*?(? str | None: + """First 4-digit run in ``value`` (handles a str, or a YAML ``datetime.date``).""" + if not value: + return None + m = re.search(r"(\d{4})", str(value)) + return m.group(1) if m else None + + +def _to_last_first(name: str) -> str: + """Normalise an author name to ``Family, Given`` for BibTeX. + + Highwire tags are already ``Surname, Forename`` (returned unchanged); arXiv + Atom gives ``Forename Surname``. Lowercase surname particles (van, de, von, + della, ...) are pulled into the family name, so ``Ludwig van Beethoven`` -> + ``van Beethoven, Ludwig`` rather than mis-splitting 'van' into the given name.""" + name = name.strip() + if "," in name or " " not in name: + return name + parts = name.split() + k = len(parts) - 1 + while k > 1 and parts[k - 1][:1].islower(): + k -= 1 + given, family = " ".join(parts[:k]), " ".join(parts[k:]) + return f"{family}, {given}" if given else family + + +def _from_blob(text: str) -> tuple[list[str], str | None, str | None, str | None]: + """Return (authors, title, year, arxiv_id) from a cached blob's Highwire tags.""" + authors = [a.strip() for a in _highwire(text, "author") if a.strip()] + titles = _highwire(text, "title") + dates = _highwire(text, "date") + ids = _highwire(text, "arxiv_id") + title = " ".join(titles[0].split()) if titles else None + year = _year(dates[0]) if dates else None + aid = ids[0].strip() if ids else None + return authors, title, year, aid + + +def _arxiv_id(url: str) -> str | None: + """Bare arXiv id (version stripped) from a URL, else None.""" + m = ARXIV_ID_RE.search(url or "") + return re.sub(r"v\d+$", "", m.group(1)) if m else None + + +def _find_abs_blob( + entry: dict[str, Any], url: str, manifest: dict[str, dict], cache_root: Path +) -> str | None: + """Text of the capture whose ``source_url == url`` and that bears author tags.""" + ordered = list(entry.get("cache_ids") or []) + ordered += [cid for cid in manifest if cid not in ordered] + for cid in ordered: + m = manifest.get(cid) + if not m or m.get("source_url") != url: + continue + tp = m.get("text_path") + if not isinstance(tp, str): + continue + blob = cache_root / tp + if blob.exists(): + text = blob.read_text(errors="replace") + if "citation_author" in text: + return text + return None + + +def _atom(arxiv_id: str) -> dict[str, Any] | None: + """Query the arXiv Atom API for the first entry's authors/title/year.""" + from urllib.request import urlopen # local: keep the hot path import-light + + try: + with urlopen(_ARXIV_ATOM_API + arxiv_id, timeout=20) as resp: # noqa: S310 + xml = resp.read().decode("utf-8", "replace") + except Exception: # noqa: BLE001 -- a live lookup must never abort the emit + return None + entry = re.search(r"(.*?)", xml, re.S) + if not entry: + return None + body = entry.group(1) + names = re.findall(r"\s*([^<]+)", body) + title = re.search(r"(.*?)", body, re.S) + published = re.search(r"(\d{4})", body) + return { + "authors": [_to_last_first(unescape(n.strip())) for n in names], + "title": " ".join(title.group(1).split()) if title else None, + "year": published.group(1) if published else None, + } + + +def _live_arxiv(arxiv_id: str, cache_root: Path, write_back: bool) -> dict[str, Any] | None: + """Recover authors for an arXiv id whose cache lacked tags. + + Preferred path re-caches the ``/abs/`` page through ``cache_source`` so the + Highwire tags become byte-anchored (write-back); the manifest line to add is + printed for the dossier to adopt deliberately (matching cache_source's own + "a skill appends them" contract). Falls back to a bare Atom read (flagged).""" + abs_url = f"https://arxiv.org/abs/{arxiv_id}" + if write_back: + try: + from scripts import cache_source # lazy: pulls heavy extractors + + entry = cache_source.cache_one( + abs_url, cache_root=cache_root, fetched_at=date.today().isoformat(), + topic="emit-bibtex", + ) + tp = entry.get("text_path") + if isinstance(tp, str): + authors, title, year, _ = _from_blob((cache_root / tp).read_text(errors="replace")) + if authors: + print( + f"note: re-cached {abs_url} for authors -- add cache_id " + f"{entry['cache_id']} to the dossier manifest to anchor it", + file=sys.stderr, + ) + return {"authors": authors, "title": title, "year": year} + except Exception as exc: # noqa: BLE001 + print(f"warning: live re-cache of {abs_url} failed: {exc}", file=sys.stderr) + atom = _atom(arxiv_id) + if atom and atom["authors"]: + print( + f"warning: resolved {arxiv_id} authors via live arXiv Atom " + f"(NOT anchored -- re-run cache-source to anchor)", + file=sys.stderr, + ) + return atom + return None + + +def resolve_entry( + entry: dict[str, Any], manifest: dict[str, dict], cache_root: Path, + *, use_live: bool, write_back: bool, +) -> dict[str, Any]: + """Resolve one ledger entry to a BibTeX record dict.""" + url = entry.get("primary_url", "") or "" + arxiv_id = _arxiv_id(url) + authors: list[str] = [] + title = year = None + + text = _find_abs_blob(entry, url, manifest, cache_root) + if text is not None: + authors, title, year, aid = _from_blob(text) + arxiv_id = aid or arxiv_id + if not authors and arxiv_id and use_live: + live = _live_arxiv(arxiv_id, cache_root, write_back) + if live: + authors = live["authors"] + title = title or live["title"] + year = year or live["year"] + + display = False + if not authors: + authors = [entry.get("authors") or entry["bibkey"]] + display = True + title = title or entry.get("title") or "" + year = year or _year(entry.get("published_online")) or _year(entry.get("authors")) or "" + return { + "bibkey": entry["bibkey"], "authors": authors, "title": str(title), + "year": year, "eprint": arxiv_id, "url": url, "display": display, + } + + +def _author_field(authors: list[str]) -> str: + """Join authors with ' and '; brace-protect any name that itself contains + ' and ' (a corporate/literal name) so BibTeX does not read it as a separator.""" + parts = [] + for a in authors: + esc = bib_escape(a) + parts.append(f"{{{esc}}}" if " and " in a else esc) + return " and ".join(parts) + + +def format_entry(record: dict[str, Any]) -> str: + """Render one resolved record as a biblatex ``@misc`` block (seed shape).""" + lines = [f"@misc{{{record['bibkey']},"] + lines.append(f" author = {{{_author_field(record['authors'])}}},") + lines.append(f" title = {{{bib_escape(record['title'])}}},") + if record["year"]: + lines.append(f" year = {{{record['year']}}},") + if record["eprint"]: + lines.append(f" eprint = {{{record['eprint']}}},") + lines.append(" archiveprefix = {arXiv},") + if record["url"]: + lines.append(f" url = {{{bib_escape(record['url'])}}},") + lines.append("}") + return "\n".join(lines) + + +def _load_dossier(src: Path) -> tuple[list[dict], dict[str, dict], Path]: + """From a dossier dir or a ledger file, return (entries, manifest_idx, cache_root).""" + ledger = src / "bib_ledger.yml" if src.is_dir() else src + manifest_path = (src if src.is_dir() else src.parent) / "cache_manifest.yml" + entries: list[dict] = [] + if ledger.exists(): + doc = yaml.safe_load(ledger.read_text()) or {} + entries = [e for e in (doc.get("entries") or []) if isinstance(e, dict) and e.get("bibkey")] + manifest: dict[str, dict] = {} + cache_root = _DEFAULT_CACHE_ROOT + if manifest_path.exists(): + mdoc = yaml.safe_load(manifest_path.read_text()) or {} + root = mdoc.get("cache_root") + if isinstance(root, str): + cache_root = Path(root).expanduser() + manifest = {c["cache_id"]: c for c in (mdoc.get("entries") or []) if isinstance(c, dict) and c.get("cache_id")} + return entries, manifest, cache_root + + +def emit( + sources: list[Path], *, use_live: bool, write_back: bool +) -> tuple[dict[str, dict], list[str], dict[str, list[str]]]: + """Resolve every ledger entry across ``sources``; dedup by bibkey. + + Dedup prefers a cache/live-resolved record over a display-string one, so the + order dossiers are passed cannot bury an authoritative record behind a + display-string duplicate. + + Returns (bibkey -> record, warnings, reverse_collisions[url -> bibkeys]).""" + by_key: dict[str, list[dict]] = {} + by_url: dict[str, list[str]] = {} + warnings: list[str] = [] + for src in sources: + entries, manifest, cache_root = _load_dossier(src) + if not entries: + warnings.append(f"no bib_ledger entries under {src}") + for entry in entries: + by_url.setdefault((entry.get("primary_url") or "").strip(), []).append(entry["bibkey"]) + rec = resolve_entry(entry, manifest, cache_root, use_live=use_live, write_back=write_back) + by_key.setdefault(entry["bibkey"], []).append(rec) + + records: dict[str, dict] = {} + for bibkey, recs in by_key.items(): + best = next((r for r in recs if not r["display"]), recs[0]) # prefer resolved + records[bibkey] = best + if best["display"]: + warnings.append(f"display-string authors (hand-fix): {bibkey}") + if len({r["title"] for r in recs if r["title"]}) > 1: + warnings.append(f"duplicate bibkey {bibkey}: differing titles across dossiers; kept the resolved/first") + reverse = {u: sorted(set(ks)) for u, ks in by_url.items() if u and len(set(ks)) > 1} + return records, warnings, reverse + + +# --------------------------------------------------------------------------- # +# --reconcile: audit our output against external .bib files, keyed by arXiv id +# --------------------------------------------------------------------------- # +_EPRINT_RE = re.compile(r"eprint\s*=\s*[{\"]?\s*([0-9]{4}\.[0-9]{4,5})", re.I) +_ARXIV_IN_URL_RE = re.compile(r"arxiv\.org/(?:abs|pdf)/([0-9]{4}\.[0-9]{4,5})", re.I) +_TITLE_RE = re.compile(r"\btitle\s*=\s*[{\"](.+?)[}\"]\s*,", re.S | re.I) + + +def _external_index(text: str, source: str) -> list[tuple[str, dict]]: + """(arxiv_id, record) for each entry in an external .bib with an eprint/arXiv URL. + + Uses the brace-balanced parser and returns a LIST, so two external entries + that share an arXiv id (a reverse collision) both survive the comparison.""" + out: list[tuple[str, dict]] = [] + for _typ, key, body in parse_entries(text): + m = _EPRINT_RE.search(body) or _ARXIV_IN_URL_RE.search(body) + if not m: + continue + t = _TITLE_RE.search(body) + out.append((m.group(1), {"source": source, "key": key, + "title": " ".join(t.group(1).split()) if t else ""})) + return out + + +def reconcile(records: dict[str, dict], external: list[Path]) -> list[str]: + """Report where the same arXiv id appears under differing keys across sources.""" + index: dict[str, list[dict]] = {} + for rec in records.values(): # iterate directly -- a shared eprint keeps BOTH keys + if rec["eprint"]: + index.setdefault(rec["eprint"], []).append( + {"source": "emit-bibtex", "key": rec["bibkey"], "title": rec["title"]}) + for path in external: + if not path.exists(): + continue + for aid, rec in _external_index(path.read_text(errors="replace"), path.name): + index.setdefault(aid, []).append(rec) + report: list[str] = [] + for aid, hits in sorted(index.items()): + keys = {h["key"] for h in hits} + if len(keys) > 1: + report.append(f"arXiv {aid}: differing keys {sorted(keys)} across {[h['source'] for h in hits]}") + return report + + +def main(argv: list[str]) -> int: + parser = argparse.ArgumentParser(prog="emit-bibtex", description=__doc__.splitlines()[0] if __doc__ else None) + parser.add_argument("sources", nargs="+", help="dossier dir(s) or bib_ledger.yml path(s)") + parser.add_argument("--out", help="write the .bib here (default: stdout)") + parser.add_argument("--seed", help="prepend a hand-verified seed .bib (e.g. fields2023vlt)") + parser.add_argument("--no-live", action="store_true", help="disable the live arXiv fallback") + parser.add_argument("--no-overwrite", action="store_true", help="refuse to overwrite an existing --out") + parser.add_argument("--reconcile", nargs="+", metavar="BIB", help="audit output against external .bib files") + args = parser.parse_args(argv) + + sources = [Path(s).expanduser().resolve() for s in args.sources] + for src in sources: + if not src.exists(): + print(f"error: source does not exist: {src}", file=sys.stderr) + return 2 + + # Usage checks run BEFORE the side-effecting emit (whose live fallback may + # re-cache): a missing --seed or a --no-overwrite refusal must not touch the + # cache or the network first. + seed_keys: set[str] = set() + seed_text = "" + if args.seed: + seed_path = Path(args.seed).expanduser().resolve() + if not seed_path.exists(): + print(f"error: seed does not exist: {seed_path}", file=sys.stderr) + return 2 + seed_text = seed_path.read_text(encoding="utf-8").rstrip() + seed_keys = {k for _t, k, _b in parse_entries(seed_text)} + out_path: Path | None = None + if args.out: + out_path = Path(args.out).expanduser().resolve() + if out_path.exists() and args.no_overwrite: + print(f"error: refusing to overwrite (--no-overwrite): {out_path}", file=sys.stderr) + return 2 + + records, warnings, reverse = emit(sources, use_live=not args.no_live, write_back=not args.no_live) + + blocks = [format_entry(records[k]) for k in sorted(records) if k not in seed_keys] + body = "\n\n".join(blocks) + "\n" + text = (seed_text + "\n\n" + body) if seed_text else body + + errors = validate_text(text) + if errors: + for err in errors: + print(f"error: emitted .bib failed validation: {err}", file=sys.stderr) + return 1 + + for warn in warnings: + print(f"warning: {warn}", file=sys.stderr) + for url, keys in sorted(reverse.items()): + print(f"note: one URL, multiple bibkeys (both emitted): {keys} -> {url}", file=sys.stderr) + if args.reconcile: + for line in reconcile(records, [Path(p).expanduser().resolve() for p in args.reconcile]): + print(f"reconcile: {line}", file=sys.stderr) + + emitted = [k for k in records if k not in seed_keys] + n_display = sum(1 for k in emitted if records[k]["display"]) + print( + f"emit-bibtex: {len(emitted) + len(seed_keys)} entries " + f"({len(emitted) - n_display} cache/live-resolved, {n_display} display-string" + f"{', ' + str(len(seed_keys)) + ' seed' if seed_keys else ''})", + file=sys.stderr, + ) + + if out_path is not None: + out_path.parent.mkdir(parents=True, exist_ok=True) + out_path.write_text(text, encoding="utf-8") + print(f"emit-bibtex: wrote {out_path}", file=sys.stderr) + else: + sys.stdout.write(text) + return 0 + + +if __name__ == "__main__": + sys.exit(main(sys.argv[1:])) diff --git a/tests/test_bibtex_out.py b/tests/test_bibtex_out.py new file mode 100644 index 0000000..0e11bc9 --- /dev/null +++ b/tests/test_bibtex_out.py @@ -0,0 +1,75 @@ +"""Tests for validators/bibtex_out.py — the emitted-.bib schema validator.""" +from __future__ import annotations + +import sys +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(REPO_ROOT)) + +from validators import bibtex_out # type: ignore[import-not-found] # noqa: E402 + +_GOOD = ( + "@misc{doe2023x,\n" + " author = {Doe, Jane and Roe, Richard},\n" + " title = {A Perfectly Valid Title},\n" + " year = {2023},\n" + "}\n" +) + + +def test_bibtex_out_accepts_a_valid_entry() -> None: + assert bibtex_out.validate_text(_GOOD) == [] + + +def test_bibtex_out_rejects_a_file_with_no_entries() -> None: + errors = bibtex_out.validate_text("% just a comment, no entries\n") + assert any("no BibTeX entries" in e for e in errors), errors + + +def test_bibtex_out_detects_duplicate_keys() -> None: + errors = bibtex_out.validate_text(_GOOD + "\n" + _GOOD) + assert any("duplicate entry key: doe2023x" in e for e in errors), errors + + +def test_bibtex_out_detects_missing_required_field() -> None: + entry = "@misc{no2023author,\n title = {T},\n year = {2023},\n}\n" + errors = bibtex_out.validate_text(entry) + assert any("missing required field 'author'" in e for e in errors), errors + + +def test_bibtex_out_detects_escaped_braces() -> None: + entry = "@misc{bad2023braces,\n author = {Doe, J},\n title = {The \\{DNA\\} of X},\n}\n" + errors = bibtex_out.validate_text(entry) + assert any("brace" in e for e in errors), errors + + +def test_bibtex_out_validate_rejects_a_directory(tmp_path) -> None: + errors = bibtex_out.validate(tmp_path) + assert any("got directory" in e for e in errors), errors + + +# --- regression tests for the adversarial-review findings --- + + +def test_bibtex_out_detects_unbalanced_braces() -> None: + errors = bibtex_out.validate_text("@misc{k,\n author={A},\n title={unterminated,\n}\n") + assert any("unbalanced braces" in e for e in errors), errors + + +def test_bibtex_out_field_name_inside_a_value_is_not_the_field() -> None: + # 'author =' / 'title =' appearing INSIDE a note value must not satisfy the + # required-field check. + errors = bibtex_out.validate_text("@misc{k,\n note={words author = and title = only},\n}\n") + assert any("missing required field 'author'" in e for e in errors), errors + assert any("missing required field 'title'" in e for e in errors), errors + + +def test_parse_entries_handles_one_line_and_skips_string() -> None: + got = bibtex_out.parse_entries('@string{p = "X"}\n@misc{k, author={A}, title={T},}\n') + assert [(t, k) for t, k, _b in got] == [("misc", "k")] # @string skipped; one-line parsed + + +def test_parse_entries_respects_nested_braces_in_a_value() -> None: + got = bibtex_out.parse_entries("@misc{k,\n title={The {DNA} of X},\n author={A},\n}\n") + assert len(got) == 1 and got[0][1] == "k" diff --git a/tests/test_cli.py b/tests/test_cli.py index abb4ff6..c5fc51d 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -143,7 +143,7 @@ def test_pyproject_declares_console_entry_point() -> None: @pytest.mark.parametrize("verb", sorted({ "cache-source", "assemble", "render-index", "build-claim-graph", "verify-citations", "build-dashboard", "freshness", "export", - "backlog-stamp", "resume-gather", "compose-kg", + "backlog-stamp", "resume-gather", "compose-kg", "emit-bibtex", })) def test_cli_core_verbs_registered(verb: str) -> None: assert verb in cli._REGISTRY diff --git a/tests/test_emit_bibtex.py b/tests/test_emit_bibtex.py new file mode 100644 index 0000000..60a1536 --- /dev/null +++ b/tests/test_emit_bibtex.py @@ -0,0 +1,182 @@ +"""Tests for scripts/emit_bibtex.py — ledger→BibTeX with cache-resolved authors. + +Hand-rolled synthetic fixtures (a ledger + manifest + a cached /abs/ blob with +Highwire tags) under tmp_path; no network (--no-live) and no unittest.mock. +""" +from __future__ import annotations + +import sys +from pathlib import Path + +import yaml + +REPO_ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(REPO_ROOT)) + +from scripts import emit_bibtex # type: ignore[import-not-found] # noqa: E402 + +_ABS_HTML = ( + "\n" + '\n' + '\n' + '\n' + '\n' + '\n' + "" +) + + +def _make_dossier(tmp_path: Path) -> Path: + """A synthetic dossier: one arXiv entry (cache HAS Highwire tags) + one non-arXiv.""" + blob = tmp_path / "cache" / "text" / "sha256" / "abc.txt" + blob.parent.mkdir(parents=True) + blob.write_text(_ABS_HTML, encoding="utf-8") + dossier = tmp_path / "dossier" + dossier.mkdir() + (dossier / "bib_ledger.yml").write_text( + yaml.safe_dump({ + "entries": [ + { + "bibkey": "siglip2", + "primary_url": "https://arxiv.org/abs/2502.14786", + "title": "wrong display title", + "authors": "Tschannen et al. (2025)", + "cache_ids": ["cache_abc"], + }, + { + "bibkey": "meta2024repo", + "primary_url": "https://github.com/meta/x", + "title": "A Repository", + "authors": "Meta AI (2024)", + "published_online": "2024-01-01", + }, + ] + }), + encoding="utf-8", + ) + (dossier / "cache_manifest.yml").write_text( + yaml.safe_dump({ + "cache_root": str(tmp_path / "cache"), + "entries": [{ + "cache_id": "cache_abc", + "source_url": "https://arxiv.org/abs/2502.14786", + "text_path": "text/sha256/abc.txt", + }], + }), + encoding="utf-8", + ) + return dossier + + +def test_emit_bibtex_resolves_authors_from_cached_highwire_tags(tmp_path) -> None: + out = tmp_path / "refs.bib" + assert emit_bibtex.main([str(_make_dossier(tmp_path)), "--no-live", "--out", str(out)]) == 0 + text = out.read_text() + assert "author = {Tschannen, Michael and Zhai, Xiaohua}" in text + assert "Tschannen et al." not in text # the ledger display string is NOT used + assert "SigLIP 2: Better Encoders" in text # cache title overrides the ledger's wrong title + assert "eprint = {2502.14786}" in text + assert "archiveprefix = {arXiv}," in text + + +def test_emit_bibtex_falls_back_to_display_string_for_non_arxiv(tmp_path, capsys) -> None: + out = tmp_path / "refs.bib" + emit_bibtex.main([str(_make_dossier(tmp_path)), "--no-live", "--out", str(out)]) + text = out.read_text() + assert "author = {Meta AI (2024)}" in text # display string kept as-is + assert "year = {2024}" in text # from published_online + err = capsys.readouterr().err + assert "display-string authors" in err and "meta2024repo" in err # flagged, not hidden + + +def test_emit_bibtex_escapes_specials_but_never_braces() -> None: + assert emit_bibtex.bib_escape("AT&T 50% #1 x_y $z") == r"AT\&T 50\% \#1 x\_y \$z" + assert emit_bibtex.bib_escape("The {DNA} of X") == "The {DNA} of X" # braces untouched + + +def test_emit_bibtex_dedups_a_bibkey_repeated_across_dossiers(tmp_path) -> None: + d1 = _make_dossier(tmp_path) + d2 = tmp_path / "d2" + d2.mkdir() + (d2 / "bib_ledger.yml").write_text( + yaml.safe_dump({"entries": [{ + "bibkey": "meta2024repo", "primary_url": "https://github.com/meta/x", + "title": "A Repository", "authors": "Meta AI (2024)", + }]}), + encoding="utf-8", + ) + out = tmp_path / "refs.bib" + emit_bibtex.main([str(d1), str(d2), "--no-live", "--out", str(out)]) + assert out.read_text().count("@misc{meta2024repo,") == 1 + + +def test_emit_bibtex_seed_wins_over_a_duplicate_ledger_key(tmp_path) -> None: + seed = tmp_path / "seed.bib" + seed.write_text("@misc{meta2024repo,\n author = {Hand, Verified},\n title = {Seed Entry},\n}\n") + out = tmp_path / "refs.bib" + emit_bibtex.main([str(_make_dossier(tmp_path)), "--no-live", "--seed", str(seed), "--out", str(out)]) + text = out.read_text() + assert text.count("@misc{meta2024repo,") == 1 # not duplicated + assert "Hand, Verified" in text # the hand-verified seed version is the one kept + + +def test_emit_bibtex_rejects_a_missing_source(tmp_path) -> None: + assert emit_bibtex.main([str(tmp_path / "does_not_exist"), "--no-live"]) == 2 + + +def test_emit_bibtex_no_overwrite_refuses_an_existing_out(tmp_path) -> None: + out = tmp_path / "refs.bib" + out.write_text("do not clobber") + rc = emit_bibtex.main([str(_make_dossier(tmp_path)), "--no-live", "--out", str(out), "--no-overwrite"]) + assert rc == 2 + assert out.read_text() == "do not clobber" + + +def test_to_last_first_normalises_atom_name_order() -> None: + assert emit_bibtex._to_last_first("Michael Tschannen") == "Tschannen, Michael" + assert emit_bibtex._to_last_first("Zhai, Xiaohua") == "Zhai, Xiaohua" # already Family, Given + assert emit_bibtex._to_last_first("Plato") == "Plato" # single token unchanged + + +# --- regression tests for the adversarial-review findings --- + + +def test_highwire_keeps_apostrophe_in_double_quoted_value() -> None: + tag = '' + assert emit_bibtex._highwire(tag, "author") == ["O'Connor, Alice"] + + +def test_highwire_ignores_data_prefixed_attributes() -> None: + assert emit_bibtex._highwire('', "author") == [] + + +def test_bib_escape_handles_backslash_tilde_caret() -> None: + assert emit_bibtex.bib_escape(r"C:\t A~B x^2") == r"C:\textbackslash{}t A\textasciitilde{}B x\textasciicircum{}2" + + +def test_author_field_brace_protects_a_corporate_and_name() -> None: + out = emit_bibtex._author_field(["Research and Development Team", "Doe, Jane"]) + assert out == "{Research and Development Team} and Doe, Jane" + + +def test_to_last_first_keeps_surname_particles() -> None: + assert emit_bibtex._to_last_first("Ludwig van Beethoven") == "van Beethoven, Ludwig" + assert emit_bibtex._to_last_first("Juan de la Cruz") == "de la Cruz, Juan" + + +def test_emit_bibtex_dedup_prefers_cache_resolved_regardless_of_order(tmp_path) -> None: + good = _make_dossier(tmp_path) # 'siglip2' with cached Highwire authors + d0 = tmp_path / "d0" + d0.mkdir() + (d0 / "bib_ledger.yml").write_text( # SAME bibkey, display-only, passed FIRST + yaml.safe_dump({"entries": [{ + "bibkey": "siglip2", "primary_url": "https://arxiv.org/abs/2502.14786", + "title": "display only", "authors": "Tschannen et al. (2025)", + }]}), + encoding="utf-8", + ) + out = tmp_path / "refs.bib" + emit_bibtex.main([str(d0), str(good), "--no-live", "--out", str(out)]) + text = out.read_text() + assert "Tschannen, Michael and Zhai, Xiaohua" in text # cache-resolved wins despite order + assert "Tschannen et al." not in text diff --git a/validators/bibtex_out.py b/validators/bibtex_out.py new file mode 100644 index 0000000..f1a5986 --- /dev/null +++ b/validators/bibtex_out.py @@ -0,0 +1,95 @@ +"""Validate an emitted BibTeX ``.bib`` (the output of ``scripts/emit_bibtex.py``). + +Schema-only, per the toolkit contract: braces balance, every ``@type{key,`` +parses with balanced braces, keys are unique, ``author``/``title`` are present as +top-level fields, and NO field value contains an escaped brace ``\\{`` / ``\\}`` +(which would destroy BibTeX brace-protection -- the bug this feature exists to +avoid). Runnable as ``python -m validators.bibtex_out ``; exits 0 clean, +1 on violation, 2 on usage/missing path. +""" +from __future__ import annotations + +import re +import sys +from pathlib import Path + +if __package__ in (None, ""): + sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from validators._common import cli_main + +_AT_ENTRY_RE = re.compile(r"@(\w+)\s*\{") +# A required field must appear at the START of a line (col 0 after indent), so a +# field NAME occurring inside another field's value ("note = {words author =}") +# is not mistaken for the field itself. +_NON_ENTRY_TYPES = {"string", "comment", "preamble"} + + +def parse_entries(text: str) -> list[tuple[str, str, str]]: + """Return ``(type, key, fields)`` for each ``@type{key, ...}`` with BALANCED + braces. Skips ``@string``/``@comment``/``@preamble`` (not reference entries). + An entry whose braces never close is dropped (``validate_text`` catches the + imbalance separately).""" + out: list[tuple[str, str, str]] = [] + i, n = 0, len(text) + while i < n: + at = text.find("@", i) + if at < 0: + break + m = _AT_ENTRY_RE.match(text, at) + if not m: + i = at + 1 + continue + typ = m.group(1).lower() + depth, j, start = 1, m.end(), m.end() + while j < n and depth: + depth += (text[j] == "{") - (text[j] == "}") + j += 1 + if depth != 0: # unterminated -- stop; imbalance reported by validate_text + break + body = text[start:j - 1] + i = j + if typ in _NON_ENTRY_TYPES: + continue + key, _, fields = body.partition(",") + out.append((typ, key.strip(), fields)) + return out + + +def validate_text(text: str) -> list[str]: + """Return a list of schema violations for the BibTeX ``text`` (empty = valid).""" + errors: list[str] = [] + if text.count("{") != text.count("}"): + errors.append("unbalanced braces in the .bib (a field value has a stray { or })") + + entries = parse_entries(text) + if not entries: + errors.append("no BibTeX entries found (expected at least one @type{key, ...})") + return errors + + seen: set[str] = set() + for _type, key, _fields in entries: + if not key: + errors.append("an entry has an empty citation key") + elif key in seen: + errors.append(f"duplicate entry key: {key}") + seen.add(key) + + for _type, key, fields in entries: + if r"\{" in fields or r"\}" in fields: + errors.append(f"{key}: escaped brace (\\{{ or \\}}) corrupts brace-protection") + for required in ("author", "title"): + if not re.search(rf"(?mi)^\s*{required}\s*=", fields): + errors.append(f"{key}: missing required field '{required}'") + return errors + + +def validate(path: Path) -> list[str]: + """Validate the ``.bib`` file at ``path``.""" + if path.is_dir(): + return [f"expected a .bib file, got directory: {path}"] + return validate_text(path.read_text(encoding="utf-8")) + + +if __name__ == "__main__": + sys.exit(cli_main(sys.argv, validate))