diff --git a/README.md b/README.md index 72d03de..0c09441 100644 --- a/README.md +++ b/README.md @@ -539,6 +539,18 @@ Additional packages: pip install openai tavily-python python-dotenv ``` +The USDA APHIS livestock scraper drives a headless browser to export the +Tableau crosstab. After installing `requirements.txt`, download the browser +once: + +```bash +playwright install chromium +``` + +If the browser is unavailable, that scraper degrades gracefully (it returns +no data and the generic fetcher is used instead) — the rest of the pipeline +is unaffected. + --- ## Environment Variables 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..23d9377 --- /dev/null +++ b/bioscancast/stages/extraction/custom_scrapers/usda_aphis_livestock.py @@ -0,0 +1,419 @@ +"""Custom scraper for USDA APHIS HPAI livestock detections. + +The public APHIS dashboard is a client-rendered Tableau page; this scraper uses +headless browser automation to export the Tableau crosstab CSV and renders +compact analytical prose/tables so the regular HTML parser + insight extraction +pipeline can operate without additional parser changes. +""" + +from __future__ import annotations + +import html +import logging +from datetime import datetime, timezone +from typing import Callable, Optional + +import numpy as np +import pandas as pd + +from bioscancast.stages.extraction.config import ExtractionConfig +from bioscancast.stages.extraction.fetcher import FetchResult + +logger = logging.getLogger(__name__) + +TABLEAU_VIEW_URL = ( + "https://publicdashboards.dl.usda.gov/t/MRP_PUB/views/" + "VS_Cattle_HPAIConfirmedDetections2024/HPAI2022ConfirmedDetections" +) + +TableauFetcher = Callable[[str, ExtractionConfig], Optional[pd.DataFrame]] + + +def _embed_url(url: str) -> str: + if ":embed=y" in url: + return url + sep = "&" if "?" in url else "?" + return f"{url}{sep}:showVizHome=no&:embed=y" + + +def _read_csv_with_fallback(path: str) -> Optional[pd.DataFrame]: + attempts = [ + {"encoding": "utf-16", "sep": "\t", "skiprows": 1}, + {"encoding": "utf-16-le", "sep": "\t", "skiprows": 1}, + {"encoding": "utf-8-sig", "sep": "\t", "skiprows": 1}, + {"encoding": "utf-8", "sep": "\t", "skiprows": 1}, + {"encoding": "utf-16"}, + {"encoding": "utf-16-le"}, + {"encoding": "utf-8-sig"}, + {"encoding": "utf-8"}, + {"encoding": "latin-1"}, + ] + for kwargs in attempts: + try: + df = pd.read_csv(path, **kwargs) + logger.info("USDA Tableau: parsed CSV with options=%s", kwargs) + return df + except UnicodeDecodeError: + logger.info("USDA Tableau: decode failed with options=%s", kwargs) + continue + except Exception as exc: + logger.info("USDA Tableau: csv parse error with options=%s: %s", kwargs, exc) + continue + return None + + +def _fetch_tableau_dataframe(url: str, cfg: ExtractionConfig) -> Optional[pd.DataFrame]: + try: + from playwright.sync_api import sync_playwright + except Exception as exc: + logger.info("USDA Tableau: Playwright unavailable: %s", exc) + return None + + timeout_ms = int(max(cfg.fetch_timeout_seconds, 45.0) * 1000) + + try: + with sync_playwright() as p: + logger.info("USDA Tableau: launching headless browser") + browser = p.chromium.launch(headless=True) + context = browser.new_context(accept_downloads=True) + page = context.new_page() + + target_url = _embed_url(url) + logger.info("USDA Tableau: navigating to %s", target_url) + page.goto(_embed_url(url), wait_until="domcontentloaded", timeout=timeout_ms) + + logger.info("USDA Tableau: waiting for download toolbar button") + for _ in range(200): + if page.locator("button[data-tb-test-id='viz-viewer-toolbar-button-download']").count() > 0: + break + page.wait_for_timeout(250) + download_btn_count = page.locator("button[data-tb-test-id='viz-viewer-toolbar-button-download']").count() + logger.info("USDA Tableau: download button count=%s", download_btn_count) + if download_btn_count == 0: + logger.info("USDA Tableau: download button not found before timeout") + context.close() + browser.close() + return None + + logger.info("USDA Tableau: opening download menu") + page.locator("button[data-tb-test-id='viz-viewer-toolbar-button-download']").first.click(timeout=20_000) + logger.info("USDA Tableau: selecting Crosstab") + page.locator("[data-tb-test-id='download-flyout-download-crosstab-MenuItem']").first.click(timeout=20_000) + page.wait_for_timeout(1500) + + logger.info("USDA Tableau: selecting worksheet 'Table Details by Date'") + page.locator("[data-tb-test-id^='sheet-thumbnail-']", has_text="Table Details by Date").first.click(timeout=20_000) + logger.info("USDA Tableau: selecting CSV export format") + page.locator("[data-tb-test-id='crosstab-options-dialog-radio-csv-Label']").first.click(timeout=20_000) + + downloads = [] + page.on("download", lambda d: downloads.append(d)) + logger.info("USDA Tableau: triggering export download") + page.locator("[data-tb-test-id='export-crosstab-export-Button']").first.click(timeout=20_000) + + waited = 0 + step = 250 + while waited <= timeout_ms and not downloads: + page.wait_for_timeout(step) + waited += step + logger.info("USDA Tableau: download events captured=%s after %sms", len(downloads), waited) + + if not downloads: + logger.info("USDA Tableau: no download event received") + context.close() + browser.close() + return None + download = downloads[0] + + path = download.path() + logger.info("USDA Tableau: download path resolved=%s", bool(path)) + if path is None: + logger.info("USDA Tableau: download path unavailable") + context.close() + browser.close() + return None + + logger.info("USDA Tableau: reading CSV from %s", path) + df = _read_csv_with_fallback(path) + context.close() + browser.close() + + if df is None: + logger.info("USDA Tableau: failed to parse CSV with fallback encodings") + return None + if df.empty: + logger.info("USDA Tableau: CSV parsed but DataFrame empty") + return None + logger.info("USDA Tableau: CSV parsed rows=%s cols=%s", df.shape[0], df.shape[1]) + return df + except Exception as exc: + logger.info("USDA Tableau: fetch failed with exception: %s", exc) + return None + + +def _resolve_column(df: pd.DataFrame, wanted: str) -> str | None: + by_lower = {str(c).strip().lower(): c for c in df.columns} + return by_lower.get(wanted.strip().lower()) + + +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, + tableau_fetcher: TableauFetcher | None = None, +) -> FetchResult | None: + cfg = config or ExtractionConfig() + fetched_at = datetime.now(timezone.utc) + + df = (tableau_fetcher or _fetch_tableau_dataframe)(TABLEAU_VIEW_URL, cfg) + if df is None or df.empty: + return None + + df = df.copy() + + # Requested behavior: ignore first row. + if df.shape[0] < 2: + return None + df = df.iloc[1:].copy() + + confirmed_col = _resolve_column(df, "Confirmed Diagnosis") + state_col = _resolve_column(df, "State") + if confirmed_col is None or state_col is None: + return None + + if confirmed_col != "Confirmed Diagnosis": + df = df.rename(columns={confirmed_col: "Confirmed Diagnosis"}) + if state_col != "State": + df = df.rename(columns={state_col: "State"}) + + 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"

Tableau source URL: {html.escape(TABLEAU_VIEW_URL)}

" + f"

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

" + "

This summary is computed from Tableau workbook data. The first data 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=TABLEAU_VIEW_URL, + final_url=TABLEAU_VIEW_URL, + status_code=200, + content_type="text/html", + content_bytes=rendered, + fetched_at=fetched_at, + error=None, + ) 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..7891033 --- /dev/null +++ b/bioscancast/tests/test_usda_aphis_livestock_scraper.py @@ -0,0 +1,208 @@ +from __future__ import annotations + +import io +from datetime import datetime, timezone + +import pandas as pd + +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 _df_fetcher(text: str = _CSV): + def _fetch(_url, _config): + return pd.read_csv(io.StringIO(text)) + + return _fetch + + +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", + tableau_fetcher=_df_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", + tableau_fetcher=_df_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"), + tableau_fetcher=_df_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_tableau_dataframe", + lambda _url, _cfg: pd.read_csv(io.StringIO(_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", + tableau_fetcher=_df_fetcher(bad), + ) + assert result is None + + +def test_usda_scraper_output_reaches_insight_llm(monkeypatch): + monkeypatch.setattr( + usda_aphis_livestock, + "_fetch_tableau_dataframe", + lambda _url, _cfg: pd.read_csv(io.StringIO(_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 + + llm = FakeLLMClient([ + LLMResponse( + content={ + "facts": [ + { + "event_type": "case_count", + "confidence": 0.9, + "location": "United States", + "pathogen": "H5N1", + "metric_name": "confirmed_cases", + "metric_value": 8, + "metric_unit": "cases", + "count_basis": "cumulative", + "time_window": "unknown", + "surveillance_method": None, + "data_quality": None, + "event_date": None, + "summary": "USDA cumulative count from dashboard extract.", + "quote": "Cumulative confirmed cases in livestock (from CSV rows): 8.", + } + ] + }, + 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 len(insight.records) == 1 + assert llm.call_count == 1 diff --git a/requirements.txt b/requirements.txt index 86323fa..bfefada 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,6 @@ rank-bm25 numpy +pandas>=2.0 # DataFrame ops in the evaluation stage and the custom Tableau/CSV scrapers (USDA APHIS, PAHO Oropouche) curl_cffi>=0.7,<1.0 # HTTP client with streaming support and browser TLS-fingerprint impersonation (Cloudflare workaround) trafilatura>=1.12,<2.0 # HTML main-content extraction (strips nav, ads, boilerplate) @@ -10,4 +11,5 @@ docling[chunking]>=2.90,<3.0 # TableFormer-based refinement for borderless / mer tiktoken>=0.7,<1.0 # Approximate token counting (cl100k_base encoding) openai>=1.0,<2.0 # OpenAI API client (used by filtering stage LLM calls) pycountry>=24.0 # ISO 3166-1 country name → alpha-2 lookup (insight stage location resolution) +playwright>=1.40,<2.0 # Headless-browser Tableau crosstab export for the USDA APHIS livestock scraper (requires a one-time `playwright install chromium`) pytest>=8.0,<9.0 # Testing