Spaces:
Sleeping
Sleeping
File size: 1,799 Bytes
2568db7 7cd4f4a 2568db7 7cd4f4a 04698c5 2568db7 04698c5 2568db7 7cd4f4a 2568db7 7cd4f4a 2568db7 7cd4f4a 2568db7 7cd4f4a 2568db7 7cd4f4a 2568db7 7cd4f4a 2568db7 7cd4f4a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 | import aiohttp
import time
from ingest.generic_public_foia import GenericFOIAAdapter
from ingest.health import HealthStatus
class ExtendedLiveAdapter(GenericFOIAAdapter):
source_type = "live"
extended_only = True
async def _safe_search(self, search_url: str, query: str):
start = time.perf_counter()
await self._rate_limit()
if not await self._guard(search_url):
self.last_health = HealthStatus(ok=False, latency_ms=0, error="robots.txt")
return []
try:
async with aiohttp.ClientSession() as session:
async with session.get(
search_url,
params={"q": query},
timeout=12,
headers={"User-Agent": "HF-FOIA-Search/1.0"}
) as resp:
text = await resp.text()
latency = int((time.perf_counter() - start) * 1000)
if query.lower() in text.lower():
self.last_health = HealthStatus(ok=True, latency_ms=latency)
return [{
"source": self.source_name,
"title": f"{self.source_name} public document mentioning '{query}'",
"url": search_url,
"snippet": "Public FOIA-released material (extended live mode).",
"live": True,
"extended": True,
"health": self.last_health.__dict__
}]
self.last_health = HealthStatus(ok=True, latency_ms=latency)
return []
except Exception as e:
latency = int((time.perf_counter() - start) * 1000)
self.last_health = HealthStatus(ok=False, latency_ms=latency, error=str(e))
return [] |