Coverage for src/local_deep_research/text_optimization/citation_formatter.py: 97%
502 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-20 01:24 +0000
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-20 01:24 +0000
1"""Citation formatter for adding hyperlinks and alternative citation styles."""
3import re
4from enum import Enum
5from typing import Any, Dict, List, Optional, Tuple
6from urllib.parse import urlparse
8from slugify import slugify
10from ..content_fetcher.url_classifier import URLClassifier, URLType
13_SOURCES_SECTION_PATTERNS = [
14 re.compile(
15 r"^#{1,3}\s*(?:Sources|References|Bibliography|Citations)",
16 re.MULTILINE | re.IGNORECASE,
17 ),
18 re.compile(
19 r"^(?:Sources|References|Bibliography|Citations):?\s*$",
20 re.MULTILINE | re.IGNORECASE,
21 ),
22]
25def find_sources_section(content: str) -> int:
26 """Find the start position of the sources/references section in *content*.
28 Returns -1 if no section is found.
29 """
30 for pattern in _SOURCES_SECTION_PATTERNS:
31 match = pattern.search(content)
32 if match:
33 return match.start()
34 return -1
37class CitationMode(Enum):
38 """Available citation formatting modes."""
40 NUMBER_HYPERLINKS = "number_hyperlinks" # [1] with hyperlinks
41 DOMAIN_HYPERLINKS = "domain_hyperlinks" # [arxiv.org] with hyperlinks
42 DOMAIN_ID_HYPERLINKS = (
43 "domain_id_hyperlinks" # [arxiv.org] or [arxiv.org-1] with smart IDs
44 )
45 DOMAIN_ID_ALWAYS_HYPERLINKS = (
46 "domain_id_always_hyperlinks" # [arxiv.org-1] always with IDs
47 )
48 SOURCE_TAGGED_HYPERLINKS = "source_tagged_hyperlinks"
49 """Preserve the global citation number and prefix it with a short source
50 tag derived from the URL: known academic sources via ``URLClassifier``
51 (``arxiv-7``, ``pubmed-3``), domain otherwise (``nytimes.com-9``), and
52 ``local-N`` for empty / local URLs. Unlike DOMAIN_ID_* modes the
53 suffix is the original citation number, so labels never collide and
54 match the bibliography order: ``[1]`` arxiv + ``[2]`` openai + ``[3]``
55 arxiv -> ``arxiv-1``, ``openai-2``, ``arxiv-3``."""
56 NO_HYPERLINKS = "no_hyperlinks" # [1] without hyperlinks
59class CitationFormatter:
60 """Formats citations in markdown documents with various styles."""
62 def __init__(self, mode: CitationMode = CitationMode.NUMBER_HYPERLINKS):
63 self.mode = mode
64 # Use negative lookbehind and lookahead to avoid matching already formatted citations
65 # Also match Unicode lenticular brackets 【】 (U+3010 and U+3011) that LLMs sometimes generate
66 self.citation_pattern = re.compile(
67 r"(?<![\[【])[\[【](\d+)[\]】](?![\]】])"
68 )
69 self.comma_citation_pattern = re.compile(
70 r"[\[【](\d+(?:,\s*\d+)+)[\]】]"
71 )
72 # Also match "Source X" or "source X" patterns
73 self.source_word_pattern = re.compile(r"\b[Ss]ource\s+(\d+)\b")
74 self.sources_pattern = re.compile(
75 r"^\[(\d+(?:,\s*\d+)*)\]\s*(.+?)(?:\n\s*URL:\s*(.+?))?$",
76 re.MULTILINE,
77 )
79 def _create_source_word_replacer(self, formatter_func):
80 """Create a replacement function for 'Source X' patterns.
82 Args:
83 formatter_func: A function that takes citation_num and returns formatted text
85 Returns:
86 A replacement function for use with regex sub
87 """
89 def replace_source_word(match):
90 citation_num = match.group(1)
91 return formatter_func(citation_num)
93 return replace_source_word
95 def _create_citation_formatter(self, sources_dict, format_pattern):
96 """Create a formatter function for citations.
98 Args:
99 sources_dict: Dictionary mapping citation numbers to data
100 format_pattern: A callable that takes (citation_num, data) and returns formatted string
102 Returns:
103 A function that formats citations or returns fallback
104 """
106 def formatter(citation_num):
107 if citation_num in sources_dict:
108 data = sources_dict[citation_num]
109 return format_pattern(citation_num, data)
110 return f"[{citation_num}]"
112 return formatter
114 def _replace_comma_citations(self, content, lookup, format_one):
115 """Replace comma-separated citations like [1, 2, 3] using *lookup* and *format_one*.
117 Args:
118 content: Text to process
119 lookup: Dict mapping citation number (str) to data
120 format_one: ``(num, data) -> str`` callback that formats a single citation
121 """
123 def _replacer(match):
124 nums = [n.strip() for n in match.group(1).split(",")]
125 parts = []
126 for num in nums:
127 if num in lookup:
128 parts.append(format_one(num, lookup[num]))
129 else:
130 parts.append(f"[{num}]")
131 return "".join(parts)
133 return self.comma_citation_pattern.sub(_replacer, content)
135 def format_document(self, content: str) -> str:
136 """Format citations and return the concatenated answer + sources blob.
138 Kept for backward compatibility — most call sites only need the
139 concatenated string. New code that needs to persist answer-only
140 should use :meth:`format_document_split` instead so the boundary
141 is returned explicitly (no re-parsing of the concatenated output).
142 """
143 formatted_answer, sources_md = self.format_document_split(content)
144 return formatted_answer + sources_md
146 def format_document_split(self, content: str) -> Tuple[str, str]:
147 """Format citations and return (answer, sources_md) separately.
149 The boundary between the LLM's answer and the trailing Sources
150 section is computed inside this method. Callers that only want
151 the answer (e.g. the chat-mode save site) get a clean split
152 without re-applying a regex on concatenated output downstream.
154 Returns ``(content, "")`` when the formatter is in NO_HYPERLINKS
155 mode or when no Sources section can be found in ``content``.
156 """
157 if self.mode == CitationMode.NO_HYPERLINKS:
158 return content, ""
160 sources_start = self._find_sources_section(content)
161 if sources_start == -1:
162 return content, ""
164 document_content = content[:sources_start]
165 sources_content = content[sources_start:]
167 sources = self._parse_sources(sources_content)
169 if self.mode == CitationMode.NUMBER_HYPERLINKS:
170 formatted_content = self._format_number_hyperlinks(
171 document_content, sources
172 )
173 elif self.mode == CitationMode.DOMAIN_HYPERLINKS:
174 formatted_content = self._format_domain_hyperlinks(
175 document_content, sources
176 )
177 elif self.mode == CitationMode.DOMAIN_ID_HYPERLINKS:
178 formatted_content = self._format_domain_id_hyperlinks(
179 document_content, sources
180 )
181 elif self.mode == CitationMode.DOMAIN_ID_ALWAYS_HYPERLINKS:
182 formatted_content = self._format_domain_id_always_hyperlinks(
183 document_content, sources
184 )
185 elif self.mode == CitationMode.SOURCE_TAGGED_HYPERLINKS: 185 ↛ 192line 185 didn't jump to line 192 because the condition on line 185 was always true
186 formatted_content = self._format_source_tagged_hyperlinks(
187 document_content,
188 sources,
189 self._parse_collections(sources_content),
190 )
191 else:
192 formatted_content = document_content
194 return formatted_content, sources_content
196 def apply_inline_hyperlinks(
197 self, content: str, sources: List[Dict[str, Any]]
198 ) -> str:
199 """Hyperlink ``[N]`` refs using a structured source list.
201 Dispatches on ``self.mode`` so the user's chosen citation
202 format (Settings → Report → Citation Format) is honored on
203 the fallback path the same way it is in
204 :meth:`format_document_split`. Inherits all the existing
205 per-mode guards (lookbehind/lookahead against ``[[1]]``,
206 comma-list handling like ``[1,2,3]``, ``Source N`` word form,
207 missing-index pass-through, lenticular bracket support).
209 Used as the safe fallback at save time when the LLM does NOT
210 emit a Sources section in its prose — the structured source
211 list (e.g. ``search_system.all_links_of_system``) is the
212 canonical source of URLs and indices.
213 """
214 if not content or not sources:
215 return content or ""
216 if self.mode == CitationMode.NO_HYPERLINKS:
217 return content
219 # Search-engine result dicts use either "url" or "link" for the
220 # destination — Searxng emits {"link": ..., "title": ..., "snippet": ...}
221 # (search_engine_searxng.py:538) and other engines use "url".
222 # Looking up only `s["url"]` silently dropped every Searxng-sourced
223 # citation, leaving the answer body with plain `[N]` brackets even
224 # though the Sources section beneath was fully populated. Accept
225 # both keys so the hyperlink fallback works regardless of engine.
226 def _src_url(s):
227 return s.get("url") or s.get("link") or ""
229 adapted: Dict[str, Tuple[str, str]] = {
230 str(s["index"]): (s.get("title", "Untitled"), _src_url(s))
231 for s in sources
232 if _src_url(s) and s.get("index") is not None
233 }
234 if not adapted:
235 return content
237 # Per-mode dispatch — mirrors format_document_split so the user's
238 # chosen citation format applies on this fallback path too.
239 # Previously this was hard-coded to _format_number_hyperlinks,
240 # which meant chat-mode answers (which always hit this fallback
241 # because the langgraph-agent synthesis doesn't emit a ## Sources
242 # block in its prose) ignored the report.citation_format setting
243 # entirely — every chat answer came out as [[N]](url) even when
244 # the user picked domain-based or source-tagged formatting.
245 if self.mode == CitationMode.DOMAIN_HYPERLINKS:
246 return self._format_domain_hyperlinks(content, adapted)
247 if self.mode == CitationMode.DOMAIN_ID_HYPERLINKS:
248 return self._format_domain_id_hyperlinks(content, adapted)
249 if self.mode == CitationMode.DOMAIN_ID_ALWAYS_HYPERLINKS: 249 ↛ 250line 249 didn't jump to line 250 because the condition on line 249 was never true
250 return self._format_domain_id_always_hyperlinks(content, adapted)
251 if self.mode == CitationMode.SOURCE_TAGGED_HYPERLINKS:
252 # Pull collection names off the structured source dicts
253 # (format_links_to_markdown uses the same shape:
254 # link["metadata"]["collection_name"]) so the SOURCE_TAGGED
255 # formatter can surface library/RAG tags as the citation
256 # label when present.
257 collections: Dict[str, str] = {}
258 for s in sources:
259 idx = s.get("index")
260 if idx is None: 260 ↛ 261line 260 didn't jump to line 261 because the condition on line 260 was never true
261 continue
262 meta = s.get("metadata") or {}
263 coll = meta.get("collection_name")
264 if coll:
265 collections.setdefault(str(idx), str(coll))
266 return self._format_source_tagged_hyperlinks(
267 content, adapted, collections
268 )
269 # NUMBER_HYPERLINKS is the default and the catch-all for any
270 # mode added later that doesn't have an explicit branch above.
271 return self._format_number_hyperlinks(content, adapted)
273 def _find_sources_section(self, content: str) -> int:
274 """Find the start of the sources/references section."""
275 return find_sources_section(content)
277 def _parse_sources(
278 self, sources_content: str
279 ) -> Dict[str, Tuple[str, str]]:
280 """
281 Parse sources section to extract citation numbers, titles, and URLs.
283 Returns:
284 Dictionary mapping citation number to (title, url) tuple
285 """
286 sources = {}
287 matches = list(self.sources_pattern.finditer(sources_content))
289 for match in matches:
290 citation_nums_str = match.group(1)
291 title = match.group(2).strip()
292 url = match.group(3).strip() if match.group(3) else ""
294 # Handle comma-separated citation numbers like [36, 3]
295 # Split by comma and strip whitespace
296 individual_nums = [
297 num.strip() for num in citation_nums_str.split(",")
298 ]
300 # Add an entry for each individual number
301 for num in individual_nums:
302 sources[num] = (title, url)
304 return sources
306 def _format_number_hyperlinks(
307 self, content: str, sources: Dict[str, Tuple[str, str]]
308 ) -> str:
309 """Replace [1] with hyperlinked version where only the number is linked."""
310 # Filter sources that have URLs
311 url_sources = {
312 num: (title, url) for num, (title, url) in sources.items() if url
313 }
315 # Create formatter for citations with number hyperlinks
316 def format_number_link(citation_num, data):
317 _, url = data
318 return f"[[{citation_num}]]({url})"
320 # Handle comma-separated citations like [1, 2, 3]
321 content = self._replace_comma_citations(
322 content, url_sources, format_number_link
323 )
325 formatter = self._create_citation_formatter(
326 url_sources, format_number_link
327 )
329 # Handle individual citations
330 def replace_citation(match):
331 return (
332 formatter(match.group(1))
333 if match.group(1) in url_sources
334 else match.group(0)
335 )
337 content = self.citation_pattern.sub(replace_citation, content)
339 # Also handle "Source X" patterns
340 return self.source_word_pattern.sub(
341 self._create_source_word_replacer(formatter), content
342 )
344 def _format_domain_hyperlinks(
345 self, content: str, sources: Dict[str, Tuple[str, str]]
346 ) -> str:
347 """Replace [1] with [domain.com] hyperlinked version."""
349 # Filter sources that have URLs
350 url_sources = {
351 num: (title, url) for num, (title, url) in sources.items() if url
352 }
354 # Pre-compute labels so [1, 2, 3] lists and standalone [N] refs
355 # use the same string for a given citation. Previously this
356 # closure called _extract_domain on every match, so a relative
357 # URL (``/library/document/...``) produced ``[[]]`` for every
358 # occurrence. _citation_label falls back to a slugified title
359 # for relative URLs, keeping the label non-empty and readable.
360 label_cache: Dict[str, str] = {
361 num: self._citation_label(num, title, url)
362 for num, (title, url) in url_sources.items()
363 }
365 # Create formatter for citations with domain hyperlinks
366 def format_domain_link(citation_num, data):
367 _, url = data
368 label = label_cache[citation_num]
369 return f"[[{label}]]({url})"
371 # Handle comma-separated citations like [1, 2, 3]
372 content = self._replace_comma_citations(
373 content, url_sources, format_domain_link
374 )
376 formatter = self._create_citation_formatter(
377 url_sources, format_domain_link
378 )
380 # Handle individual citations
381 def replace_citation(match):
382 return (
383 formatter(match.group(1))
384 if match.group(1) in url_sources
385 else match.group(0)
386 )
388 content = self.citation_pattern.sub(replace_citation, content)
390 # Also handle "Source X" patterns
391 return self.source_word_pattern.sub(
392 self._create_source_word_replacer(formatter), content
393 )
395 def _format_domain_id_hyperlinks(
396 self, content: str, sources: Dict[str, Tuple[str, str]]
397 ) -> str:
398 """Replace [1] with [domain.com-1] hyperlinked version with hyphen-separated IDs."""
399 # First, create a mapping of domains to their citation numbers.
400 # _citation_label (vs raw _extract_domain) means relative URLs
401 # like ``/library/document/...`` get a slugified title as the
402 # "domain" instead of an empty string. Without this, a RAG
403 # result with one citation rendered as ``[[]](url)`` (the
404 # label disappeared), and a multi-citation set rendered as
405 # ``[[-1]](url)`` (the orphaned ``-N`` suffix leaked through).
406 domain_citations: dict[str, list[Any]] = {}
408 for citation_num, (title, url) in sources.items():
409 if url: 409 ↛ 408line 409 didn't jump to line 408 because the condition on line 409 was always true
410 domain = self._citation_label(citation_num, title, url)
411 if domain not in domain_citations:
412 domain_citations[domain] = []
413 domain_citations[domain].append((citation_num, url))
415 # Create a mapping from citation number to domain with ID
416 citation_to_domain_id = {}
417 for domain, citations in domain_citations.items():
418 if len(citations) > 1:
419 # Multiple citations from same domain - add hyphen and number
420 for idx, (citation_num, url) in enumerate(citations, 1):
421 citation_to_domain_id[citation_num] = (
422 f"{domain}-{idx}",
423 url,
424 )
425 else:
426 # Single citation from domain - no ID needed
427 citation_num, url = citations[0]
428 citation_to_domain_id[citation_num] = (domain, url)
430 # Create formatter for citations with domain_id hyperlinks
431 def format_domain_id_link(citation_num, data):
432 domain_id, url = data
433 return f"[[{domain_id}]]({url})"
435 # Handle comma-separated citations
436 content = self._replace_comma_citations(
437 content, citation_to_domain_id, format_domain_id_link
438 )
440 formatter = self._create_citation_formatter(
441 citation_to_domain_id, format_domain_id_link
442 )
444 # Handle individual citations
445 def replace_citation(match):
446 return (
447 formatter(match.group(1))
448 if match.group(1) in citation_to_domain_id
449 else match.group(0)
450 )
452 content = self.citation_pattern.sub(replace_citation, content)
454 # Also handle "Source X" patterns
455 return self.source_word_pattern.sub(
456 self._create_source_word_replacer(formatter), content
457 )
459 def _format_domain_id_always_hyperlinks(
460 self, content: str, sources: Dict[str, Tuple[str, str]]
461 ) -> str:
462 """Replace [1] with [domain.com-1] hyperlinked version, always with IDs."""
463 # Use _citation_label so relative URLs (e.g. RAG /library/document/...
464 # results) produce a slugified-title prefix instead of an empty
465 # string. Without this, every label here collapses to ``-N``
466 # (e.g. ``[[-1]](url)``) and the user has no idea what the link
467 # references. See _format_domain_id_hyperlinks for the same fix.
468 domain_citations: dict[str, list[Any]] = {}
470 for citation_num, (title, url) in sources.items():
471 if url: 471 ↛ 470line 471 didn't jump to line 470 because the condition on line 471 was always true
472 domain = self._citation_label(citation_num, title, url)
473 if domain not in domain_citations:
474 domain_citations[domain] = []
475 domain_citations[domain].append((citation_num, url))
477 # Create a mapping from citation number to domain with ID
478 citation_to_domain_id = {}
479 for domain, citations in domain_citations.items():
480 # Always add hyphen and number for consistency
481 for idx, (citation_num, url) in enumerate(citations, 1):
482 citation_to_domain_id[citation_num] = (f"{domain}-{idx}", url)
484 # Create formatter for citations with domain_id hyperlinks
485 def format_domain_id_link(citation_num, data):
486 domain_id, url = data
487 return f"[[{domain_id}]]({url})"
489 # Handle comma-separated citations
490 content = self._replace_comma_citations(
491 content, citation_to_domain_id, format_domain_id_link
492 )
494 formatter = self._create_citation_formatter(
495 citation_to_domain_id, format_domain_id_link
496 )
498 # Handle individual citations
499 def replace_citation(match):
500 return (
501 formatter(match.group(1))
502 if match.group(1) in citation_to_domain_id
503 else match.group(0)
504 )
506 content = self.citation_pattern.sub(replace_citation, content)
508 # Also handle "Source X" patterns
509 return self.source_word_pattern.sub(
510 self._create_source_word_replacer(formatter), content
511 )
513 # Sources section may carry a "Collection: <name>" line for RAG /
514 # library hits (emitted by ``utilities/search_utilities.format_links_to_markdown``).
515 # The line sits between this ``[N]`` entry's ``URL:`` line and the
516 # next ``[N+1]`` entry. We anchor the match on a non-greedy span up
517 # to the next citation header (or end of string) to scope correctly.
518 _collection_line_pattern = re.compile(
519 r"^\[(\d+(?:,\s*\d+)*)\][^\n]*\n" # the [N] header line
520 r"(?:[^\n\[]*\n)*?" # any non-[ lines (typically URL: ...)
521 r"\s*Collection:\s*(.+?)\s*$",
522 re.MULTILINE,
523 )
525 def _parse_collections(self, sources_content: str) -> Dict[str, str]:
526 """Extract ``{citation_num: collection_name}`` from a sources
527 block. Returns an empty dict when no ``Collection:`` lines exist
528 — the absence of collection info is the common case (web URLs)
529 and must never raise."""
530 collections: Dict[str, str] = {}
531 for match in self._collection_line_pattern.finditer(sources_content):
532 citation_nums_str = match.group(1)
533 collection = match.group(2).strip()
534 if not collection: 534 ↛ 535line 534 didn't jump to line 535 because the condition on line 534 was never true
535 continue
536 for num in (n.strip() for n in citation_nums_str.split(",")):
537 collections[num] = collection
538 return collections
540 def _format_source_tagged_hyperlinks(
541 self,
542 content: str,
543 sources: Dict[str, Tuple[str, str]],
544 collections: Dict[str, str],
545 ) -> str:
546 """Replace ``[N]`` with ``[[source-N]](url)``.
548 ``source`` resolves to (in order): the RAG ``Collection:``
549 tag for library hits, the short URLClassifier tag for known
550 academic sources (``arxiv``, ``pubmed``, ...), the cleaned
551 domain otherwise, or ``local`` for empty/file URLs. ``N`` is
552 the original global citation number — labels never collide and
553 the suffix always matches the bibliography ordering.
555 Args:
556 content: Document body (sources section already split off).
557 sources: ``{citation_num: (title, url)}`` parsed from the
558 sources block.
559 collections: ``{citation_num: collection_name}`` parsed from
560 optional ``Collection:`` lines in the sources block
561 (empty dict when no library/RAG hits are cited). Wins
562 over URL-derived tags when present for a given citation.
563 """
565 # Pre-compute tags so each URL-having source resolves its label
566 # once, however many [N] / [1, 2, 3] refs cite it.
567 tag_cache: Dict[str, str] = {
568 num: self._extract_source_label(
569 url, collection=collections.get(num)
570 )
571 for num, (_, url) in sources.items()
572 if url
573 }
575 def format_link(citation_num, data):
576 _, url = data
577 label = tag_cache.get(citation_num)
578 if label is None:
579 # Only URL-less sources are absent from tag_cache. Don't
580 # pass this as dict.get()'s default — that would evaluate
581 # the expensive lookup eagerly on every cache hit too.
582 label = self._extract_source_label(
583 url, collection=collections.get(citation_num)
584 )
585 tag = f"{label}-{citation_num}"
586 # Only emit a hyperlink for http(s) URLs — local/file URLs are
587 # rendered as plain bracketed tags so the markdown stays clean
588 # and viewers don't try to navigate to a server-local path.
589 return (
590 f"[[{tag}]]({url})"
591 if self._is_linkable_url(url)
592 else f"[{tag}]"
593 )
595 # Handle comma-separated citations like [1, 2, 3]
596 content = self._replace_comma_citations(content, sources, format_link)
598 formatter = self._create_citation_formatter(sources, format_link)
600 # Handle individual citations
601 def replace_citation(match):
602 return (
603 formatter(match.group(1))
604 if match.group(1) in sources
605 else match.group(0)
606 )
608 content = self.citation_pattern.sub(replace_citation, content)
610 # Also handle "Source X" patterns
611 return self.source_word_pattern.sub(
612 self._create_source_word_replacer(formatter), content
613 )
615 @staticmethod
616 def _slugify_collection(name: str) -> str:
617 """Make a user-set collection name safe for inline citations.
619 Collection names are free-form strings (``"My Papers"``,
620 ``"team/finance"``). Citations need a compact token that won't
621 break markdown — strip whitespace, lowercase, replace runs of
622 non-alphanumeric chars with a single hyphen, trim leading and
623 trailing hyphens, and fall back to ``"local"`` if the result is
624 empty. ``-N`` is appended downstream so we strip trailing
625 hyphens to keep the join clean.
626 """
627 slug = re.sub(r"[^a-z0-9]+", "-", name.strip().lower()).strip("-")
628 return slug or "local"
630 @staticmethod
631 def _slugify_title(title: str, max_length: int = 32) -> Optional[str]:
632 """Generate a slug from a document title using python-slugify.
634 Returns ``None`` when ``title`` is empty / whitespace / pure
635 punctuation, or when ``slugify`` produces an empty result. The
636 ``None`` sentinel lets callers distinguish "no meaningful slug"
637 from "slug happens to be a real word" (e.g. a document literally
638 titled ``"Doc"`` must yield ``"doc"`` as its label, not fall
639 through to the citation-number fallback).
640 """
641 if not title:
642 return None
643 slug = slugify(title, max_length=max_length, separator="-")
644 return slug if slug else None
646 def _citation_label(self, citation_num: str, title: str, url: str) -> str:
647 """Return the inline-citation label for a source.
649 Order of preference:
650 1. The URL's domain (``arxiv.org``, ``example.com``) — used by
651 web citations. Falls through when the URL is relative or
652 empty, as is common for RAG / library documents.
653 2. A slugified version of the document title (truncated to
654 keep labels compact) — gives users a readable label for
655 local-library hits.
656 3. The citation number itself — last-resort guarantee the
657 label is never empty (an empty label produces ``[[]]``
658 which renders as an uninformative ``[]`` link).
659 """
660 domain = self._extract_domain(url) if url else ""
661 if domain:
662 return domain
663 slug = self._slugify_title(title)
664 if slug:
665 return slug
666 return str(citation_num)
668 @staticmethod
669 def _is_linkable_url(url: str) -> bool:
670 """Return True iff ``url`` is a http(s) URL safe to wrap in a
671 markdown hyperlink. Empty strings and file:// / local: schemes
672 are not linkable."""
673 if not url:
674 return False
675 try:
676 scheme = (urlparse(url).scheme or "").lower()
677 except (ValueError, AttributeError):
678 return False
679 return scheme in ("http", "https")
681 def _extract_source_label(
682 self, url: str, collection: str | None = None
683 ) -> str:
684 """Return a short source tag for ``url``.
686 Resolution order:
687 1. ``collection`` (when supplied) wins outright — RAG / library
688 hits surface their collection name as the citation tag
689 (``mypapers``, ``personal-notes``, ...). The renderer in
690 ``utilities/search_utilities.format_links_to_markdown``
691 emits a ``Collection:`` line per source for library results,
692 which the formatter parses back into this argument.
693 2. Empty URL or non-http(s) scheme (``file://``, ``local:``, ...) →
694 ``"local"``. Uniform fallback when no collection name is
695 available.
696 3. ``URLClassifier`` matches a known academic source → use the
697 enum value (``arxiv``, ``pubmed``, ``pmc``, ``biorxiv``,
698 ``medrxiv``, ``semantic_scholar``, ``doi``).
699 4. Otherwise → fall back to ``_extract_domain`` (e.g.
700 ``arxiv.org``, ``nytimes.com``).
701 """
702 if collection:
703 return self._slugify_collection(collection)
704 if not url:
705 return "local"
706 try:
707 parsed = urlparse(url)
708 except (ValueError, AttributeError):
709 return "local"
710 scheme = (parsed.scheme or "").lower()
711 if scheme not in ("http", "https"):
712 return "local"
714 url_type = URLClassifier.classify(url)
715 # Generic HTML/PDF/INVALID → fall back to domain. Everything else
716 # is a known academic source whose enum value is the short tag.
717 if url_type in (URLType.HTML, URLType.PDF, URLType.INVALID):
718 return self._extract_domain(url)
719 return str(url_type.value)
721 def _extract_domain(self, url: str) -> str:
722 """Extract domain name from URL."""
723 try:
724 parsed = urlparse(url)
725 domain = parsed.netloc
726 # Remove www. prefix if present
727 if domain.startswith("www."):
728 domain = domain[4:]
729 # Keep known domains as-is
730 known_domains = {
731 "arxiv.org": "arxiv.org",
732 "github.com": "github.com",
733 "reddit.com": "reddit.com",
734 "youtube.com": "youtube.com",
735 "pypi.org": "pypi.org",
736 "milvus.io": "milvus.io",
737 "medium.com": "medium.com",
738 }
740 for known, display in known_domains.items():
741 if known in domain:
742 return display
744 # For other domains, extract main domain
745 parts = domain.split(".")
746 if len(parts) >= 2:
747 return ".".join(parts[-2:])
748 return domain
749 except (ValueError, AttributeError):
750 return "source"
753class QuartoExporter:
754 """Export markdown documents to Quarto (.qmd) format."""
756 def __init__(self):
757 # Also match Unicode lenticular brackets 【】 (U+3010 and U+3011) that LLMs sometimes generate
758 self.citation_pattern = re.compile(
759 r"(?<![\[【])[\[【](\d+)[\]】](?![\]】])"
760 )
761 self.comma_citation_pattern = re.compile(
762 r"[\[【](\d+(?:,\s*\d+)+)[\]】]"
763 )
765 def export_to_quarto(self, content: str, title: str | None = None) -> str:
766 """
767 Convert markdown document to Quarto format.
769 Args:
770 content: Markdown content
771 title: Document title (if None, will extract from content)
773 Returns:
774 Quarto formatted content
775 """
776 # Extract title from markdown if not provided
777 if not title:
778 title_match = re.search(r"^#\s+(.+)$", content, re.MULTILINE)
779 title = title_match.group(1) if title_match else "Research Report"
781 # Create Quarto YAML header
782 from datetime import UTC, datetime
784 current_date = datetime.now(UTC).strftime("%Y-%m-%d")
785 yaml_header = f"""---
786title: "{title}"
787author: "Local Deep Research"
788date: "{current_date}"
789format:
790 html:
791 toc: true
792 toc-depth: 3
793 number-sections: true
794 pdf:
795 toc: true
796 number-sections: true
797 colorlinks: true
798bibliography: references.bib
799csl: apa.csl
800---
802"""
804 # Process content
805 processed_content = content
807 # First handle comma-separated citations like [1, 2, 3]
808 def replace_comma_citations(match):
809 citation_nums = match.group(1)
810 # Split by comma and strip whitespace
811 nums = [num.strip() for num in citation_nums.split(",")]
812 refs = [f"@ref{num}" for num in nums]
813 return f"[{', '.join(refs)}]"
815 processed_content = self.comma_citation_pattern.sub(
816 replace_comma_citations, processed_content
817 )
819 # Then convert individual citations to Quarto format [@citation]
820 def replace_citation(match):
821 citation_num = match.group(1)
822 return f"[@ref{citation_num}]"
824 processed_content = self.citation_pattern.sub(
825 replace_citation, processed_content
826 )
828 # Generate bibliography file content
829 bib_content = self._generate_bibliography(content)
831 # Add note about bibliography file
832 bibliography_note = (
833 "\n\n::: {.callout-note}\n## Bibliography File Required\n\nThis document requires a `references.bib` file in the same directory with the following content:\n\n```bibtex\n"
834 + bib_content
835 + "\n```\n:::\n"
836 )
838 return yaml_header + processed_content + bibliography_note
840 def _generate_bibliography(self, content: str) -> str:
841 """Generate BibTeX bibliography from sources."""
842 sources_pattern = re.compile(
843 r"^\[(\d+)\]\s*(.+?)(?:\n\s*URL:\s*(.+?))?$", re.MULTILINE
844 )
846 bibliography = ""
847 matches = list(sources_pattern.finditer(content))
849 for match in matches:
850 citation_num = match.group(1)
851 title = match.group(2).strip()
852 url = match.group(3).strip() if match.group(3) else ""
854 # Generate BibTeX entry
855 bib_entry = f"@misc{{ref{citation_num},\n"
856 bib_entry += f' title = "{{{title}}}",\n'
857 if url:
858 bib_entry += f" url = {{{url}}},\n"
859 bib_entry += f' howpublished = "\\url{{{url}}}",\n'
860 bib_entry += f" year = {{{2024}}},\n"
861 bib_entry += ' note = "Accessed: \\today"\n'
862 bib_entry += "}\n"
864 bibliography += bib_entry + "\n"
866 return bibliography.strip()
869class RISExporter:
870 """Export references to RIS format for reference managers like Zotero."""
872 def __init__(self):
873 self.sources_pattern = re.compile(
874 r"^\[(\d+(?:,\s*\d+)*)\]\s*(.+?)(?:\n\s*URL:\s*(.+?))?$",
875 re.MULTILINE,
876 )
878 def export_to_ris(self, content: str) -> str:
879 """
880 Extract references from markdown and convert to RIS format.
882 Args:
883 content: Markdown content with sources
885 Returns:
886 RIS formatted references
887 """
888 # Find sources section
889 sources_start = find_sources_section(content)
890 if sources_start == -1:
891 return ""
893 # Find the end of the first sources section (before any other major section)
894 sources_content = content[sources_start:]
896 # Look for the next major section to avoid duplicates
897 next_section_markers = [
898 "\n## ALL SOURCES",
899 "\n### ALL SOURCES",
900 "\n## Research Metrics",
901 "\n### Research Metrics",
902 "\n## SEARCH QUESTIONS",
903 "\n### SEARCH QUESTIONS",
904 "\n## DETAILED FINDINGS",
905 "\n### DETAILED FINDINGS",
906 "\n---", # Horizontal rule often separates sections
907 ]
909 sources_end = len(sources_content)
910 for marker in next_section_markers:
911 pos = sources_content.find(marker)
912 if pos != -1 and pos < sources_end:
913 sources_end = pos
915 sources_content = sources_content[:sources_end]
917 # Parse sources and generate RIS entries
918 ris_entries = []
919 seen_refs = set() # Track which references we've already processed
921 # Split sources into individual entries
922 import re
924 # Pattern to match each source entry. Accept both ASCII "[N]" and
925 # lenticular "【N】" openers/closers — the inline citation patterns
926 # in this file already handle lenticular brackets (some LLMs emit
927 # them), so the source-list parser must stay consistent or it would
928 # silently drop lenticular-bracketed source entries.
929 source_entry_pattern = re.compile(
930 r"^[\[【](\d+)[\]】]\s*(.+?)(?=^[\[【]\d+[\]】]|\Z)",
931 re.MULTILINE | re.DOTALL,
932 )
934 for match in source_entry_pattern.finditer(sources_content):
935 citation_num = match.group(1)
936 entry_text = match.group(2).strip()
938 # Extract the title (first line)
939 lines = entry_text.split("\n")
940 title = lines[0].strip()
942 # Extract URL, DOI, and other metadata from subsequent lines
943 url = ""
944 metadata = {}
945 for line in lines[1:]:
946 line = line.strip()
947 if line.startswith("URL:"):
948 url = line[4:].strip()
949 elif line.startswith("DOI:"):
950 metadata["doi"] = line[4:].strip()
951 elif line.startswith("Published in"):
952 metadata["journal"] = line[12:].strip()
953 # Add more metadata parsing as needed
954 elif line: 954 ↛ 945line 954 didn't jump to line 945 because the condition on line 954 was always true
955 # Store other lines as additional metadata
956 if "additional" not in metadata: 956 ↛ 958line 956 didn't jump to line 958 because the condition on line 956 was always true
957 metadata["additional"] = []
958 additional = metadata["additional"]
959 if isinstance(additional, list): 959 ↛ 945line 959 didn't jump to line 945 because the condition on line 959 was always true
960 additional.append(line)
962 # Combine title with additional metadata lines for full context
963 full_text = entry_text
965 # Create a unique key to avoid duplicates
966 ref_key = (citation_num, title, url)
967 if ref_key not in seen_refs: 967 ↛ 934line 967 didn't jump to line 934 because the condition on line 967 was always true
968 seen_refs.add(ref_key)
969 # Create RIS entry with full text for metadata extraction
970 ris_entry = self._create_ris_entry(
971 citation_num, full_text, url, metadata
972 )
973 ris_entries.append(ris_entry)
975 return "\n".join(ris_entries)
977 def _create_ris_entry(
978 self,
979 ref_id: str,
980 full_text: str,
981 url: str = "",
982 metadata: dict | None = None,
983 ) -> str:
984 """Create a single RIS entry."""
985 lines = []
987 # Parse metadata from full text
988 import re
990 if metadata is None:
991 metadata = {}
993 # Extract title from first line. NB: split into a *separate* variable —
994 # ``lines`` is the RIS-output accumulator initialized above and appended
995 # to below; reusing it here previously overwrote it with the source
996 # text, so every entry emitted the raw source body before the mandatory
997 # leading ``TY - `` tag and reference managers rejected the file.
998 text_lines = full_text.split("\n")
999 title = text_lines[0].strip()
1001 # Extract year from full text (looks for 4-digit year)
1002 year_match = re.search(r"\b(19\d{2}|20\d{2})\b", full_text)
1003 year = year_match.group(1) if year_match else None
1005 # Extract authors if present (looks for "by Author1, Author2")
1006 authors_match = re.search(
1007 r"\bby\s+([^.\n]+?)(?:\.|\n|$)", full_text, re.IGNORECASE
1008 )
1009 authors = []
1010 if authors_match:
1011 authors_text = authors_match.group(1)
1012 # Split by 'and' or ','
1013 author_parts = re.split(r"\s*(?:,|\sand\s|&)\s*", authors_text)
1014 authors = [a.strip() for a in author_parts if a.strip()]
1016 # Extract DOI from metadata or text
1017 doi = metadata.get("doi")
1018 if not doi:
1019 doi_match = re.search(
1020 r"DOI:\s*([^\s\n]+)", full_text, re.IGNORECASE
1021 )
1022 doi = doi_match.group(1) if doi_match else None
1024 # Clean title - remove author and metadata info for cleaner title
1025 clean_title = title
1026 if authors_match and authors_match.start() < len(title):
1027 clean_title = (
1028 title[: authors_match.start()] + title[authors_match.end() :]
1029 if authors_match.end() < len(title)
1030 else title[: authors_match.start()]
1031 )
1032 clean_title = re.sub(
1033 r"\s*DOI:\s*[^\s]+", "", clean_title, flags=re.IGNORECASE
1034 )
1035 clean_title = re.sub(
1036 r"\s*Published in.*", "", clean_title, flags=re.IGNORECASE
1037 )
1038 clean_title = re.sub(
1039 r"\s*Volume.*", "", clean_title, flags=re.IGNORECASE
1040 )
1041 clean_title = re.sub(
1042 r"\s*Pages.*", "", clean_title, flags=re.IGNORECASE
1043 )
1044 clean_title = clean_title.strip()
1046 # TY - Type of reference (ELEC for electronic source/website)
1047 lines.append("TY - ELEC")
1049 # ID - Reference ID
1050 lines.append(f"ID - ref{ref_id}")
1052 # TI - Title
1053 lines.append(f"TI - {clean_title if clean_title else title}")
1055 # AU - Authors
1056 for author in authors:
1057 lines.append(f"AU - {author}")
1059 # DO - DOI
1060 if doi:
1061 lines.append(f"DO - {doi}")
1063 # PY - Publication year (if found in title)
1064 if year:
1065 lines.append(f"PY - {year}")
1067 # UR - URL
1068 if url:
1069 lines.append(f"UR - {url}")
1071 # Try to extract domain as publisher
1072 try:
1073 from urllib.parse import urlparse
1075 parsed = urlparse(url)
1076 domain = parsed.netloc
1077 if domain.startswith("www."):
1078 domain = domain[4:]
1079 # Extract readable publisher name from domain
1080 if domain == "github.com" or domain.endswith(".github.com"):
1081 lines.append("PB - GitHub")
1082 elif domain == "arxiv.org" or domain.endswith(".arxiv.org"):
1083 lines.append("PB - arXiv")
1084 elif domain == "reddit.com" or domain.endswith(".reddit.com"):
1085 lines.append("PB - Reddit")
1086 elif (
1087 domain == "youtube.com"
1088 or domain == "m.youtube.com"
1089 or domain.endswith(".youtube.com")
1090 ):
1091 lines.append("PB - YouTube")
1092 elif domain == "medium.com" or domain.endswith(".medium.com"):
1093 lines.append("PB - Medium")
1094 elif domain == "pypi.org" or domain.endswith(".pypi.org"):
1095 lines.append("PB - Python Package Index (PyPI)")
1096 else:
1097 # Use domain as publisher
1098 lines.append(f"PB - {domain}")
1099 except (ValueError, AttributeError):
1100 pass
1102 # Y1 - Year accessed (current year)
1103 from datetime import UTC, datetime
1105 current_year = datetime.now(UTC).year
1106 lines.append(f"Y1 - {current_year}")
1108 # DA - Date accessed
1109 current_date = datetime.now(UTC).strftime("%Y/%m/%d")
1110 lines.append(f"DA - {current_date}")
1112 # LA - Language
1113 lines.append("LA - en")
1115 # ER - End of reference
1116 lines.append("ER - ")
1118 return "\n".join(lines)
1121class LaTeXExporter:
1122 """Export markdown documents to LaTeX format."""
1124 def __init__(self):
1125 # Also match Unicode lenticular brackets 【】 (U+3010 and U+3011) that LLMs sometimes generate
1126 self.citation_pattern = re.compile(r"[\[【](\d+)[\]】]")
1127 self.heading_patterns = [
1128 (re.compile(r"^# (.+)$", re.MULTILINE), r"\\section{\1}"),
1129 (re.compile(r"^## (.+)$", re.MULTILINE), r"\\subsection{\1}"),
1130 (re.compile(r"^### (.+)$", re.MULTILINE), r"\\subsubsection{\1}"),
1131 ]
1132 self.emphasis_patterns = [
1133 (re.compile(r"\*\*(.+?)\*\*"), r"\\textbf{\1}"),
1134 (re.compile(r"\*(.+?)\*"), r"\\textit{\1}"),
1135 (re.compile(r"`(.+?)`"), r"\\texttt{\1}"),
1136 ]
1138 def export_to_latex(self, content: str) -> str:
1139 """
1140 Convert markdown document to LaTeX format.
1142 Args:
1143 content: Markdown content
1145 Returns:
1146 LaTeX formatted content
1147 """
1148 latex_content = self._create_latex_header()
1150 # Convert markdown to LaTeX
1151 body_content = content
1153 # Escape special LaTeX characters but preserve math mode
1154 # Split by $ to preserve math sections
1155 parts = body_content.split("$")
1156 for i in range(len(parts)):
1157 # Even indices are outside math mode
1158 if i % 2 == 0:
1159 # Only escape if not inside $$
1160 if not (
1161 i > 0
1162 and parts[i - 1] == ""
1163 and i < len(parts) - 1
1164 and parts[i + 1] == ""
1165 ):
1166 # Preserve certain patterns that will be processed later
1167 # like headings (#), emphasis (*), and citations ([n])
1168 lines = parts[i].split("\n")
1169 for j, line in enumerate(lines):
1170 # Don't escape lines that start with # (headings)
1171 if not line.strip().startswith("#"):
1172 # Don't escape emphasis markers or citations for now
1173 # They'll be handled by their own patterns
1174 temp_line = line
1175 # Escape special chars except *, #, [, ]
1176 temp_line = temp_line.replace("&", r"\&")
1177 temp_line = temp_line.replace("%", r"\%")
1178 temp_line = temp_line.replace("_", r"\_")
1179 # Don't escape { } inside citations
1180 lines[j] = temp_line
1181 parts[i] = "\n".join(lines)
1182 body_content = "$".join(parts)
1184 # Convert headings
1185 for pattern, replacement in self.heading_patterns:
1186 body_content = pattern.sub(replacement, body_content)
1188 # Convert emphasis
1189 for pattern, replacement in self.emphasis_patterns:
1190 body_content = pattern.sub(replacement, body_content)
1192 # Convert citations to LaTeX \cite{} format
1193 body_content = self.citation_pattern.sub(r"\\cite{\1}", body_content)
1195 # Convert lists
1196 body_content = self._convert_lists(body_content)
1198 # Add body content
1199 latex_content += body_content
1201 # Add bibliography section
1202 latex_content += self._create_bibliography(content)
1204 # Add footer
1205 latex_content += self._create_latex_footer()
1207 return latex_content
1209 def _create_latex_header(self) -> str:
1210 """Create LaTeX document header."""
1211 return r"""\documentclass[12pt]{article}
1212\usepackage[utf8]{inputenc}
1213\usepackage{hyperref}
1214\usepackage{cite}
1215\usepackage{url}
1217\title{Research Report}
1218\date{\today}
1220\begin{document}
1221\maketitle
1223"""
1225 def _create_latex_footer(self) -> str:
1226 """Create LaTeX document footer."""
1227 return "\n\\end{document}\n"
1229 def _escape_latex(self, text: str) -> str:
1230 """Escape special LaTeX characters in text."""
1231 # Escape special LaTeX characters
1232 replacements = [
1233 ("\\", r"\textbackslash{}"), # Must be first
1234 ("&", r"\&"),
1235 ("%", r"\%"),
1236 ("$", r"\$"),
1237 ("#", r"\#"),
1238 ("_", r"\_"),
1239 ("{", r"\{"),
1240 ("}", r"\}"),
1241 ("~", r"\textasciitilde{}"),
1242 ("^", r"\textasciicircum{}"),
1243 ]
1245 for old, new in replacements:
1246 text = text.replace(old, new)
1248 return text
1250 def _convert_lists(self, content: str) -> str:
1251 """Convert markdown lists to LaTeX format."""
1252 # Simple conversion for bullet points
1253 content = re.sub(r"^- (.+)$", r"\\item \1", content, flags=re.MULTILINE)
1255 # Add itemize environment around list items
1256 lines = content.split("\n")
1257 result = []
1258 in_list = False
1260 for line in lines:
1261 if line.strip().startswith("\\item"):
1262 if not in_list:
1263 result.append("\\begin{itemize}")
1264 in_list = True
1265 result.append(line)
1266 else:
1267 if in_list and line.strip():
1268 result.append("\\end{itemize}")
1269 in_list = False
1270 result.append(line)
1272 if in_list:
1273 result.append("\\end{itemize}")
1275 return "\n".join(result)
1277 def _create_bibliography(self, content: str) -> str:
1278 """Extract sources and create LaTeX bibliography."""
1279 sources_start = find_sources_section(content)
1280 if sources_start == -1:
1281 return ""
1283 sources_content = content[sources_start:]
1284 pattern = re.compile(
1285 r"^\[(\d+)\]\s*(.+?)(?:\n\s*URL:\s*(.+?))?$", re.MULTILINE
1286 )
1288 bibliography = "\n\\begin{thebibliography}{99}\n"
1290 for match in pattern.finditer(sources_content):
1291 citation_num = match.group(1)
1292 title = match.group(2).strip()
1293 url = match.group(3).strip() if match.group(3) else ""
1295 # Escape special LaTeX characters in title
1296 escaped_title = self._escape_latex(title)
1298 if url:
1299 bibliography += f"\\bibitem{{{citation_num}}} {escaped_title}. \\url{{{url}}}\n"
1300 else:
1301 bibliography += (
1302 f"\\bibitem{{{citation_num}}} {escaped_title}.\n"
1303 )
1305 bibliography += "\\end{thebibliography}\n"
1307 return bibliography