Spaces:
Sleeping
Sleeping
| # src/analyzer/net/fetcher.py | |
| from __future__ import annotations | |
| import os, time, hashlib | |
| from typing import Optional, Dict | |
| import httpx | |
| from bs4 import BeautifulSoup | |
| CACHE_DIR = os.getenv("LINK_CACHE_DIR", "data/link_cache") | |
| HEADERS = {"User-Agent": "grant-analyst/1.0 (+polite; contact: engineering@example.com)"} | |
| TIMEOUT = httpx.Timeout(25.0) | |
| def _sha(s: str) -> str: | |
| return hashlib.sha1(s.encode("utf-8", "ignore")).hexdigest() | |
| def _path(stem: str, ext: str) -> str: | |
| os.makedirs(CACHE_DIR, exist_ok=True) | |
| return os.path.join(CACHE_DIR, f"{stem}.{ext}") | |
| def _html2text(html: str) -> str: | |
| soup = BeautifulSoup(html, "lxml") | |
| for sel in ["nav", "header", "footer", ".govuk-footer", ".govuk-phase-banner"]: | |
| for n in soup.select(sel): | |
| n.decompose() | |
| txt = "\n".join(p.get_text(" ", strip=True) for p in soup.find_all(["h1","h2","h3","p","li","dt","dd"])) | |
| return txt.strip() | |
| def fetch_link(url: str, *, force: bool = False) -> Dict: | |
| """ | |
| Fetch a URL (HTML or PDF), parse to text, and cache results. | |
| Returns: {url, ok, kind, text, cached_txt, cached_html, fetched_at} | |
| """ | |
| key = _sha(url) | |
| cached_txt = _path(key, "txt") | |
| cached_html = _path(key, "html") | |
| meta_path = _path(key, "meta") | |
| if not force and os.path.exists(cached_txt): | |
| with open(cached_txt, "r", encoding="utf-8") as f: | |
| text = f.read() | |
| return {"url": url, "ok": True, "kind": "cached", "text": text, | |
| "cached_txt": cached_txt, "cached_html": cached_html, | |
| "fetched_at": os.path.getmtime(cached_txt)} | |
| kind = "html" | |
| try: | |
| with httpx.Client(timeout=TIMEOUT, follow_redirects=True, headers=HEADERS) as cli: | |
| r = cli.get(url) | |
| ctype = r.headers.get("content-type","").lower() | |
| if "application/pdf" in ctype or url.lower().endswith(".pdf") or ".pdf?" in url.lower(): | |
| kind = "pdf" | |
| pdf_path = _path(key, "pdf") | |
| with open(pdf_path, "wb") as f: | |
| f.write(r.content) | |
| try: | |
| import fitz # PyMuPDF | |
| with fitz.open(pdf_path) as doc: | |
| pages = [p.get_text() for p in doc] | |
| text = "\n".join(pages).strip() | |
| except Exception: | |
| text = "(PDF saved; text extraction failed)" | |
| else: | |
| html = r.text | |
| text = _html2text(html) | |
| with open(cached_html, "w", encoding="utf-8") as f: | |
| f.write(html) | |
| except Exception as e: | |
| return {"url": url, "ok": False, "error": str(e), "kind": kind} | |
| with open(cached_txt, "w", encoding="utf-8") as f: | |
| f.write(text) | |
| with open(meta_path, "w", encoding="utf-8") as f: | |
| f.write(f"{time.time()}\n{url}\n{kind}\n") | |
| return {"url": url, "ok": True, "kind": kind, "text": text, | |
| "cached_txt": cached_txt, "cached_html": cached_html, | |
| "fetched_at": time.time()} |