Coverage for src/local_deep_research/research_library/zotero/sync_service.py: 83%
686 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-19 23:35 +0000
1"""
2Zotero sync service.
4Imports items from a Zotero library/collection into the LDR document library
5and keeps them in sync. Items become :class:`Document` rows (with extracted
6text, optional encrypted PDF blob, and RAG auto-indexing) inside a dedicated
7:class:`Collection`, exactly like uploads/research downloads — so they are
8searchable during research with no extra plumbing.
10Sync algorithm (per collection), using cheap version maps:
121. Ask Zotero for ``{itemKey: version}`` of current top-level items.
132. Map keys not present anymore -> remove the local document.
143. Map keys that are new, version-bumped, or whose local document went
15 missing -> (re)fetch full data, download the best PDF (or import
16 metadata-only), and upsert the document + mapping row.
174. Persist the library version as the new high-water mark.
19Known limitations (v1):
21- With MULTIPLE synced collections, an item moved from collection A to B is
22 removed by A's sync before B has imported it: the document is deleted and
23 re-created under a new id once B next syncs successfully (briefly absent
24 from the library if B's sync fails first). Whole-library (single-target)
25 configs are unaffected.
26- Toggling ``import_items_without_pdf`` (or the tag filter) does not apply
27 to already-skipped items on SCHEDULED syncs — run a manual "Sync now"
28 (which re-examines skipped items), or wait for the items to change in
29 Zotero.
30"""
32import hashlib
33import os
34import re
35import threading
36import uuid
37from dataclasses import dataclass, field
38from datetime import date, datetime, UTC
39from pathlib import Path
40from typing import Dict, List, Optional, Tuple
42from loguru import logger
44from ...constants import FILE_PATH_TEXT_ONLY
45from ...config.paths import get_library_directory
46from ...database.library_init import get_source_type_id
47from ...database.models.library import (
48 Collection,
49 Document,
50 DocumentBlob,
51 DocumentChunk,
52 DocumentCollection,
53 DocumentStatus,
54)
55from ...database.models.zotero import ZoteroItemMap, ZoteroSyncState
56from ...database.session_context import get_user_db_session, safe_rollback
57from ...security import (
58 sanitize_error_for_client,
59 sanitize_filename,
60 UnsafeFilenameError,
61)
62from ...settings.manager import SettingsManager
63from ...utilities.type_utils import to_bool
64from ..deletion.utils.cascade_helper import CascadeHelper
65from ..services.pdf_storage_manager import (
66 DEFAULT_MAX_PDF_SIZE_MB,
67 PDFStorageManager,
68)
69from ..utils import ensure_in_collection
70from .client import (
71 ZoteroClient,
72 ZoteroError,
73 ZoteroItem,
74 ZoteroTransientError,
75)
78@dataclass
79class ZoteroConfig:
80 """Snapshot of the user's Zotero settings."""
82 # On by default — inert until an API key is configured; the toggle is an
83 # opt-out kill-switch, not a required opt-in step.
84 enabled: bool = True
85 api_key: str = ""
86 library_type: str = "user"
87 library_id: str = ""
88 collection_keys: List[str] = field(default_factory=list)
89 import_tags: List[str] = field(default_factory=list)
90 # Off by default: metadata/abstract-only documents carry no full text,
91 # which is what LDR's research actually needs.
92 import_items_without_pdf: bool = False
93 import_annotations: bool = False
94 # Text-only by default: the extracted text is what research uses, and it
95 # keeps the encrypted DB small. "database" additionally keeps the PDF.
96 pdf_storage_mode: str = "none"
97 auto_sync_enabled: bool = False
98 sync_interval_minutes: int = 360
99 use_local_api: bool = False
101 @property
102 def is_configured(self) -> bool:
103 # Local API mode needs no API key (it talks to the running desktop
104 # Zotero on localhost); the web API needs one. A library ID is only
105 # required for GROUP libraries: a personal library is identified by
106 # the key itself (resolved via /keys/current), and the local API
107 # addresses the current desktop user as user 0.
108 has_auth = self.use_local_api or bool(self.api_key)
109 needs_id = self.library_type == "group"
110 return bool(
111 self.enabled and has_auth and (self.library_id or not needs_id)
112 )
114 @property
115 def targets(self) -> List[str]:
116 """Collection keys to sync; ``[""]`` means the whole library."""
117 return self.collection_keys or [""]
120# Per-user sync locks. A single non-blocking lock per username serialises ALL
121# sync entry points (the manual "Sync now" daemon thread AND the scheduled
122# auto-sync job), so two concurrent runs can't write the same per-user
123# SQLCipher DB and race on the ZoteroItemMap unique constraint.
124_user_sync_locks: dict[str, threading.Lock] = {}
125_user_sync_locks_guard = threading.Lock()
127# Live progress of the currently running sync, per user — read by the status
128# endpoint while a sync runs so the UI can show a real progress bar.
129# In-memory on purpose: progress is transient, and both sync entry points
130# (manual daemon thread, scheduler job) run in this process.
131_user_sync_progress: dict[str, Dict] = {}
132_user_sync_progress_guard = threading.Lock()
135def _get_user_sync_lock(username: str) -> threading.Lock:
136 with _user_sync_locks_guard:
137 lock = _user_sync_locks.get(username)
138 if lock is None:
139 lock = threading.Lock()
140 _user_sync_locks[username] = lock
141 return lock
144class ZoteroSyncService:
145 """Service that imports / syncs Zotero items into the library."""
147 def __init__(
148 self,
149 username: str,
150 password: Optional[str] = None, # gitleaks:allow
151 ):
152 self.username = username
153 self.password = password # gitleaks:allow
154 # Cached /keys/current lookup ("" = resolution failed) so one service
155 # instance resolves the key's userID at most once.
156 self._resolved_user_id: Optional[str] = None
158 # -- settings ----------------------------------------------------------
160 def get_config(self) -> ZoteroConfig:
161 """Read the user's Zotero settings from their encrypted database."""
162 with get_user_db_session(self.username, self.password) as session:
163 sm = SettingsManager(session)
164 raw_keys = sm.get_setting("zotero.collection_keys", "") or ""
165 collection_keys = [
166 k.strip() for k in str(raw_keys).split(",") if k.strip()
167 ]
168 raw_tags = sm.get_setting("zotero.import_tags", "") or ""
169 import_tags = [
170 t.strip() for t in str(raw_tags).split(",") if t.strip()
171 ]
172 try:
173 interval = int(
174 sm.get_setting("zotero.sync_interval_minutes", 360) or 360
175 )
176 except (TypeError, ValueError):
177 interval = 360
178 return ZoteroConfig(
179 enabled=sm.get_bool_setting("zotero.enabled", True),
180 api_key=str(sm.get_setting("zotero.api_key", "") or "").strip(),
181 library_type=str(
182 sm.get_setting("zotero.library_type", "user") or "user"
183 ).strip(),
184 library_id=str(
185 sm.get_setting("zotero.library_id", "") or ""
186 ).strip(),
187 collection_keys=collection_keys,
188 import_tags=import_tags,
189 import_items_without_pdf=sm.get_bool_setting(
190 "zotero.import_items_without_pdf", False
191 ),
192 import_annotations=sm.get_bool_setting(
193 "zotero.import_annotations", False
194 ),
195 pdf_storage_mode=str(
196 sm.get_setting("zotero.pdf_storage_mode", "none") or "none"
197 ).strip(),
198 auto_sync_enabled=sm.get_bool_setting(
199 "zotero.auto_sync_enabled", False
200 ),
201 sync_interval_minutes=interval,
202 use_local_api=sm.get_bool_setting(
203 "zotero.use_local_api", False
204 ),
205 )
207 def _make_client(self, cfg: ZoteroConfig) -> ZoteroClient:
208 return ZoteroClient(
209 api_key=cfg.api_key,
210 library_type=cfg.library_type,
211 library_id=cfg.library_id,
212 local=cfg.use_local_api,
213 )
215 def _resolve_library_id(self, cfg: ZoteroConfig) -> Optional[str]:
216 """Normalize ``cfg.library_id`` in place; returns an error or None.
218 A personal library needs no manually entered ID: the API key
219 identifies its owner, so an empty or non-numeric value (a username,
220 say) is resolved to the numeric userID via ``/keys/current``. Local
221 mode addresses the desktop's current user as ``0``. Only group
222 libraries must carry an explicit numeric group ID.
224 Called by the operational entry points (test/list/sync) — NOT by
225 :meth:`get_config`, which must stay network-free (the config
226 endpoint reads it on every page load).
227 """
228 lid = cfg.library_id
229 if cfg.library_type == "group":
230 if not lid.isdigit():
231 return (
232 "Zotero group libraries need the numeric group ID — use "
233 "the group picker, or the number in the group's URL on "
234 "zotero.org."
235 )
236 return None
237 if cfg.use_local_api:
238 if not lid: 238 ↛ 241line 238 didn't jump to line 241 because the condition on line 238 was always true
239 # The local API addresses the desktop's current user as 0.
240 cfg.library_id = "0"
241 return None
242 if lid.isdigit():
243 return None
244 # Personal library with an empty or non-numeric ID (e.g. a username):
245 # the key knows its owner — resolve and use the real userID.
246 if self._resolved_user_id is None:
247 try:
248 probe = ZoteroClient(
249 api_key=cfg.api_key, library_type="user", library_id="0"
250 )
251 with probe:
252 user_id = probe.get_current_key_info().get("userID")
253 self._resolved_user_id = str(user_id) if user_id else ""
254 except ZoteroError:
255 logger.debug(
256 "Zotero: could not resolve the key's userID via "
257 "/keys/current",
258 exc_info=True,
259 )
260 self._resolved_user_id = ""
261 if self._resolved_user_id:
262 if lid:
263 logger.info(
264 f"Zotero: library ID {lid!r} is not numeric — using the "
265 f"API key's userID {self._resolved_user_id} instead"
266 )
267 cfg.library_id = self._resolved_user_id
268 return None
269 return (
270 "Could not determine the Zotero userID from the API key — check "
271 "the key, or enter the numeric userID shown at "
272 "zotero.org/settings/keys."
273 )
275 def _library_root(self) -> str:
276 with get_user_db_session(self.username, self.password) as session:
277 sm = SettingsManager(session)
278 storage_path = sm.get_setting(
279 "research_library.storage_path", str(get_library_directory())
280 )
281 return str(
282 Path(os.path.expandvars(storage_path)).expanduser().resolve()
283 )
285 # -- public operations -------------------------------------------------
287 def test_connection(self) -> Dict:
288 """Validate credentials. Returns ``{success, library_version, ...}``."""
289 cfg = self.get_config()
290 if not cfg.use_local_api and not cfg.api_key: 290 ↛ 291line 290 didn't jump to line 291 because the condition on line 290 was never true
291 return {
292 "success": False,
293 "error": "A Zotero API key is required (unless using the "
294 "local API).",
295 }
296 resolve_error = self._resolve_library_id(cfg)
297 if resolve_error: 297 ↛ 298line 297 didn't jump to line 298 because the condition on line 297 was never true
298 return {"success": False, "error": resolve_error}
299 try:
300 with self._make_client(cfg) as client:
301 version = client.test_connection()
302 collections = client.get_collections()
303 return {
304 "success": True,
305 "library_version": version,
306 "collection_count": len(collections),
307 }
308 except ZoteroError as exc:
309 return {
310 "success": False,
311 "error": sanitize_error_for_client(str(exc)),
312 }
314 def list_collections(self) -> List[Dict]:
315 """List Zotero collections (for the UI picker)."""
316 cfg = self.get_config()
317 resolve_error = self._resolve_library_id(cfg)
318 if resolve_error:
319 raise ZoteroError(resolve_error)
320 with self._make_client(cfg) as client:
321 return client.get_collections()
323 def list_groups(self) -> List[Dict]:
324 """List the groups the configured API key can access (UI picker)."""
325 cfg = self.get_config()
326 resolve_error = self._resolve_library_id(cfg)
327 if resolve_error:
328 raise ZoteroError(resolve_error)
329 with self._make_client(cfg) as client:
330 return client.get_groups()
332 @classmethod
333 def is_user_syncing(cls, username: str) -> bool:
334 """True if a sync is currently in progress for this user."""
335 lock = _user_sync_locks.get(username)
336 return bool(lock and lock.locked())
338 @classmethod
339 def get_sync_progress(cls, username: str) -> Optional[Dict]:
340 """Live progress of the user's running sync, or None when idle."""
341 if not cls.is_user_syncing(username):
342 return None
343 with _user_sync_progress_guard:
344 progress = _user_sync_progress.get(username)
345 return dict(progress) if progress else None
347 def _set_progress(self, **fields) -> None:
348 with _user_sync_progress_guard:
349 _user_sync_progress.setdefault(self.username, {}).update(fields)
351 def _clear_progress(self) -> None:
352 with _user_sync_progress_guard:
353 _user_sync_progress.pop(self.username, None)
355 def sync_all(self, reprocess_skipped: bool = False) -> Dict:
356 """Sync every configured collection (or the whole library).
358 Serialised per user: if a sync (manual or scheduled) is already
359 running for this user, returns immediately rather than starting a
360 second concurrent writer against the same encrypted DB.
362 ``reprocess_skipped`` re-examines items previously recorded without
363 a document (skipped strays, no-PDF items, tag-filter misses) even
364 though their Zotero version is unchanged. Manual "Sync now" sets it
365 so settings changes and newly importable item shapes apply without
366 waiting for the item to change in Zotero; scheduled syncs keep the
367 cheap version-diff.
368 """
369 cfg = self.get_config()
370 if not cfg.is_configured: 370 ↛ 371line 370 didn't jump to line 371 because the condition on line 370 was never true
371 return {
372 "success": False,
373 "error": "Zotero is not enabled/configured.",
374 }
375 # Resolve BEFORE the per-collection loop so the sync-state rows are
376 # always keyed on the real numeric library ID, whatever the setting
377 # holds (empty for personal libraries, or even a pasted username).
378 resolve_error = self._resolve_library_id(cfg)
379 if resolve_error: 379 ↛ 380line 379 didn't jump to line 380 because the condition on line 379 was never true
380 return {"success": False, "error": resolve_error}
382 lock = _get_user_sync_lock(self.username)
383 if not lock.acquire(blocking=False): 383 ↛ 399line 383 didn't jump to line 399 because the condition on line 383 was always true
384 logger.info(
385 f"Zotero: a sync is already running for {self.username}; "
386 "skipping this run"
387 )
388 return {
389 "success": True,
390 "already_running": True,
391 "imported": 0,
392 "updated": 0,
393 "removed": 0,
394 "skipped": 0,
395 "errors": 0,
396 "collections": [],
397 }
399 totals = {
400 "imported": 0,
401 "updated": 0,
402 "removed": 0,
403 "skipped": 0,
404 "errors": 0,
405 "collections": [],
406 }
407 try:
408 self._set_progress(phase="starting", done=0, total=None)
409 with self._make_client(cfg) as client:
410 for collection_key in cfg.targets:
411 stats = self._sync_one(
412 client,
413 cfg,
414 collection_key,
415 reprocess_skipped=reprocess_skipped,
416 )
417 totals["imported"] += stats.get("imported", 0)
418 totals["updated"] += stats.get("updated", 0)
419 totals["removed"] += stats.get("removed", 0)
420 totals["skipped"] += stats.get("skipped", 0)
421 totals["errors"] += stats.get("errors", 0)
422 totals["collections"].append(stats)
423 finally:
424 self._clear_progress()
425 lock.release()
426 totals["success"] = all(
427 c.get("status") != "failed" for c in totals["collections"]
428 )
429 return totals
431 def get_status(self) -> List[Dict]:
432 """Return the stored sync state for all configured collections."""
433 with get_user_db_session(self.username, self.password) as session:
434 rows = session.query(ZoteroSyncState).all()
435 return [
436 {
437 "library_type": r.library_type,
438 "library_id": r.library_id,
439 "collection_key": r.collection_key or None,
440 "ldr_collection_id": r.ldr_collection_id,
441 "last_version": r.last_version,
442 "last_status": r.last_status,
443 "last_error": r.last_error,
444 "item_count": r.item_count,
445 "last_synced_at": r.last_synced_at.isoformat()
446 if r.last_synced_at
447 else None,
448 }
449 for r in rows
450 ]
452 # -- per-collection sync ----------------------------------------------
454 def _sync_one(
455 self,
456 client: ZoteroClient,
457 cfg: ZoteroConfig,
458 collection_key: str,
459 reprocess_skipped: bool = False,
460 ) -> Dict:
461 """Sync a single collection (``""`` = whole library)."""
462 stats = {
463 "collection_key": collection_key or None,
464 "imported": 0,
465 "updated": 0,
466 "removed": 0,
467 # Deliberate skips (tag filter, no-PDF import disabled).
468 "skipped": 0,
469 # Contained per-item failures (rolled back, retried next sync).
470 # Kept separate from "skipped" so they can be SURFACED: a sync
471 # that completes with errors must not look identical to a clean
472 # one.
473 "errors": 0,
474 "status": "completed",
475 }
476 new_or_updated_doc_ids: List[str] = []
477 # Docs whose content changed in place: their stale FAISS vectors must be
478 # purged before the re-index rebuilds them (see _ingest_item).
479 reindexed_doc_ids: List[str] = []
480 # Prior documents orphaned when an in-place edit dedups onto a different
481 # existing doc: their stale vectors are purged like a removal.
482 released_doc_ids: List[str] = []
483 # Each id is appended only AFTER its own removal committed, so the
484 # finally-block vector purge never touches a rolled-back removal.
485 removed_doc_ids: List[str] = []
486 # Subset of removed/released docs whose Document row was HARD-deleted:
487 # their vectors must also leave the default Library collection's index
488 # (a kept/foreign doc's default-collection vectors stay valid).
489 hard_deleted_doc_ids: List[str] = []
490 # (document_id, collection_id) -> FAISS int chunk ids, captured by
491 # _release_document / _purge_collection_chunks just BEFORE they
492 # eagerly delete the matching DocumentChunk rows (each in its own
493 # per-item transaction, for that transaction's atomicity — a cascade
494 # failure must roll the rows back too). The finally-block vector
495 # purge below runs AFTER those rows are gone, so it needs these
496 # captured ids to find the FAISS vectors instead of the row lookup
497 # VectorIndex.delete would otherwise do. See _cleanup_removed_vectors.
498 pending_chunk_ids: Dict[Tuple[str, str], List[int]] = {}
499 ldr_collection_id: Optional[str] = None
500 default_collection_id: Optional[str] = None
501 try:
502 with get_user_db_session(self.username, self.password) as session:
503 ldr_collection_id = self._ensure_ldr_collection(
504 session, client, cfg, collection_key
505 )
506 state = self._get_or_create_state(
507 session, cfg, collection_key, ldr_collection_id
508 )
509 state.last_status = "syncing"
510 session.commit()
512 self._set_progress(
513 phase="checking",
514 collection_key=collection_key or None, # gitleaks:allow
515 done=0,
516 total=None,
517 current=None,
518 )
519 remote_versions, library_version = client.get_top_item_versions(
520 collection_key or None
521 )
522 rows = (
523 session.query(ZoteroItemMap)
524 .filter_by(ldr_collection_id=ldr_collection_id)
525 .all()
526 )
527 mapped = {r.zotero_item_key: r for r in rows}
528 remote_keys = set(remote_versions)
530 source_type_id = get_source_type_id(
531 self.username, "zotero", self.password
532 )
534 # The default "Library" collection shows ALL documents, so
535 # imports are linked into it too (like uploads and research
536 # downloads). That membership is AMBIENT: the ownership /
537 # removal logic below ignores it when deciding whether a
538 # document is shared.
539 default_row = (
540 session.query(Collection.id)
541 .filter_by(is_default=True)
542 .first()
543 )
544 default_collection_id = default_row[0] if default_row else None
546 # 1) Removals — but NEVER read an empty remote set as "delete
547 # everything". A truncated/transient empty response would
548 # otherwise wipe the whole collection (docs + blobs + chunks).
549 to_remove = set(mapped) - remote_keys
550 if to_remove and (remote_keys or not mapped):
551 self._set_progress(
552 phase="removing", done=0, total=len(to_remove)
553 )
554 if remote_keys or not mapped:
555 for removal_index, key in enumerate(to_remove, start=1):
556 # Per-item transaction: one document whose cascade
557 # fails must not abort the whole batch (or the sync).
558 # Its removal rolls back cleanly — the map row
559 # survives, so it is re-detected and retried next
560 # sync — while every other removal proceeds.
561 try:
562 rid = self._remove_item(
563 session,
564 mapped[key],
565 source_type_id,
566 default_collection_id,
567 pending_chunk_ids,
568 )
569 session.commit()
570 except Exception:
571 safe_rollback(session, "zotero removal")
572 logger.exception(
573 f"Zotero: failed to remove item {key}"
574 )
575 stats["errors"] += 1
576 continue
577 if rid: 577 ↛ 581line 577 didn't jump to line 581 because the condition on line 577 was always true
578 removed_doc_ids.append(rid)
579 if not self._document_exists(session, rid):
580 hard_deleted_doc_ids.append(rid)
581 stats["removed"] += 1
582 self._set_progress(done=removal_index)
583 else:
584 logger.warning(
585 f"Zotero: empty item set for "
586 f"{collection_key or 'ALL'} while {len(mapped)} items "
587 "are mapped — skipping removals to avoid mass-deletion"
588 )
590 # 2) Additions / changes. Reprocess a key when it is new, its
591 # version changed, or its previously-imported document went
592 # missing. On a scheduled sync we intentionally do NOT
593 # reprocess solely because document_id is None: that is the
594 # case for deliberately skipped (no-PDF, import-disabled,
595 # stray) items, which would otherwise be re-fetched on every
596 # sync forever. A manual sync (reprocess_skipped) re-examines
597 # them so settings changes and newly importable shapes take
598 # effect without waiting for the item to change in Zotero.
599 to_process = [
600 key
601 for key, version in remote_versions.items()
602 if key not in mapped
603 or mapped[key].zotero_version != version
604 or (reprocess_skipped and mapped[key].document_id is None)
605 or (
606 mapped[key].document_id is not None
607 and not self._document_exists(
608 session, mapped[key].document_id
609 )
610 )
611 ]
613 pdf_storage = PDFStorageManager(
614 library_root=self._library_root(),
615 storage_mode=cfg.pdf_storage_mode,
616 max_pdf_size_mb=DEFAULT_MAX_PDF_SIZE_MB,
617 )
619 self._set_progress(
620 phase="importing",
621 done=0,
622 total=len(to_process),
623 current=None,
624 )
625 returned_keys: set = set()
626 for item in client.get_items(to_process):
627 returned_keys.add(item.key)
628 self._set_progress(
629 done=len(returned_keys),
630 current=(item.data.get("title") or item.key)[:120],
631 )
632 # A top-level stored PDF with no bibliographic parent (a
633 # PDF dropped straight into Zotero) IS importable — it
634 # falls through to ingestion as its own document. Other
635 # attachment/note strays are not.
636 if not item.is_standalone_pdf_attachment and (
637 item.data.get("parentItem")
638 or item.item_type in ("attachment", "note")
639 ):
640 # Record the version so it isn't re-fetched every
641 # sync. A child that slipped into /top is plumbing
642 # (not counted); a genuine top-level stray (a note,
643 # or a non-PDF / linked-file attachment) counts as
644 # skipped so the sync result doesn't silently read
645 # as "nothing found".
646 self._touch_map_version(
647 session,
648 ldr_collection_id,
649 item.key,
650 item.version,
651 )
652 if not item.data.get("parentItem"):
653 stats["skipped"] += 1
654 continue
655 if cfg.import_tags and not self._item_has_tags(
656 item, cfg.import_tags
657 ):
658 # Doesn't match the tag filter — record the version so
659 # it isn't re-fetched every sync. Additive filter: we
660 # do NOT remove items imported before the filter was
661 # set (removals come only from the version-map diff).
662 self._touch_map_version(
663 session,
664 ldr_collection_id,
665 item.key,
666 item.version,
667 )
668 stats["skipped"] += 1
669 continue
670 # The document this item currently maps to (if any) before
671 # ingest may repoint it onto a DIFFERENT existing doc via
672 # content-hash dedup.
673 prior_row = mapped.get(item.key)
674 prior_doc_id = prior_row.document_id if prior_row else None
675 try:
676 doc_id, action, reindexed = self._ingest_item(
677 session,
678 client,
679 cfg,
680 item,
681 ldr_collection_id,
682 source_type_id,
683 pdf_storage,
684 default_collection_id,
685 pending_chunk_ids,
686 )
687 except ZoteroTransientError:
688 # A rate-limit / 5xx (from a PDF/full-text fetch inside
689 # _ingest_item) will hit every remaining item too. Abort
690 # the batch and let the outer handler mark the sync
691 # failed so it retries next run, rather than hammering
692 # items 2..N against the limit and burning the budget.
693 safe_rollback(session, "zotero ingest")
694 raise
695 except Exception:
696 safe_rollback(session, "zotero ingest")
697 logger.exception(
698 f"Zotero: failed to ingest item {item.key}"
699 )
700 # Don't record a version — let it retry next sync.
701 stats["errors"] += 1
702 continue
704 # Map upsert + (on a dedup remap) release of the orphaned
705 # prior doc, in ONE transaction: committing the repoint on
706 # its own first would strand the prior doc forever if the
707 # release then failed — the item's new version would
708 # already be recorded, so no later sync would retry it.
709 # Per-item isolation: a failure rolls this item back
710 # cleanly (ingest included; retried next sync since its
711 # version was not recorded) without aborting the batch.
712 try:
713 self._upsert_map(
714 session, ldr_collection_id, item, doc_id
715 )
716 released = None
717 if (
718 prior_row
719 and prior_doc_id
720 and prior_doc_id != doc_id
721 ):
722 released = self._release_document(
723 session,
724 prior_doc_id,
725 ldr_collection_id,
726 prior_row.id,
727 source_type_id,
728 default_collection_id,
729 pending_chunk_ids,
730 )
731 session.commit()
732 except Exception:
733 safe_rollback(session, "zotero map update")
734 logger.exception(
735 f"Zotero: failed to record mapping for item "
736 f"{item.key}"
737 )
738 stats["errors"] += 1
739 continue
740 if released:
741 released_doc_ids.append(released)
742 if not self._document_exists(session, released): 742 ↛ 745line 742 didn't jump to line 745 because the condition on line 742 was always true
743 hard_deleted_doc_ids.append(released)
745 if action == "imported":
746 stats["imported"] += 1
747 new_or_updated_doc_ids.append(doc_id)
748 elif action == "updated":
749 stats["updated"] += 1
750 new_or_updated_doc_ids.append(doc_id)
751 if reindexed: 751 ↛ 752line 751 didn't jump to line 752 because the condition on line 751 was never true
752 reindexed_doc_ids.append(doc_id)
753 else:
754 stats["skipped"] += 1
756 # Keys we asked for but Zotero didn't return (trashed,
757 # permission-changed, etc.): record their version so they
758 # aren't re-requested on every subsequent sync.
759 for key in set(to_process) - returned_keys:
760 self._touch_map_version(
761 session,
762 ldr_collection_id,
763 key,
764 remote_versions.get(key, 0),
765 )
766 session.commit()
768 # 3) Persist high-water mark. Contained per-item failures
769 # must stay visible: a run that skipped items because their
770 # removal/ingest kept failing would otherwise look identical
771 # to a clean one (green badge, last_error wiped) while the
772 # same items silently fail on every sync.
773 state.last_version = library_version
774 state.last_status = "completed"
775 state.last_error = (
776 f"{stats['errors']} item(s) failed and will be retried "
777 "next sync (see server log)"
778 if stats["errors"]
779 else None
780 )
781 state.last_synced_at = datetime.now(UTC)
782 state.item_count = (
783 session.query(ZoteroItemMap)
784 .filter_by(ldr_collection_id=ldr_collection_id)
785 .filter(ZoteroItemMap.document_id.isnot(None))
786 .count()
787 )
788 session.commit()
789 except ZoteroError as exc:
790 stats["status"] = "failed"
791 stats["error"] = sanitize_error_for_client(str(exc))
792 self._mark_failed(cfg, collection_key, stats["error"])
793 return stats
794 except Exception as exc:
795 stats["status"] = "failed"
796 stats["error"] = sanitize_error_for_client(str(exc))
797 logger.exception(
798 f"Zotero: sync failed for collection {collection_key or 'ALL'}"
799 )
800 self._mark_failed(cfg, collection_key, stats["error"])
801 return stats
802 finally:
803 # Purge stale FAISS vectors for DURABLY-COMMITTED removals and
804 # in-place content updates — even when the sync later fails partway
805 # through the additions phase. Those docs' DB chunks are already
806 # gone, so skipping this on the failure path would strand ORPHANED
807 # vectors that keep surfacing in semantic search and are never
808 # re-detected by a future sync (the map row was deleted). Runs
809 # before the auto-index below, so an edited doc's old vectors are
810 # cleared first. _cleanup_removed_vectors is best-effort and no-ops
811 # when the collection was never RAG-indexed.
812 #
813 # Every id in these lists was appended only AFTER its own per-item
814 # commit, so each is durable — a rolled-back removal/update never
815 # reaches here and cannot lose a still-valid doc's vectors.
816 if ldr_collection_id and self.password:
817 if reindexed_doc_ids: 817 ↛ 818line 817 didn't jump to line 818 because the condition on line 817 was never true
818 self._cleanup_removed_vectors(
819 reindexed_doc_ids,
820 ldr_collection_id,
821 pending_chunk_ids=pending_chunk_ids,
822 )
823 if released_doc_ids:
824 self._cleanup_removed_vectors(
825 released_doc_ids,
826 ldr_collection_id,
827 pending_chunk_ids=pending_chunk_ids,
828 )
829 if removed_doc_ids:
830 self._cleanup_removed_vectors(
831 removed_doc_ids,
832 ldr_collection_id,
833 pending_chunk_ids=pending_chunk_ids,
834 )
835 # The default Library collection carries every synced doc too
836 # (ambient membership), so its index needs the same care:
837 # in-place updates invalidate its copy of the vectors, and a
838 # hard-deleted document must not linger as dead hits there.
839 # Kept/foreign releases are deliberately NOT purged from it —
840 # their default-collection vectors are still valid.
841 if (
842 default_collection_id
843 and default_collection_id != ldr_collection_id
844 ):
845 if reindexed_doc_ids: 845 ↛ 846line 845 didn't jump to line 846 because the condition on line 845 was never true
846 self._cleanup_removed_vectors(
847 reindexed_doc_ids,
848 default_collection_id,
849 pending_chunk_ids=pending_chunk_ids,
850 )
851 if hard_deleted_doc_ids:
852 self._cleanup_removed_vectors(
853 hard_deleted_doc_ids,
854 default_collection_id,
855 pending_chunk_ids=pending_chunk_ids,
856 )
858 # Auto-index outside the sync transaction (best-effort, async).
859 # Deliberately in the finally, ON FAILURE PATHS TOO: every id in
860 # new_or_updated_doc_ids was committed per-item, and a sync that
861 # fails on item N must not leave items 1..N-1 permanently
862 # unindexed — their versions already match, so no future sync
863 # reprocesses them, and the background reconciler only runs when
864 # the document-scheduler settings are enabled.
865 if new_or_updated_doc_ids and ldr_collection_id and self.password:
866 try:
867 from ..routes.rag_routes import trigger_auto_index
869 trigger_auto_index(
870 new_or_updated_doc_ids,
871 ldr_collection_id,
872 self.username,
873 self.password,
874 )
875 # Keep the default Library collection's index fresh too.
876 if (
877 default_collection_id
878 and default_collection_id != ldr_collection_id
879 ):
880 trigger_auto_index(
881 new_or_updated_doc_ids,
882 default_collection_id,
883 self.username,
884 self.password,
885 )
886 except Exception:
887 logger.exception("Zotero: failed to trigger auto-indexing")
889 return stats
891 def _cleanup_removed_vectors(
892 self,
893 document_ids: List[str],
894 ldr_collection_id: str,
895 pending_chunk_ids: Optional[Dict[Tuple[str, str], List[int]]] = None,
896 ) -> None:
897 """Remove the given documents' vectors from the collection's FAISS
898 index. No-op if the collection was never RAG-indexed. Best-effort —
899 never raises (it runs inside a ``finally`` where a propagating exception
900 would mask the real sync error), so the imports live inside the try.
902 The DocumentChunk rows for these documents are ALREADY GONE by the
903 time this runs (the per-item transactions that released/removed/
904 reindexed them deleted them eagerly, for their own atomicity — see
905 ``_release_document`` / ``_purge_collection_chunks``), so the usual
906 source_id lookup in ``VectorIndex.delete`` would find nothing.
907 ``pending_chunk_ids`` (keyed by ``(document_id, ldr_collection_id)``)
908 carries the ids those per-item transactions captured BEFORE deleting
909 the rows, so ``remove_documents_from_index`` can purge the exact
910 FAISS vectors via ``VectorIndex.delete_ids`` instead.
911 """
912 try:
913 from ...database.models.library import RAGIndex
914 from ..services.library_rag_service import LibraryRAGService
916 collection_name = f"collection_{ldr_collection_id}"
917 with get_user_db_session(self.username, self.password) as session:
918 rag_index = (
919 session.query(RAGIndex)
920 .filter_by(collection_name=collection_name, is_current=True)
921 .first()
922 )
923 if not rag_index:
924 return # collection isn't indexed — nothing to clean
925 embedding_model = rag_index.embedding_model
926 embedding_provider = (
927 rag_index.embedding_model_type.value
928 if rag_index.embedding_model_type
929 else None
930 )
931 chunk_size = rag_index.chunk_size
932 chunk_overlap = rag_index.chunk_overlap
933 # This is a DELETE path, so the stored index_type MUST be
934 # threaded through: LibraryRAGService defaults index_type="flat",
935 # and removing from a real HNSW collection under the wrong type
936 # would take the in-place remove_ids path (unsupported for HNSW)
937 # and silently fail instead of the rebuild path. distance_metric
938 # / normalize_vectors are threaded too for parity with the
939 # search engines. NULL (legacy rows) → the service defaults.
940 index_type = rag_index.index_type or "flat"
941 distance_metric = rag_index.distance_metric or "cosine"
942 normalize_vectors = to_bool(
943 rag_index.normalize_vectors, default=True
944 )
946 chunk_ids_by_document = None
947 if pending_chunk_ids:
948 chunk_ids_by_document = {
949 doc_id: ids
950 for doc_id in document_ids
951 if (
952 ids := pending_chunk_ids.get(
953 (doc_id, ldr_collection_id)
954 )
955 )
956 }
958 with LibraryRAGService(
959 username=self.username,
960 embedding_model=embedding_model,
961 embedding_provider=embedding_provider,
962 chunk_size=chunk_size,
963 chunk_overlap=chunk_overlap,
964 index_type=index_type,
965 distance_metric=distance_metric,
966 normalize_vectors=normalize_vectors,
967 db_password=self.password,
968 ) as rag_service:
969 removed = rag_service.remove_documents_from_index(
970 document_ids,
971 ldr_collection_id,
972 chunk_ids_by_document=chunk_ids_by_document,
973 )
974 logger.info(
975 f"Zotero: cleared {removed} FAISS vectors for "
976 f"{len(document_ids)} document(s) in {collection_name}"
977 )
978 except Exception:
979 logger.warning(
980 "Zotero: failed to clear FAISS vectors for removed documents"
981 )
983 def _mark_failed(
984 self, cfg: ZoteroConfig, collection_key: str, error: str
985 ) -> None:
986 """Best-effort: record a failed status on the sync-state row."""
987 ck = collection_key or ""
988 try:
989 with get_user_db_session(self.username, self.password) as session:
990 state = (
991 session.query(ZoteroSyncState)
992 .filter_by(
993 library_type=cfg.library_type,
994 library_id=cfg.library_id,
995 collection_key=ck,
996 )
997 .first()
998 )
999 if state: 999 ↛ exitline 999 didn't jump to the function exit
1000 state.last_status = "failed"
1001 state.last_error = error[:2000]
1002 session.commit()
1003 except Exception:
1004 logger.debug("Zotero: could not record failed state", exc_info=True)
1006 # -- LDR collection / state helpers -----------------------------------
1008 def _ensure_ldr_collection(
1009 self,
1010 session,
1011 client: ZoteroClient,
1012 cfg: ZoteroConfig,
1013 collection_key: str,
1014 ) -> str:
1015 """Find or create the LDR collection backing this Zotero collection."""
1016 ck = collection_key or ""
1017 state = (
1018 session.query(ZoteroSyncState)
1019 .filter_by(
1020 library_type=cfg.library_type,
1021 library_id=cfg.library_id,
1022 collection_key=ck,
1023 )
1024 .first()
1025 )
1026 if state:
1027 existing = (
1028 session.query(Collection)
1029 .filter_by(id=state.ldr_collection_id)
1030 .first()
1031 )
1032 if existing: 1032 ↛ 1036line 1032 didn't jump to line 1036 because the condition on line 1032 was always true
1033 return existing.id
1035 # Determine a friendly name.
1036 if collection_key: 1036 ↛ 1037line 1036 didn't jump to line 1037 because the condition on line 1036 was never true
1037 name = client.get_collection_name(collection_key) or collection_key
1038 display = f"Zotero: {name}"
1039 description = (
1040 f"Imported from Zotero collection '{name}' ({collection_key})."
1041 )
1042 else:
1043 display = "Zotero Library"
1044 description = "Imported from the entire Zotero library."
1046 collection_id = str(uuid.uuid4())
1047 session.add(
1048 Collection(
1049 id=collection_id,
1050 name=display,
1051 description=description,
1052 collection_type="zotero",
1053 is_default=False,
1054 )
1055 )
1056 session.flush()
1057 return collection_id
1059 def _get_or_create_state(
1060 self,
1061 session,
1062 cfg: ZoteroConfig,
1063 collection_key: str,
1064 ldr_collection_id: str,
1065 ) -> ZoteroSyncState:
1066 ck = collection_key or ""
1067 state = (
1068 session.query(ZoteroSyncState)
1069 .filter_by(
1070 library_type=cfg.library_type,
1071 library_id=cfg.library_id,
1072 collection_key=ck,
1073 )
1074 .first()
1075 )
1076 if state:
1077 # Keep the collection pointer fresh in case it was recreated.
1078 state.ldr_collection_id = ldr_collection_id
1079 return state
1080 state = ZoteroSyncState(
1081 library_type=cfg.library_type,
1082 library_id=cfg.library_id,
1083 collection_key=ck,
1084 ldr_collection_id=ldr_collection_id,
1085 last_version=0,
1086 last_status="pending",
1087 )
1088 session.add(state)
1089 session.flush()
1090 return state
1092 # -- item ingestion ----------------------------------------------------
1094 def _ingest_item(
1095 self,
1096 session,
1097 client: ZoteroClient,
1098 cfg: ZoteroConfig,
1099 item: ZoteroItem,
1100 ldr_collection_id: str,
1101 source_type_id: str,
1102 pdf_storage: PDFStorageManager,
1103 default_collection_id: Optional[str] = None,
1104 pending_chunk_ids: Optional[Dict[Tuple[str, str], List[int]]] = None,
1105 ) -> Tuple[Optional[str], str, bool]:
1106 """Create or update the LDR document for a Zotero item.
1108 ``pending_chunk_ids`` is forwarded to ``_purge_collection_chunks`` —
1109 see the note on ``_release_document``.
1111 Returns ``(document_id, action, reindexed_in_place)`` where action is
1112 ``"imported"`` | ``"updated"`` | ``"skipped"``. ``reindexed_in_place``
1113 is True only when an existing document's content changed and its stale
1114 RAG chunks were purged here — the caller must then clear the matching
1115 FAISS vectors before the re-index rebuilds them.
1116 """
1117 if item.is_standalone_pdf_attachment:
1118 # A PDF dropped into Zotero with no bibliographic parent: the
1119 # item IS its own attachment (it has no children to fetch).
1120 item.attachments = [item.as_attachment()]
1121 else:
1122 item.attachments = client.get_children(item.key)
1123 pdf_att = item.best_pdf_attachment()
1125 pdf_bytes = None
1126 zotero_fulltext = None
1127 if pdf_att:
1128 # In text-only mode we don't keep the PDF, so prefer Zotero's
1129 # already-extracted full text and skip the download + local parse
1130 # entirely. Falls back to downloading if Zotero has no indexed
1131 # text for the attachment.
1132 if cfg.pdf_storage_mode == "none":
1133 zotero_fulltext = client.get_attachment_fulltext(pdf_att.key)
1134 if not zotero_fulltext:
1135 pdf_bytes = client.download_attachment(pdf_att.key)
1136 # Validate the downloaded body really is a PDF. The /file
1137 # endpoint can redirect to an HTML error page; without this
1138 # we'd store that as file_type="pdf" and extract garbage.
1139 if pdf_bytes is not None and b"%PDF" not in pdf_bytes[:1024]:
1140 logger.warning(
1141 f"Zotero: attachment {pdf_att.key} did not return PDF "
1142 "content; treating item as having no PDF"
1143 )
1144 pdf_bytes = None
1146 if pdf_bytes:
1147 content = pdf_bytes
1148 file_type = "pdf"
1149 mime_type = "application/pdf"
1150 text, extraction_method = self._safe_extract_text(pdf_bytes, item)
1151 elif zotero_fulltext:
1152 from ...text_processing import remove_surrogates
1154 text = remove_surrogates(zotero_fulltext)
1155 content = text.encode("utf-8", errors="ignore")
1156 file_type = "pdf" # it is a PDF paper; we just reused Zotero's text
1157 mime_type = "application/pdf"
1158 extraction_method = "zotero_fulltext"
1159 else:
1160 if not cfg.import_items_without_pdf:
1161 return None, "skipped", False
1162 text = self._metadata_text(item)
1163 content = text.encode("utf-8", errors="ignore")
1164 file_type = "text"
1165 mime_type = "text/plain"
1166 extraction_method = "metadata"
1168 # Optionally append the user's annotations (highlights/comments) and
1169 # child notes to the searchable text. Kept out of the content hash so
1170 # it doesn't perturb dedup of the primary document.
1171 if cfg.import_annotations:
1172 extra = self._collect_annotations_text(client, item)
1173 if extra:
1174 text = f"{text}\n\n{extra}" if text else extra
1176 doc_hash = hashlib.sha256(content).hexdigest()
1178 # The document this item currently maps to (if any) — needed both to
1179 # detect an in-place content change and, on unchanged content, to
1180 # decide whether the searchable text may be refreshed safely.
1181 prior = (
1182 session.query(ZoteroItemMap)
1183 .filter_by(
1184 ldr_collection_id=ldr_collection_id, zotero_item_key=item.key
1185 )
1186 .first()
1187 )
1188 # -1 matches no row (Integer PK starts at 1) so the "any OTHER
1189 # mapping?" exclusion below stays correct when there is no prior row.
1190 prior_mapping_id = prior.id if prior else -1
1192 # Dedup: an identical document (same bytes) may already exist.
1193 # document_hash is globally UNIQUE, so this can match a document from
1194 # a different source (a manual upload or research download). Only
1195 # rewrite metadata on documents WE own (Zotero source); for a foreign
1196 # document just add the collection link — never clobber the user's
1197 # title/tags or take ownership of it.
1198 existing = (
1199 session.query(Document).filter_by(document_hash=doc_hash).first()
1200 )
1201 if existing:
1202 if existing.source_type_id == source_type_id:
1203 # Deliberately gated on ownership only (NOT exclusivity):
1204 # metadata edits must keep propagating to a doc shared across
1205 # two synced collections. For byte-identical twin ITEMS this
1206 # means last-synced-twin-wins on title/tags — an accepted
1207 # trade-off for content that is by definition the same paper.
1208 self._apply_metadata(existing, item)
1209 ensure_in_collection(session, existing.id, ldr_collection_id)
1210 if default_collection_id:
1211 # The default "Library" collection lists all documents.
1212 ensure_in_collection(
1213 session, existing.id, default_collection_id
1214 )
1215 # The content bytes are unchanged, but the searchable text may
1216 # not be: annotations/notes are appended AFTER hashing (so they
1217 # don't perturb dedup), and a previously failed PDF extraction
1218 # may have fallen back to metadata text. Refresh it — but only on
1219 # a document this item exclusively owns; rewriting a shared or
1220 # foreign doc's text would change what its other references
1221 # surface.
1222 if (text or "") != (
1223 existing.text_content or ""
1224 ) and self._exclusive_to_mapping(
1225 session,
1226 existing,
1227 prior_mapping_id,
1228 ldr_collection_id,
1229 source_type_id,
1230 default_collection_id,
1231 ):
1232 existing.text_content = text
1233 existing.character_count = len(text) if text else 0
1234 existing.word_count = len(text.split()) if text else 0
1235 existing.extraction_method = extraction_method
1236 self._purge_collection_chunks(
1237 session,
1238 existing.id,
1239 ldr_collection_id,
1240 default_collection_id,
1241 pending_chunk_ids,
1242 )
1243 return existing.id, "updated", True
1244 # Content is byte-identical (matched by hash), so the existing
1245 # chunks/vectors are still valid — no purge needed.
1246 return existing.id, "updated", False
1248 # Is this a re-import of an item we already mapped (content changed)?
1249 doc = None
1250 if prior and prior.document_id:
1251 doc = (
1252 session.query(Document).filter_by(id=prior.document_id).first()
1253 )
1254 # An in-place rewrite is only safe on a document this item
1255 # exclusively owns. A dedup-adopted user upload, a byte-identical
1256 # twin's shared doc, or a doc linked into another collection must
1257 # never be mutated under its other references — fork a fresh
1258 # Document instead; the caller's remap logic then releases this
1259 # item's claim on the old one (_release_document keeps shared and
1260 # foreign documents intact).
1261 if doc is not None and not self._exclusive_to_mapping(
1262 session,
1263 doc,
1264 prior_mapping_id,
1265 ldr_collection_id,
1266 source_type_id,
1267 default_collection_id,
1268 ):
1269 doc = None
1271 created = doc is None
1272 if doc is None:
1273 doc = Document(
1274 id=str(uuid.uuid4()),
1275 source_type_id=source_type_id,
1276 )
1277 session.add(doc)
1279 store_pdf_in_db = (
1280 cfg.pdf_storage_mode == "database" and file_type == "pdf"
1281 )
1283 doc.document_hash = doc_hash
1284 doc.filename = self._filename_for(item, file_type)
1285 doc.file_size = len(content)
1286 doc.file_type = file_type
1287 doc.mime_type = mime_type
1288 doc.text_content = text
1289 doc.character_count = len(text) if text else 0
1290 doc.word_count = len(text.split()) if text else 0
1291 doc.extraction_method = extraction_method
1292 doc.extraction_source = "zotero"
1293 doc.status = DocumentStatus.COMPLETED
1294 doc.processed_at = datetime.now(UTC)
1295 doc.storage_mode = "database" if store_pdf_in_db else "none"
1296 doc.file_path = None if store_pdf_in_db else FILE_PATH_TEXT_ONLY
1297 self._apply_metadata(doc, item)
1298 session.flush()
1300 if store_pdf_in_db:
1301 try:
1302 pdf_storage.save_pdf(
1303 pdf_content=content,
1304 document=doc,
1305 session=session,
1306 filename=doc.filename,
1307 )
1308 except Exception:
1309 logger.exception(
1310 f"Zotero: failed to store PDF blob for {item.key}"
1311 )
1312 # The blob write failed, so there is no PDF to serve. Leave the
1313 # doc searchable via its extracted text but downgrade it to
1314 # text-only, otherwise ZoteroItemMap.has_pdf (keyed off
1315 # file_type=="pdf") would advertise a PDF that load_pdf() — which
1316 # looks up the missing blob by document_id — can never return.
1317 doc.file_type = "text"
1318 doc.mime_type = "text/plain"
1319 doc.storage_mode = "none"
1320 doc.file_path = FILE_PATH_TEXT_ONLY
1321 # On a re-sync a stale blob from a previous import may linger;
1322 # drop it so load_pdf() can't serve an outdated PDF for what is
1323 # now a text-only doc.
1324 session.query(DocumentBlob).filter_by(
1325 document_id=doc.id
1326 ).delete(synchronize_session=False)
1327 elif not created:
1328 # Re-sync of an item previously imported WITH a database-stored
1329 # PDF that now has no usable PDF (removed in Zotero, or the
1330 # re-downloaded body failed the %PDF check). The doc is being
1331 # marked text-only (storage_mode="none"), so drop the stale
1332 # encrypted blob — otherwise PDFStorageManager.load_pdf(), which
1333 # keys off document_id and ignores file_type, would keep serving
1334 # the orphaned PDF.
1335 removed = (
1336 session.query(DocumentBlob)
1337 .filter_by(document_id=doc.id)
1338 .delete(synchronize_session=False)
1339 )
1340 if removed:
1341 logger.info(
1342 f"Zotero: removed stale PDF blob for {item.key} "
1343 "(item no longer has a usable PDF)"
1344 )
1346 ensure_in_collection(session, doc.id, ldr_collection_id)
1347 if default_collection_id:
1348 # The default "Library" collection lists all documents.
1349 ensure_in_collection(session, doc.id, default_collection_id)
1350 # On an in-place content change the document's previous RAG chunks are
1351 # now stale: chunking is content-derived and _store_chunks_to_db dedups
1352 # by chunk hash, so a re-index would ADD the new chunks while the old
1353 # ones linger and keep surfacing outdated text in search. Purge this
1354 # collection's chunks for the doc now (atomic with the content update)
1355 # and force a re-index by clearing the indexed flag; the caller clears
1356 # the matching FAISS vectors before that re-index runs.
1357 if not created:
1358 self._purge_collection_chunks(
1359 session,
1360 doc.id,
1361 ldr_collection_id,
1362 default_collection_id,
1363 pending_chunk_ids,
1364 )
1366 return doc.id, ("imported" if created else "updated"), (not created)
1368 @staticmethod
1369 def _exclusive_to_mapping(
1370 session,
1371 doc: Document,
1372 mapping_id,
1373 coll_id: str,
1374 source_type_id: str,
1375 default_collection_id: Optional[str] = None,
1376 ) -> bool:
1377 """True when ``doc`` belongs to this item mapping alone: Zotero owns
1378 it, no OTHER :class:`ZoteroItemMap` row points at it, and it isn't
1379 linked into any other collection. Only such a document may be
1380 rewritten in place — anything else (a dedup-adopted user upload, a
1381 byte-identical twin's shared doc, a doc the user linked into another
1382 collection) would be corrupted under its other references.
1384 Membership in the default "Library" collection is AMBIENT (every
1385 import is linked there so the main library view lists it) and does
1386 not count as another reference.
1387 """
1388 if doc.source_type_id != source_type_id:
1389 return False
1390 shared_mapping = (
1391 session.query(ZoteroItemMap.id)
1392 .filter(
1393 ZoteroItemMap.document_id == doc.id,
1394 ZoteroItemMap.id != mapping_id,
1395 )
1396 .first()
1397 is not None
1398 )
1399 if shared_mapping:
1400 return False
1401 ambient = {coll_id}
1402 if default_collection_id:
1403 ambient.add(default_collection_id)
1404 linked_elsewhere = (
1405 session.query(DocumentCollection.document_id)
1406 .filter(
1407 DocumentCollection.document_id == doc.id,
1408 DocumentCollection.collection_id.notin_(ambient),
1409 )
1410 .first()
1411 is not None
1412 )
1413 return not linked_elsewhere
1415 @staticmethod
1416 def _purge_collection_chunks(
1417 session,
1418 doc_id: str,
1419 coll_id: str,
1420 default_collection_id: Optional[str] = None,
1421 pending_chunk_ids: Optional[Dict[Tuple[str, str], List[int]]] = None,
1422 ) -> None:
1423 """Purge a doc's RAG chunks and clear the link's indexed flag, forcing
1424 a re-index after an in-place text change — for this collection AND
1425 (when given) the default Library collection, whose ambient copy of
1426 the chunks is derived from the same now-changed text.
1428 Raises on failure so the whole per-item transaction rolls back (the
1429 item retries next sync) — swallowing here would let the caller purge
1430 FAISS vectors for chunks that still exist.
1432 ``pending_chunk_ids`` — see the note on ``_release_document``: when
1433 supplied, each collection's chunk ids are captured into it BEFORE
1434 being eagerly deleted below, so the later batched
1435 ``_cleanup_removed_vectors`` pass can purge those exact FAISS ids
1436 via ``VectorIndex.delete_ids`` instead of a now-futile source_id
1437 lookup. ``None`` (what direct unit tests pass) skips capture.
1438 """
1439 targets = [coll_id]
1440 if default_collection_id and default_collection_id != coll_id:
1441 targets.append(default_collection_id)
1442 for cid in targets:
1443 if pending_chunk_ids is not None: 1443 ↛ 1444line 1443 didn't jump to line 1444 because the condition on line 1443 was never true
1444 ZoteroSyncService._capture_chunk_ids(
1445 pending_chunk_ids, session, doc_id, cid
1446 )
1447 CascadeHelper.delete_document_chunks(
1448 session, doc_id, collection_name=f"collection_{cid}"
1449 )
1450 link = (
1451 session.query(DocumentCollection)
1452 .filter_by(document_id=doc_id, collection_id=cid)
1453 .first()
1454 )
1455 if link: 1455 ↛ 1442line 1455 didn't jump to line 1442 because the condition on line 1455 was always true
1456 link.indexed = False
1458 def _collect_annotations_text(
1459 self, client: ZoteroClient, item: ZoteroItem
1460 ) -> str:
1461 """Gather the item's PDF annotations + child notes as one text block.
1463 Best-effort: a failure fetching either must not break ingestion.
1464 """
1465 from ...text_processing import remove_surrogates
1467 annotations: List[str] = []
1468 for att in item.attachments or []:
1469 try:
1470 annotations.extend(client.get_annotations(att.key))
1471 except Exception:
1472 logger.debug(
1473 f"Zotero: could not fetch annotations for {att.key}",
1474 exc_info=True,
1475 )
1476 try:
1477 notes = client.get_notes(item.key)
1478 except Exception:
1479 logger.debug(
1480 f"Zotero: could not fetch notes for {item.key}", exc_info=True
1481 )
1482 notes = []
1484 sections: List[str] = []
1485 if annotations:
1486 sections.append(
1487 "Annotations:\n" + "\n".join(f"- {a}" for a in annotations)
1488 )
1489 if notes:
1490 sections.append("Notes:\n" + "\n\n".join(notes))
1491 return remove_surrogates("\n\n".join(sections)) if sections else ""
1493 def _safe_extract_text(
1494 self, pdf_bytes: bytes, item: ZoteroItem
1495 ) -> Tuple[Optional[str], str]:
1496 """Extract text from PDF bytes; returns ``(text, extraction_method)``.
1498 Falls back to metadata text (method ``"metadata"``) when extraction
1499 fails or yields nothing, so the document is still searchable — the
1500 method string keeps downstream consumers from assuming real PDF text
1501 when only the fallback was stored.
1502 """
1503 from ...document_loaders import extract_text_from_bytes
1504 from ...text_processing import remove_surrogates
1506 try:
1507 text = extract_text_from_bytes(pdf_bytes, ".pdf", item.key)
1508 except Exception:
1509 logger.debug(
1510 f"Zotero: PDF text extraction failed for {item.key}",
1511 exc_info=True,
1512 )
1513 text = None
1514 if not text: 1514 ↛ 1517line 1514 didn't jump to line 1517 because the condition on line 1514 was always true
1515 # Fall back to metadata so the document is still searchable.
1516 return self._metadata_text(item), "metadata"
1517 return remove_surrogates(text), "pdf_extraction"
1519 # -- metadata helpers --------------------------------------------------
1521 def _apply_metadata(self, doc: Document, item: ZoteroItem) -> None:
1522 data = item.data
1523 doc.title = (data.get("title") or doc.filename or item.key)[:5000]
1524 authors = self._authors(item)
1525 if authors: 1525 ↛ 1526line 1525 didn't jump to line 1526 because the condition on line 1525 was never true
1526 doc.authors = authors
1527 doi = (data.get("DOI") or "").strip()
1528 if doi:
1529 doc.doi = doi[:255]
1530 isbn = (data.get("ISBN") or "").strip()
1531 if isbn: 1531 ↛ 1532line 1531 didn't jump to line 1532 because the condition on line 1531 was never true
1532 doc.isbn = isbn[:20]
1533 published = self._parse_date(data.get("date"))
1534 if published:
1535 doc.published_date = published
1536 tags = [
1537 t.get("tag")
1538 for t in data.get("tags", [])
1539 if isinstance(t, dict) and t.get("tag")
1540 ]
1541 if tags:
1542 doc.tags = tags
1543 url = self._source_url(item)
1544 if url:
1545 doc.original_url = url
1547 @staticmethod
1548 def _item_has_tags(item: ZoteroItem, required: List[str]) -> bool:
1549 """True if the item carries ALL required tags (case-insensitive)."""
1550 item_tags = {
1551 (t.get("tag") or "").strip().lower()
1552 for t in item.data.get("tags", []) or []
1553 if isinstance(t, dict)
1554 }
1555 return all(r.strip().lower() in item_tags for r in required)
1557 @staticmethod
1558 def _authors(item: ZoteroItem) -> List[str]:
1559 names: List[str] = []
1560 for creator in item.data.get("creators", []) or []:
1561 if not isinstance(creator, dict): 1561 ↛ 1562line 1561 didn't jump to line 1562 because the condition on line 1561 was never true
1562 continue
1563 if creator.get("name"):
1564 names.append(creator["name"].strip())
1565 else:
1566 full = " ".join(
1567 p
1568 for p in (
1569 creator.get("firstName", ""),
1570 creator.get("lastName", ""),
1571 )
1572 if p
1573 ).strip()
1574 if full: 1574 ↛ 1560line 1574 didn't jump to line 1560 because the condition on line 1574 was always true
1575 names.append(full)
1576 return names
1578 @staticmethod
1579 def _source_url(item: ZoteroItem) -> Optional[str]:
1580 data = item.data
1581 if data.get("url"):
1582 return str(data["url"]).strip()
1583 doi = (data.get("DOI") or "").strip()
1584 if doi:
1585 return f"https://doi.org/{doi}"
1586 return None
1588 @staticmethod
1589 def _parse_date(raw) -> Optional[date]:
1590 """Best-effort parse of Zotero's freeform date field."""
1591 if not raw:
1592 return None
1593 text = str(raw).strip()
1594 # Try full ISO first.
1595 for fmt in ("%Y-%m-%d", "%Y-%m"):
1596 try:
1597 return datetime.strptime(text[: len(fmt) + 2], fmt).date()
1598 except ValueError:
1599 continue
1600 match = re.search(r"(\d{4})", text)
1601 if match:
1602 year = int(match.group(1))
1603 if 1000 <= year <= 9999: 1603 ↛ 1605line 1603 didn't jump to line 1605 because the condition on line 1603 was always true
1604 return date(year, 1, 1)
1605 return None
1607 def _metadata_text(self, item: ZoteroItem) -> str:
1608 data = item.data
1609 parts: List[str] = []
1610 if data.get("title"): 1610 ↛ 1612line 1610 didn't jump to line 1612 because the condition on line 1610 was always true
1611 parts.append(f"Title: {data['title']}")
1612 authors = self._authors(item)
1613 if authors: 1613 ↛ 1614line 1613 didn't jump to line 1614 because the condition on line 1613 was never true
1614 parts.append(f"Authors: {', '.join(authors)}")
1615 if data.get("date"):
1616 parts.append(f"Date: {data['date']}")
1617 if data.get("publicationTitle"): 1617 ↛ 1618line 1617 didn't jump to line 1618 because the condition on line 1617 was never true
1618 parts.append(f"Publication: {data['publicationTitle']}")
1619 if data.get("DOI"):
1620 parts.append(f"DOI: {data['DOI']}")
1621 if data.get("url"): 1621 ↛ 1622line 1621 didn't jump to line 1622 because the condition on line 1621 was never true
1622 parts.append(f"URL: {data['url']}")
1623 tags = [
1624 t.get("tag")
1625 for t in data.get("tags", [])
1626 if isinstance(t, dict) and t.get("tag")
1627 ]
1628 if tags:
1629 parts.append(f"Tags: {', '.join(tags)}")
1630 if data.get("abstractNote"): 1630 ↛ 1631line 1630 didn't jump to line 1631 because the condition on line 1630 was never true
1631 parts.append(f"\nAbstract:\n{data['abstractNote']}")
1633 from ...text_processing import remove_surrogates
1635 return remove_surrogates("\n".join(parts)) if parts else item.key
1637 @staticmethod
1638 def _filename_for(item: ZoteroItem, file_type: str) -> str:
1639 title = (item.data.get("title") or "").strip()
1640 # Standalone attachments carry the filename as their title — don't
1641 # produce "paper.pdf.pdf".
1642 if title.lower().endswith(f".{file_type}".lower()):
1643 title = title[: -(len(file_type) + 1)]
1644 candidate = f"{title[:120]}.{file_type}" if title else None
1645 if candidate:
1646 try:
1647 return sanitize_filename(candidate)
1648 except UnsafeFilenameError:
1649 pass
1650 return f"{item.key}.{file_type}"
1652 # -- mapping / removal -------------------------------------------------
1654 def _upsert_map(
1655 self,
1656 session,
1657 ldr_collection_id: str,
1658 item: ZoteroItem,
1659 document_id: Optional[str],
1660 ) -> None:
1661 row = (
1662 session.query(ZoteroItemMap)
1663 .filter_by(
1664 ldr_collection_id=ldr_collection_id,
1665 zotero_item_key=item.key,
1666 )
1667 .first()
1668 )
1669 has_pdf = bool(
1670 document_id
1671 and (
1672 session.query(Document)
1673 .filter_by(id=document_id, file_type="pdf")
1674 .first()
1675 )
1676 )
1677 if row:
1678 row.zotero_version = item.version
1679 row.document_id = document_id
1680 row.has_pdf = has_pdf
1681 row.last_synced_at = datetime.now(UTC)
1682 else:
1683 session.add(
1684 ZoteroItemMap(
1685 ldr_collection_id=ldr_collection_id,
1686 zotero_item_key=item.key,
1687 zotero_version=item.version,
1688 document_id=document_id,
1689 has_pdf=has_pdf,
1690 last_synced_at=datetime.now(UTC),
1691 )
1692 )
1694 def _touch_map_version(
1695 self,
1696 session,
1697 ldr_collection_id: str,
1698 zotero_item_key: str,
1699 version: int,
1700 ) -> None:
1701 """Record a key's version without importing a document.
1703 Used for keys we deliberately don't import (a child/note that slipped
1704 into the top-level list, or a key Zotero didn't return) so the
1705 version-changed predicate doesn't re-queue them on every sync. Does
1706 not touch ``document_id`` (stays None / unchanged).
1707 """
1708 ik = zotero_item_key
1709 row = (
1710 session.query(ZoteroItemMap)
1711 .filter_by(
1712 ldr_collection_id=ldr_collection_id,
1713 zotero_item_key=ik,
1714 )
1715 .first()
1716 )
1717 if row: 1717 ↛ 1718line 1717 didn't jump to line 1718 because the condition on line 1717 was never true
1718 row.zotero_version = version
1719 row.last_synced_at = datetime.now(UTC)
1720 else:
1721 session.add(
1722 ZoteroItemMap(
1723 ldr_collection_id=ldr_collection_id,
1724 zotero_item_key=ik,
1725 zotero_version=version,
1726 document_id=None,
1727 has_pdf=False,
1728 last_synced_at=datetime.now(UTC),
1729 )
1730 )
1732 def _remove_item(
1733 self,
1734 session,
1735 row: ZoteroItemMap,
1736 source_type_id: str,
1737 default_collection_id: Optional[str] = None,
1738 pending_chunk_ids: Optional[Dict[Tuple[str, str], List[int]]] = None,
1739 ) -> Optional[str]:
1740 """Remove an item that no longer exists in Zotero.
1742 The document is only hard-deleted when nothing else references it AND
1743 it is a Zotero-owned document. Because dedup is global by content hash,
1744 one Document can be shared by multiple collections / Zotero mappings, or
1745 can be a user upload / research download that Zotero merely adopted via
1746 dedup; in those cases we must only drop *this* collection's link and the
1747 mapping row, never destroy a shared or foreign-owned document (and its
1748 blob/chunks) out from under the other references / the user.
1750 ``pending_chunk_ids`` — see :meth:`_release_document`.
1752 Returns the RELEASED document_id (so the caller can clear its FAISS
1753 vectors for this collection after the transaction), or None when the
1754 document is kept — either it has no document, or a byte-identical twin
1755 still maps to it in this collection (in which case its vectors must NOT
1756 be purged, or the surviving twin would vanish from search).
1757 """
1758 released = self._release_document(
1759 session,
1760 row.document_id,
1761 row.ldr_collection_id,
1762 row.id,
1763 source_type_id,
1764 default_collection_id,
1765 pending_chunk_ids,
1766 )
1767 session.delete(row)
1768 return released
1770 def _release_document(
1771 self,
1772 session,
1773 doc_id: Optional[str],
1774 coll_id: str,
1775 keep_mapping_id,
1776 source_type_id: str,
1777 default_collection_id: Optional[str] = None,
1778 pending_chunk_ids: Optional[Dict[Tuple[str, str], List[int]]] = None,
1779 ) -> Optional[str]:
1780 """Release a document from this collection: drop its collection link and
1781 RAG chunks, hard-deleting the document only when nothing else references
1782 it AND Zotero owns it. Never destroys a shared or foreign-owned document
1783 (dedup is global by content hash, so one Document may be shared across
1784 collections/mappings, or be a user upload Zotero merely adopted).
1786 ``keep_mapping_id`` is the ZoteroItemMap row that is going away or being
1787 repointed — excluded from the "is it shared?" check. Used both when an
1788 item is removed (:meth:`_remove_item`) and when an in-place edit dedups
1789 onto a *different* existing document, orphaning the item's prior doc.
1790 Returns ``doc_id`` so the caller can clear its FAISS vectors after the
1791 transaction.
1793 The chunk rows this method deletes below are the SAME rows a later,
1794 BATCHED ``_cleanup_removed_vectors`` call needs to find (by
1795 source_id + collection_name) to know which FAISS vectors to remove
1796 — but by the time that batched call runs, deleting the rows here
1797 (eagerly, so a cascade failure rolls back this whole transaction —
1798 see the comment above the ``if referenced_elsewhere`` branch below)
1799 has already made them unfindable. When the caller supplies
1800 ``pending_chunk_ids`` (keyed by ``(doc_id, collection_id)``), this
1801 captures each affected collection's chunk ids into it BEFORE
1802 deleting the rows, so the caller can purge those exact FAISS ids
1803 later via ``VectorIndex.delete_ids`` instead of the now-futile
1804 source_id lookup. ``None`` (the default, and what direct unit tests
1805 pass) skips capture and behaves exactly as before.
1806 """
1807 if not doc_id: 1807 ↛ 1808line 1807 didn't jump to line 1808 because the condition on line 1807 was never true
1808 return None
1809 # Another item in THIS collection still maps to the doc (byte-identical
1810 # dedup): it must stay linked + indexed here, so there is nothing to
1811 # release and no vectors to purge.
1812 if (
1813 session.query(ZoteroItemMap)
1814 .filter(
1815 ZoteroItemMap.document_id == doc_id,
1816 ZoteroItemMap.id != keep_mapping_id,
1817 ZoteroItemMap.ldr_collection_id == coll_id,
1818 )
1819 .first()
1820 is not None
1821 ):
1822 return None
1823 doc_owned_by_zotero = (
1824 session.query(Document.id)
1825 .filter_by(id=doc_id, source_type_id=source_type_id)
1826 .first()
1827 is not None
1828 )
1829 referenced_elsewhere = (
1830 session.query(ZoteroItemMap)
1831 .filter(
1832 ZoteroItemMap.document_id == doc_id,
1833 ZoteroItemMap.id != keep_mapping_id,
1834 )
1835 .first()
1836 is not None
1837 ) or (
1838 # Membership in the default "Library" collection is ambient
1839 # (every import is linked there) — it must not keep a removed
1840 # item's document alive.
1841 session.query(DocumentCollection)
1842 .filter(
1843 DocumentCollection.document_id == doc_id,
1844 DocumentCollection.collection_id.notin_(
1845 {coll_id, default_collection_id}
1846 if default_collection_id
1847 else {coll_id}
1848 ),
1849 )
1850 .first()
1851 is not None
1852 )
1853 # A cascade failure below must PROPAGATE so the whole release (and the
1854 # map change it shares a transaction with) rolls back and the item is
1855 # retried next sync. Swallowing it would commit a half-release — the
1856 # map row gone but the doc/chunks still present — and the caller would
1857 # then purge FAISS vectors for a document that still exists.
1858 if referenced_elsewhere or not doc_owned_by_zotero:
1859 # Keep the document (shared with another collection/mapping, or a
1860 # user upload / research download Zotero only adopted via
1861 # content-hash dedup — never Zotero's to destroy), but drop this
1862 # collection's link AND its RAG chunks so it stops surfacing here.
1863 # The caller clears this collection's FAISS vectors for the returned
1864 # doc_id after the transaction.
1865 if pending_chunk_ids is not None:
1866 self._capture_chunk_ids(
1867 pending_chunk_ids,
1868 session,
1869 doc_id,
1870 coll_id,
1871 )
1872 CascadeHelper.delete_document_chunks(
1873 session, doc_id, collection_name=f"collection_{coll_id}"
1874 )
1875 session.query(DocumentCollection).filter_by(
1876 document_id=doc_id, collection_id=coll_id
1877 ).delete(synchronize_session=False)
1878 else:
1879 doc = session.query(Document).filter_by(id=doc_id).first()
1880 if doc: 1880 ↛ 1902line 1880 didn't jump to line 1902 because the condition on line 1880 was always true
1881 # DocumentChunk has no FK cascade — delete RAG chunks
1882 # explicitly before removing the document. Capture ids for
1883 # BOTH collections this doc can possibly be in — established
1884 # above (doc_owned_by_zotero and not referenced_elsewhere)
1885 # to be at most coll_id and default_collection_id.
1886 if pending_chunk_ids is not None:
1887 self._capture_chunk_ids(
1888 pending_chunk_ids, session, doc_id, coll_id
1889 )
1890 if (
1891 default_collection_id
1892 and default_collection_id != coll_id
1893 ):
1894 self._capture_chunk_ids(
1895 pending_chunk_ids,
1896 session,
1897 doc_id,
1898 default_collection_id,
1899 )
1900 CascadeHelper.delete_document_chunks(session, doc.id)
1901 CascadeHelper.delete_document_completely(session, doc.id)
1902 return doc_id
1904 @staticmethod
1905 def _capture_chunk_ids(
1906 pending_chunk_ids: Dict[Tuple[str, str], List[int]],
1907 session,
1908 doc_id: str,
1909 collection_id: str,
1910 ) -> None:
1911 """Snapshot ``doc_id``'s chunk-row int ids for ``collection_id`` into
1912 ``pending_chunk_ids`` BEFORE they are eagerly deleted by the caller.
1914 A plain SELECT (no locking/mutation) — safe to run just ahead of the
1915 row-deleting statement in the same transaction. See the note on
1916 ``pending_chunk_ids`` in :meth:`_release_document`.
1917 """
1918 ids = [
1919 row.id
1920 for row in session.query(DocumentChunk.id).filter_by(
1921 source_type="document",
1922 source_id=doc_id,
1923 collection_name=f"collection_{collection_id}",
1924 )
1925 ]
1926 if ids: 1926 ↛ 1927line 1926 didn't jump to line 1927 because the condition on line 1926 was never true
1927 pending_chunk_ids.setdefault((doc_id, collection_id), []).extend(
1928 ids
1929 )
1931 @staticmethod
1932 def _document_exists(session, document_id: Optional[str]) -> bool:
1933 if not document_id: 1933 ↛ 1934line 1933 didn't jump to line 1934 because the condition on line 1933 was never true
1934 return False
1935 return (
1936 session.query(Document.id).filter_by(id=document_id).first()
1937 is not None
1938 )