Coverage for src/local_deep_research/utilities/search_utilities.py: 97%
179 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-20 01:24 +0000
1import re
2from typing import Dict, List
4from loguru import logger
6from .url_utils import canonical_url_key
9LANGUAGE_CODE_MAP = {
10 "english": "en",
11 "french": "fr",
12 "german": "de",
13 "spanish": "es",
14 "italian": "it",
15 "japanese": "ja",
16 "chinese": "zh",
17 "hindi": "hi",
18 "arabic": "ar",
19 "bengali": "bn",
20 "portuguese": "pt",
21 "russian": "ru",
22 "korean": "ko",
23}
26def remove_think_tags(text: str) -> str:
27 # NOTE: Fresh LLM responses from get_llm() are already <think>-stripped
28 # centrally by ProcessingLLMWrapper (config/llm_config.py). Use this only on
29 # text NOT from a fresh wrapped invoke (accumulated/concatenated text, or
30 # agent/bind_tools output that bypasses the wrapper).
31 # Remove paired <think>...</think> tags
32 text = re.sub(r"<think>.*?</think>", "", text, flags=re.DOTALL)
33 # Remove any orphaned opening or closing think tags
34 text = re.sub(r"</think>", "", text)
35 text = re.sub(r"<think>", "", text)
36 return text.strip()
39# Sentinel values used by the journal reputation filter alongside the
40# numeric 1-10 quality scores. Distinguish structurally different
41# "not scored" cases so the renderer can show the user *why* the tag
42# isn't a numeric quality tier:
43#
44# - QUALITY_PENDING: reference DB hadn't finished building when the
45# search ran (first-search-during-install case).
46# - QUALITY_PREPRINT: result has no journal_ref at all (pure arxiv
47# preprint or similar); there's no venue to score. Distinct from
48# "venue unknown to our catalog" (that becomes score 3, rendered
49# as Unranked).
50QUALITY_PENDING = "pending"
51QUALITY_PREPRINT = "preprint"
54def _format_quality_tag(quality) -> str:
55 """Format a journal quality score as a compact tag for source lists.
57 The output is plaintext / Markdown. **Do NOT** render the containing
58 string through a template filter like ``{{ foo|safe }}`` or
59 ``DOMPurify.sanitize(..., {ALLOWED_TAGS:['a']})`` without first HTML-
60 escaping the surrounding title — the tag itself is safe, but a
61 downstream caller that concatenates ``title + quality_tag`` and
62 emits the result as HTML will leak any tags in ``title`` (XSS).
64 See :func:`_format_quality_tag_html` for the HTML-safe variant.
66 Accepts int | None for scored journals, plus the string sentinels
67 ``QUALITY_PENDING`` and ``QUALITY_PREPRINT``. Every numeric value
68 in VALID_QUALITY_SCORES has its own explicit branch so a bad
69 scoring-logic change can't silently rebucket a score — unexpected
70 values fall through to a debug tag that shows the raw value.
71 """
72 if quality is None:
73 return ""
74 if quality == QUALITY_PENDING:
75 return (
76 " [journal quality data is downloading in the background; "
77 "by the time you open /metrics/journals it may already "
78 "be complete — re-run this search in a minute to get "
79 "real quality scores]"
80 )
81 if quality == QUALITY_PREPRINT:
82 # No venue at all (arxiv preprint / working paper / dataset).
83 # Distinct from score 3 ("we looked and didn't find the
84 # venue") — here there's nothing *to* look up.
85 return " [preprint — not in journal catalog]"
86 # Numeric tiers. Explicit per-score branches instead of ``>=``
87 # ranges so boundary changes can't silently shift a bucket.
88 if quality == 10:
89 return " [Q1 ★★★★★]"
90 # KNOWN-DEFERRED: quality == 9 is a dead branch —
91 # constants.VALID_QUALITY_SCORES excludes 9 and the filter rejects
92 # any LLM output of that value. Kept defensively so a future change
93 # to VALID_QUALITY_SCORES does not require editing the formatter.
94 # Post-merge candidate for removal together with any score-9
95 # reintroduction work.
96 if quality == 9:
97 return " [Q1 ★★★★★]"
98 if quality == 8:
99 return " [Q1 ★★★★]"
100 if quality == 7:
101 return " [Q1 ★★★★]"
102 if quality == 6:
103 return " [Q2 ★★★]"
104 if quality == 5:
105 return " [Q2 ★★★]"
106 if quality == 4:
107 # JOURNAL_QUALITY_DEFAULT — venue found in the catalog but
108 # with no h-index / quartile / DOAJ signal.
109 return " [Unranked ★]"
110 if quality == 3:
111 # Low-confidence fallback — venue didn't match any tier. We
112 # don't know the journal, not "we know it's low-quality".
113 return " [Unranked ★]"
114 if quality == 2: 114 ↛ 115line 114 didn't jump to line 115 because the condition on line 114 was never true
115 return " [Q4 ★]"
116 if quality == 1:
117 # Predatory. Usually auto-removed before this renderer sees
118 # it, but surfaces if whitelisted or the threshold is 1.
119 return " [Q4 ★]"
120 # Out-of-set value — VALID_QUALITY_SCORES gates the inputs so this
121 # is unreachable in normal operation. Show the raw value so bad
122 # data surfaces visibly instead of silently bucketing into Q4.
123 return f" [quality={quality!r}]"
126def _format_quality_tag_html(quality, *, title: str = "") -> str:
127 """HTML-safe wrapper for :func:`_format_quality_tag`.
129 Callers that render search-result titles + quality tags into an
130 HTML page must use this variant and pass the raw ``title`` so both
131 are escaped together. The quality tag itself is plaintext, but the
132 brackets and stars are safe to emit verbatim — the danger is the
133 untrusted ``title`` that a downstream HTML template may concatenate
134 alongside the tag.
136 Returns:
137 ``"{escaped_title}{quality_tag}"`` where ``escaped_title`` is
138 HTML-escaped with ``html.escape(..., quote=True)`` so quotes,
139 angle brackets, and ampersands are rendered as text.
140 """
141 import html as _html
143 return _html.escape(title, quote=True) + _format_quality_tag(quality)
146def extract_links_from_search_results(search_results: List[Dict]) -> List[Dict]:
147 """
148 Extracts links and titles from a list of search result dictionaries.
150 Each dictionary is expected to have at least the keys "title" and "link".
152 Returns a list of dictionaries with 'title' and 'url' keys.
153 """
154 links = []
155 if not search_results:
156 return links
158 for result in search_results:
159 try:
160 # Ensure we handle None values safely before calling strip()
161 title = result.get("title", "")
162 url = result.get("link", "")
163 index = result.get("index", "")
165 # Apply strip() only if the values are not None
166 title = title.strip() if title is not None else ""
167 url = url.strip() if url is not None else ""
168 index = index.strip() if index is not None else ""
170 if title and url:
171 link = {
172 "title": title,
173 "url": url,
174 "index": index,
175 "journal_quality": result.get("journal_quality"),
176 }
177 # Preserve citation-relevant fields from search engines
178 # so they reach the database (previously lost here)
179 for key in (
180 "doi",
181 "authors",
182 "published",
183 "publication_date",
184 "year",
185 "date",
186 "volume",
187 "issue",
188 "pages",
189 "journal_ref",
190 "journal",
191 "venue",
192 "publisher",
193 "source_type",
194 "openalex_source_id",
195 "source",
196 "source_engine",
197 "pmid",
198 "pmcid",
199 "arxiv_id",
200 "isbn",
201 "citations",
202 "is_open_access",
203 "abstract",
204 "metadata",
205 ):
206 val = result.get(key)
207 if val is not None:
208 link[key] = val
209 links.append(link)
210 except Exception:
211 # Log the specific error for debugging
212 logger.exception("Error extracting link from result")
213 continue
214 return links
217def format_links_to_markdown(all_links: List[Dict]) -> str:
218 parts: list[str] = []
219 logger.info(f"Formatting {len(all_links)} links to markdown...")
221 if all_links:
222 # Group links by canonical URL (collapses trailing slash, utm
223 # params, fragments, default ports, scheme/host case, userinfo).
224 # The canonical form is also what gets displayed so the Sources
225 # section stays clean — no utm_*/fbclid clutter, no embedded
226 # credentials, no scheme/host casing noise. Click-through is
227 # unaffected (tracking params carry no content).
228 url_to_indices: dict[str, list] = {}
229 canon_to_title: dict[str, str] = {}
230 canon_to_quality: dict[str, int] = {}
231 # Track the RAG/library collection name per canonical URL so the
232 # citation formatter's source-tagged mode can surface it as the
233 # citation tag (e.g. `[mypapers-7]`) instead of falling back to
234 # the generic `local` label.
235 canon_to_collection: dict[str, str] = {}
236 for link in all_links:
237 raw = link.get("url") or link.get("link") or ""
238 canon = canonical_url_key(raw)
239 if not canon:
240 continue
241 url_to_indices.setdefault(canon, []).append(link.get("index", ""))
242 canon_to_title.setdefault(canon, link.get("title", "Untitled"))
243 # Track journal quality per canonical URL (first non-None wins)
244 if canon not in canon_to_quality and link.get("journal_quality"): 244 ↛ 245line 244 didn't jump to line 245 because the condition on line 244 was never true
245 canon_to_quality[canon] = link["journal_quality"]
246 # First non-empty collection name wins (mirrors title/quality).
247 if canon not in canon_to_collection:
248 metadata = link.get("metadata") or {}
249 collection = metadata.get("collection_name")
250 if collection:
251 canon_to_collection[canon] = str(collection)
253 # Emit each unique source once, in first-seen order.
254 seen: set[str] = set()
255 for link in all_links:
256 raw = link.get("url") or link.get("link") or ""
257 canon = canonical_url_key(raw)
258 if not canon or canon in seen:
259 continue
260 title = canon_to_title[canon]
261 # Indices arrive as int (from strategy enumeration) or str (from
262 # _build_sources_markdown's fallback). Coerce so dedup collapses
263 # 1 and "1", and sorted() doesn't TypeError on mixed types.
264 indices = sorted(
265 {str(i) for i in url_to_indices[canon]},
266 key=lambda s: (0, int(s)) if s.isdigit() else (1, s),
267 )
268 indices_str = f"[{', '.join(indices)}]"
269 quality_tag = _format_quality_tag(canon_to_quality.get(canon))
270 collection_line = (
271 f" Collection: {canon_to_collection[canon]}\n"
272 if canon in canon_to_collection
273 else ""
274 )
275 parts.append(
276 f"{indices_str} {title}{quality_tag} "
277 f"(source nr: {', '.join(map(str, indices))})\n"
278 f" URL: {canon}\n"
279 f"{collection_line}"
280 f"\n"
281 )
282 seen.add(canon)
284 parts.append("\n")
286 return "".join(parts)
289def format_findings(
290 findings_list: List[Dict],
291 synthesized_content: str,
292 questions_by_iteration: Dict[int, List[str]],
293) -> str:
294 """Format findings into a detailed text output.
296 Args:
297 findings_list: List of finding dictionaries
298 synthesized_content: The synthesized content from the LLM.
299 questions_by_iteration: Dictionary mapping iteration numbers to lists of questions
301 Returns:
302 str: Formatted text output
303 """
304 logger.info(
305 f"Inside format_findings utility. Findings count: {len(findings_list)}, Questions iterations: {len(questions_by_iteration)}"
306 )
307 parts: list[str] = []
309 # Extract all sources from findings
310 all_links = []
311 for finding in findings_list:
312 search_results = finding.get("search_results", [])
313 if search_results:
314 try:
315 links = extract_links_from_search_results(search_results)
316 all_links.extend(links)
317 except Exception:
318 logger.exception("Error processing search results/links")
320 # Start with the synthesized content (passed as synthesized_content)
321 parts.append(f"{synthesized_content}\n\n")
323 # Add sources section after synthesized content if sources exist
324 parts.append(format_links_to_markdown(all_links))
326 parts.append("\n\n") # Separator after synthesized content
328 # Add Search Questions by Iteration section
329 if questions_by_iteration:
330 parts.append("## SEARCH QUESTIONS BY ITERATION\n")
331 parts.append("\n")
332 for iter_num, questions in questions_by_iteration.items():
333 parts.append(f"\n #### Iteration {iter_num}:\n")
334 for i, q in enumerate(questions, 1):
335 parts.append(f"{i}. {q}\n")
336 parts.append("\n\n\n")
337 else:
338 logger.warning("No questions by iteration found to format.")
340 # Add Detailed Findings section
341 if findings_list:
342 parts.append("## DETAILED FINDINGS\n\n")
343 logger.info(f"Formatting {len(findings_list)} detailed finding items.")
345 for idx, finding in enumerate(findings_list):
346 logger.debug(
347 f"Formatting finding item {idx}. Keys: {list(finding.keys())}"
348 )
349 # Use .get() for safety
350 phase = finding.get("phase", "Unknown Phase")
351 content = finding.get("content", "No content available.")
352 search_results = finding.get("search_results", [])
354 # Phase header
355 parts.append(f"\n### {phase}\n\n\n")
357 question_displayed = False
358 # If this is a follow-up phase, try to show the corresponding question
359 if isinstance(phase, str) and phase.startswith("Follow-up"):
360 try:
361 phase_parts = phase.replace(
362 "Follow-up Iteration ", ""
363 ).split(".")
364 if len(phase_parts) == 2:
365 iteration = int(phase_parts[0])
366 question_index = int(phase_parts[1]) - 1
367 if (
368 iteration in questions_by_iteration
369 and 0
370 <= question_index
371 < len(questions_by_iteration[iteration])
372 ):
373 parts.append(
374 f"#### {questions_by_iteration[iteration][question_index]}\n\n"
375 )
376 question_displayed = True
377 else:
378 logger.warning(
379 f"Could not find matching question for phase: {phase}"
380 )
381 else:
382 logger.warning(
383 f"Could not parse iteration/index from phase: {phase}"
384 )
385 except ValueError:
386 logger.warning(
387 f"Could not parse iteration/index from phase: {phase}"
388 )
389 # Handle Sub-query phases from IterDRAG strategy
390 elif isinstance(phase, str) and phase.startswith("Sub-query"):
391 try:
392 # Extract the index number from "Sub-query X"
393 query_index = int(phase.replace("Sub-query ", "")) - 1
394 # In IterDRAG, sub-queries are stored in iteration 0
395 if 0 in questions_by_iteration and query_index < len(
396 questions_by_iteration[0]
397 ):
398 parts.append(
399 f"#### {questions_by_iteration[0][query_index]}\n\n"
400 )
401 question_displayed = True
402 else:
403 logger.warning(
404 f"Could not find matching question for phase: {phase}"
405 )
406 except ValueError:
407 logger.warning(
408 f"Could not parse question index from phase: {phase}"
409 )
411 # If the question is in the finding itself, display it
412 if (
413 not question_displayed
414 and "question" in finding
415 and finding["question"]
416 ):
417 parts.append(f"### SEARCH QUESTION:\n{finding['question']}\n\n")
419 # Content
420 parts.append(f"\n\n{content}\n\n")
422 # Search results if they exist
423 if search_results:
424 try:
425 links = extract_links_from_search_results(search_results)
426 if links:
427 parts.append("### SOURCES USED IN THIS SECTION:\n")
428 parts.append(format_links_to_markdown(links) + "\n\n")
429 except Exception:
430 logger.exception(
431 f"Error processing search results/links for finding {idx}"
432 )
433 else:
434 logger.debug(f"No search_results found for finding item {idx}.")
436 parts.append(f"{'_' * 80}\n\n")
437 else:
438 logger.warning("No detailed findings found to format.")
440 # Add summary of all sources at the end
441 if all_links:
442 parts.append("## ALL SOURCES:\n")
443 parts.append(format_links_to_markdown(all_links))
444 else:
445 logger.info("No unique sources found across all findings to list.")
447 logger.info("Finished format_findings utility.")
448 return "".join(parts)