Spaces:
Running
Running
File size: 4,933 Bytes
4939f56 | 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 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 | """Page capture service — techspec.md §3.
For every external source a pipeline run queries, capture the *human-facing*
page (not just the API endpoint): page title, readable text sections and
figure image URLs. Rows land in `page_captures` keyed by (job_id, source) so
the final synthesis can cite real pages.
Safety: every fetch goes through `ssrf.validate_url()` (allowlist) and a
per-host rate limiter. Captures are strictly best-effort — they must never
break a pipeline run.
"""
import asyncio
import html as htmllib
import logging
import re
import time
import httpx
from app.services.ssrf import validate_url
logger = logging.getLogger(__name__)
# Minimum delay between consecutive fetches of the same host.
MIN_INTERVAL_S = 1.5
_host_lock = asyncio.Lock()
_last_hit: dict[str, float] = {}
_MAX_SECTIONS = 8
_SECTION_CHARS = 600
_TIMEOUT_S = 20
_TAG_RE = re.compile(r"<[^>]+>")
_SCRIPT_RE = re.compile(r"<(script|style)\b[^>]*>.*?</\1>", re.IGNORECASE | re.DOTALL)
_HEADING_RE = re.compile(r"<h([23])\b[^>]*>(.*?)</h\1>", re.IGNORECASE | re.DOTALL)
_TITLE_RE = re.compile(r"<title[^>]*>(.*?)</title>", re.IGNORECASE | re.DOTALL)
_IMG_RE = re.compile(r'<img\b[^>]*?\bsrc=["\']([^"\']+)["\']', re.IGNORECASE)
_OG_IMAGE_RE = re.compile(
r'<meta\s+[^>]*?property=["\']og:image["\'][^>]*?content=["\']([^"\']+)["\']', re.IGNORECASE
)
def _clean(text: str) -> str:
return htmllib.unescape(re.sub(r"\s+", " ", _TAG_RE.sub("", text))).strip()
def extract_page_content(html: str) -> dict:
"""Stdlib-only extraction: title, heading-led text sections, figure URLs."""
body = _SCRIPT_RE.sub("", html)
title_match = _TITLE_RE.search(body)
title = _clean(title_match.group(1)) if title_match else None
figure_urls: list[str] = []
og = _OG_IMAGE_RE.search(body)
if og:
figure_urls.append(og.group(1))
for fig_block in re.findall(r"<figure\b[^>]*>(.*?)</figure>", body, re.IGNORECASE | re.DOTALL):
for src in _IMG_RE.findall(fig_block)[:3]:
if src.startswith("http") and src not in figure_urls:
figure_urls.append(src)
sections: list[dict] = []
headings = list(_HEADING_RE.finditer(body))
for i, match in enumerate(headings):
if len(sections) >= _MAX_SECTIONS:
break
start = match.end()
end = headings[i + 1].start() if i + 1 < len(headings) else min(len(body), start + 4000)
text = _clean(body[start:end])[:_SECTION_CHARS]
if text:
sections.append({"heading": _clean(match.group(2))[:200], "text": text})
if not sections and not headings:
fallback = _clean(body)[:_SECTION_CHARS]
if fallback:
sections.append({"heading": None, "text": fallback})
return {
"title": title,
"text_sections": sections,
"figure_urls": figure_urls[:6],
}
async def _throttle(host: str) -> None:
async with _host_lock:
now = time.monotonic()
last = _last_hit.get(host)
wait = 0.0 if last is None else max(0.0, MIN_INTERVAL_S - (now - last))
_last_hit[host] = now + wait
if wait > 0:
await asyncio.sleep(wait)
def _get_supabase():
from app.services.supabase import get_client
return get_client()
async def capture_page(
job_id: str,
source: str,
page_url: str,
user_id: str | None = None,
) -> dict | None:
"""Fetch + store one page capture. Never raises."""
row = {"job_id": job_id, "source": source, "page_url": page_url, "user_id": user_id}
try:
validate_url(page_url)
host = httpx.URL(page_url).host
await _throttle(host)
async with httpx.AsyncClient(timeout=_TIMEOUT_S, follow_redirects=True) as client:
res = await client.get(page_url, headers={"User-Agent": "BioNexus/1.0 (research tool)"})
res.raise_for_status()
extracted = extract_page_content(res.text)
row.update(extracted, fetch_status="captured")
except Exception as exc:
logger.info("Page capture %s/%s failed: %s", job_id, source, exc)
row.update(fetch_status="failed", error_note=str(exc)[:300])
try:
(
_get_supabase()
.table("page_captures")
.upsert(row, on_conflict="job_id,source")
.execute()
)
except Exception as exc:
logger.warning("Could not persist page capture %s/%s: %s", job_id, source, exc)
return row
def capture_bg(job_id: str, source: str, page_url: str, user_id: str | None = None):
"""Fire-and-forget capture for pipeline steps (never blocks the run)."""
if not job_id or "-" not in str(job_id): # skip test/in-memory ids
return
try:
loop = asyncio.get_running_loop()
loop.create_task(capture_page(job_id, source, page_url, user_id))
except RuntimeError:
logger.debug("No running loop for capture %s/%s", job_id, source)
|