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