Coverage for src/local_deep_research/web/services/research_sources_service.py: 88%
296 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +0000
1"""
2Service for managing research sources/resources in the database.
4This service handles saving and retrieving sources from research
5in a proper relational way using the ResearchResource table.
6"""
8from typing import List, Dict, Any, Optional
9from datetime import datetime, UTC
10from loguru import logger
11from sqlalchemy import or_
12import numbers
14from sqlalchemy.exc import IntegrityError
15from sqlalchemy.orm import Session, load_only
17from ...database.models import (
18 Journal,
19 Paper,
20 PaperAppearance,
21 ResearchResource,
22 ResearchHistory,
23)
24from ...database.session_context import get_user_db_session
25from ...utilities.citation_normalizer import normalize_citation
26from ...utilities.url_utils import (
27 CHUNK_DISPLAY_KEY,
28 canonical_url_key,
29 library_display_url,
30 preferred_chunk_display,
31)
34def _as_text(value: object) -> str:
35 """Coerce a stored-dict value to ``str`` for slicing and ``len()``.
37 Every read in the per-source loop comes from a JSON blob that
38 round-tripped through the database, so an ``int`` or ``list`` in a
39 text field is possible. Slicing one raises inside the broad
40 ``except`` below, which rolls back and drops the whole citation with
41 no message — the same silent-loss failure the url guards prevent.
42 """
43 if value is None:
44 return ""
45 if isinstance(value, str):
46 return value
47 try:
48 return str(value)
49 except Exception:
50 # A coercion helper on a silent-drop path must not be able to
51 # raise: an object whose ``__str__`` fails would otherwise take
52 # out the exception handler that reports the failure, and with it
53 # every sibling source in the batch.
54 return ""
57# Every field in this loop that is read as text — by normalize_citation,
58# by a slice, as a dict key, or as a Text column. Coercing them once at
59# the top of the loop is what stops the "guard the read a review named,
60# miss its sibling" cycle: nine commits guarded url, then snippet, then
61# ct_matched, then title, while pmid, arxiv_id and normalize_citation's
62# own re-read of url stayed raw and kept dropping the citation.
63_TEXT_FIELDS = (
64 "url",
65 "link",
66 "title",
67 "name",
68 "snippet",
69 "content_preview",
70 "description",
71 "source_type",
72 "source_engine",
73 "source",
74 "container_title",
75 "container-title",
76 "journal",
77 "venue",
78 "journal_ref",
79 "journal_name_matched",
80 "volume",
81 "issue",
82 "pages",
83 "publisher",
84)
87# Identifier fields. These are NOT text-coerced: they feed ``unique``
88# columns on ``Paper`` and the dedup SELECT that decides whether two
89# engines found the same paper. ``_extract_doi`` documents and tests a
90# LIST doi (it takes ``doi[0]``), so stringifying one writes
91# ``"['10.1/x', '10.2/y']"`` into a unique key — worse than the drop it
92# replaced, because it is permanent and defeats dedup. Unwrap instead.
93_IDENTIFIER_FIELDS = ("doi", "pmid", "pmcid", "arxiv_id")
96# The two fields whose contents ``_parse_authors_list`` turns into author
97# records. Only inside these does an empty name key mean "junk author".
98_AUTHOR_LIST_KEYS = ("authors", "authors_csl")
100# Author fields whose emptiness produces a junk author record.
101# ``name``/``display_name`` are the two ``citation_normalizer._parse_name``
102# strips — ``display_name`` is the OpenAlex shape, missing from an earlier
103# spelling of this tuple. ``family``/``given``/``suffix`` are copied
104# verbatim by ``_parse_authors_list``'s CSL branch, which tests only
105# ``"family" in author``, so an empty one of those is embedded in the
106# stored csl_json rather than ignored. ``literal`` is deliberately absent:
107# ``_parse_name`` only ever EMITS it, never reads it from input, and an
108# earlier spelling carried it (plus two more) as keys that did nothing.
109_AUTHOR_NAME_KEYS = ("name", "display_name", "family", "given", "suffix")
112# Where the walk currently is, relative to an author record. A bool was
113# not enough: it could say "inside authors" but not "inside an author
114# record's OWN keys", so once set it stayed set for every descendant and
115# deleted an empty ``given`` from ``authors[0].affiliation`` or an
116# OpenAlex ``institutions[].display_name`` — neither of which
117# ``_parse_authors_list`` ever reads.
118_AUTHORS_OUTSIDE = 0
119_AUTHORS_LIST = 1 # this value IS an authors/authors_csl list
120_AUTHORS_RECORD = 2 # this value IS one author record
123def _json_text_safe(
124 value: object, _depth: int = 0, _authors: int = _AUTHORS_OUTSIDE
125) -> object:
126 """Recursively make *value* safe to serialize and to read as text.
128 The ingest boundary is the TREE, not the top-level dict. ``metadata``
129 and ``authors`` come from the same untrusted producer, are read by the
130 same ``normalize_citation`` call, and are serialized into the same JSON
131 column — so a ``set`` under ``metadata["journal"]`` or an ``int`` under
132 ``authors[0]["name"]`` raises exactly where a top-level one used to.
134 ``_authors`` scopes the empty-name drop to an author record's OWN
135 keys, which are the only ones ``_parse_authors_list`` reads. Anything
136 nested below them is inert data the producer may want preserved.
137 """
138 if _depth > 6: 138 ↛ 139line 138 didn't jump to line 139 because the condition on line 138 was never true
139 return str(value)
140 if value is None or isinstance(value, (str, int, float, bool)):
141 return value
142 if isinstance(value, dict):
143 in_record = _authors == _AUTHORS_RECORD
144 # Author name fields are coerced ONCE and reused. Calling
145 # ``_as_text`` separately in the filter and the value expression
146 # meant a non-deterministic ``__str__`` could pass the filter on
147 # one call and write "" from the next.
148 coerced = (
149 {
150 str(k): _as_text(v)
151 for k, v in value.items()
152 if k in _AUTHOR_NAME_KEYS
153 }
154 if in_record
155 else {}
156 )
157 return {
158 str(k): (
159 coerced[str(k)]
160 if in_record and k in _AUTHOR_NAME_KEYS
161 # Descending out of the record: re-arm only on another
162 # authors key, never inherit.
163 else _json_text_safe(
164 v,
165 _depth + 1,
166 _AUTHORS_LIST
167 if k in _AUTHOR_LIST_KEYS
168 else _AUTHORS_OUTSIDE,
169 )
170 )
171 for k, v in value.items()
172 # An author name that coerces to nothing is DROPPED, not kept.
173 # ``_parse_name(None)`` raises and ``_parse_name("")``
174 # synthesises ``{"literal": ""}``; the CSL branch of
175 # ``_parse_authors_list`` copies ``family``/``given``/``suffix``
176 # verbatim, so an empty one of those lands as a junk author in
177 # the stored csl_json just the same. ``.strip()`` because
178 # ``_parse_name`` strips before its own emptiness test, so a
179 # whitespace-only name produces the same junk.
180 if not (
181 in_record
182 and k in _AUTHOR_NAME_KEYS
183 and not (coerced.get(str(k)) or "").strip()
184 )
185 }
186 if isinstance(value, (list, tuple)):
187 # The ELEMENTS of an authors list are the author records.
188 child = (
189 _AUTHORS_RECORD if _authors == _AUTHORS_LIST else _AUTHORS_OUTSIDE
190 )
191 return [_json_text_safe(v, _depth + 1, child) for v in value]
192 return str(value)
195# SQLite's INTEGER column is 64-bit. Python's int is not, so a value can
196# pass ``int()`` inside citation_normalizer and still raise OverflowError
197# at flush() — and because that is not an IntegrityError, the per-source
198# SAVEPOINT retry does not apply and the Session is left rolled back, so
199# every LATER source in the batch is lost too and nothing commits.
200_SQLITE_INT_MIN = -(2**63)
201_SQLITE_INT_MAX = 2**63 - 1
204def _coerce_year(value: object) -> object:
205 """Drop a year that cannot be stored, rather than losing the batch."""
206 if value is None or isinstance(value, bool): 206 ↛ 207line 206 didn't jump to line 207 because the condition on line 206 was never true
207 return value
208 # Duck-typed on the operation, not on a type list. ``_parse_date`` does
209 # ``int(raw_year)``, which accepts anything with ``__int__`` — numpy
210 # scalars, Decimal, Fraction, even UUID — and this file's own comments
211 # name numpy as an expected producer. Gating on ``(int, float, str)``
212 # let every one of those past the bound and into flush().
213 try:
214 as_int = int(float(value) if isinstance(value, float) else value)
215 except (ValueError, OverflowError, TypeError, ArithmeticError):
216 # Not int()-able, so ``_parse_date`` cannot overflow on it either:
217 # leave strings for its regex, drop anything else.
218 return value if isinstance(value, str) else None
219 return value if _SQLITE_INT_MIN <= as_int <= _SQLITE_INT_MAX else None
222def _coerce_identifier(value: object) -> object:
223 """Return *value* fit for a ``unique`` identifier column, else ``None``.
225 Unwrap-or-DROP, never unwrap-or-stringify. These feed ``Paper.doi``,
226 ``.pmid`` and ``.arxiv_id`` and the dedup SELECT, where a repr is worse
227 than absence twice over: it is permanent, and two DIFFERENT papers
228 carrying the same unparseable value match each other and collapse into
229 one row. Absence just means "this engine gave us no identifier", which
230 ``_extract_doi`` already handles by falling through to its next
231 channel.
233 A list is unwrapped once and its element re-checked, so ``[["10.1/x"]]``
234 drops rather than stringifying the inner list.
235 """
236 if isinstance(value, (list, tuple)):
237 value = value[0] if value else None
238 if value is None or isinstance(value, str):
239 return value or None
240 # Integral-ness, not ``isinstance(int)``: a PMID legitimately arrives
241 # as an int, and numpy.int64 / Decimal("12345") are integers that are
242 # NOT int subclasses. Rejecting them dropped real identifiers and
243 # broke dedup — the same harm as stringifying, in the other direction.
244 # ``str()`` on a large int is exact, so there is nothing to protect
245 # against here.
246 # A numpy boolean is neither a Python ``bool`` nor a
247 # ``numbers.Real``/``Integral``, so it slips both gates while
248 # ``int()``-ing to 1 and comparing equal to it — a True in a pmid
249 # field was stored as the identifier "1".
250 #
251 # Matched on the dtype kind rather than the type name: numpy 1.x calls
252 # the scalar ``bool_`` and numpy 2.x calls it ``bool``, so a name check
253 # written against either version silently misses the other.
254 if isinstance(value, bool):
255 return None
256 try:
257 # ``getattr`` only swallows a MISSING attribute; a ``.kind``
258 # property that raises would propagate out of a coercion helper
259 # that every other step in this function keeps total.
260 if getattr(getattr(value, "dtype", None), "kind", None) == "b":
261 return None
262 except Exception:
263 return None
264 # ``numbers.Real`` rather than ``isinstance(value, float)``: numpy's
265 # float32 is not a float subclass, so an exactly-integral one slipped
266 # through as an identifier. Engines send ints and strings, never reals.
267 if isinstance(value, numbers.Real) and not isinstance(
268 value, numbers.Integral
269 ):
270 return None
271 try:
272 as_int = int(value)
273 except (ValueError, OverflowError, TypeError, ArithmeticError):
274 return None
275 # Exactness. ``numbers.Integral`` first, because ``as_int != value``
276 # relies on a symmetric ``__eq__``: numpy ints and Decimal have one,
277 # so both are accepted. An object that defines ``__int__`` but neither
278 # registers as Integral nor compares equal to its own integer is
279 # DROPPED — deliberately. This feeds a unique column, and a value that
280 # cannot be shown equal to the integer it claims to be is not one to
281 # guess at. So this is int()-ability plus provable exactness, not
282 # int()-ability alone.
283 if not isinstance(value, numbers.Integral) and as_int != value: 283 ↛ 284line 283 didn't jump to line 284 because the condition on line 283 was never true
284 return None
285 return str(as_int) if as_int else None
288def normalize_source_fields(source: dict) -> dict:
289 """Return a copy of *source* with its text fields coerced to ``str``.
291 Engine dicts reach this service from arbitrary producers — a LangChain
292 retriever's ``Document.metadata`` is whatever the user put there — and
293 a non-str reaching a slice, a dict key, a regex or a Text column raises
294 inside the per-source ``except``, which rolls back and drops the whole
295 citation with only a counter to show for it.
297 A falsy non-str becomes ``""``, not its repr: ``{"link": []}`` should
298 keep skipping the source, as it did before any coercion existed, rather
299 than persisting the string ``"[]"`` and rendering it as a link.
301 Typed fields are left alone — ``authors`` stays a list, ``year`` an
302 int, ``score`` a float — and only a non-dict ``metadata`` is replaced,
303 since callers index into it.
304 """
305 normalized = dict(source)
306 for field in _TEXT_FIELDS:
307 value = normalized.get(field)
308 if value is None or isinstance(value, str):
309 continue
310 normalized[field] = _as_text(value) if value else ""
311 for field in ("year", "publication_year"):
312 if field in normalized:
313 normalized[field] = _coerce_year(normalized[field])
315 for field in _IDENTIFIER_FIELDS:
316 # ``in``, not ``.get``: materialising ``doi: None`` on every plain
317 # web result changes what ``original_data`` stores for rows that
318 # never had an academic identity.
319 if field not in normalized:
320 continue
321 normalized[field] = _coerce_identifier(normalized[field])
323 if "metadata" in normalized:
324 metadata = normalized.get("metadata")
325 normalized["metadata"] = (
326 _json_text_safe(metadata) if isinstance(metadata, dict) else {}
327 )
329 # D2: ``authors_csl`` is read BEFORE ``authors`` by
330 # ``normalize_citation``, in the same expression, and is NASA ADS's
331 # primary author channel — normalizing only one of the two left the
332 # other reaching the same json.dumps.
333 for field in _AUTHOR_LIST_KEYS:
334 value = normalized.get(field)
335 if isinstance(value, (list, tuple)):
336 normalized[field] = _json_text_safe(
337 list(value), _authors=_AUTHORS_LIST
338 )
340 # D1: ``external_ids`` is the second DOI source ``_extract_doi``
341 # documents. Leaving it raw put a list DOI into the unique Paper.doi
342 # by the very channel the identifier unwrap above exists to protect.
343 for field in ("external_ids", "externalIds"):
344 value = normalized.get(field)
345 if value is not None:
346 normalized[field] = (
347 _json_text_safe(value) if isinstance(value, dict) else {}
348 )
349 for key in ("DOI", "doi"):
350 if key in normalized[field]:
351 normalized[field][key] = _coerce_identifier(
352 normalized[field][key]
353 )
354 return normalized
357def select_source_url(source: dict) -> object:
358 """Choose the URL to persist for *source*.
360 Extracted so the test suite can exercise THIS function. It previously
361 lived inline and was pinned by a test that reimplemented it; the mirror
362 stayed faithful right up until it faithfully mirrored a bug, and went
363 green while the hole was open.
365 Fails CLOSED on a non-str url: an earlier version made the isinstance
366 check one conjunct of the rejection condition, so a non-str url made
367 the whole ``and``-chain false, the ownership check never ran, and a
368 recorded anchor naming a DIFFERENT document was persisted unchecked.
369 """
370 own_url = source.get("url", "") or source.get("link", "")
371 own_key = canonical_url_key(own_url) if isinstance(own_url, str) else None
373 raw_key = source.get(CHUNK_DISPLAY_KEY)
374 recorded = (
375 preferred_chunk_display(raw_key) if isinstance(raw_key, str) else None
376 )
377 # No usable key for this entry means no way to prove ownership, so the
378 # recorded anchor is refused rather than trusted.
379 if recorded and (own_key is None or canonical_url_key(recorded) != own_key):
380 recorded = None
382 own_anchor = (
383 preferred_chunk_display(own_url) if isinstance(own_url, str) else None
384 )
385 # Normalise before falling back to the raw string. Without this the
386 # absolute alias is persisted as typed, and the ``:443``/userinfo
387 # spellings — which the alias parser accepts as the same document but
388 # ``library_resolver`` refuses, since it compares netloc exactly —
389 # become dead links in ``research_resources.url``. Returns None for a
390 # non-library URL, so external sources keep their own spelling.
391 normalised = (
392 library_display_url(own_url) if isinstance(own_url, str) else None
393 )
394 return own_anchor or recorded or normalised or own_url
397class ResearchSourcesService:
398 """Service for managing research sources in the database."""
400 @staticmethod
401 def save_research_sources(
402 research_id: str,
403 sources: List[Dict[str, Any]],
404 username: Optional[str] = None,
405 ) -> int:
406 """
407 Save sources from research to the ResearchResource table.
409 Args:
410 research_id: The UUID of the research
411 sources: List of source dictionaries with url, title, snippet, etc.
412 username: Username for database access
414 Returns:
415 Number of sources saved
416 """
417 if not sources:
418 logger.info(f"No sources to save for research {research_id}")
419 return 0
421 saved_count = 0
422 # Failed-source counter. The per-source try/except below catches
423 # broad exceptions to keep one bad source from killing the batch,
424 # but without this counter the caller had no way to distinguish
425 # "all N saved" from "some silently dropped". Emitted in the
426 # final log line so admins can spot save-failure trends.
427 failed_count = 0
429 try:
430 with get_user_db_session(username) as db_session:
431 # First check if resources already exist for this research
432 existing = (
433 db_session.query(ResearchResource)
434 .filter_by(research_id=research_id)
435 .count()
436 )
438 if existing > 0:
439 logger.info(
440 f"Research {research_id} already has {existing} resources, skipping save"
441 )
442 return int(existing)
444 # Save each source as a ResearchResource.
445 # Each source runs inside a SAVEPOINT so a per-source
446 # failure can be rolled back cleanly without losing
447 # any previously saved sources in this batch.
448 # Per-batch memoization: container_title → journal_id
449 # avoids redundant Journal lookups when multiple sources
450 # share the same venue (common in topic-focused searches).
451 journal_id_cache: Dict[Optional[str], Optional[int]] = {}
452 for source in sources:
453 sp = None
454 try:
455 # Once, before anything reads it — see
456 # normalize_source_fields.
457 source = normalize_source_fields(source)
458 # Extract fields from various possible formats
459 # A chunk-anchored spelling recorded by the
460 # collector wins. After canonical-key dedup only
461 # one entry per library document survives, and it
462 # may be the anchor-less view — persisting that
463 # loses the #chunk-<n> permanently, since this row
464 # is what later renders the source.
465 # Validated, not trusted: this value is written
466 # to the database and later rendered as a link, so
467 # an unchecked read would let whatever set the key
468 # choose the stored URL.
469 # ``_as_text`` here, not just at the slices: these
470 # three reach ``Text`` columns, and a list or dict
471 # raises at ``flush()`` rather than at the read —
472 # so guarding the read alone only moved the
473 # exception, and the citation was dropped just the
474 # same.
475 url = _as_text(select_source_url(source))
476 title = _as_text(
477 source.get("title", "") or source.get("name", "")
478 )
479 # Coerced, not assumed: these come from the same
480 # stored dict as the url, and a non-str raises
481 # inside the per-source ``except`` below — which
482 # drops the whole citation silently.
483 snippet = _as_text(
484 source.get("snippet", "")
485 or source.get("content_preview", "")
486 or source.get("description", "")
487 )
488 source_type = _as_text(source.get("source_type", "web"))
490 # Skip if no URL
491 if not url:
492 continue
494 # Start savepoint for this source — any rollback
495 # inside this block (including the IntegrityError
496 # retry path below) only affects this source.
497 sp = db_session.begin_nested()
499 # Create resource record.
500 # Sanitize the source dict before embedding it in
501 # resource_metadata — raw engine dicts can contain
502 # non-JSON-serializable values (nested objects,
503 # numpy types, affiliation sub-dicts, etc.) which
504 # would crash json.dumps() at flush time.
505 safe_source = _json_safe(source)
506 resource = ResearchResource(
507 research_id=research_id,
508 title=title or "Untitled",
509 url=url,
510 content_preview=snippet[:1000]
511 if snippet
512 else None, # Limit preview length
513 source_type=source_type,
514 resource_metadata={
515 "added_at": datetime.now(UTC).isoformat(),
516 "original_data": safe_source,
517 },
518 created_at=datetime.now(UTC).isoformat(),
519 )
521 db_session.add(resource)
522 db_session.flush() # Get resource.id for FK
524 # Create or reuse Paper for academic sources
525 citation_fields = normalize_citation(source)
526 if citation_fields:
527 source_engine = citation_fields.pop(
528 "source_engine", None
529 )
530 # Try to link to existing Journal record
531 # (container_title stays in citation_fields so
532 # it ends up in the metadata blob for citation
533 # export). Memoized per batch to avoid repeat
534 # lookups for the same venue.
535 ct = citation_fields.get("container_title")
536 # Coerced before use as a dict key: CSL
537 # ``container-title`` is legitimately an array,
538 # and an unhashable value raises here — 34
539 # lines before the guard on the same value.
540 ct = _as_text(ct) or None
541 if ct in journal_id_cache:
542 journal_id = journal_id_cache[ct]
543 else:
544 journal_id = _resolve_journal_id(db_session, ct)
545 journal_id_cache[ct] = journal_id
547 # Separate indexed columns from metadata blob.
548 # Only doi/arxiv_id/pmid/journal_id/
549 # container_title/year are real columns on
550 # Paper; everything else is bundled into
551 # the metadata JSON blob. Quality is NOT
552 # stored per-Paper — the dashboard resolves
553 # it live (Tier 4: journals.quality via
554 # container_title lookup; Tier 1-3: bundled
555 # reference DB) so a re-scored journal
556 # propagates automatically.
557 # container_title: prefer the filter's
558 # cleaned matched name (what actually keyed
559 # the successful score); fall back to the raw
560 # CSL container_title if the filter didn't
561 # run (e.g. journal_reputation disabled).
562 #
563 # .pop() removes it from citation_fields so it
564 # doesn't end up duplicated in paper_metadata
565 # JSON. The Paper column is the sole source
566 # of truth; CSL-JSON export already captured
567 # the raw value inside citation_fields[
568 # "csl_json"] during normalize_citation.
569 ct_raw = citation_fields.pop(
570 "container_title", None
571 )
572 ct_matched = (
573 source.get("journal_name_matched") or ct_raw
574 )
575 # ``or None`` like its sibling above: an empty
576 # container_title must stay NULL, or every
577 # venue-less paper forms an empty-string group
578 # in the journal metrics instead of being
579 # excluded by ``isnot(None)``.
580 ct_matched = _as_text(ct_matched) or None
581 if ct_matched and len(ct_matched) > 500: 581 ↛ 582line 581 didn't jump to line 582 because the condition on line 581 was never true
582 logger.debug(
583 f"Truncating container_title to 500 "
584 f"chars: {ct_matched[:80]}..."
585 )
586 ct_matched = ct_matched[:500]
587 # `year` intentionally stays in citation_fields
588 # (JSON blob) AND is copied to the indexed column.
589 # The JSON blob remains the CSL-JSON source of
590 # truth; the column is a denormalized index
591 # surface for dashboard year queries.
592 indexed = {
593 "doi": citation_fields.pop("doi", None),
594 "arxiv_id": citation_fields.pop(
595 "arxiv_id", None
596 ),
597 "pmid": citation_fields.pop("pmid", None),
598 "journal_id": journal_id,
599 "container_title": ct_matched,
600 "year": citation_fields.get("year"),
601 }
603 # Dedup: find existing paper by DOI/arxiv/pmid.
604 # The UNIQUE constraints on doi/arxiv_id/pmid
605 # prevent duplicates from concurrent writers,
606 # but we still need to handle the race where
607 # our SELECT missed and another writer's
608 # INSERT succeeds first — catch IntegrityError
609 # and re-query.
610 paper = _find_existing_paper(db_session, indexed)
611 if paper is not None:
612 _merge_identifiers(
613 paper, indexed, citation_fields
614 )
615 else:
616 paper = Paper(
617 **indexed,
618 paper_metadata=citation_fields or None,
619 )
620 db_session.add(paper)
621 try:
622 db_session.flush()
623 except IntegrityError:
624 # Concurrent writer inserted same
625 # paper. Roll back this SAVEPOINT
626 # only (not the whole batch), then
627 # restart a nested one and re-fetch
628 # the existing row for merging.
629 sp.rollback()
630 sp = db_session.begin_nested()
631 # After savepoint rollback we also
632 # need to re-create the resource
633 # since its flush was undone.
634 resource = ResearchResource(
635 research_id=research_id,
636 title=title or "Untitled",
637 url=url,
638 content_preview=snippet[:1000]
639 if snippet
640 else None,
641 source_type=source_type,
642 resource_metadata={
643 "added_at": datetime.now(
644 UTC
645 ).isoformat(),
646 "original_data": safe_source,
647 },
648 created_at=datetime.now(
649 UTC
650 ).isoformat(),
651 )
652 db_session.add(resource)
653 db_session.flush()
654 paper = _find_existing_paper(
655 db_session, indexed
656 )
657 if paper is None:
658 # Truly unexpected — concurrent
659 # writer's row is gone.
660 raise
661 _merge_identifiers(
662 paper, indexed, citation_fields
663 )
665 # Link paper to this resource
666 appearance = PaperAppearance(
667 paper_id=paper.id,
668 resource_id=resource.id,
669 source_engine=source_engine,
670 )
671 db_session.add(appearance)
673 # Commit the savepoint so this source's writes
674 # persist even if a later source fails.
675 sp.commit()
676 saved_count += 1
678 except Exception:
679 # Roll back just this source's savepoint; earlier
680 # sources in the batch stay committed at the
681 # outer transaction level.
682 #
683 # Unconditional, NOT gated on ``sp.is_active``: a
684 # failure inside flush() deactivates the savepoint,
685 # so the guard skipped the rollback in exactly the
686 # case that needs it. The Session then stayed in
687 # its failed state, every later source raised
688 # PendingRollbackError, and the final commit()
689 # raised out of the function — one bad source lost
690 # the whole batch and the caller got an exception
691 # instead of a reduced count.
692 if sp is not None: 692 ↛ 701line 692 didn't jump to line 701 because the condition on line 692 was always true
693 try:
694 sp.rollback()
695 except Exception:
696 # Already unwound; nothing further to undo
697 # for this source.
698 logger.debug(
699 "savepoint rollback was already done"
700 )
701 failed_count += 1
702 # ``_as_text``: the only unguarded read of url
703 # left, and it runs when something has ALREADY
704 # failed — an exception raised here escapes the
705 # handler and loses the sibling sources too.
706 logger.exception(
707 "Failed to save source "
708 + _as_text(source.get("url", "unknown"))
709 )
710 continue
712 # Commit all resources
713 if saved_count > 0:
714 db_session.commit()
715 if failed_count > 0:
716 logger.warning(
717 f"Saved {saved_count} sources for research "
718 f"{research_id} — {failed_count} source(s) "
719 f"failed and were skipped (see earlier "
720 f"ERROR logs for per-source stack traces)"
721 )
722 else:
723 logger.info(
724 f"Saved {saved_count} sources for research {research_id}"
725 )
726 elif failed_count > 0: 726 ↛ 727line 726 didn't jump to line 727 because the condition on line 726 was never true
727 logger.warning(
728 f"No sources saved for research {research_id} — "
729 f"all {failed_count} sources in the batch failed "
730 f"(see earlier ERROR logs for per-source stack "
731 f"traces)"
732 )
734 except Exception:
735 logger.exception("Error saving research sources")
736 raise
738 return saved_count
740 @staticmethod
741 def get_research_sources(
742 research_id: str, username: Optional[str] = None
743 ) -> List[Dict[str, Any]]:
744 """
745 Get all sources for a research from the database.
747 Args:
748 research_id: The UUID of the research
749 username: Username for database access
751 Returns:
752 List of source dictionaries
753 """
754 sources = []
756 try:
757 with get_user_db_session(username) as db_session:
758 resources = (
759 db_session.query(ResearchResource)
760 .filter_by(research_id=research_id)
761 .order_by(ResearchResource.id.asc())
762 .all()
763 )
765 for resource in resources:
766 sources.append(
767 {
768 "id": resource.id,
769 "url": resource.url,
770 "title": resource.title,
771 "snippet": resource.content_preview,
772 "content_preview": resource.content_preview,
773 "source_type": resource.source_type,
774 "metadata": resource.resource_metadata or {},
775 "created_at": resource.created_at,
776 }
777 )
779 logger.info(
780 f"Retrieved {len(sources)} sources for research {research_id}"
781 )
783 except Exception:
784 logger.exception("Error retrieving research sources")
785 raise
787 return sources
789 @staticmethod
790 def update_research_with_sources(
791 research_id: str,
792 all_links_of_system: List[Dict[str, Any]],
793 username: Optional[str] = None,
794 ) -> bool:
795 """
796 Update a completed research with its sources.
797 This should be called when research completes.
799 Args:
800 research_id: The UUID of the research
801 all_links_of_system: List of all sources found during research
802 username: Username for database access
804 Returns:
805 True if successful
806 """
807 try:
808 # Save sources to ResearchResource table
809 saved_count = ResearchSourcesService.save_research_sources(
810 research_id, all_links_of_system, username
811 )
813 # Also update the research metadata to include source count
814 with get_user_db_session(username) as db_session:
815 research = (
816 db_session.query(ResearchHistory)
817 .filter_by(id=research_id)
818 .first()
819 )
821 if research:
822 if not research.research_meta: 822 ↛ 826line 822 didn't jump to line 826 because the condition on line 822 was always true
823 research.research_meta = {}
825 # Update metadata with source information
826 research.research_meta["sources_count"] = saved_count
827 research.research_meta["has_sources"] = saved_count > 0
829 db_session.commit()
830 logger.info(
831 f"Updated research {research_id} with {saved_count} sources"
832 )
833 return True
834 logger.warning(
835 f"Research {research_id} not found for source update"
836 )
837 return False
839 except Exception:
840 logger.exception("Error updating research with sources")
841 return False
844def _json_safe(value: Any, _depth: int = 0, _seen: Optional[set] = None) -> Any:
845 """Recursively coerce a value into a JSON-serializable form.
847 Used before embedding arbitrary engine result dicts into JSON
848 columns. Non-primitive values (datetime, date, set, tuple,
849 custom objects) are converted to strings or dropped. This is a
850 last-resort sanitizer — callers should still prefer structured
851 whitelisting (e.g., Paper.paper_metadata only stores known CSL
852 fields).
854 Depth limit and cycle detection prevent RecursionError on
855 pathological input (circular dict/list references).
856 """
857 # Depth limit as a belt-and-braces guard
858 if _depth > 32: 858 ↛ 859line 858 didn't jump to line 859 because the condition on line 858 was never true
859 return str(value)
861 # JSON primitives pass through unchanged
862 if value is None or isinstance(value, (bool, int, float, str)):
863 return value
865 # Container cycle detection via id() tracking
866 if isinstance(value, (dict, list, tuple, set, frozenset)):
867 if _seen is None:
868 _seen = set()
869 if id(value) in _seen: 869 ↛ 870line 869 didn't jump to line 870 because the condition on line 869 was never true
870 return "<circular>"
871 _seen = _seen | {id(value)}
873 if isinstance(value, dict):
874 return {
875 str(k): _json_safe(v, _depth + 1, _seen)
876 for k, v in value.items()
877 if isinstance(k, (str, int, float, bool))
878 }
879 if isinstance(value, (list, tuple, set, frozenset)):
880 return [_json_safe(v, _depth + 1, _seen) for v in value]
881 # datetime/date and anything else: coerce to string
882 return str(value)
885def _resolve_journal_id(
886 db_session: Session, container_title: Optional[str]
887) -> Optional[int]:
888 """Look up a Journal record by name. Returns journal.id or None.
890 The journal reputation filter writes Journal rows using the
891 cleaned journal name as returned by its regex cleanup (NFKC-
892 normalized, whitespace-stripped, but NOT lowercased). We match
893 against that by applying the same NFKC+strip normalization here
894 and using a case-insensitive comparison so mismatched capitalization
895 in the container_title doesn't break the lookup.
896 """
897 if not container_title:
898 return None
899 import unicodedata
901 name_norm = unicodedata.normalize("NFKC", container_title).strip()
902 # Query name_lower, not func.lower(name): expression-wrapping the
903 # indexed column forces a full scan.
904 row = (
905 db_session.query(Journal.id)
906 .filter(Journal.name_lower == name_norm.lower())
907 .first()
908 )
909 return row[0] if row else None
912def _find_existing_paper(
913 db_session: Session, fields: dict
914) -> Optional["Paper"]:
915 """Find an existing Paper by any of DOI, arXiv ID, or PMID.
917 Issues a single OR-query across all provided identifiers so that a
918 caller with multiple IDs doesn't miss dedup because the first one
919 (e.g. DOI) is absent from the stored row but a later one (e.g.
920 arXiv) would have matched. The previous waterfall short-circuited
921 on the first non-null input and never tried the remaining IDs.
923 Uses load_only to skip the ``paper_metadata`` JSON blob on the
924 dedup lookup — we only need the identifier columns. The blob is
925 lazy-loaded if the caller later touches ``paper.paper_metadata``.
926 """
927 id_only = load_only(
928 Paper.id,
929 Paper.doi,
930 Paper.arxiv_id,
931 Paper.pmid,
932 Paper.journal_id,
933 )
935 conditions = []
936 doi = fields.get("doi")
937 arxiv_id = fields.get("arxiv_id")
938 pmid = fields.get("pmid")
939 if doi:
940 conditions.append(Paper.doi == doi)
941 if arxiv_id:
942 conditions.append(Paper.arxiv_id == arxiv_id)
943 if pmid:
944 conditions.append(Paper.pmid == pmid)
946 if not conditions:
947 return None
949 matches = (
950 db_session.query(Paper).options(id_only).filter(or_(*conditions)).all()
951 )
953 if not matches:
954 return None
955 if len(matches) == 1: 955 ↛ 961line 955 didn't jump to line 961 because the condition on line 955 was always true
956 return matches[0]
958 # Multiple distinct rows matched different identifiers of the same
959 # incoming record. This indicates a prior mismerge; deterministic
960 # tie-break on oldest (lowest) id so repeat runs don't oscillate.
961 winner = min(matches, key=lambda p: p.id)
962 logger.warning(
963 f"Paper dedup conflict on {sorted(k for k in ('doi', 'arxiv_id', 'pmid') if fields.get(k))}: "
964 f"{len(matches)} rows (ids {sorted(m.id for m in matches)}); "
965 f"using id {winner.id}. Manual review recommended."
966 )
967 return winner
970def _merge_identifiers(paper: "Paper", indexed: dict, metadata: dict) -> None:
971 """Enrich an existing Paper with identifiers from a new encounter.
973 E.g., an ArXiv paper later found via OpenAlex gains a DOI.
975 Args:
976 paper: The existing Paper row to enrich.
977 indexed: New values for the real columns (doi, arxiv_id,
978 pmid, journal_id). Only applied if the column is
979 currently empty.
980 metadata: Additional bibliographic fields (pmcid, authors,
981 csl_json, etc.) to merge into paper.paper_metadata. Only
982 keys that aren't already present in the existing blob
983 are added — first write wins, to preserve the original
984 enrichment.
985 """
986 # Indexed columns — first write wins. Avoids churning rows when
987 # the same paper turns up across many research sessions with
988 # slightly different scoring / cleaned names.
989 if indexed.get("doi") and not paper.doi:
990 paper.doi = indexed["doi"]
991 if indexed.get("arxiv_id") and not paper.arxiv_id: 991 ↛ 992line 991 didn't jump to line 992 because the condition on line 991 was never true
992 paper.arxiv_id = indexed["arxiv_id"]
993 if indexed.get("pmid") and not paper.pmid: 993 ↛ 994line 993 didn't jump to line 994 because the condition on line 993 was never true
994 paper.pmid = indexed["pmid"]
995 if indexed.get("journal_id") and not paper.journal_id: 995 ↛ 996line 995 didn't jump to line 996 because the condition on line 995 was never true
996 paper.journal_id = indexed["journal_id"]
997 if indexed.get("container_title") and not paper.container_title:
998 paper.container_title = indexed["container_title"]
999 if indexed.get("year") is not None and paper.year is None: 999 ↛ 1000line 999 didn't jump to line 1000 because the condition on line 999 was never true
1000 paper.year = indexed["year"]
1002 # Metadata blob — merge any missing keys.
1003 # IMPORTANT: we must build a NEW dict and reassign the attribute so
1004 # that SQLAlchemy's plain JSON column marks it dirty. In-place
1005 # mutation of the existing dict is not detected without
1006 # MutableDict.as_mutable() — which this column does not use, to
1007 # stay consistent with other JSON columns in the project.
1008 if metadata:
1009 existing = dict(paper.paper_metadata) if paper.paper_metadata else {}
1010 changed = False
1011 for key, value in metadata.items():
1012 if value is not None and key not in existing: 1012 ↛ 1013line 1012 didn't jump to line 1013 because the condition on line 1012 was never true
1013 existing[key] = value
1014 changed = True
1015 if changed: 1015 ↛ 1016line 1015 didn't jump to line 1016 because the condition on line 1015 was never true
1016 paper.paper_metadata = existing or None