Coverage for src/local_deep_research/web_search_engines/engines/search_engine_federal_register.py: 92%
273 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1import re
2import time
3from dataclasses import dataclass, field
4from typing import Any, Dict, Final, List, Optional, cast
5from urllib.parse import quote
7import requests
8from bs4 import BeautifulSoup
9from langchain_core.language_models import BaseLLM
10from urllib3.exceptions import LocationParseError
11from urllib3.util import parse_url
13from ...security.safe_requests import safe_get
14from ...security.secure_logging import logger
15from ...security.ssrf_validator import RFC_FORBIDDEN_URL_CHARS_RE
16from ..engine_availability import retry_after_from_error
17from ..rate_limiting import RateLimitError
18from ..search_engine_base import (
19 BaseSearchEngine,
20 Exposure,
21 Sensitivity,
22 _SearchRunState,
23)
26FULL_CONTENT_DEADLINE_SECONDS: Final = 120
27FULL_CONTENT_MAX_CHARS: Final = 50000
30class _FullContentDeadlineReached(Exception):
31 """Internal signal that the full-content budget ran out mid-item.
33 Raised by ``_request_timeout`` and caught by ``_get_full_content``, so
34 every way of running out of time unwinds through a single path that
35 still returns one result per input item.
36 """
39class _UnexpectedResponseHost(Exception):
40 """A list response whose redirect chain ended off Federal Register."""
43@dataclass
44class _FederalRegisterRunState(_SearchRunState):
45 full_content_deadline: float | None = None
46 full_content_fetch_cache: dict[str, tuple[str | None, str | None]] = field(
47 default_factory=dict
48 )
51class FederalRegisterSearchEngine(BaseSearchEngine):
52 """Search rules, proposed rules, notices, and presidential documents."""
54 is_public = True
55 is_generic = False
56 is_scientific = False
57 is_lexical = True
58 needs_llm_relevance_filter = True
59 egress_sensitivity = Sensitivity.NON_SENSITIVE
60 egress_exposure = Exposure.EXPOSING
62 LIST_URL = "https://www.federalregister.gov/api/v1/documents.json"
63 DETAIL_URL = "https://www.federalregister.gov/api/v1/documents"
64 MAX_QUERY_LENGTH = 400
65 MAX_RESULTS = 20
66 REQUEST_TIMEOUT = 30
67 RAW_TEXT_HOSTS = frozenset(
68 {"www.federalregister.gov", "federalregister.gov"}
69 )
70 # The values the API's ``conditions[type][]`` filter accepts: the four
71 # the ``/documents/facets/type`` endpoint lists, plus CORRECT
72 # ("Correction"), UNKNOWN ("Uncategorized Document") and SUNSHINE
73 # ("Sunshine Act Document"), which the filter also recognises (its
74 # reply names the type) though the facet omits them. SUNSHINE is a
75 # legacy type that currently matches no documents. The API answers
76 # any other value (``"rule"``, ``"Rule"``, ``"RULES"``) with HTTP 200
77 # ``{"description": "Documents of type ", "count": 0}`` —
78 # indistinguishable from a query that matched nothing — so an unknown
79 # type must never be sent.
80 VALID_DOCUMENT_TYPES: Final = (
81 "RULE",
82 "PRORULE",
83 "NOTICE",
84 "PRESDOCU",
85 "CORRECT",
86 "UNKNOWN",
87 "SUNSHINE",
88 )
90 def __init__(
91 self,
92 max_results: int = 10,
93 document_types: Optional[List[str]] = None,
94 llm: Optional[BaseLLM] = None,
95 max_filtered_results: Optional[int] = None,
96 settings_snapshot: Optional[Dict[str, Any]] = None,
97 search_snippets_only: bool = True,
98 max_content_chars: int = FULL_CONTENT_MAX_CHARS,
99 **kwargs,
100 ):
101 super().__init__(
102 llm=llm,
103 max_filtered_results=max_filtered_results,
104 max_results=max_results,
105 settings_snapshot=settings_snapshot,
106 search_snippets_only=search_snippets_only,
107 **kwargs,
108 )
109 self.document_types = self._normalize_document_types(document_types)
110 self.max_content_chars = max_content_chars
112 def _new_search_run_state(self) -> _FederalRegisterRunState:
113 return _FederalRegisterRunState()
115 def _get_previews(self, query: str) -> List[Dict[str, Any]]:
116 result_limit = min(self.max_results, self.MAX_RESULTS)
118 try:
119 params: Dict[str, Any] = {
120 "conditions[term]": query[: self.MAX_QUERY_LENGTH],
121 "per_page": result_limit,
122 "order": "newest",
123 }
124 if self.document_types:
125 params["conditions[type][]"] = self.document_types
126 self._last_wait_time = self.rate_tracker.apply_rate_limit(
127 self.engine_type
128 )
129 response = safe_get(
130 self.LIST_URL,
131 params=params,
132 timeout=self.REQUEST_TIMEOUT,
133 )
134 # Checked before the 429: a rate limit (and Retry-After) from
135 # an off-host redirect target says nothing about Federal
136 # Register's own limit, and its body is not a Federal Register
137 # answer either.
138 self._raise_if_list_ended_off_host(response)
139 self._raise_if_rate_limited(response)
140 response.raise_for_status()
141 data = response.json()
142 except RateLimitError:
143 raise
144 except _UnexpectedResponseHost as error:
145 self._search_failed = True
146 self._record_search_failure(error)
147 logger.warning(
148 "Federal Register request ended on an unexpected host"
149 )
150 return []
151 except requests.exceptions.RequestException as error:
152 self._search_failed = True
153 self._record_search_failure(error)
154 status_code = getattr(
155 getattr(error, "response", None), "status_code", None
156 )
157 logger.warning(
158 "Federal Register request failed "
159 f"(status={status_code if status_code is not None else 'unknown'})"
160 )
161 return []
162 except (AttributeError, KeyError, TypeError, ValueError) as error:
163 self._search_failed = True
164 self._record_search_failure(error)
165 logger.warning("Federal Register response parsing failed")
166 return []
168 if not isinstance(data, dict):
169 self._search_failed = True
170 self._record_search_failure(
171 ValueError("Federal Register response was not an object")
172 )
173 logger.warning(
174 "Federal Register response did not contain an object"
175 )
176 return []
177 documents = data.get("results")
178 if documents is None and self._is_zero_match(data):
179 # FederalRegister.gov answers a query that matched nothing
180 # with ``{"description": ..., "count": 0}`` and no
181 # ``results`` key. That is a valid empty result, not a
182 # provider failure: it must neither fail the search in
183 # metrics nor cool the engine down for the rest of the run.
184 return []
185 if not isinstance(documents, list):
186 self._search_failed = True
187 self._record_search_failure(
188 ValueError("Federal Register response had no results list")
189 )
190 logger.warning("Federal Register response did not contain results")
191 return []
193 previews = []
194 for document in documents[:result_limit]:
195 if not isinstance(document, dict): 195 ↛ 196line 195 didn't jump to line 196 because the condition on line 195 was never true
196 continue
197 document_number = document.get("document_number")
198 if not isinstance(document_number, str) or not document_number: 198 ↛ 199line 198 didn't jump to line 199 because the condition on line 198 was never true
199 continue
200 previews.append(
201 {
202 "id": document_number,
203 "title": document.get("title") or "Untitled",
204 "link": document.get("html_url") or "",
205 "snippet": document.get("abstract")
206 or document.get("excerpts")
207 or "",
208 "type": document.get("type") or "",
209 "agencies": document.get("agencies") or [],
210 "publication_date": document.get("publication_date") or "",
211 "pdf_url": document.get("pdf_url") or "",
212 }
213 )
214 return previews
216 def _get_full_content(
217 self, relevant_items: List[Dict[str, Any]]
218 ) -> List[Dict[str, Any]]:
219 state = cast(_FederalRegisterRunState, self._get_search_run_state())
220 deadline = state.full_content_deadline
221 if deadline is None:
222 deadline = time.monotonic() + FULL_CONTENT_DEADLINE_SECONDS
223 state.full_content_deadline = deadline
225 fetch_cache = state.full_content_fetch_cache
227 pending = list(relevant_items)
228 results: List[Dict[str, Any]] = []
229 for position, item in enumerate(pending):
230 result = self._preview_fields(item)
231 document_number = result.get("id")
232 if not isinstance(document_number, str) or not document_number: 232 ↛ 233line 232 didn't jump to line 233 because the condition on line 232 was never true
233 results.append(result)
234 continue
236 cached_fetch = fetch_cache.get(document_number)
237 if cached_fetch is not None and cached_fetch[1] is not None:
238 cached_content = cached_fetch[1]
239 result["full_content"] = cached_content
240 results.append(result)
241 continue
243 if time.monotonic() > deadline:
244 results.append(result)
245 results.extend(
246 self._drain_without_fetching(
247 pending[position + 1 :], fetch_cache
248 )
249 )
250 return results
252 try:
253 if cached_fetch is None: 253 ↛ 297line 253 didn't jump to line 297 because the condition on line 253 was always true
254 self._last_wait_time = self.rate_tracker.apply_rate_limit(
255 self.engine_type
256 )
257 detail_response = safe_get(
258 f"{self.DETAIL_URL}/{quote(document_number, safe='')}.json",
259 timeout=self._request_timeout(deadline),
260 )
261 if self._ended_off_host(detail_response):
262 # Same reasoning as the list request: neither a 429
263 # nor a body from another host is Federal
264 # Register's.
265 logger.warning(
266 "Federal Register detail request ended on an "
267 "unexpected host"
268 )
269 fetch_cache[document_number] = (None, None)
270 results.append(result)
271 continue
272 self._raise_if_rate_limited(detail_response)
273 detail_response.raise_for_status()
274 detail = detail_response.json()
275 if not isinstance(detail, dict): 275 ↛ 276line 275 didn't jump to line 276 because the condition on line 275 was never true
276 logger.warning(
277 "Federal Register detail response did not contain an object"
278 )
279 fetch_cache[document_number] = (None, None)
280 results.append(result)
281 continue
283 raw_text_url = detail.get("raw_text_url")
284 if not isinstance(raw_text_url, str) or not raw_text_url: 284 ↛ 285line 284 didn't jump to line 285 because the condition on line 284 was never true
285 fetch_cache[document_number] = (None, None)
286 results.append(result)
287 continue
288 if not self._is_allowed_raw_text_url(raw_text_url):
289 logger.warning(
290 "Federal Register raw-text URL had an unexpected host"
291 )
292 fetch_cache[document_number] = (None, None)
293 results.append(result)
294 continue
295 fetch_cache[document_number] = (raw_text_url, None)
296 else:
297 raw_text_url = cached_fetch[0]
299 if raw_text_url is None: 299 ↛ 300line 299 didn't jump to line 300 because the condition on line 299 was never true
300 results.append(result)
301 continue
303 self._last_wait_time = self.rate_tracker.apply_rate_limit(
304 self.engine_type
305 )
306 try:
307 text_response = safe_get(
308 raw_text_url,
309 timeout=self._request_timeout(deadline),
310 require_https=True,
311 )
312 except ValueError as refusal:
313 # safe_get refuses (rather than fails) with ValueError:
314 # a redirect hop to cleartext, SSRF validation, or an
315 # oversized body. Log that as a refusal, not a parse
316 # failure, and never echo the URL (its query string can
317 # carry tokens). The result keeps its preview fields.
318 if "downgrade to cleartext" in str(refusal): 318 ↛ 324line 318 didn't jump to line 324 because the condition on line 318 was always true
319 logger.warning(
320 "Federal Register raw-text fetch refused: "
321 "redirect to non-https"
322 )
323 else:
324 logger.warning(
325 "Federal Register raw-text fetch refused by the "
326 "request guard"
327 )
328 fetch_cache[document_number] = (None, None)
329 results.append(result)
330 continue
331 # Host check first: a 429 (and its Retry-After) from an
332 # off-host redirect target says nothing about Federal
333 # Register's own rate limit.
334 if not self._is_allowed_raw_text_url(
335 getattr(text_response, "url", None)
336 ):
337 logger.warning(
338 "Federal Register raw-text fetch ended on an "
339 "unexpected host"
340 )
341 fetch_cache[document_number] = (None, None)
342 results.append(result)
343 continue
344 self._raise_if_rate_limited(text_response)
345 text_response.raise_for_status()
346 full_content = self._truncate_full_content(
347 self._normalize_full_content(text_response.text)
348 )
349 result["full_content"] = full_content
350 fetch_cache[document_number] = (raw_text_url, full_content)
351 except RateLimitError:
352 # The previews are already paid for. Re-raising would make
353 # ``BaseSearchEngine.run()`` retry the whole search and,
354 # once retries ran out, return [] — losing every preview
355 # over an optional enrichment. So a 429 here only stops
356 # further full-content fetches, exactly like the deadline.
357 # The availability cooldown is not recorded: usable results
358 # supersede a failure signal, and the next search's preview
359 # request raises its own RateLimitError if the limit holds.
360 # The adaptive rate limiter must still learn of the 429,
361 # though, or a sustained limit is re-hit by every search
362 # without the wait ever growing: the flag makes ``run()``
363 # record this attempt as a rate-limit failure instead of a
364 # success.
365 self._rate_limited_after_previews = True
366 results.append(result)
367 results.extend(
368 self._drain_without_fetching(
369 pending[position + 1 :],
370 fetch_cache,
371 reason="was rate limited (HTTP 429)",
372 )
373 )
374 return results
375 except _FullContentDeadlineReached:
376 results.append(result)
377 results.extend(
378 self._drain_without_fetching(
379 pending[position + 1 :], fetch_cache
380 )
381 )
382 return results
383 except requests.exceptions.RequestException as error:
384 status_code = getattr(
385 getattr(error, "response", None), "status_code", None
386 )
387 logger.warning(
388 "Federal Register full-content request failed "
389 f"(status={status_code if status_code is not None else 'unknown'})"
390 )
391 except (AttributeError, KeyError, TypeError, ValueError):
392 logger.warning("Federal Register full-content parsing failed")
393 results.append(result)
394 return results
396 @classmethod
397 def _normalize_document_types(cls, document_types: Any) -> List[str]:
398 """Strip, uppercase, de-duplicate and validate *document_types*.
400 Values outside ``VALID_DOCUMENT_TYPES`` are dropped with a warning
401 rather than sent: the API turns an unknown type into an empty
402 result, so one typo would otherwise make every search return
403 nothing. If every configured value is invalid the filter is
404 removed, i.e. all document types are searched.
405 """
406 normalized: List[str] = []
407 invalid_count = 0
408 for value in cls._ensure_list(document_types):
409 candidate = value.strip().upper() if isinstance(value, str) else ""
410 if candidate not in cls.VALID_DOCUMENT_TYPES:
411 invalid_count += 1
412 elif candidate not in normalized:
413 normalized.append(candidate)
414 if invalid_count:
415 logger.warning(
416 f"Ignoring {invalid_count} invalid Federal Register document "
417 f"type(s); valid values are {', '.join(cls.VALID_DOCUMENT_TYPES)}"
418 + ("" if normalized else ". Searching all document types.")
419 )
420 return normalized
422 @staticmethod
423 def _is_zero_match(data: Dict[str, Any]) -> bool:
424 """True for the API's no-match reply: an integer ``count`` of 0."""
425 count = data.get("count")
426 return (
427 isinstance(count, int)
428 and not isinstance(count, bool)
429 and (count == 0)
430 )
432 @staticmethod
433 def _raise_if_rate_limited(response: Any) -> None:
434 """Raise ``RateLimitError`` when *response* is an HTTP 429.
436 Decided by the status code alone. The base class's
437 ``_raise_if_rate_limit`` also matches phrases such as "throttl" or
438 "ratelimit" in an exception's text, and a ``requests`` exception's
439 text carries the request URL — here, the user's query — so a 400,
440 5xx or connection error on a query like "FCC throttling rules"
441 would be misread as a rate limit (retries, a 300-second cooldown).
442 A 429's ``Retry-After`` is kept on the error; for the preview
443 request engine availability uses it as the cooldown (in the
444 full-content phase ``_get_full_content`` catches the error and only
445 stops fetching). (A 503 is not a rate limit here:
446 ``_record_search_failure`` classifies it as provider-unavailable.)
447 """
448 if getattr(response, "status_code", None) != 429:
449 return
450 raise RateLimitError(
451 "Federal Register rate limit reached (HTTP 429)",
452 retry_after=retry_after_from_error(
453 requests.HTTPError(response=response)
454 ),
455 )
457 def _raise_if_list_ended_off_host(self, response: Any) -> None:
458 """Raise ``_UnexpectedResponseHost`` for an off-host list response."""
459 if self._ended_off_host(response):
460 raise _UnexpectedResponseHost(
461 "Federal Register list request ended on an unexpected host"
462 )
464 def _ended_off_host(self, response: Any) -> bool:
465 """True when *response*'s final URL is not on a Federal Register host.
467 ``safe_get`` follows redirects, so a response can come from
468 wherever the last hop pointed. A response whose ``url`` is not a
469 string (a test double that never set one) is taken to come from
470 the requested Federal Register URL; ``requests`` always sets it.
471 """
472 final_url = getattr(response, "url", None)
473 return isinstance(final_url, str) and not self._is_allowed_raw_text_url(
474 final_url
475 )
477 @staticmethod
478 def _preview_fields(item: Dict[str, Any]) -> Dict[str, Any]:
479 """Copy an item's public fields, dropping internal ``_`` keys."""
480 return {
481 key: value for key, value in item.items() if not key.startswith("_")
482 }
484 def _drain_without_fetching(
485 self,
486 items: List[Dict[str, Any]],
487 fetch_cache: dict[str, tuple[str | None, str | None]],
488 reason: str = "exceeded its deadline",
489 ) -> List[Dict[str, Any]]:
490 """Return the items an expired deadline or a 429 stops us fetching.
492 ``BaseSearchEngine._get_full_content`` returns one result per input
493 item, and ``run()`` records the returned count as the search's
494 result count, so running out of time may only cost the
495 ``full_content`` key — never the title/link/snippet/agencies the
496 preview phase already paid for. Text already in ``fetch_cache``
497 costs nothing to hand back, so it still travels.
498 """
499 logger.info(
500 f"Full-content phase {reason}, returning the remaining "
501 "items without full content"
502 )
503 drained = []
504 for item in items:
505 result = self._preview_fields(item)
506 document_number = result.get("id")
507 cached_fetch = (
508 fetch_cache.get(document_number)
509 if isinstance(document_number, str)
510 else None
511 )
512 if cached_fetch is not None and cached_fetch[1] is not None: 512 ↛ 513line 512 didn't jump to line 513 because the condition on line 512 was never true
513 result["full_content"] = cached_fetch[1]
514 drained.append(result)
515 return drained
517 def _request_timeout(self, deadline: float) -> int:
518 """Seconds to allow the next request, bounded by *deadline*.
520 Raises ``_FullContentDeadlineReached`` once the budget is spent.
521 The one-second floor keeps a sub-second remainder from truncating
522 to ``0``: ``int(0.4)`` is ``0``, and a zero timeout is not "no
523 limit" but an immediate failure, so the item burned its turn for
524 nothing.
525 """
526 remaining = deadline - time.monotonic()
527 if remaining <= 0:
528 raise _FullContentDeadlineReached
529 return min(self.REQUEST_TIMEOUT, max(1, int(remaining)))
531 def _is_allowed_raw_text_url(self, url: Any) -> bool:
532 """True when *url* is an ``https`` URL on a Federal Register host.
534 Parsed with ``urllib3.util.parse_url`` — the parser ``requests``
535 resolves the destination with — because ``urlparse`` disagrees
536 with it on backslash authorities (GHSA-g23j-2vwm-5c25), the same
537 reason ``security/ssrf_validator.py`` validates against urllib3:
538 ``https://evil.example\\@www.federalregister.gov/x`` has host
539 ``www.federalregister.gov`` to ``urlparse`` but connects to
540 ``evil.example``. The scheme and the RFC 3986 character class are
541 checked here too — the same rule ``validate_url`` applies — so
542 ``file://``, ``ftp://`` and every parser-differential payload are
543 refused by this guard alone rather than by a downstream layer.
544 """
545 if not isinstance(url, str) or not url:
546 return False
547 if RFC_FORBIDDEN_URL_CHARS_RE.search(url):
548 return False
549 try:
550 parsed = parse_url(url)
551 except LocationParseError:
552 return False
553 if (parsed.scheme or "").lower() != "https":
554 return False
555 return (parsed.host or "").lower() in self.RAW_TEXT_HOSTS
557 @staticmethod
558 def _normalize_full_content(text: str) -> str:
559 """Unwrap the provider's HTML/pre representation without reflowing it.
561 ``raw_text_url`` can serve an HTML document as ``text/plain``.
562 Its preformatted body carries paragraph breaks and aligned tables,
563 so keep its whitespace and inline text. Ordinary plain text must
564 not be HTML-decoded or lose literal angle-bracketed expressions.
565 """
566 text = text.replace("\x00", "")
567 if not re.match(
568 r"\s*(?:<!doctype\s+html\b[^>]*>\s*)?<html(?:\s|>)",
569 text.lstrip("\ufeff"),
570 re.IGNORECASE,
571 ):
572 return text
574 soup = BeautifulSoup(text, "html.parser")
575 if soup.body is None:
576 return text
577 pre_blocks = soup.body.find_all("pre", recursive=False)
578 # An unfamiliar layout must not silently lose document sections.
579 if len(pre_blocks) != 1:
580 return text
581 document = pre_blocks[0]
582 if soup.body.get_text().strip() != document.get_text().strip():
583 return text
584 for line_break in document.find_all("br"):
585 line_break.replace_with("\n")
586 for bullet in document.find_all("bullet"):
587 # GPO's nonstandard <bullet> tags are unclosed. Replacing the
588 # tag itself would discard the remainder nested inside it.
589 bullet.insert_before("•")
590 bullet.unwrap()
591 return document.get_text()
593 def _truncate_full_content(self, text: str) -> str:
594 """Bound one document's text before it reaches an LLM prompt.
596 Mirrors ``search_engine_gutenberg.py``'s ``max_content_chars``:
597 the raw-text endpoint serves whole rules, and the string is
598 inlined verbatim into the citation prompt.
599 """
600 if len(text) > self.max_content_chars:
601 return text[: self.max_content_chars] + "\n\n[... truncated ...]"
602 return text