79 lines
3.2 KiB
Python
79 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from .config import Settings
|
|
from .http import BotanicalHTTPClient
|
|
from .models import evidence, utc_now_iso
|
|
|
|
FIRECRAWL_SOURCE = "Local Firecrawl"
|
|
|
|
|
|
class FirecrawlClient:
|
|
def __init__(self, http: BotanicalHTTPClient, settings: Settings):
|
|
self.http = http
|
|
self.settings = settings
|
|
|
|
async def search(self, query: str, limit: int = 5, domains: list[str] | None = None) -> dict[str, Any]:
|
|
if not self.settings.firecrawl_api_url:
|
|
return {
|
|
"status": "not_configured",
|
|
"source": FIRECRAWL_SOURCE,
|
|
"retrieved_at": utc_now_iso(),
|
|
"reason": "Set FIRECRAWL_API_URL to enable targeted source search.",
|
|
"suggested_queries": self._suggested_web_queries(query, domains),
|
|
"records": [],
|
|
}
|
|
base = self.settings.firecrawl_api_url.rstrip("/")
|
|
headers = {}
|
|
if self.settings.firecrawl_api_key:
|
|
headers["Authorization"] = f"Bearer {self.settings.firecrawl_api_key}"
|
|
allowed = domains or self.settings.allowed_domains
|
|
constrained_query = query if not allowed else f"{query} ({' OR '.join('site:' + d for d in allowed[:8])})"
|
|
payload = {"query": constrained_query, "limit": min(max(limit, 1), 10)}
|
|
endpoints = [f"{base}/v1/search", f"{base}/search"]
|
|
last_error: str | None = None
|
|
for endpoint in endpoints:
|
|
try:
|
|
data = await self.http.post_json(endpoint, payload, headers=headers)
|
|
return self._normalize(data, query, endpoint)
|
|
except httpx.HTTPError as exc:
|
|
last_error = str(exc)
|
|
return {
|
|
"status": "error",
|
|
"source": FIRECRAWL_SOURCE,
|
|
"retrieved_at": utc_now_iso(),
|
|
"reason": last_error or "Unknown Firecrawl error",
|
|
"records": [],
|
|
}
|
|
|
|
def _normalize(self, data: dict[str, Any], query: str, endpoint: str) -> dict[str, Any]:
|
|
rows = data.get("data") or data.get("results") or []
|
|
if isinstance(rows, dict):
|
|
rows = rows.get("results", [])
|
|
records = []
|
|
for row in rows if isinstance(rows, list) else []:
|
|
url = row.get("url") or row.get("sourceURL")
|
|
records.append(
|
|
{
|
|
"title": evidence(row.get("title"), source=FIRECRAWL_SOURCE, url=url, confidence="medium"),
|
|
"url": evidence(url, source=FIRECRAWL_SOURCE, url=url, confidence="medium"),
|
|
"snippet": evidence(row.get("description") or row.get("markdown") or row.get("content"), source=FIRECRAWL_SOURCE, url=url, confidence="medium", raw=row),
|
|
}
|
|
)
|
|
return {
|
|
"status": "ok",
|
|
"query": query,
|
|
"source": FIRECRAWL_SOURCE,
|
|
"firecrawl_endpoint": endpoint,
|
|
"retrieved_at": utc_now_iso(),
|
|
"records": records,
|
|
"raw": data,
|
|
}
|
|
|
|
def _suggested_web_queries(self, query: str, domains: list[str] | None) -> list[str]:
|
|
allowed = domains or self.settings.allowed_domains
|
|
return [f"{query} site:{domain}" for domain in allowed[:8]]
|