57 lines
1.7 KiB
Python
57 lines
1.7 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
|
|
import httpx
|
|
from bs4 import BeautifulSoup
|
|
|
|
from app.services.ai_tools.schemas import FetchedEvidence
|
|
|
|
|
|
class WebFetchError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def _extract_title_and_text(html: str) -> tuple[str, str]:
|
|
soup = BeautifulSoup(html, "html.parser")
|
|
for tag in soup(["script", "style", "noscript", "svg"]):
|
|
tag.decompose()
|
|
title = soup.title.get_text(" ", strip=True) if soup.title else ""
|
|
main = soup.find("main") or soup.find("article") or soup.body or soup
|
|
text = main.get_text("\n", strip=True)
|
|
lines = [line.strip() for line in text.splitlines() if line.strip()]
|
|
return title, "\n".join(lines)
|
|
|
|
|
|
async def fetch_url_evidence(
|
|
url: str,
|
|
*,
|
|
timeout_seconds: int = 20,
|
|
max_bytes: int = 1_500_000,
|
|
) -> FetchedEvidence:
|
|
if not url:
|
|
raise WebFetchError("url is required")
|
|
try:
|
|
async with httpx.AsyncClient(
|
|
timeout=timeout_seconds,
|
|
follow_redirects=True,
|
|
headers={"User-Agent": "PlanetEvidenceFetcher/1.0"},
|
|
) as client:
|
|
response = await client.get(url)
|
|
response.raise_for_status()
|
|
content = response.content[:max_bytes]
|
|
except httpx.HTTPError as exc:
|
|
raise WebFetchError(f"failed to fetch page: {exc}") from exc
|
|
|
|
title, text = _extract_title_and_text(content.decode(response.encoding or "utf-8", errors="ignore"))
|
|
content_hash = hashlib.sha256(text.encode("utf-8")).hexdigest()
|
|
return FetchedEvidence(
|
|
url=url,
|
|
final_url=str(response.url),
|
|
title=title,
|
|
text=text,
|
|
content_hash=content_hash,
|
|
extractor="beautifulsoup_basic",
|
|
)
|
|
|