From dfd28ad409bb2fd8810c756a5732fe49e32f8c49 Mon Sep 17 00:00:00 2001 From: rapsoj Date: Tue, 14 Jul 2026 22:20:12 +0100 Subject: [PATCH 1/2] Added custom scrapers for Tableau insights --- .DS_Store | Bin 6148 -> 6148 bytes bioscancast/datasets/sources.yaml | 5 + .../custom_scrapers/paho_oropouche_portal.py | 318 ++++++++++++++++++ .../custom_scrapers/usda_aphis_livestock.py | 307 +++++++++++++++++ .../text_extraction/chunk_extractor.py | 16 +- .../test_paho_oropouche_portal_scraper.py | 156 +++++++++ .../test_usda_aphis_livestock_scraper.py | 176 ++++++++++ 7 files changed, 977 insertions(+), 1 deletion(-) create mode 100644 bioscancast/stages/extraction/custom_scrapers/paho_oropouche_portal.py create mode 100644 bioscancast/stages/extraction/custom_scrapers/usda_aphis_livestock.py create mode 100644 bioscancast/tests/test_paho_oropouche_portal_scraper.py create mode 100644 bioscancast/tests/test_usda_aphis_livestock_scraper.py diff --git a/.DS_Store b/.DS_Store index 5008ddfcf53c02e82d7eee2e57c38e5672ef89f6..d2944c18b9b9c9eb673cb6cf1c41873d007c516b 100644 GIT binary patch literal 6148 zcmeHLu}T9$5S@)D21T^AJpBMMt<@!*je-b%fJuxbVv;*d42TNb*jf1n+FRN93HbtB z|3Ynjv%APmE{RqWWCwQN-OkL;yLa4fHbkT<7mX6pDiIaY7=uN0V~qXWa#nC7Ye2mfYwrw%n%7-$}A;f0H5j(*g6J z2mb-McWEH_-Y>pNujHq1uilMznS9A-T!g59Q|0hX zyr_9*b(qV8kFbr+06=g=xe z^`5A%uDA0Ml7kMmKFs?XnMX326eai+_+`wDVb5kOE?M-wSHLUa75G+w*M|^|F>(9>S>u2q4*(;F+9F0^%BBKss&YpRWz#Vpy12-pwrJBynaY^QRaWkV zqTC9rLK99ZvgmuSfLA~&FlTR9dH>&;eEye7{?04l75G;Qs31O!D|jTgx8@#>_u3Gx sipIuq+M-E8=eA>W!CNu^uV4(mkS~CdLv0ZwF#kiq$lyD#z@IAc1xrlLPXGV_ delta 65 zcmZoMXfc=|#>AjHF;Q%yo+1YW5HK<@2y7PQ5M$X`FpGIJI|n}pP#!4ooq009h$187 SWK$94$^JYXn`1;)FarR`U=BS1 diff --git a/bioscancast/datasets/sources.yaml b/bioscancast/datasets/sources.yaml index dfa2a22..5a590d0 100644 --- a/bioscancast/datasets/sources.yaml +++ b/bioscancast/datasets/sources.yaml @@ -217,6 +217,11 @@ specific_pathogen_sources: url: "https://www.paho.org/en/topics/oropouche-virus-disease" geography: "Americas" + - id: "paho_oropouche_portal" + name: "ARBO Portal - Oropouche" + url: "https://www.paho.org/en/arbo-portal/arbo-portal-oropouche" + geography: "Americas" + enteric: cholera: - id: "who_cholera" diff --git a/bioscancast/stages/extraction/custom_scrapers/paho_oropouche_portal.py b/bioscancast/stages/extraction/custom_scrapers/paho_oropouche_portal.py new file mode 100644 index 0000000..ae81f12 --- /dev/null +++ b/bioscancast/stages/extraction/custom_scrapers/paho_oropouche_portal.py @@ -0,0 +1,318 @@ +"""Custom scraper for PAHO ARBO Portal Oropouche weekly CSV. + +The PAHO Oropouche portal is dashboard-backed. This scraper pulls the downloadable +underlying CSV and renders computed weekly/country analytics as compact HTML so +standard extraction + insight stages can consume the information. +""" + +from __future__ import annotations + +import html +import io +import re +import unicodedata +from datetime import datetime, timezone +from typing import Callable, Optional + +import numpy as np +import pandas as pd +from curl_cffi import requests as curl_requests + +from bioscancast.stages.extraction.config import ExtractionConfig +from bioscancast.stages.extraction.fetcher import FetchResult + +CSV_URL = ( + "https://phip.paho.org/vizql/w/AME_OROV_Cases/v/Oropouche/vudcsv/" + "sessions/21F758F6EAB9431B932EB51F61EBDE01-1:0/views/" + "10956961424773455462_12644916002446576356" + "?underlying_table_id=FZ_AME_OROV_Cases_SE_12E30DDB8F874DC6AE5888461B08AB74" + "&underlying_table_caption=Full%20Data" +) + +CsvFetcher = Callable[[str, ExtractionConfig], Optional[str]] + + +def _fetch_csv_text(url: str, cfg: ExtractionConfig) -> Optional[str]: + try: + resp = curl_requests.get( + url, + timeout=max(cfg.fetch_timeout_seconds, 30.0), + impersonate=cfg.impersonate, + allow_redirects=True, + ) + except Exception: + return None + + if resp.status_code != 200: + return None + + text = (resp.text or "").strip() + return text or None + + +def _normalize_col(name: str) -> str: + folded = unicodedata.normalize("NFKD", name or "") + folded = "".join(ch for ch in folded if not unicodedata.combining(ch)) + folded = folded.lower().strip() + folded = re.sub(r"[^a-z0-9]+", "_", folded) + return folded.strip("_") + + +def _resolve_columns(df: pd.DataFrame) -> tuple[str, str, str, str] | None: + mapping = {_normalize_col(c): c for c in df.columns} + + year = mapping.get("ano") + week = mapping.get("semanas_epi") + confirmed = mapping.get("week_lab_confirmed") + country = mapping.get("pais") + + if not all((year, week, confirmed, country)): + return None + return year, week, confirmed, country + + +def _r2(y: np.ndarray, yhat: np.ndarray) -> float: + ss_res = float(np.sum((y - yhat) ** 2)) + ss_tot = float(np.sum((y - np.mean(y)) ** 2)) + if ss_tot <= 0.0: + return 1.0 if ss_res <= 1e-12 else 0.0 + return max(0.0, min(1.0, 1.0 - (ss_res / ss_tot))) + + +def _fit_linear(counts: pd.Series) -> dict[str, float] | None: + if counts.shape[0] < 2: + return None + y = counts.to_numpy(dtype=float) + x = np.arange(y.shape[0], dtype=float) + slope, intercept = np.polyfit(x, y, 1) + yhat = slope * x + intercept + return { + "intercept": float(intercept), + "slope": float(slope), + "r2": _r2(y, yhat), + } + + +def _fit_exponential(counts: pd.Series) -> dict[str, float] | None: + if counts.shape[0] < 2: + return None + y = counts.to_numpy(dtype=float) + x = np.arange(y.shape[0], dtype=float) + + mask = y > 0 + if int(np.sum(mask)) < 2: + return None + + x_pos = x[mask] + y_pos = y[mask] + b, ln_a = np.polyfit(x_pos, np.log(y_pos), 1) + a = float(np.exp(ln_a)) + yhat = a * np.exp(b * x_pos) + + return { + "a": a, + "b": float(b), + "r2": _r2(y_pos, yhat), + } + + +def _fmt_stats(series: pd.Series) -> str: + if series.empty: + return "n=0" + desc = series.describe() + std = desc["std"] if pd.notna(desc["std"]) else 0.0 + return ( + f"n={int(desc['count'])}, mean={desc['mean']:.2f}, std={std:.2f}, " + f"min={desc['min']:.0f}, median={series.median():.0f}, max={desc['max']:.0f}" + ) + + +def _render_counts_table(title: str, counts: pd.Series, key_header: str) -> str: + rows = [] + for idx, value in counts.items(): + rows.append(f"{html.escape(str(idx))}{int(value)}") + body = "".join(rows) if rows else "No data" + return ( + f"

{html.escape(title)}

" + "" + f"" + f"{body}
{html.escape(key_header)}sum_week_lab_confirmed
" + ) + + +def _render_model_section( + title: str, + counts: pd.Series, + *, + linear: dict[str, float] | None, + exp: dict[str, float] | None, +) -> str: + lines = [f"

{html.escape(title)}

"] + lines.append(f"

Input points: {counts.shape[0]}.

") + if linear is None: + lines.append("

Linear model: unavailable (insufficient data).

") + else: + lines.append( + "

Linear model y = intercept + slope*x: " + f"intercept={linear['intercept']:.6f}, slope={linear['slope']:.6f}, R^2={linear['r2']:.6f}.

" + ) + + if exp is None: + lines.append( + "

Exponential model y = a*exp(b*x): unavailable " + "(insufficient strictly-positive points).

" + ) + else: + lines.append( + "

Exponential model y = a*exp(b*x): " + f"a={exp['a']:.6f}, b={exp['b']:.6f}, R^2={exp['r2']:.6f}.

" + ) + return "".join(lines) + + +def _render_country_summary(df: pd.DataFrame, country_col: str, value_col: str) -> str: + if df.empty: + return "

Per-country summary

No country data available.

" + + g = df.groupby(country_col)[value_col].sum().sort_values(ascending=False) + summary = g.describe() + std = summary["std"] if pd.notna(summary["std"]) else 0.0 + + rows = [] + for country, value in g.items(): + rows.append( + f"{html.escape(str(country))}{int(value)}" + ) + + return ( + "

Per-country summary

" + f"

Affected states/countries (Pais values): {int(g.shape[0])}. " + f"Per-country cumulative summary: n={int(summary['count'])}, mean={summary['mean']:.2f}, " + f"std={std:.2f}, min={summary['min']:.0f}, median={g.median():.0f}, max={summary['max']:.0f}. " + f"Cumulative confirmed count across all Pais: {int(g.sum())}.

" + "" + "" + f"{''.join(rows)}
Paiscumulative_week_lab_confirmed
" + ) + + +def fetch( + url: str, + *, + config: ExtractionConfig | None = None, + as_of_date: datetime | None = None, + region: str | None = None, + question_text: str | None = None, + csv_fetcher: CsvFetcher | None = None, +) -> FetchResult | None: + cfg = config or ExtractionConfig() + fetched_at = datetime.now(timezone.utc) + + text = (csv_fetcher or _fetch_csv_text)(CSV_URL, cfg) + if not text: + return None + + try: + df = pd.read_csv(io.StringIO(text)) + except Exception: + return None + + cols = _resolve_columns(df) + if cols is None: + return None + year_col, week_col, confirmed_col, country_col = cols + + df = df[[year_col, week_col, confirmed_col, country_col]].copy() + df[year_col] = pd.to_numeric(df[year_col], errors="coerce") + df[week_col] = pd.to_numeric(df[week_col], errors="coerce") + + value_text = ( + df[confirmed_col] + .astype(str) + .str.replace(",", "", regex=False) + .str.strip() + ) + df[confirmed_col] = pd.to_numeric(value_text, errors="coerce").fillna(0) + + df = df.dropna(subset=[year_col, week_col]).copy() + if df.empty: + return None + + df[year_col] = df[year_col].astype(int) + df[week_col] = df[week_col].astype(int) + df = df[(df[week_col] >= 1) & (df[week_col] <= 53)].copy() + if df.empty: + return None + + df["week_start"] = [ + datetime.fromisocalendar(int(y), int(w), 1).date() + for y, w in zip(df[year_col], df[week_col]) + ] + + if as_of_date is not None: + cutoff = as_of_date.astimezone(timezone.utc).date() + df = df[df["week_start"] <= cutoff] + if df.empty: + return None + + df["year_week"] = [f"{int(y)}-W{int(w):02d}" for y, w in zip(df[year_col], df[week_col])] + + weekly_counts = ( + df.groupby(["week_start", "year_week"], as_index=False)[confirmed_col] + .sum() + .sort_values("week_start") + ) + if weekly_counts.empty: + return None + + weekly_series = pd.Series( + weekly_counts[confirmed_col].to_numpy(), + index=weekly_counts["year_week"].tolist(), + ) + + recent_24 = weekly_counts.tail(24) + recent_24_series = pd.Series( + recent_24[confirmed_col].to_numpy(), + index=recent_24["year_week"].tolist(), + ) + + lin = _fit_linear(recent_24_series) + exp = _fit_exponential(recent_24_series) + + country_week = ( + df.groupby(["year_week", country_col], as_index=False)[confirmed_col] + .sum() + .sort_values(["year_week", country_col]) + ) + + cumulative_all = int(weekly_series.sum()) + latest_week = str(weekly_counts["year_week"].iloc[-1]) + + rendered = ( + "" + "PAHO Oropouche portal - weekly CSV analytics" + "" + "

PAHO Oropouche weekly analytics snapshot

" + f"

Source portal URL: {html.escape(url)}

" + f"

CSV source URL: {html.escape(CSV_URL)}

" + f"

Retrieved at: {fetched_at.isoformat()} | latest year_week: {html.escape(latest_week)}.

" + "

Weekly counts from Año + Semanas Epi

" + f"

Summary statistics (weekly summed Week Lab Confirmed): {_fmt_stats(weekly_series)}. " + f"Cumulative confirmed count across all weeks: {cumulative_all}.

" + f"{_render_counts_table('Counts by year_week', weekly_series, 'year_week')}" + f"{_render_model_section('Model fit on past 24 weeks (approx past 6 months)', recent_24_series, linear=lin, exp=exp)}" + "

Country/state grouping from year_week + Pais

" + f"

Grouped rows (year_week x Pais): {int(country_week.shape[0])}.

" + f"{_render_country_summary(df, country_col, confirmed_col)}" + "" + ).encode("utf-8") + + return FetchResult( + url=CSV_URL, + final_url=CSV_URL, + status_code=200, + content_type="text/html", + content_bytes=rendered, + fetched_at=fetched_at, + error=None, + ) diff --git a/bioscancast/stages/extraction/custom_scrapers/usda_aphis_livestock.py b/bioscancast/stages/extraction/custom_scrapers/usda_aphis_livestock.py new file mode 100644 index 0000000..a673396 --- /dev/null +++ b/bioscancast/stages/extraction/custom_scrapers/usda_aphis_livestock.py @@ -0,0 +1,307 @@ +"""Custom scraper for USDA APHIS HPAI livestock detections. + +The public APHIS dashboard is a client-rendered Tableau page; this scraper reads +its downloadable CSV export and renders compact analytical prose/tables so the +regular HTML parser + insight extraction pipeline can operate without additional +CSV-specific parser changes. +""" + +from __future__ import annotations + +import html +import io +from datetime import datetime, timezone +from typing import Callable, Optional + +import numpy as np +import pandas as pd +from curl_cffi import requests as curl_requests + +from bioscancast.stages.extraction.config import ExtractionConfig +from bioscancast.stages.extraction.fetcher import FetchResult + +CSV_URL = ( + "https://publicdashboards.dl.usda.gov/vizql/t/MRP_PUB/" + "w/VS_Cattle_HPAIConfirmedDetections2024/v/HPAI2022ConfirmedDetections/" + "tempfile/sessions/89200926388D46C4A9F17603C7900132-1:0/" + "?key=2503115340&keepfile=yes&attachment=yes" +) + +CsvFetcher = Callable[[str, ExtractionConfig], Optional[str]] + + +def _fetch_csv_text(url: str, cfg: ExtractionConfig) -> Optional[str]: + try: + resp = curl_requests.get( + url, + timeout=max(cfg.fetch_timeout_seconds, 30.0), + impersonate=cfg.impersonate, + allow_redirects=True, + ) + except Exception: + return None + + if resp.status_code != 200: + return None + + text = (resp.text or "").strip() + return text or None + + +def _r2(y: np.ndarray, yhat: np.ndarray) -> float: + ss_res = float(np.sum((y - yhat) ** 2)) + ss_tot = float(np.sum((y - np.mean(y)) ** 2)) + if ss_tot <= 0.0: + return 1.0 if ss_res <= 1e-12 else 0.0 + return max(0.0, min(1.0, 1.0 - (ss_res / ss_tot))) + + +def _fit_linear(counts: pd.Series) -> dict[str, float] | None: + if counts.shape[0] < 2: + return None + y = counts.to_numpy(dtype=float) + x = np.arange(y.shape[0], dtype=float) + slope, intercept = np.polyfit(x, y, 1) + yhat = slope * x + intercept + return { + "intercept": float(intercept), + "slope": float(slope), + "r2": _r2(y, yhat), + } + + +def _fit_exponential(counts: pd.Series) -> dict[str, float] | None: + if counts.shape[0] < 2: + return None + y = counts.to_numpy(dtype=float) + x = np.arange(y.shape[0], dtype=float) + + # Log-linear fit requires strictly positive observations. + mask = y > 0 + if int(np.sum(mask)) < 2: + return None + + x_pos = x[mask] + y_pos = y[mask] + + b, ln_a = np.polyfit(x_pos, np.log(y_pos), 1) + a = float(np.exp(ln_a)) + yhat = a * np.exp(b * x_pos) + + return { + "a": a, + "b": float(b), + "r2": _r2(y_pos, yhat), + } + + +def _fmt_stats(series: pd.Series) -> str: + if series.empty: + return "n=0" + desc = series.describe() + return ( + f"n={int(desc['count'])}, mean={desc['mean']:.2f}, std={desc['std'] if pd.notna(desc['std']) else 0.0:.2f}, " + f"min={desc['min']:.0f}, median={series.median():.0f}, max={desc['max']:.0f}" + ) + + +def _render_counts_table(title: str, counts: pd.Series, key_header: str) -> str: + rows = [] + for idx, value in counts.items(): + rows.append( + f"{html.escape(str(idx))}{int(value)}" + ) + body = "".join(rows) if rows else "No data" + return ( + f"

{html.escape(title)}

" + "" + f"" + f"{body}
{html.escape(key_header)}count
" + ) + + +def _render_state_summary(df: pd.DataFrame) -> str: + if df.empty: + return "

Per-state summary

No state data available.

" + + g = df.groupby("State").size().sort_values(ascending=False) + summary = g.describe() + + rows = [] + for state, value in g.items(): + rows.append(f"{html.escape(str(state))}{int(value)}") + + return ( + "

Per-state summary

" + f"

State-case-count statistics: n={int(summary['count'])}, " + f"mean={summary['mean']:.2f}, std={summary['std'] if pd.notna(summary['std']) else 0.0:.2f}, " + f"min={summary['min']:.0f}, median={g.median():.0f}, max={summary['max']:.0f}.

" + "" + "" + f"{''.join(rows)}
Statecase_count
" + ) + + +def _render_first_case_by_state(df: pd.DataFrame) -> str: + if df.empty: + return "

Affected states and first detected dates

No state data available.

" + + first_by_state = ( + df.groupby("State")["Confirmed Diagnosis"] + .min() + .sort_values() + ) + + rows = [] + for state, dt in first_by_state.items(): + rows.append( + "" + f"{html.escape(str(state))}" + f"{html.escape(dt.date().isoformat())}" + "" + ) + + return ( + "

Affected states and first detected dates

" + f"

Total affected states: {int(first_by_state.shape[0])}.

" + "" + "" + f"{''.join(rows)}
Statefirst_confirmed_diagnosis_date
" + ) + + +def _render_model_section( + title: str, + counts: pd.Series, + *, + linear: dict[str, float] | None, + exp: dict[str, float] | None, +) -> str: + lines = [f"

{html.escape(title)}

"] + # Repeat the section title in prose so it survives heading-only drops in + # some HTML extraction paths and remains quotable for insight extraction. + lines.append( + f"

Section title: {html.escape(title)}. " + f"Input points: {counts.shape[0]}.

" + ) + + if linear is None: + lines.append("

Linear model: unavailable (insufficient data).

") + else: + lines.append( + "

Linear model y = intercept + slope*x: " + f"intercept={linear['intercept']:.6f}, slope={linear['slope']:.6f}, " + f"R^2={linear['r2']:.6f}.

" + ) + + if exp is None: + lines.append( + "

Exponential model y = a*exp(b*x): unavailable " + "(insufficient strictly-positive points).

" + ) + else: + lines.append( + "

Exponential model y = a*exp(b*x): " + f"a={exp['a']:.6f}, b={exp['b']:.6f}, R^2={exp['r2']:.6f}.

" + ) + + return "".join(lines) + + +def fetch( + url: str, + *, + config: ExtractionConfig | None = None, + as_of_date: datetime | None = None, + region: str | None = None, + question_text: str | None = None, + csv_fetcher: CsvFetcher | None = None, +) -> FetchResult | None: + cfg = config or ExtractionConfig() + fetched_at = datetime.now(timezone.utc) + + text = (csv_fetcher or _fetch_csv_text)(CSV_URL, cfg) + if not text: + return None + + try: + df = pd.read_csv(io.StringIO(text)) + except Exception: + return None + + # Requested behavior: ignore first row. + if df.shape[0] < 2: + return None + df = df.iloc[1:].copy() + + if "Confirmed Diagnosis" not in df.columns or "State" not in df.columns: + return None + + df["Confirmed Diagnosis"] = pd.to_datetime( + df["Confirmed Diagnosis"], errors="coerce" + ) + df = df.dropna(subset=["Confirmed Diagnosis"]).copy() + if df.empty: + return None + + if as_of_date is not None: + cutoff = as_of_date.astimezone(timezone.utc).replace(tzinfo=None) + df = df[df["Confirmed Diagnosis"] <= cutoff] + if df.empty: + return None + + # Normalize to date only for groupings. + df["diagnosis_date"] = df["Confirmed Diagnosis"].dt.date + df["month"] = df["Confirmed Diagnosis"].dt.to_period("M").astype(str) + + monthly_counts = df.groupby("month").size().sort_index() + daily_counts = df.groupby("diagnosis_date").size().sort_index() + cumulative_case_count = int(df.shape[0]) + + if monthly_counts.empty or daily_counts.empty: + return None + + month_window = monthly_counts.iloc[-6:] + day_window = daily_counts.iloc[-30:] + + lin_month = _fit_linear(month_window) + exp_month = _fit_exponential(month_window) + lin_day = _fit_linear(day_window) + exp_day = _fit_exponential(day_window) + + latest = str(daily_counts.index.max()) + + rendered = ( + "" + "USDA APHIS HPAI Confirmed Cases in Livestock - CSV analytics" + "" + "

USDA APHIS HPAI Confirmed Cases in Livestock - analytics snapshot

" + f"

Source dashboard URL: {html.escape(url)}

" + f"

CSV source URL: {html.escape(CSV_URL)}

" + f"

Retrieved at: {fetched_at.isoformat()} | latest confirmed diagnosis date: {html.escape(latest)}.

" + "

This summary is computed from the downloadable APHIS CSV. The first CSV row was ignored by design.

" + "

Cumulative and state coverage summary

" + f"

Cumulative confirmed cases in livestock (from CSV rows): {cumulative_case_count}.

" + f"{_render_first_case_by_state(df)}" + "

Monthly counts from Confirmed Diagnosis

" + f"

Summary statistics (monthly case counts): {_fmt_stats(monthly_counts)}.

" + f"{_render_counts_table('Counts by month', monthly_counts, 'month')}" + f"{_render_model_section('Model fit on past 6 months (monthly counts)', month_window, linear=lin_month, exp=exp_month)}" + "

Daily counts from Confirmed Diagnosis

" + f"

Summary statistics (daily case counts): {_fmt_stats(daily_counts)}.

" + f"{_render_counts_table('Counts by day', daily_counts, 'date')}" + f"{_render_model_section('Model fit on past 30 days (daily counts)', day_window, linear=lin_day, exp=exp_day)}" + "

State-level counts from Confirmed Diagnosis + State

" + f"{_render_state_summary(df)}" + "" + ).encode("utf-8") + + return FetchResult( + url=CSV_URL, + final_url=CSV_URL, + status_code=200, + content_type="text/html", + content_bytes=rendered, + fetched_at=fetched_at, + error=None, + ) diff --git a/bioscancast/stages/insight/text_extraction/chunk_extractor.py b/bioscancast/stages/insight/text_extraction/chunk_extractor.py index 4fa8d67..5434374 100644 --- a/bioscancast/stages/insight/text_extraction/chunk_extractor.py +++ b/bioscancast/stages/insight/text_extraction/chunk_extractor.py @@ -414,6 +414,20 @@ def extract_facts_from_chunk( # and content-insertion hallucinations. See ``_quote_matches`` for # the rationale and the layers. canonical_quote = _quote_matches(raw_quote, chunk.text) + source_chunk_id = chunk.chunk_id + if canonical_quote is None: + # Minimal fallback for chunk-routing misses: if retrieval sends a + # chunk adjacent to the one the model quoted, accept only when the + # quote appears verbatim in some other chunk of the same document. + for alt_chunk in document.chunks: + if alt_chunk.chunk_id == chunk.chunk_id: + continue + alt_quote = _quote_matches(raw_quote, alt_chunk.text) + if alt_quote is not None: + canonical_quote = alt_quote + source_chunk_id = alt_chunk.chunk_id + break + if canonical_quote is None: logger.warning( "Hallucination guard: dropping fact with non-matching quote. " @@ -456,7 +470,7 @@ def extract_facts_from_chunk( sources=[ ChunkReference( document_id=document.id, - chunk_id=chunk.chunk_id, + chunk_id=source_chunk_id, source_url=document.source_url, quote=canonical_quote[:200], ), diff --git a/bioscancast/tests/test_paho_oropouche_portal_scraper.py b/bioscancast/tests/test_paho_oropouche_portal_scraper.py new file mode 100644 index 0000000..62e3900 --- /dev/null +++ b/bioscancast/tests/test_paho_oropouche_portal_scraper.py @@ -0,0 +1,156 @@ +from __future__ import annotations + +from datetime import datetime, timezone + +from bioscancast.llm.base import LLMResponse +from bioscancast.llm.fake_client import FakeLLMClient +from bioscancast.stages.extraction.config import ExtractionConfig +from bioscancast.stages.extraction.custom_scrapers import paho_oropouche_portal +from bioscancast.stages.extraction.pipeline import ExtractionPipeline +from bioscancast.stages.filtering.models import FilteredDocument, ForecastQuestion +from bioscancast.stages.insight.config import InsightConfig +from bioscancast.stages.insight.pipeline import InsightPipeline + + +_CSV = """Año,Semanas Epi,Week Lab Confirmed,Pais +2026,1,10,Brazil +2026,1,5,Peru +2026,2,7,Brazil +2026,3,0,Bolivia +2026,20,9,Brazil +2026,21,4,Peru +2026,22,3,Colombia +2026,23,8,Brazil +2026,24,2,Peru +""" + + +def _fetcher(text: str = _CSV): + return lambda url, config: text + + +def _html(result) -> str: + assert result is not None + assert result.content_type == "text/html" + return result.content_bytes.decode("utf-8") + + +def _asof(s: str) -> datetime: + return datetime.strptime(s, "%Y-%m-%d").replace(tzinfo=timezone.utc) + + +def test_renders_weekly_and_country_analytics(): + result = paho_oropouche_portal.fetch( + "https://www.paho.org/en/arbo-portal/arbo-portal-oropouche", + csv_fetcher=_fetcher(), + ) + body = _html(result) + + assert "Weekly counts from Año + Semanas Epi" in body + assert "Counts by year_week" in body + assert "2026-W01" in body + assert "Cumulative confirmed count across all weeks" in body + assert "Model fit on past 24 weeks" in body + assert "Country/state grouping from year_week + Pais" in body + assert "Per-country summary" in body + assert "Affected states/countries (Pais values):" in body + + +def test_as_of_date_filters_future_weeks(): + result = paho_oropouche_portal.fetch( + "https://www.paho.org/en/arbo-portal/arbo-portal-oropouche", + as_of_date=_asof("2026-05-25"), + csv_fetcher=_fetcher(), + ) + body = _html(result) + + assert "2026-W23" not in body + assert "2026-W24" not in body + + +def test_dispatcher_uses_source_id_custom_scraper(monkeypatch): + from bioscancast.stages.extraction import fetcher as fetcher_mod + + monkeypatch.setattr(paho_oropouche_portal, "_fetch_csv_text", lambda _url, _cfg: _CSV) + + result = fetcher_mod.fetch( + "https://www.paho.org/en/arbo-portal/arbo-portal-oropouche", + config=ExtractionConfig(), + source_id="paho_oropouche_portal", + ) + assert result is not None + assert result.fetch_strategy == "custom:paho_oropouche_portal" + assert "PAHO Oropouche weekly analytics snapshot" in _html(result) + + +def test_returns_none_on_missing_required_columns(): + bad = "year,week,cases,country\n2026,1,10,Brazil\n" + result = paho_oropouche_portal.fetch( + "https://www.paho.org/en/arbo-portal/arbo-portal-oropouche", + csv_fetcher=_fetcher(bad), + ) + assert result is None + + +def test_oropouche_scraper_output_reaches_insight_llm(monkeypatch): + monkeypatch.setattr(paho_oropouche_portal, "_fetch_csv_text", lambda _url, _cfg: _CSV) + + fdoc = FilteredDocument( + result_id="r-orov-1", + question_id="q-orov-1", + url="https://www.paho.org/en/arbo-portal/arbo-portal-oropouche", + canonical_url="https://www.paho.org/en/arbo-portal/arbo-portal-oropouche", + domain="www.paho.org", + title="PAHO Oropouche portal", + snippet="Dashboard", + published_date=None, + file_type="html", + relevance_score=1.0, + credibility_score=1.0, + final_score=1.0, + source_tier="official", + is_official_domain=True, + selection_reasons=["test"], + extraction_priority=1, + extraction_mode="full", + expected_value="high", + source_id="paho_oropouche_portal", + region="Americas", + question_text="How many countries/territories report Oropouche?", + ) + + question = ForecastQuestion( + id="q-orov-1", + text="How many Americas countries or territories report confirmed Oropouche?", + created_at=datetime(2026, 1, 1, tzinfo=timezone.utc), + region="Americas", + pathogen="Oropouche", + event_type="case_count", + ) + + docs = ExtractionPipeline(config=ExtractionConfig()).run([fdoc]) + assert len(docs) == 1 + assert docs[0].status == "success" + assert docs[0].chunks + + llm = FakeLLMClient([ + LLMResponse( + content={"facts": []}, + input_tokens=100, + output_tokens=10, + model="gpt-4o-mini", + raw_text='{"facts": []}', + ) + ]) + + insight = InsightPipeline( + llm_client=llm, + config=InsightConfig( + retrieval_top_k=1, + max_chunks_per_document=1, + low_survival_top_k=1, + ), + ).run(question, docs) + + assert insight.documents_processed == 1 + assert llm.call_count == 1 diff --git a/bioscancast/tests/test_usda_aphis_livestock_scraper.py b/bioscancast/tests/test_usda_aphis_livestock_scraper.py new file mode 100644 index 0000000..60ab2ba --- /dev/null +++ b/bioscancast/tests/test_usda_aphis_livestock_scraper.py @@ -0,0 +1,176 @@ +from __future__ import annotations + +from datetime import datetime, timezone + +from bioscancast.llm.fake_client import FakeLLMClient +from bioscancast.llm.base import LLMResponse +from bioscancast.stages.extraction.config import ExtractionConfig +from bioscancast.stages.extraction.pipeline import ExtractionPipeline +from bioscancast.stages.extraction.custom_scrapers import usda_aphis_livestock +from bioscancast.stages.filtering.models import FilteredDocument, ForecastQuestion +from bioscancast.stages.insight.config import InsightConfig +from bioscancast.stages.insight.pipeline import InsightPipeline + + +_CSV = """Confirmed Diagnosis,State,County +2024-01-01,IGNORE,NA +2026-01-05,CA,A +2026-01-15,CA,B +2026-02-10,TX,C +2026-03-12,TX,D +2026-03-20,TX,E +2026-04-02,NY,F +2026-04-02,CA,G +2026-04-03,CA,H +""" + + +def _fetcher(text: str = _CSV): + return lambda url, config: text + + +def _html(result) -> str: + assert result is not None + assert result.content_type == "text/html" + return result.content_bytes.decode("utf-8") + + +def _asof(s: str) -> datetime: + return datetime.strptime(s, "%Y-%m-%d").replace(tzinfo=timezone.utc) + + +def test_renders_requested_analytics_sections(): + result = usda_aphis_livestock.fetch( + "https://www.aphis.usda.gov/livestock-poultry-disease/avian/avian-influenza/hpai-detections/hpai-confirmed-cases-livestock", + csv_fetcher=_fetcher(), + ) + body = _html(result) + + assert "Monthly counts from Confirmed Diagnosis" in body + assert "Model fit on past 6 months (monthly counts)" in body + assert "Daily counts from Confirmed Diagnosis" in body + assert "Model fit on past 30 days (daily counts)" in body + assert "Per-state summary" in body + assert "Linear model y = intercept + slope*x" in body + assert "Exponential model y = a*exp(b*x)" in body + assert "Cumulative confirmed cases in livestock (from CSV rows): 8" in body + assert "Affected states and first detected dates" in body + assert "CA2026-01-05" in body + assert "TX2026-02-10" in body + assert "NY2026-04-02" in body + + +def test_ignores_first_row_before_aggregation(): + result = usda_aphis_livestock.fetch( + "https://www.aphis.usda.gov/livestock-poultry-disease/avian/avian-influenza/hpai-detections/hpai-confirmed-cases-livestock", + csv_fetcher=_fetcher(), + ) + body = _html(result) + + # The first row (2024-01-01, IGNORE) should not appear in output. + assert "2024-01" not in body + assert "IGNORE" not in body + # Remaining months do appear. + assert "2026-01" in body + assert "2026-04" in body + + +def test_as_of_date_applies_cutoff(): + result = usda_aphis_livestock.fetch( + "https://www.aphis.usda.gov/livestock-poultry-disease/avian/avian-influenza/hpai-detections/hpai-confirmed-cases-livestock", + as_of_date=_asof("2026-03-31"), + csv_fetcher=_fetcher(), + ) + body = _html(result) + + assert "2026-04" not in body + assert "2026-03" in body + + +def test_dispatcher_uses_source_id_custom_scraper(monkeypatch): + from bioscancast.stages.extraction import fetcher as fetcher_mod + + monkeypatch.setattr(usda_aphis_livestock, "_fetch_csv_text", lambda _url, _cfg: _CSV) + + result = fetcher_mod.fetch( + "https://www.aphis.usda.gov/livestock-poultry-disease/avian/avian-influenza/hpai-detections/hpai-confirmed-cases-livestock", + config=ExtractionConfig(), + source_id="usda_aphis_livestock", + ) + assert result is not None + assert result.fetch_strategy == "custom:usda_aphis_livestock" + assert "USDA APHIS HPAI Confirmed Cases in Livestock - analytics snapshot" in _html(result) + + +def test_returns_none_on_missing_columns(): + bad = "date,state\n2026-01-01,CA\n2026-01-02,TX\n" + result = usda_aphis_livestock.fetch( + "https://www.aphis.usda.gov/livestock-poultry-disease/avian/avian-influenza/hpai-detections/hpai-confirmed-cases-livestock", + csv_fetcher=_fetcher(bad), + ) + assert result is None + + +def test_usda_scraper_output_reaches_insight_llm(monkeypatch): + monkeypatch.setattr(usda_aphis_livestock, "_fetch_csv_text", lambda _url, _cfg: _CSV) + + fdoc = FilteredDocument( + result_id="r-usda-1", + question_id="q-usda-1", + url="https://www.aphis.usda.gov/livestock-poultry-disease/avian/avian-influenza/hpai-detections/hpai-confirmed-cases-livestock", + canonical_url="https://www.aphis.usda.gov/livestock-poultry-disease/avian/avian-influenza/hpai-detections/hpai-confirmed-cases-livestock", + domain="aphis.usda.gov", + title="USDA APHIS HPAI Confirmed Cases in Livestock", + snippet="Dashboard", + published_date=None, + file_type="html", + relevance_score=1.0, + credibility_score=1.0, + final_score=1.0, + source_tier="official", + is_official_domain=True, + selection_reasons=["test"], + extraction_priority=1, + extraction_mode="full", + expected_value="high", + source_id="usda_aphis_livestock", + region="United States", + question_text="How many livestock detections are reported?", + ) + + question = ForecastQuestion( + id="q-usda-1", + text="How many HPAI confirmed cases in livestock are reported in the US?", + created_at=datetime(2026, 1, 1, tzinfo=timezone.utc), + region="United States", + pathogen="H5N1", + event_type="case_count", + ) + + docs = ExtractionPipeline(config=ExtractionConfig()).run([fdoc]) + assert len(docs) == 1 + assert docs[0].status == "success" + assert docs[0].chunks + + # Empty fact response is enough to prove the USDA-derived chunks are sent + # to extraction LLM calls in the insight stage. + llm = FakeLLMClient([ + LLMResponse( + content={"facts": []}, + input_tokens=100, + output_tokens=10, + model="gpt-4o-mini", + raw_text='{"facts": []}', + ) + ]) + insight = InsightPipeline( + llm_client=llm, + config=InsightConfig( + retrieval_top_k=1, + max_chunks_per_document=1, + low_survival_top_k=1, + ), + ).run(question, docs) + + assert insight.documents_processed == 1 + assert llm.call_count == 1 From 4f574d743ae88343ad64519a1c90b72c78884d57 Mon Sep 17 00:00:00 2001 From: rapsoj Date: Tue, 14 Jul 2026 23:42:59 +0100 Subject: [PATCH 2/2] Improve OWID extraction to get trend info from fit models --- .../custom_scrapers/_owid_common.py | 175 +++++++++++++++++- .../tests/test_owid_custom_scrapers.py | 5 + 2 files changed, 176 insertions(+), 4 deletions(-) diff --git a/bioscancast/stages/extraction/custom_scrapers/_owid_common.py b/bioscancast/stages/extraction/custom_scrapers/_owid_common.py index 6a51de2..f7a33a4 100644 --- a/bioscancast/stages/extraction/custom_scrapers/_owid_common.py +++ b/bioscancast/stages/extraction/custom_scrapers/_owid_common.py @@ -33,6 +33,8 @@ import html import io import logging +import math +import statistics from dataclasses import dataclass from datetime import datetime, timezone from typing import Callable, Optional @@ -203,6 +205,89 @@ def _series_delta(series: list[tuple[datetime, float]], n_back: int) -> float | return series[-1][1] - series[-(n_back + 1)][1] +def _r2(y: list[float], yhat: list[float]) -> float: + if not y or len(y) != len(yhat): + return 0.0 + y_mean = sum(y) / len(y) + ss_res = sum((a - b) ** 2 for a, b in zip(y, yhat)) + ss_tot = sum((a - y_mean) ** 2 for a in y) + if ss_tot <= 0.0: + return 1.0 if ss_res <= 1e-12 else 0.0 + return max(0.0, min(1.0, 1.0 - (ss_res / ss_tot))) + + +def _fit_linear(values: list[float]) -> dict[str, float] | None: + if len(values) < 2: + return None + n = len(values) + xs = list(range(n)) + x_mean = sum(xs) / n + y_mean = sum(values) / n + denom = sum((x - x_mean) ** 2 for x in xs) + if denom <= 0.0: + return None + slope = sum((x - x_mean) * (y - y_mean) for x, y in zip(xs, values)) / denom + intercept = y_mean - (slope * x_mean) + yhat = [intercept + slope * x for x in xs] + return { + "intercept": float(intercept), + "slope": float(slope), + "r2": _r2(values, yhat), + } + + +def _fit_exponential(values: list[float]) -> dict[str, float] | None: + if len(values) < 2: + return None + points = [(idx, v) for idx, v in enumerate(values) if v > 0] + if len(points) < 2: + return None + + xs = [float(idx) for idx, _ in points] + ys = [float(v) for _, v in points] + ln_ys = [math.log(v) for v in ys] + + n = len(xs) + x_mean = sum(xs) / n + y_mean = sum(ln_ys) / n + denom = sum((x - x_mean) ** 2 for x in xs) + if denom <= 0.0: + return None + + b = sum((x - x_mean) * (y - y_mean) for x, y in zip(xs, ln_ys)) / denom + ln_a = y_mean - (b * x_mean) + a = math.exp(ln_a) + yhat = [a * math.exp(b * x) for x in xs] + return { + "a": float(a), + "b": float(b), + "r2": _r2(ys, yhat), + } + + +def _fmt_num(value: float | None, *, decimals: int = 2) -> str: + if value is None: + return "n/a" + if abs(value - round(value)) < 1e-9: + return f"{int(round(value)):,}" + return f"{value:,.{decimals}f}" + + +def _stats_summary(values: list[float]) -> str: + if not values: + return "n=0" + n = len(values) + mean = sum(values) / n + std = statistics.stdev(values) if n > 1 else 0.0 + min_v = min(values) + med = statistics.median(values) + max_v = max(values) + return ( + f"n={n}, mean={_fmt_num(mean)}, std={_fmt_num(std)}, " + f"min={_fmt_num(min_v)}, median={_fmt_num(med)}, max={_fmt_num(max_v)}" + ) + + def _cumulative_line(dataset: OWIDDataset, entity: str, dt: datetime, row: dict[str, str]) -> str: """One prose sentence stating an entity's cumulative figures as of a date.""" parts = [f"As of {dt.date().isoformat()}, {html.escape(entity)}:"] @@ -213,7 +298,13 @@ def _cumulative_line(dataset: OWIDDataset, entity: str, dt: datetime, row: dict[ return " ".join(parts).rstrip(";") -def _trend_table_html(dataset: OWIDDataset, entity: str, ordered_rows: list[tuple[datetime, dict[str, str]]]) -> list[str]: +def _trend_table_html( + dataset: OWIDDataset, + entity: str, + ordered_rows: list[tuple[datetime, dict[str, str]]], + *, + is_target_entity: bool, +) -> list[str]: cols = [c for c in dataset.trend_columns if ordered_rows and c in ordered_rows[-1][1]] if not cols: return [] @@ -221,16 +312,85 @@ def _trend_table_html(dataset: OWIDDataset, entity: str, ordered_rows: list[tupl value_series = [(dt, _parse_float(row.get(dataset.value_col))) for dt, row in series] d4 = _series_delta([(dt, v) for dt, v in value_series if v is not None], 4) d12 = _series_delta([(dt, v) for dt, v in value_series if v is not None], 12) + d7 = _series_delta([(dt, v) for dt, v in value_series if v is not None], 7) + d30 = _series_delta([(dt, v) for dt, v in value_series if v is not None], 30) lines = [f"

{html.escape(entity)} — recent trend

"] - if value_series and value_series[-1][1] is not None: - latest_dt, latest_val = value_series[-1] + clean_series = [(dt, v) for dt, v in value_series if v is not None] + if clean_series and clean_series[-1][1] is not None: + latest_dt, latest_val = clean_series[-1] + row_span_days = ( + (clean_series[-1][0] - clean_series[0][0]).days + if len(clean_series) > 1 + else 0 + ) lines.append( f"

Latest {html.escape(dataset.value_col)} " f"({latest_dt.date().isoformat()}): {latest_val:,.0f}; " f"delta_4_rows: {d4 if d4 is not None else 'n/a'}; " f"delta_12_rows: {d12 if d12 is not None else 'n/a'}

" ) + + cumulative_values = [v for _, v in clean_series] + cumulative_fit_window = cumulative_values[-30:] + cumulative_linear = _fit_linear(cumulative_fit_window) + increments = [ + cumulative_values[i] - cumulative_values[i - 1] + for i in range(1, len(cumulative_values)) + ] + + lines.append( + f"

Trend summary ({html.escape(dataset.value_col)}, {html.escape(entity)}): " + f"points={len(cumulative_values)}, span_days={row_span_days}, " + f"delta_7_rows={_fmt_num(d7)}, delta_30_rows={_fmt_num(d30)}, " + f"recent_increment_stats={_stats_summary(increments)}.

" + ) + + if cumulative_linear is None: + lines.append( + "

Linear fit on recent cumulative trend: unavailable " + "(insufficient points).

" + ) + else: + lines.append( + "

Linear fit on recent cumulative trend " + f"({html.escape(dataset.value_col)}, last {len(cumulative_fit_window)} rows): " + f"intercept={cumulative_linear['intercept']:.6f}, " + f"slope={cumulative_linear['slope']:.6f} per row, " + f"R^2={cumulative_linear['r2']:.6f}.

" + ) + + if "new_cases" in cols: + new_case_values = [ + _parse_float(row.get("new_cases")) + for _dt, row in ordered_rows + ] + new_case_values = [v for v in new_case_values if v is not None] + new_case_fit_window = new_case_values[-30:] + exp_fit = _fit_exponential(new_case_fit_window) + lines.append( + f"

Incident trend stats (new_cases, {html.escape(entity)}): " + f"{_stats_summary(new_case_fit_window)}.

" + ) + if exp_fit is None: + lines.append( + "

Exponential fit on recent new_cases: unavailable " + "(insufficient strictly-positive points).

" + ) + else: + lines.append( + "

Exponential fit on recent new_cases " + f"(last {len(new_case_fit_window)} rows): " + f"a={exp_fit['a']:.6f}, b={exp_fit['b']:.6f}, " + f"R^2={exp_fit['r2']:.6f}.

" + ) + + if is_target_entity: + lines.append( + f"

Region/question-target focus: {html.escape(entity)} trend " + "statistics and model-fit outputs are included in this section.

" + ) + header = "".join(f"{html.escape(c)}" for c in cols) lines.append(f"{header}") for _dt, row in ordered_rows[-dataset.trend_rows:]: @@ -348,7 +508,14 @@ def latest_value(loc: str) -> float: f"{_cumulative_line(dataset, entity, dt, row)}. " f"Source: {html.escape(dataset.csv_url)}.

" ) - summary_lines.extend(_trend_table_html(dataset, entity, rows_by_location[entity])) + summary_lines.extend( + _trend_table_html( + dataset, + entity, + rows_by_location[entity], + is_target_entity=entity in target_locations, + ) + ) if top_locations: summary_lines.append("

Top locations by latest cumulative value

") diff --git a/bioscancast/tests/test_owid_custom_scrapers.py b/bioscancast/tests/test_owid_custom_scrapers.py index 6ddf1c9..1e72d48 100644 --- a/bioscancast/tests/test_owid_custom_scrapers.py +++ b/bioscancast/tests/test_owid_custom_scrapers.py @@ -112,6 +112,9 @@ def test_emits_prose_and_trend_table(self): # Trend columns present, and the last World trend row is the cutoff date. assert "" in body assert "2025-03-05" in body + assert "Trend summary (total_cases, World):" in body + assert "Linear fit on recent cumulative trend (total_cases" in body + assert "Incident trend stats (new_cases, World):" in body def test_returns_none_when_no_rows_before_cutoff(self): assert ( @@ -146,6 +149,8 @@ def test_region_surfaces_target_entity(self): ) assert "

Africa

" in body assert "cumulative confirmed cases (Africa): 41,000" in body + assert "Region/question-target focus: Africa trend statistics" in body + assert "Trend summary (total_cases, Africa):" in body def test_question_text_infers_target_entity(self): body = _html(
total_cases