Coverage for src/local_deep_research/utilities/search_utilities.py: 97%
234 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +0000
1import re
2from typing import Dict, List
4from loguru import logger
6from local_deep_research.text_optimization.citation_formatter import (
7 is_line_breaking_char,
8 LDR_APPENDED_SOURCES_SENTINEL,
9)
10from .url_utils import (
11 CHUNK_DISPLAY_KEY,
12 canonical_url_key,
13 library_display_url,
14 preferred_chunk_display,
15)
18LANGUAGE_CODE_MAP = {
19 "english": "en",
20 "french": "fr",
21 "german": "de",
22 "spanish": "es",
23 "italian": "it",
24 "japanese": "ja",
25 "chinese": "zh",
26 "hindi": "hi",
27 "arabic": "ar",
28 "bengali": "bn",
29 "portuguese": "pt",
30 "russian": "ru",
31 "korean": "ko",
32}
35def remove_think_tags(text: str) -> str:
36 # NOTE: Fresh LLM responses from get_llm() are already <think>-stripped
37 # centrally by ProcessingLLMWrapper (config/llm_config.py). Use this only on
38 # text NOT from a fresh wrapped invoke (accumulated/concatenated text, or
39 # agent/bind_tools output that bypasses the wrapper).
40 # Remove paired <think>...</think> tags
41 text = re.sub(r"<think>.*?</think>", "", text, flags=re.DOTALL)
42 # Remove any orphaned opening or closing think tags
43 text = re.sub(r"</think>", "", text)
44 text = re.sub(r"<think>", "", text)
45 return text.strip()
48# Sentinel values used by the journal reputation filter alongside the
49# numeric 1-10 quality scores. Distinguish structurally different
50# "not scored" cases so the renderer can show the user *why* the tag
51# isn't a numeric quality tier:
52#
53# - QUALITY_PENDING: reference DB hadn't finished building when the
54# search ran (first-search-during-install case).
55# - QUALITY_PREPRINT: result has no journal_ref at all (pure arxiv
56# preprint or similar); there's no venue to score. Distinct from
57# "venue unknown to our catalog" (that becomes score 3, rendered
58# as Unranked).
59QUALITY_PENDING = "pending"
60QUALITY_PREPRINT = "preprint"
63def _format_quality_tag(quality) -> str:
64 """Format a journal quality score as a compact tag for source lists.
66 The output is plaintext / Markdown. **Do NOT** render the containing
67 string through a template filter like ``{{ foo|safe }}`` or
68 ``DOMPurify.sanitize(..., {ALLOWED_TAGS:['a']})`` without first HTML-
69 escaping the surrounding title — the tag itself is safe, but a
70 downstream caller that concatenates ``title + quality_tag`` and
71 emits the result as HTML will leak any tags in ``title`` (XSS).
73 See :func:`_format_quality_tag_html` for the HTML-safe variant.
75 Accepts int | None for scored journals, plus the string sentinels
76 ``QUALITY_PENDING`` and ``QUALITY_PREPRINT``. Every numeric value
77 in VALID_QUALITY_SCORES has its own explicit branch so a bad
78 scoring-logic change can't silently rebucket a score — unexpected
79 values fall through to a debug tag that shows the raw value.
80 """
81 if quality is None:
82 return ""
83 if quality == QUALITY_PENDING:
84 return (
85 " [journal quality data is downloading in the background; "
86 "by the time you open /metrics/journals it may already "
87 "be complete — re-run this search in a minute to get "
88 "real quality scores]"
89 )
90 if quality == QUALITY_PREPRINT:
91 # No venue at all (arxiv preprint / working paper / dataset).
92 # Distinct from score 3 ("we looked and didn't find the
93 # venue") — here there's nothing *to* look up.
94 return " [preprint — not in journal catalog]"
95 # Numeric tiers. Explicit per-score branches instead of ``>=``
96 # ranges so boundary changes can't silently shift a bucket.
97 if quality == 10:
98 return " [Q1 ★★★★★]"
99 # KNOWN-DEFERRED: quality == 9 is a dead branch —
100 # constants.VALID_QUALITY_SCORES excludes 9 and the filter rejects
101 # any LLM output of that value. Kept defensively so a future change
102 # to VALID_QUALITY_SCORES does not require editing the formatter.
103 # Post-merge candidate for removal together with any score-9
104 # reintroduction work.
105 if quality == 9:
106 return " [Q1 ★★★★★]"
107 if quality == 8:
108 return " [Q1 ★★★★]"
109 if quality == 7:
110 return " [Q1 ★★★★]"
111 if quality == 6:
112 return " [Q2 ★★★]"
113 if quality == 5:
114 return " [Q2 ★★★]"
115 if quality == 4:
116 # JOURNAL_QUALITY_DEFAULT — venue found in the catalog but
117 # with no h-index / quartile / DOAJ signal.
118 return " [Unranked ★]"
119 if quality == 3:
120 # Low-confidence fallback — venue didn't match any tier. We
121 # don't know the journal, not "we know it's low-quality".
122 return " [Unranked ★]"
123 if quality == 2: 123 ↛ 124line 123 didn't jump to line 124 because the condition on line 123 was never true
124 return " [Q4 ★]"
125 if quality == 1:
126 # Predatory. Usually auto-removed before this renderer sees
127 # it, but surfaces if whitelisted or the threshold is 1.
128 return " [Q4 ★]"
129 # Out-of-set value — VALID_QUALITY_SCORES gates the inputs so this
130 # is unreachable in normal operation. Show the raw value so bad
131 # data surfaces visibly instead of silently bucketing into Q4.
132 return f" [quality={quality!r}]"
135def _format_quality_tag_html(quality, *, title: str = "") -> str:
136 """HTML-safe wrapper for :func:`_format_quality_tag`.
138 Callers that render search-result titles + quality tags into an
139 HTML page must use this variant and pass the raw ``title`` so both
140 are escaped together. The quality tag itself is plaintext, but the
141 brackets and stars are safe to emit verbatim — the danger is the
142 untrusted ``title`` that a downstream HTML template may concatenate
143 alongside the tag.
145 Returns:
146 ``"{escaped_title}{quality_tag}"`` where ``escaped_title`` is
147 HTML-escaped with ``html.escape(..., quote=True)`` so quotes,
148 angle brackets, and ampersands are rendered as text.
149 """
150 import html as _html
152 return _html.escape(title, quote=True) + _format_quality_tag(quality)
155def extract_links_from_search_results(search_results: List[Dict]) -> List[Dict]:
156 """
157 Extracts links and titles from a list of search result dictionaries.
159 Each dictionary is expected to have at least the keys "title" and "link".
161 Returns a list of dictionaries with 'title' and 'url' keys.
162 """
163 links = []
164 if not search_results:
165 return links
167 for result in search_results:
168 try:
169 # Ensure we handle None values safely before calling strip()
170 title = result.get("title", "")
171 url = result.get("link", "")
172 index = result.get("index", "")
174 # Apply strip() only if the values are not None
175 title = title.strip() if title is not None else ""
176 url = url.strip() if url is not None else ""
177 index = index.strip() if index is not None else ""
179 if title and url:
180 link = {
181 "title": title,
182 "url": url,
183 "index": index,
184 "journal_quality": result.get("journal_quality"),
185 }
186 # Preserve citation-relevant fields from search engines
187 # so they reach the database (previously lost here)
188 for key in (
189 "doi",
190 "authors",
191 "published",
192 "publication_date",
193 "year",
194 "date",
195 "volume",
196 "issue",
197 "pages",
198 "journal_ref",
199 "journal",
200 "venue",
201 "publisher",
202 "source_type",
203 "openalex_source_id",
204 "source",
205 "source_engine",
206 "pmid",
207 "pmcid",
208 "arxiv_id",
209 "isbn",
210 "citations",
211 "is_open_access",
212 "abstract",
213 "metadata",
214 ):
215 val = result.get(key)
216 if val is not None:
217 link[key] = val
218 links.append(link)
219 except Exception:
220 # Log the specific error for debugging
221 logger.exception("Error extracting link from result")
222 continue
223 return links
226def _sanitize_sources_field(value: str) -> str:
227 """Flatten a value being rendered into the Sources block.
229 Titles come from search-result metadata — i.e. from whatever a page
230 calls itself — and were previously rendered verbatim, so a crafted
231 title could forge a whole extra numbered citation pointing anywhere.
232 URLs get the same treatment because the non-library canonical-key
233 fallback returns arbitrary scheme-less paths unchanged.
235 Control characters are replaced with a space rather than dropped, so a
236 forged ``\n[9] Fake`` degrades to visible text on the same line instead
237 of silently vanishing.
238 """
239 if not value:
240 return value
241 return "".join(
242 " " if is_line_breaking_char(ch) else ch for ch in value
243 ).strip()
246def _owned_chunk_display(recorded: object, canon: str) -> str | None:
247 """Return a recorded chunk anchor only if it names *this* citation.
249 .. note::
250 This check is the PRIMARY control, not defence in depth. Only
251 ``SearchResultsCollector`` strips a producer-supplied key at
252 ingest, and it exists solely in the LangGraph strategy —
253 ``source_based``, ``focused_iteration`` and
254 ``topic_organization`` all extend ``all_links_of_system`` with
255 raw engine dicts. So on those paths a key present here is
256 producer-supplied by construction, and the residual is bounded to
257 what this function permits: a wrong chunk or view segment of the
258 correctly-identified document, never a foreign document or an
259 arbitrary URL.
261 Shape validation alone is not enough. The value is preferred over the
262 entry's own url, so a well-formed anchor for a DIFFERENT document —
263 which arrives for free, since ``add_results`` copies the engine's dict
264 — would render and persist under this citation. The writer refuses to
265 record a foreign anchor; the reader has to refuse to read one, or the
266 two disagree about which spelling is authoritative.
268 Comparing canonical keys is exact here: ``canonical_url_key`` collapses
269 every view of a library document onto one key, so an anchor for the
270 same document matches and an anchor for any other does not.
271 """
272 if not isinstance(recorded, str):
273 return None
274 display = preferred_chunk_display(recorded)
275 if display is None: 275 ↛ 276line 275 didn't jump to line 276 because the condition on line 275 was never true
276 return None
277 return display if canonical_url_key(display) == canon else None
280def source_url_field(link: Dict) -> object:
281 """The field a source's identity is read from: ``link`` first.
283 ``SearchResultsCollector`` keys citations on
284 ``_citation_dedup_key(result["link"])``, and records the authoritative
285 chunk anchor or normalized URL in ``link``. A result carrying BOTH
286 fields with divergent values (e.g. ``link`` carrying the collector's
287 rebuilt anchor while ``url`` retains an anchor-less engine string) must
288 consistently read ``link`` so the entry groups under its intended chunk
289 anchor rather than fanning out into an unintended second group.
291 ``link`` wins because it is what the collector keys on, what
292 ``_format_results`` shows the agent next to the ``[N]`` marker, and
293 what ``find_by_url`` resolves — i.e. it is what the citation index
294 actually means. ``url`` remains the fallback for the raw engine dicts
295 the non-LangGraph strategies extend ``all_links_of_system`` with,
296 some of which set only that key.
297 """
298 return link.get("link") or link.get("url") or ""
301def count_distinct_sources(all_links: List[Dict]) -> int:
302 """How many distinct sources a report cites (document-level count).
304 ``len(all_links_of_system)`` counts OCCURRENCES, not sources: the
305 LangGraph collector stores one entry per ``(url, snippet)`` pair, and
306 ``source_based`` / ``focused_iteration`` / ``topic_organization``
307 extend the list with raw engine dicts and no URL dedup at all. Both
308 shapes put one source in the list several times, so every "N sources"
309 number must group at the document level rather than take a length.
311 Counts by canonical URL key, so distinct library documents are counted
312 once regardless of how many distinct chunks or views are cited.
313 The relation across reporting layers is:
314 distinct sources <= bibliography lines <= citation indices.
315 """
316 seen: set[str] = set()
317 for link in all_links or []:
318 if not isinstance(link, dict):
319 continue
320 raw = source_url_field(link)
321 if not isinstance(raw, str):
322 continue
323 canon = canonical_url_key(raw)
324 if canon:
325 seen.add(canon)
326 return len(seen)
329def format_links_to_markdown(all_links: List[Dict]) -> str:
330 parts: list[str] = []
331 logger.info(f"Formatting {len(all_links)} links to markdown...")
333 if all_links:
334 # Group links by canonical URL (collapses trailing slash, utm
335 # params, fragments, default ports, scheme/host case, userinfo).
336 # The canonical form is also what gets displayed so the Sources
337 # section stays clean — no utm_*/fbclid clutter, no embedded
338 # credentials, no scheme/host casing noise. Click-through is
339 # unaffected (tracking params carry no content).
340 # Group links by (canonical URL, chunk display URL).
341 # Canonical key remains per-document so count_distinct_sources,
342 # MCP sources, and news metrics do not inflate, while grouping on
343 # chunk anchor ensures distinct cited chunks (/chunks#chunk-N) each
344 # render their own Sources entry with their own anchor. Unanchored
345 # views of the same document (e.g. /pdf and base route) merge together;
346 # an unanchored /pdf view deliberately renders on its own line when
347 # cited alongside chunks (reversing #5685's collapse so anchorless
348 # indices honestly link to /pdf rather than mis-pointing at an arbitrary chunk).
349 # Non-library sources display their canonical URL, so tracking params
350 # and credentials stay out of the report.
351 url_to_indices: dict[tuple[str, str], list] = {}
352 group_to_title: dict[tuple[str, str], str] = {}
353 group_to_quality: dict[tuple[str, str], int] = {}
354 group_to_collection: dict[tuple[str, str], str] = {}
355 group_to_display: dict[tuple[str, str], str] = {}
356 for link in all_links:
357 raw = source_url_field(link)
358 # Skipped, not coerced. These dicts reach here straight from
359 # engine output on the non-LangGraph strategies, and
360 # canonical_url_key raises on a non-str. Stringifying instead
361 # renders a Python repr as a clickable URL and can merge two
362 # distinct sources onto one citation — and it would disagree
363 # with ``_citation_dedup_key``, which refuses a non-str link
364 # outright and whose docstring requires the two to group
365 # identically.
366 if not isinstance(raw, str):
367 continue
368 canon = canonical_url_key(raw)
369 if not canon:
370 continue
371 chunk_disp = preferred_chunk_display(raw) or _owned_chunk_display(
372 link.get(CHUNK_DISPLAY_KEY), canon
373 )
374 if chunk_disp:
375 key = (canon, chunk_disp)
376 disp = chunk_disp
377 else:
378 key = (canon, "")
379 disp = library_display_url(raw) or canon
381 url_to_indices.setdefault(key, []).append(link.get("index", ""))
382 group_to_title.setdefault(key, link.get("title", "Untitled"))
383 # Prefer /pdf over bare /library/document/<id> for unanchored display
384 curr_disp = group_to_display.get(key)
385 if curr_disp is None or (
386 curr_disp.endswith(canon) and "/pdf" in disp
387 ):
388 group_to_display[key] = disp
390 # Track journal quality per group (first non-None wins)
391 if key not in group_to_quality and link.get("journal_quality"): 391 ↛ 392line 391 didn't jump to line 392 because the condition on line 391 was never true
392 group_to_quality[key] = link["journal_quality"]
393 # First non-empty collection name wins (mirrors title/quality).
394 # Note: per-group collection tracking is display-only today and
395 # anticipates #5722 encoding collection identity into chunk anchors.
396 if key not in group_to_collection:
397 metadata = link.get("metadata") or {}
398 if not isinstance(metadata, dict):
399 metadata = {}
400 collection = metadata.get("collection_name")
401 if collection:
402 group_to_collection[key] = str(collection)
404 # Emit each unique source once, in first-seen order.
405 seen: set[tuple[str, str]] = set()
406 for link in all_links:
407 raw = source_url_field(link)
408 if not isinstance(raw, str):
409 continue
410 canon = canonical_url_key(raw)
411 if not canon:
412 continue
413 chunk_disp = preferred_chunk_display(raw) or _owned_chunk_display(
414 link.get(CHUNK_DISPLAY_KEY), canon
415 )
416 key = (canon, chunk_disp) if chunk_disp else (canon, "")
417 if key in seen:
418 continue
419 title = group_to_title[key]
420 # Coerced for the same reason the url is skipped: it comes
421 # from the same engine dict, and ``.replace`` below raises on
422 # a non-str, taking the whole Sources block with it.
423 if title is not None and not isinstance(title, str):
424 title = str(title)
425 if title:
426 title = _sanitize_sources_field(
427 title.replace(LDR_APPENDED_SOURCES_SENTINEL, "")
428 )
429 # Indices arrive as int (from strategy enumeration) or str (from
430 # _build_sources_markdown's fallback). Coerce so dedup collapses
431 # 1 and "1", and sorted() doesn't TypeError on mixed types.
432 indices = sorted(
433 {str(i) for i in url_to_indices[key]},
434 # ``isdigit()`` is True for non-ASCII digit characters that
435 # ``int()`` rejects — e.g. the superscript "\u00b9" — so it
436 # alone would raise ValueError here and crash the whole
437 # bibliography. Require ASCII before converting.
438 key=lambda s: (
439 (0, int(s)) if s.isascii() and s.isdigit() else (1, s)
440 ),
441 )
442 # Sanitised like the title, collection and URL. Indices are
443 # LDR-internal enumerations today, but ``_create_documents``
444 # PRESERVES a pre-existing ``index`` on a raw engine dict and
445 # ``_build_sources_markdown`` reads it back out of persisted
446 # ``original_data`` — so "no engine emits this key" is a
447 # coincidence, not a control.
448 #
449 # ``quality_tag`` below is the one field still interpolated
450 # raw. It reaches the renderer by the same persisted route,
451 # and is safe only incidentally: the fall-through branch that
452 # builds it uses ``repr()``, which escapes newlines and
453 # non-printables. Incidental, not designed — do not remove
454 # that ``repr()`` without sanitising here instead.
455 indices = [_sanitize_sources_field(i) for i in indices]
456 indices_str = f"[{', '.join(indices)}]"
457 quality_tag = _format_quality_tag(group_to_quality.get(key))
458 collection = group_to_collection.get(key, "")
459 if collection:
460 collection = _sanitize_sources_field(
461 collection.replace(LDR_APPENDED_SOURCES_SENTINEL, "")
462 )
463 collection_line = (
464 f" Collection: {collection}\n" if collection else ""
465 )
466 display = group_to_display[key]
467 parts.append(
468 f"{indices_str} {title}{quality_tag} "
469 f"(source nr: {', '.join(map(str, indices))})\n"
470 f" URL: {_sanitize_sources_field(display)}\n"
471 f"{collection_line}"
472 f"\n"
473 )
474 seen.add(key)
476 parts.append("\n")
478 return "".join(parts)
481def format_findings(
482 findings_list: List[Dict],
483 synthesized_content: str,
484 questions_by_iteration: Dict[int, List[str]],
485) -> str:
486 """Format findings into a detailed text output.
488 Args:
489 findings_list: List of finding dictionaries
490 synthesized_content: The synthesized content from the LLM.
491 questions_by_iteration: Dictionary mapping iteration numbers to lists of questions
493 Returns:
494 str: Formatted text output
495 """
496 logger.info(
497 f"Inside format_findings utility. Findings count: {len(findings_list)}, Questions iterations: {len(questions_by_iteration)}"
498 )
499 parts: list[str] = []
501 # Extract all sources from findings
502 all_links = []
503 for finding in findings_list:
504 search_results = finding.get("search_results", [])
505 if search_results:
506 try:
507 links = extract_links_from_search_results(search_results)
508 all_links.extend(links)
509 except Exception:
510 logger.exception("Error processing search results/links")
512 # Start with the synthesized content (passed as synthesized_content)
513 parts.append(f"{synthesized_content}\n\n")
515 # Add sources section after synthesized content if sources exist
516 parts.append(format_links_to_markdown(all_links))
518 parts.append("\n\n") # Separator after synthesized content
520 # Add Search Questions by Iteration section
521 if questions_by_iteration:
522 parts.append("## SEARCH QUESTIONS BY ITERATION\n")
523 parts.append("\n")
524 for iter_num, questions in questions_by_iteration.items():
525 parts.append(f"\n #### Iteration {iter_num}:\n")
526 for i, q in enumerate(questions, 1):
527 parts.append(f"{i}. {q}\n")
528 parts.append("\n\n\n")
529 else:
530 logger.warning("No questions by iteration found to format.")
532 # Add Detailed Findings section
533 if findings_list:
534 parts.append("## DETAILED FINDINGS\n\n")
535 logger.info(f"Formatting {len(findings_list)} detailed finding items.")
537 for idx, finding in enumerate(findings_list):
538 logger.debug(
539 f"Formatting finding item {idx}. Keys: {list(finding.keys())}"
540 )
541 # Use .get() for safety
542 phase = finding.get("phase", "Unknown Phase")
543 content = finding.get("content", "No content available.")
544 search_results = finding.get("search_results", [])
546 # Phase header
547 parts.append(f"\n### {phase}\n\n\n")
549 question_displayed = False
550 # If this is a follow-up phase, try to show the corresponding question
551 if isinstance(phase, str) and phase.startswith("Follow-up"):
552 try:
553 phase_parts = phase.replace(
554 "Follow-up Iteration ", ""
555 ).split(".")
556 if len(phase_parts) == 2:
557 iteration = int(phase_parts[0])
558 question_index = int(phase_parts[1]) - 1
559 if (
560 iteration in questions_by_iteration
561 and 0
562 <= question_index
563 < len(questions_by_iteration[iteration])
564 ):
565 parts.append(
566 f"#### {questions_by_iteration[iteration][question_index]}\n\n"
567 )
568 question_displayed = True
569 else:
570 logger.warning(
571 f"Could not find matching question for phase: {phase}"
572 )
573 else:
574 logger.warning(
575 f"Could not parse iteration/index from phase: {phase}"
576 )
577 except ValueError:
578 logger.warning(
579 f"Could not parse iteration/index from phase: {phase}"
580 )
581 # Handle Sub-query phases from IterDRAG strategy
582 elif isinstance(phase, str) and phase.startswith("Sub-query"):
583 try:
584 # Extract the index number from "Sub-query X"
585 query_index = int(phase.replace("Sub-query ", "")) - 1
586 # In IterDRAG, sub-queries are stored in iteration 0
587 if 0 in questions_by_iteration and query_index < len(
588 questions_by_iteration[0]
589 ):
590 parts.append(
591 f"#### {questions_by_iteration[0][query_index]}\n\n"
592 )
593 question_displayed = True
594 else:
595 logger.warning(
596 f"Could not find matching question for phase: {phase}"
597 )
598 except ValueError:
599 logger.warning(
600 f"Could not parse question index from phase: {phase}"
601 )
603 # If the question is in the finding itself, display it
604 if (
605 not question_displayed
606 and "question" in finding
607 and finding["question"]
608 ):
609 parts.append(f"### SEARCH QUESTION:\n{finding['question']}\n\n")
611 # Content
612 parts.append(f"\n\n{content}\n\n")
614 # Search results if they exist
615 if search_results:
616 try:
617 links = extract_links_from_search_results(search_results)
618 if links:
619 parts.append("### SOURCES USED IN THIS SECTION:\n")
620 parts.append(format_links_to_markdown(links) + "\n\n")
621 except Exception:
622 logger.exception(
623 f"Error processing search results/links for finding {idx}"
624 )
625 else:
626 logger.debug(f"No search_results found for finding item {idx}.")
628 parts.append(f"{'_' * 80}\n\n")
629 else:
630 logger.warning("No detailed findings found to format.")
632 # Add summary of all sources at the end
633 if all_links:
634 parts.append("## ALL SOURCES:\n")
635 parts.append(format_links_to_markdown(all_links))
636 else:
637 logger.info("No unique sources found across all findings to list.")
639 logger.info("Finished format_findings utility.")
640 return "".join(parts)