d042a47fe2
- pragma: no cover on ImportError fallbacks (packages always installed), __main__ guards, timing-dependent sleep loops, unreachable dead code, pypdf fallback (not installed) - New tests: cloudflare _headers, cli.main(), perf show no-spans, zot.ops remaining collection keys, seed read_excel Tracks #353
363 lines
11 KiB
Python
363 lines
11 KiB
Python
"""Serialize one Zotero item to an LLM-ready markdown document.
|
|
|
|
Shape::
|
|
|
|
---
|
|
zot_key: ABCD1234
|
|
title: "Long study title"
|
|
authors:
|
|
- Smith J
|
|
- Lee R
|
|
year: 2024
|
|
journal: "N Engl J Med"
|
|
doi: "10.1056/…"
|
|
url: "https://…"
|
|
tags: [module:skin-subs, source:pubmed, type:rct, year:2024]
|
|
collections: [Healthcare Data Platform > Skin Substitutes]
|
|
---
|
|
|
|
# Abstract
|
|
|
|
<abstract text>
|
|
|
|
# Full text
|
|
<PDF text — only when include_fulltext=True>
|
|
|
|
# Reviewer annotations
|
|
<existing Zotero notes and PDF highlights, if any>
|
|
|
|
Stage 2 (title/abstract screening) asks for ``include_fulltext=False``
|
|
— fast, cheap, many items per prompt. Stage 3 (eligibility + data
|
|
extraction) passes ``True``; PDF text comes from the attachment via
|
|
pdfminer (Zotero's own fulltext tables store a bag-of-words index
|
|
useful for search, not for LLM context).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING
|
|
|
|
from zot.db import FIELD_MAP
|
|
|
|
if TYPE_CHECKING:
|
|
from zot.db import Db
|
|
|
|
|
|
@dataclass
|
|
class ItemSnapshot:
|
|
"""Everything we know about a Zotero item in one place."""
|
|
|
|
zot_id: int
|
|
zot_key: str
|
|
title: str
|
|
authors: list[str]
|
|
year: str
|
|
journal: str
|
|
doi: str
|
|
url: str
|
|
abstract: str
|
|
tags: list[str]
|
|
collections: list[str] # "Parent > Child > Grandchild" paths
|
|
attachment_paths: list[Path] # local PDFs/HTMLs
|
|
notes: list[str] # existing Zotero notes (HTML stripped)
|
|
annotations: list[str] # PDF annotation text (itemAnnotations)
|
|
|
|
|
|
# ── Loading ──────────────────────────────────────────────────────
|
|
|
|
|
|
def load_item(db: Db, zot_id: int, storage_dir: Path) -> ItemSnapshot:
|
|
"""Pull every relevant field for one item from the Zotero DB."""
|
|
|
|
# Core fields
|
|
field_of = {fid: name for name, fid in FIELD_MAP.items()}
|
|
raw: dict[str, str] = {}
|
|
for row in db.con.execute(
|
|
"""SELECT id.fieldID, idv.value
|
|
FROM itemData id
|
|
JOIN itemDataValues idv ON id.valueID = idv.valueID
|
|
WHERE id.itemID = ?""",
|
|
(zot_id,),
|
|
).fetchall():
|
|
fname = field_of.get(row[0])
|
|
if fname:
|
|
raw[fname] = row[1]
|
|
|
|
# Authors
|
|
authors: list[str] = []
|
|
for row in db.con.execute(
|
|
"""SELECT c.firstName, c.lastName
|
|
FROM itemCreators ic
|
|
JOIN creators c ON ic.creatorID = c.creatorID
|
|
WHERE ic.itemID = ?
|
|
ORDER BY ic.orderIndex""",
|
|
(zot_id,),
|
|
).fetchall():
|
|
name = " ".join(part for part in (row[0], row[1]) if part)
|
|
if name:
|
|
authors.append(name)
|
|
|
|
# Tags
|
|
tags = [
|
|
r[0]
|
|
for r in db.con.execute(
|
|
"""SELECT t.name FROM itemTags it
|
|
JOIN tags t ON it.tagID = t.tagID
|
|
WHERE it.itemID = ? ORDER BY t.name""",
|
|
(zot_id,),
|
|
).fetchall()
|
|
]
|
|
|
|
# Collection paths
|
|
collections = [
|
|
_collection_path(db, row[0])
|
|
for row in db.con.execute(
|
|
"SELECT collectionID FROM collectionItems WHERE itemID = ?",
|
|
(zot_id,),
|
|
).fetchall()
|
|
]
|
|
|
|
# Attachments — local paths resolve against the storage dir
|
|
att_paths: list[Path] = []
|
|
for att_row in db.con.execute(
|
|
"""SELECT ia.itemID, ia.path, i.key
|
|
FROM itemAttachments ia
|
|
JOIN items i ON ia.itemID = i.itemID
|
|
WHERE ia.parentItemID = ?""",
|
|
(zot_id,),
|
|
).fetchall():
|
|
path = _resolve_attachment(storage_dir, att_row[2], att_row[1])
|
|
if path is not None:
|
|
att_paths.append(path)
|
|
|
|
# Reviewer notes (HTML → plain text via a crude strip — LLM can
|
|
# handle the noise, and we avoid another dep).
|
|
notes: list[str] = []
|
|
for row in db.con.execute(
|
|
"SELECT note FROM itemNotes WHERE parentItemID = ?",
|
|
(zot_id,),
|
|
).fetchall():
|
|
txt = _strip_html((row[0] or "").strip())
|
|
if txt:
|
|
notes.append(txt)
|
|
|
|
# PDF annotations (itemAnnotations.text + comment)
|
|
annotations: list[str] = []
|
|
for row in db.con.execute(
|
|
"""SELECT ia.text, ia.comment
|
|
FROM itemAnnotations ia
|
|
JOIN itemAttachments att ON ia.parentItemID = att.itemID
|
|
WHERE att.parentItemID = ?""",
|
|
(zot_id,),
|
|
).fetchall():
|
|
parts = [p for p in (row[0], row[1]) if p]
|
|
if parts:
|
|
annotations.append(" — ".join(parts))
|
|
|
|
return ItemSnapshot(
|
|
zot_id=zot_id,
|
|
zot_key=_item_key(db, zot_id),
|
|
title=raw.get("title", ""),
|
|
authors=authors,
|
|
year=_year_from(raw.get("date", "")),
|
|
journal=raw.get("publicationTitle", raw.get("seriesTitle", "")),
|
|
doi=raw.get("DOI", ""),
|
|
url=raw.get("url", ""),
|
|
abstract=raw.get("abstractNote", ""),
|
|
tags=tags,
|
|
collections=collections,
|
|
attachment_paths=att_paths,
|
|
notes=notes,
|
|
annotations=annotations,
|
|
)
|
|
|
|
|
|
# ── Rendering ────────────────────────────────────────────────────
|
|
|
|
|
|
def to_markdown(snap: ItemSnapshot, *, include_fulltext: bool = False) -> str:
|
|
"""Render an item as the LLM-ready markdown format."""
|
|
lines: list[str] = ["---"]
|
|
lines.append(f'zot_key: "{snap.zot_key}"')
|
|
lines.append(f"title: {_yaml_scalar(snap.title)}")
|
|
if snap.authors:
|
|
lines.append("authors:")
|
|
for a in snap.authors:
|
|
lines.append(f" - {_yaml_scalar(a)}")
|
|
if snap.year:
|
|
lines.append(f'year: "{snap.year}"')
|
|
if snap.journal:
|
|
lines.append(f"journal: {_yaml_scalar(snap.journal)}")
|
|
if snap.doi:
|
|
lines.append(f'doi: "{snap.doi}"')
|
|
if snap.url:
|
|
lines.append(f'url: "{snap.url}"')
|
|
if snap.tags:
|
|
lines.append(f"tags: [{', '.join(_yaml_scalar(t) for t in snap.tags)}]")
|
|
if snap.collections:
|
|
lines.append("collections:")
|
|
for c in snap.collections:
|
|
lines.append(f" - {_yaml_scalar(c)}")
|
|
lines.append("---")
|
|
lines.append("")
|
|
|
|
lines.append("# Abstract")
|
|
lines.append("")
|
|
lines.append(snap.abstract.strip() or "_(no abstract)_")
|
|
lines.append("")
|
|
|
|
if include_fulltext:
|
|
text = _extract_fulltext(snap.attachment_paths)
|
|
if text:
|
|
lines.append("# Full text")
|
|
lines.append("")
|
|
lines.append(text)
|
|
lines.append("")
|
|
|
|
if snap.notes:
|
|
lines.append("# Reviewer notes")
|
|
lines.append("")
|
|
for i, n in enumerate(snap.notes, 1):
|
|
lines.append(f"## Note {i}")
|
|
lines.append("")
|
|
lines.append(n)
|
|
lines.append("")
|
|
|
|
if snap.annotations:
|
|
lines.append("# PDF annotations")
|
|
lines.append("")
|
|
for ann in snap.annotations:
|
|
lines.append(f"- {ann}")
|
|
lines.append("")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
# ── Helpers ──────────────────────────────────────────────────────
|
|
|
|
|
|
def _item_key(db: Db, zot_id: int) -> str:
|
|
r = db.con.execute("SELECT key FROM items WHERE itemID = ?", (zot_id,)).fetchone()
|
|
return r[0] if r else ""
|
|
|
|
|
|
def _collection_path(db: Db, collection_id: int) -> str:
|
|
"""Render a collection's lineage as ``Root > Child > Grandchild``."""
|
|
names: list[str] = []
|
|
cid: int | None = collection_id
|
|
while cid is not None:
|
|
row = db.con.execute(
|
|
"SELECT collectionName, parentCollectionID FROM collections "
|
|
"WHERE collectionID = ?",
|
|
(cid,),
|
|
).fetchone()
|
|
if not row:
|
|
break
|
|
names.append(row[0])
|
|
cid = row[1]
|
|
return " > ".join(reversed(names))
|
|
|
|
|
|
def _resolve_attachment(
|
|
storage_dir: Path,
|
|
attachment_key: str,
|
|
path_field: str | None,
|
|
) -> Path | None:
|
|
"""Map Zotero's ``storage:<filename>`` path form into an absolute file."""
|
|
if not path_field:
|
|
return None
|
|
if path_field.startswith("storage:"):
|
|
name = path_field[len("storage:") :]
|
|
cand = storage_dir / attachment_key / name
|
|
return cand if cand.is_file() else None
|
|
# Linked-file modes store an absolute path directly.
|
|
p = Path(path_field)
|
|
return p if p.is_absolute() and p.is_file() else None
|
|
|
|
|
|
_PDF_LIB_ATTEMPTS = ("pdfminer", "pypdf", "pypdfium2")
|
|
|
|
|
|
def _extract_fulltext(paths: list[Path], max_chars: int = 240_000) -> str:
|
|
"""Pull text out of PDF/HTML attachments with whatever library is
|
|
available. Stage 3 prompts can easily hit 200K+ tokens of context,
|
|
so we cap at ~60K tokens (240K chars) to leave headroom for the
|
|
criteria + system prompt in the cached prefix."""
|
|
chunks: list[str] = []
|
|
for p in paths:
|
|
text = _extract_one(p) or ""
|
|
if not text:
|
|
continue
|
|
chunks.append(f"\n\n---\n\n_(source: {p.name})_\n\n{text}")
|
|
if sum(len(c) for c in chunks) >= max_chars:
|
|
break
|
|
joined = "".join(chunks)
|
|
return joined[:max_chars]
|
|
|
|
|
|
def _extract_one(path: Path) -> str | None:
|
|
suffix = path.suffix.lower()
|
|
if suffix in (".html", ".htm"):
|
|
try:
|
|
return _strip_html(path.read_text(encoding="utf-8", errors="ignore"))
|
|
except OSError:
|
|
return None
|
|
if suffix != ".pdf":
|
|
return None
|
|
# Prefer pdfminer (deterministic, dep-free flow). Fall back on
|
|
# whatever's installed.
|
|
try:
|
|
from pdfminer.high_level import extract_text # type: ignore[import-not-found]
|
|
|
|
return extract_text(str(path)) or None
|
|
except ImportError:
|
|
pass
|
|
try:
|
|
from pypdf import PdfReader # type: ignore[import-not-found]
|
|
|
|
reader = PdfReader(str(path)) # pragma: no cover — pypdf not installed
|
|
return (
|
|
"\n\n".join(page.extract_text() or "" for page in reader.pages) or None
|
|
) # pragma: no cover
|
|
except ImportError:
|
|
return None
|
|
|
|
|
|
def _strip_html(s: str) -> str:
|
|
"""Tiny HTML→text — good enough for Zotero notes and basic HTML
|
|
attachments; we don't need anything fancy for LLM consumption."""
|
|
import re as _re
|
|
|
|
# Drop script/style blocks entirely.
|
|
s = _re.sub(r"<(script|style)[^>]*>.*?</\1>", "", s, flags=_re.S | _re.I)
|
|
# Break on block boundaries.
|
|
s = _re.sub(r"<(p|div|br|li|h[1-6])[^>]*>", "\n", s, flags=_re.I)
|
|
# Strip remaining tags.
|
|
s = _re.sub(r"<[^>]+>", "", s)
|
|
# HTML entities the cheap way.
|
|
s = (
|
|
s.replace(" ", " ")
|
|
.replace("&", "&")
|
|
.replace("<", "<")
|
|
.replace(">", ">")
|
|
.replace(""", '"')
|
|
.replace("'", "'")
|
|
)
|
|
return _re.sub(r"\n{3,}", "\n\n", s).strip()
|
|
|
|
|
|
def _year_from(date_str: str) -> str:
|
|
"""Zotero `date` can be '2024', '2024-05', 'May 2024', '2024-05-12T…'.
|
|
Grab the first 4-digit year we see."""
|
|
import re as _re
|
|
|
|
m = _re.search(r"\b(19|20)\d{2}\b", date_str or "")
|
|
return m.group(0) if m else ""
|
|
|
|
|
|
def _yaml_scalar(s: str) -> str:
|
|
"""Quote + escape a YAML scalar string."""
|
|
return '"' + (s or "").replace("\\", "\\\\").replace('"', '\\"') + '"'
|