")
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/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"| {html.escape(key_header)} | sum_week_lab_confirmed |
|---|
"
+ f"{body}
"
+ )
+
+
+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())}.
"
+ ""
+ "| Pais | cumulative_week_lab_confirmed |
|---|
"
+ f"{''.join(rows)}
"
+ )
+
+
+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"| {html.escape(key_header)} | count |
|---|
"
+ f"{body}
"
+ )
+
+
+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}.
"
+ ""
+ "| State | case_count |
|---|
"
+ f"{''.join(rows)}
"
+ )
+
+
+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])}.
"
+ ""
+ "| State | first_confirmed_diagnosis_date |
|---|
"
+ f"{''.join(rows)}
"
+ )
+
+
+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_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 "total_cases | " 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(
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 "CA | 2026-01-05 | " in body
+ assert "TX | 2026-02-10 | " in body
+ assert "NY | 2026-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