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