fix: bib.sync and XML parsers — Zotero key validation, creators, journalArticle, itertext
bib.sync: - Keys validated against Zotero allowedKeyChars (no 0, 1, O, lowercase) - Creators parsed from extra into Zotero creators/itemCreators tables - PubMed articles pushed as journalArticle (typeID=22) not document - DOI, PMID, Journal parsed into dedicated Zotero fields (8, 86, 41) - Authors stripped from extra to avoid duplication XML parsers (search_pubmed, snowball_citations): - _text() uses itertext() instead of .text to handle mixed-content XML (sup, i, sub tags inside ArticleTitle) — fixes 20 blank titles Verified: 10,262 items, 0 blank titles, 0 invalid keys, 60,866 creators, 9,658 DOIs, 10,232 PMIDs in proper Zotero fields.
This commit is contained in:
kert
@@ -198,11 +198,18 @@ def efetch_articles(
|
||||
|
||||
|
||||
def _text(el: ET.Element | None, path: str, default: str = "") -> str:
|
||||
"""Get text from an XML element, handling None."""
|
||||
"""Get all text from an XML element, including mixed content children.
|
||||
|
||||
PubMed XML uses inline markup like ``<sup>``, ``<i>``, ``<sub>``
|
||||
inside ``<ArticleTitle>`` — ``.text`` only returns text before the
|
||||
first child element. ``itertext()`` concatenates all text nodes.
|
||||
"""
|
||||
if el is None:
|
||||
return default
|
||||
node = el.find(path)
|
||||
return (node.text or default) if node is not None else default
|
||||
if node is None:
|
||||
return default
|
||||
return "".join(node.itertext()).strip() or default
|
||||
|
||||
|
||||
def _parse_pubmed_xml(xml_text: str) -> list[Article]:
|
||||
|
||||
@@ -96,10 +96,13 @@ def elink_cited_by(pmids: list[str], batch_size: int = 50) -> dict[str, list[str
|
||||
|
||||
|
||||
def _text(el: ET.Element | None, path: str, default: str = "") -> str:
|
||||
"""Get all text including mixed-content children (sup, i, sub, etc.)."""
|
||||
if el is None:
|
||||
return default
|
||||
node = el.find(path)
|
||||
return (node.text or default) if node is not None else default
|
||||
if node is None:
|
||||
return default
|
||||
return "".join(node.itertext()).strip() or default
|
||||
|
||||
|
||||
def efetch_basic(pmids: list[str], batch_size: int = 100) -> list[dict]:
|
||||
@@ -152,7 +155,7 @@ def efetch_basic(pmids: list[str], batch_size: int = 100) -> list[dict]:
|
||||
|
||||
# Journal + year
|
||||
journal_el = article_el.find("Journal")
|
||||
journal = _text(journal_el, "Title") if journal_el else ""
|
||||
journal = _text(journal_el, "Title") if journal_el is not None else ""
|
||||
year = ""
|
||||
pub_date = article_el.find(".//PubDate")
|
||||
if pub_date is not None:
|
||||
|
||||
164
src/bib/sync.py
164
src/bib/sync.py
@@ -34,6 +34,7 @@ _TYPE_MAP = {
|
||||
"manual": 34, # report
|
||||
"download": 40, # webpage
|
||||
"source": 14, # document
|
||||
"journal-article": 22, # journalArticle
|
||||
}
|
||||
|
||||
# fieldName → fieldID (from Zotero's fields table)
|
||||
@@ -62,13 +63,34 @@ _FIELD_IDS = {
|
||||
"place": 26,
|
||||
"websiteTitle": 123,
|
||||
"websiteType": 42,
|
||||
# journalArticle fields
|
||||
"DOI": 8,
|
||||
"PMID": 86,
|
||||
"PMCID": 87,
|
||||
"ISSN": 44,
|
||||
"publicationTitle": 41,
|
||||
"journalAbbreviation": 85,
|
||||
"volume": 22,
|
||||
"issue": 67,
|
||||
}
|
||||
|
||||
|
||||
_ALLOWED_KEY_CHARS = "23456789ABCDEFGHIJKLMNPQRSTUVWXYZ"
|
||||
|
||||
|
||||
def _zotero_key() -> str:
|
||||
"""Generate an 8-char Zotero key using the allowed character set."""
|
||||
chars = "23456789ABCDEFGHIJKLMNPQRSTUVWXYZ"
|
||||
return "".join(random.choices(chars, k=8)) # noqa: S311
|
||||
"""Generate an 8-char Zotero key using the allowed character set.
|
||||
|
||||
Zotero keys must be exactly 8 chars from ``23456789ABCDEFGHIJKLMNPQRSTUVWXYZ``.
|
||||
No ``0``, ``1``, ``O``, or lowercase.
|
||||
Source: ``Zotero.Utilities.allowedKeyChars`` in utilities.js.
|
||||
"""
|
||||
return "".join(random.choices(_ALLOWED_KEY_CHARS, k=8)) # noqa: S311
|
||||
|
||||
|
||||
def _is_valid_zotero_key(key: str) -> bool:
|
||||
"""Check if a key is a valid Zotero object key."""
|
||||
return len(key) == 8 and all(c in _ALLOWED_KEY_CHARS for c in key)
|
||||
|
||||
|
||||
def _now_iso() -> str:
|
||||
@@ -115,6 +137,97 @@ def _set_field(
|
||||
)
|
||||
|
||||
|
||||
def _ensure_creator(con: sqlite3.Connection, first_name: str, last_name: str) -> int:
|
||||
"""Find or create a creator row. Returns creatorID."""
|
||||
row = con.execute(
|
||||
"SELECT creatorID FROM creators WHERE firstName = ? AND lastName = ?",
|
||||
(first_name, last_name),
|
||||
).fetchone()
|
||||
if row:
|
||||
return row[0]
|
||||
cur = con.execute(
|
||||
"INSERT INTO creators (firstName, lastName, fieldMode) VALUES (?, ?, 0)",
|
||||
(first_name, last_name),
|
||||
)
|
||||
return cur.lastrowid
|
||||
|
||||
|
||||
def _parse_authors_from_extra(extra: str) -> tuple[list[tuple[str, str]], str]:
|
||||
"""Parse 'Authors: Last First; Last First' from extra, return (authors, cleaned_extra).
|
||||
|
||||
Returns a list of (firstName, lastName) tuples and the extra text
|
||||
with the Authors line removed.
|
||||
"""
|
||||
authors: list[tuple[str, str]] = []
|
||||
cleaned_lines: list[str] = []
|
||||
for line in extra.split("\n"):
|
||||
if line.startswith("Authors:"):
|
||||
raw = line[8:].strip()
|
||||
for entry in raw.split(";"):
|
||||
entry = entry.strip()
|
||||
if not entry or entry.startswith("(+"):
|
||||
continue
|
||||
parts = entry.split(None, 1)
|
||||
if len(parts) == 2:
|
||||
authors.append((parts[1].strip(), parts[0].strip()))
|
||||
elif parts:
|
||||
authors.append(("", parts[0].strip()))
|
||||
else:
|
||||
cleaned_lines.append(line)
|
||||
return authors, "\n".join(cleaned_lines).strip()
|
||||
|
||||
|
||||
def _add_creators(
|
||||
con: sqlite3.Connection,
|
||||
item_id: int,
|
||||
authors: list[tuple[str, str]],
|
||||
creator_type_id: int = 10, # 10 = author
|
||||
) -> int:
|
||||
"""Add creators to a Zotero item. Returns count added."""
|
||||
for idx, (first_name, last_name) in enumerate(authors):
|
||||
creator_id = _ensure_creator(con, first_name, last_name)
|
||||
con.execute(
|
||||
"INSERT OR IGNORE INTO itemCreators "
|
||||
"(itemID, creatorID, creatorTypeID, orderIndex) "
|
||||
"VALUES (?, ?, ?, ?)",
|
||||
(item_id, creator_id, creator_type_id, idx),
|
||||
)
|
||||
return len(authors)
|
||||
|
||||
|
||||
def _parse_extra_to_journal_fields(extra: str) -> dict[str, str]:
|
||||
"""Parse structured extra text into Zotero journalArticle fields.
|
||||
|
||||
Extracts PMID, DOI, Journal, PubTypes, MeSH from lines like::
|
||||
|
||||
PMID: 12345678
|
||||
DOI: 10.1234/example
|
||||
Journal: The Lancet
|
||||
PubTypes: Randomized Controlled Trial; Journal Article
|
||||
MeSH: Skin Substitutes; Wound Healing
|
||||
|
||||
Returns a field dict with proper Zotero field names and a
|
||||
cleaned ``extra`` containing only the leftover lines.
|
||||
"""
|
||||
fields: dict[str, str] = {}
|
||||
leftover: list[str] = []
|
||||
|
||||
for line in extra.split("\n"):
|
||||
if line.startswith("PMID:"):
|
||||
fields["PMID"] = line[5:].strip()
|
||||
elif line.startswith("DOI:"):
|
||||
fields["DOI"] = line[4:].strip()
|
||||
elif line.startswith("Journal:"):
|
||||
fields["publicationTitle"] = line[8:].strip()
|
||||
elif line.startswith("Authors:"):
|
||||
pass # handled separately by _parse_authors_from_extra
|
||||
else:
|
||||
leftover.append(line)
|
||||
|
||||
fields["extra"] = "\n".join(leftover).strip()
|
||||
return fields
|
||||
|
||||
|
||||
def _item_to_zotero_fields(item: Item) -> dict[str, str]:
|
||||
"""Convert a bib Item to Zotero EAV field dict."""
|
||||
ej = json.loads(item.to_row().get("extra_json", "{}"))
|
||||
@@ -199,10 +312,24 @@ def _item_to_zotero_fields(item: Item) -> dict[str, str]:
|
||||
"extra": extra,
|
||||
}
|
||||
|
||||
# source (document)
|
||||
doc_type = ej.get("doc_type", "")
|
||||
|
||||
# Journal articles get full bibliographic fields
|
||||
if doc_type == "journal-article":
|
||||
fields = _parse_extra_to_journal_fields(item.extra)
|
||||
fields["title"] = item.title
|
||||
fields["date"] = item.date_published
|
||||
fields["url"] = item.url
|
||||
fields["accessDate"] = item.access_date
|
||||
fields["abstractNote"] = item.abstract
|
||||
if item.institution:
|
||||
fields.setdefault("publicationTitle", item.institution)
|
||||
return fields
|
||||
|
||||
# Generic source (document)
|
||||
return {
|
||||
"title": item.title,
|
||||
"type": ej.get("doc_type", ""),
|
||||
"type": doc_type,
|
||||
"publisher": item.institution,
|
||||
"date": item.date_published,
|
||||
"url": item.url,
|
||||
@@ -280,19 +407,21 @@ def push_to_zotero(
|
||||
stats["skipped"] += 1
|
||||
continue
|
||||
|
||||
type_id = _TYPE_MAP.get(item.item_type)
|
||||
# Resolve Zotero item type. journal-article sources become
|
||||
# journalArticle (22); other sources stay document (14).
|
||||
ej = json.loads(item.to_row().get("extra_json", "{}"))
|
||||
doc_type = ej.get("doc_type", "")
|
||||
type_id = _TYPE_MAP.get(doc_type) or _TYPE_MAP.get(item.item_type)
|
||||
if type_id is None:
|
||||
stats["skipped"] += 1
|
||||
continue
|
||||
|
||||
# Use the bib key if it looks like a Zotero key (8 alphanum),
|
||||
# otherwise generate a new one
|
||||
key = item.key
|
||||
if not key or len(key) != 8:
|
||||
key = _zotero_key()
|
||||
# Only reuse bib key if it passes Zotero's key validation;
|
||||
# otherwise generate a fresh one.
|
||||
key = item.key if _is_valid_zotero_key(item.key) else _zotero_key()
|
||||
|
||||
# Check key collision
|
||||
if con.execute("SELECT 1 FROM items WHERE key = ?", (key,)).fetchone():
|
||||
# Avoid key collision
|
||||
while con.execute("SELECT 1 FROM items WHERE key = ?", (key,)).fetchone():
|
||||
key = _zotero_key()
|
||||
|
||||
# Insert item
|
||||
@@ -305,11 +434,20 @@ def push_to_zotero(
|
||||
)
|
||||
item_id = cur.lastrowid
|
||||
|
||||
# Parse authors from the original item extra (before field
|
||||
# mapping strips them).
|
||||
authors, _ = _parse_authors_from_extra(item.extra)
|
||||
|
||||
# Set fields via EAV
|
||||
fields = _item_to_zotero_fields(item)
|
||||
for field_name, value in fields.items():
|
||||
_set_field(con, item_id, field_name, value)
|
||||
|
||||
# Creators (authors)
|
||||
if authors:
|
||||
_add_creators(con, item_id, authors)
|
||||
stats["creators"] = stats.get("creators", 0) + len(authors)
|
||||
|
||||
# Tags
|
||||
_sync_tags(con, item_id, item.tags)
|
||||
stats["tags"] += len(item.tags)
|
||||
|
||||
Reference in New Issue
Block a user