Add stale-scrape health alerting (email + in-app banner)

Silent scraper breakage (200-with-empty-body) went unnoticed for 16 days
because nothing flagged it. Now, after each daily direct scrape, a health
check compares every enabled competitor's freshest captured rate against a
staleness threshold (default 5 days, config: scrape_health_stale_days).

State lives in a new direct_scrape_health table so we alert on transitions
only — one email when a competitor goes stale, one when it recovers, never a
daily repeat. Email uses the stack's shared SMTP integration via
central_settings (recipient: scrape_health_alert_email, else SMTP reply_to/
from) — no new mail secrets in this app. The current stale set also drives an
in-app warning banner on the Direct Rates page (GET /direct/health), with an
on-demand re-check endpoint (POST /direct/health/check).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

backend/api/direct.py

@@ -121,6 +121,31 @@ async def list_hotels(user=Depends(get_current_user)):
         return [dict(r) for r in rows]
 
 
+@router.get("/health")
+async def scrape_health(user=Depends(get_current_user)):
+    """Competitor hotels currently flagged as stale — drives the Direct Rates banner."""
+    require_cap(user, "view_direct_rates")
+    async with AsyncSessionLocal() as db:
+        rows = (await db.execute(text("""
+            SELECT h.id AS hotel_id, h.name,
+                   hh.fresh_as_of, hh.stale_since
+            FROM direct_competitor_hotels h
+            JOIN direct_scrape_health hh ON hh.hotel_id = h.id
+            WHERE hh.state = 'stale' AND h.scrape_enabled = true
+            ORDER BY h.name
+        """))).mappings().all()
+        return {"stale": [dict(r) for r in rows]}
+
+
+@router.post("/health/check")
+async def scrape_health_check(background_tasks: BackgroundTasks, user=Depends(get_current_user)):
+    """Re-evaluate scrape freshness on demand (also runs automatically after each scrape)."""
+    require_cap(user, "manage_hotels")
+    from jobs.check_scrape_health import check_scrape_health
+    background_tasks.add_task(check_scrape_health)
+    return {"status": "checking"}
+
+
 @router.post("/hotels", status_code=201)
 async def create_hotel(body: HotelCreate, user=Depends(get_current_user)):
     require_cap(user, "manage_hotels")

backend/jobs/check_scrape_health.py

@@ -0,0 +1,133 @@
+"""
+Direct scrape health check.
+
+Runs after the daily direct scrape. For each enabled competitor hotel it
+compares the freshest captured rate (direct_rates.scraped_at) against a
+staleness threshold. A hotel with no fresh data for longer than the threshold
+is flagged 'stale' — this is exactly the failure mode that let the Guestline /
+Mews engine changes go unnoticed for 16 days (scrapes "ran" but saved nothing).
+
+State is tracked in direct_scrape_health so we alert on *transitions* only —
+one email when a competitor goes stale, one when it recovers — never a daily
+repeat. The current stale set also drives the in-app banner (GET /direct/health).
+
+Config (system_config):
+  scrape_health_stale_days    days without fresh data before alerting (default 5)
+  scrape_health_alert_email   recipient; falls back to the SMTP reply_to/from
+"""
+import logging
+from datetime import datetime, timezone, timedelta
+
+from sqlalchemy import text
+
+from database import SyncSessionLocal
+from services.mailer import send_mail
+from services.central_settings import get_integration_sync
+
+log = logging.getLogger(__name__)
+
+DEFAULT_STALE_DAYS = 5
+
+
+def _cfg(db, key: str, default):
+    row = db.execute(
+        text("SELECT config_value FROM system_config WHERE config_key = :k"), {"k": key}
+    ).fetchone()
+    return row[0] if row and row[0] not in (None, "") else default
+
+
+def check_scrape_health():
+    db = SyncSessionLocal()
+    try:
+        try:
+            stale_days = int(_cfg(db, "scrape_health_stale_days", DEFAULT_STALE_DAYS) or DEFAULT_STALE_DAYS)
+        except (ValueError, TypeError):
+            stale_days = DEFAULT_STALE_DAYS
+        now = datetime.now(timezone.utc)
+        cutoff = now - timedelta(days=stale_days)
+
+        rows = db.execute(text("""
+            SELECT h.id, h.name, h.scrape_enabled,
+                   (SELECT MAX(scraped_at) FROM direct_rates r      WHERE r.hotel_id = h.id) AS fresh_as_of,
+                   (SELECT MIN(scraped_at) FROM direct_scrape_runs s WHERE s.hotel_id = h.id) AS first_attempt,
+                   hh.state AS prev_state
+            FROM direct_competitor_hotels h
+            LEFT JOIN direct_scrape_health hh ON hh.hotel_id = h.id
+            ORDER BY h.id
+        """)).mappings().fetchall()
+
+        transitions = []  # (row, desired_state)
+        for h in rows:
+            prev = h["prev_state"] or "ok"
+            first = h["first_attempt"]
+            fresh = h["fresh_as_of"]
+
+            # Not yet eligible for staleness alerting: disabled, or too new to
+            # have had a fair chance (no attempt, or first attempt within window).
+            if not h["scrape_enabled"] or first is None or first > cutoff:
+                desired = "ok"
+            else:
+                desired = "stale" if (fresh is None or fresh < cutoff) else "ok"
+
+            if desired != prev:
+                transitions.append((h, desired))
+
+            db.execute(text("""
+                INSERT INTO direct_scrape_health (hotel_id, state, fresh_as_of, stale_since, updated_at)
+                VALUES (:id, :st, :fresh, :now, :now)
+                ON CONFLICT (hotel_id) DO UPDATE SET
+                    state       = EXCLUDED.state,
+                    fresh_as_of = EXCLUDED.fresh_as_of,
+                    stale_since = CASE WHEN EXCLUDED.state = 'stale'
+                                       THEN COALESCE(direct_scrape_health.stale_since, EXCLUDED.stale_since)
+                                       ELSE NULL END,
+                    updated_at  = EXCLUDED.updated_at
+            """), {"id": h["id"], "st": desired, "fresh": fresh, "now": now})
+
+        db.commit()
+
+        if transitions:
+            _notify(db, transitions, stale_days)
+    except Exception as e:
+        log.error(f"Scrape health check failed: {e}")
+    finally:
+        db.close()
+
+
+def _notify(db, transitions, stale_days: int):
+    stale = [h for h, s in transitions if s == "stale"]
+    recovered = [h for h, s in transitions if s == "ok"]
+
+    lines = []
+    if stale:
+        lines.append(f"The following competitor scrapes have gone STALE — no fresh rates in over {stale_days} days:")
+        for h in stale:
+            fa = h["fresh_as_of"]
+            when = fa.strftime("%Y-%m-%d %H:%M UTC") if fa else "never"
+            lines.append(f"  • {h['name']} — last fresh data: {when}")
+        lines.append("")
+    if recovered:
+        lines.append("Recovered — fresh data flowing again:")
+        for h in recovered:
+            lines.append(f"  • {h['name']}")
+        lines.append("")
+    lines.append("Check the Direct Rates tab in Rate Monitor for detail.")
+    body = "\n".join(lines)
+
+    subject = (f"⚠ Rate Monitor: {len(stale)} competitor scrape(s) stale"
+               if stale else "Rate Monitor: competitor scrape(s) recovered")
+
+    recipient = _cfg(db, "scrape_health_alert_email", "")
+    if not recipient:
+        smtp = get_integration_sync("smtp") or {}
+        recipient = smtp.get("reply_to") or smtp.get("from") or ""
+
+    if recipient:
+        send_mail(recipient, subject, body)
+        log.info(f"Scrape-health alert emailed to {recipient}: {len(stale)} stale, {len(recovered)} recovered")
+    else:
+        log.warning("No scrape-health alert recipient configured — email skipped (banner still updated)")
+
+    ids = [h["id"] for h, _ in transitions]
+    db.execute(text("UPDATE direct_scrape_health SET notified_at = NOW() WHERE hotel_id = ANY(:ids)"), {"ids": ids})
+    db.commit()

backend/jobs/scrape_direct_rates.py

@@ -41,3 +41,10 @@ def run_scrape_all_direct():
             log.error(f"Direct scrape failed for hotel {hotel['id']} ({hotel['name']}): {e}")
 
     log.info("Direct rate scrape complete")
+
+    # Flag any competitor whose data has gone stale (and alert on transitions).
+    try:
+        from jobs.check_scrape_health import check_scrape_health
+        check_scrape_health()
+    except Exception as e:
+        log.error(f"Scrape health check failed: {e}")

backend/schema.sql

@@ -331,6 +331,19 @@ CREATE INDEX IF NOT EXISTS idx_direct_rates_scraped    ON direct_rates(scraped_a
 ALTER TABLE booking_com_hotels ADD COLUMN IF NOT EXISTS direct_hotel_id INTEGER
     REFERENCES direct_competitor_hotels(id) ON DELETE SET NULL;
 
+-- ============================================
+-- DIRECT SCRAPE HEALTH (per-hotel freshness state, drives stale-data alerting)
+-- ============================================
+
+CREATE TABLE IF NOT EXISTS direct_scrape_health (
+    hotel_id     INTEGER PRIMARY KEY REFERENCES direct_competitor_hotels(id) ON DELETE CASCADE,
+    state        TEXT NOT NULL DEFAULT 'ok',   -- 'ok' | 'stale'
+    fresh_as_of  TIMESTAMPTZ,                  -- freshest direct_rates.scraped_at seen for this hotel
+    stale_since  TIMESTAMPTZ,                  -- when it first went stale (cleared on recovery)
+    notified_at  TIMESTAMPTZ,                  -- last time a transition email was sent
+    updated_at   TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
 -- Pin the Booking.com destination: free-text ss= searches non-deterministically
 -- resolve to the wrong place (Stow on the Wold once matched St. Wolfgang, AT).
 ALTER TABLE booking_scrape_config ADD COLUMN IF NOT EXISTS dest_id VARCHAR(32);

backend/services/mailer.py

@@ -0,0 +1,52 @@
+"""
+Minimal SMTP sender for the rates app.
+
+Reuses the stack's shared SMTP integration (managed once in the Settings
+app, same as the tasks app's mailer) via central_settings — no per-app mail
+secrets. Fire-and-forget: a mail failure must never break the scheduler.
+"""
+import logging
+import os
+import smtplib
+from email.message import EmailMessage
+
+from services.central_settings import get_integration_sync
+
+logger = logging.getLogger(__name__)
+
+HOTEL_NAME = os.getenv("VITE_HOTEL_NAME", "Hotel")
+
+
+def send_mail(to: str, subject: str, body: str) -> bool:
+    """Send a plain-text email through the central SMTP integration.
+    Returns True on success. Never raises — logs and returns False instead."""
+    if not to:
+        return False
+    cfg = get_integration_sync("smtp")
+    if not cfg or not cfg.get("host"):
+        logger.warning("SMTP not configured in central settings — email skipped")
+        return False
+
+    try:
+        msg = EmailMessage()
+        msg["From"] = cfg.get("from") or f"{HOTEL_NAME} Rates <noreply@localhost>"
+        msg["To"] = to
+        msg["Subject"] = subject
+        if cfg.get("reply_to"):
+            msg["Reply-To"] = cfg["reply_to"]
+        msg.set_content(body)
+
+        port = int(cfg.get("port") or 587)
+        with smtplib.SMTP(cfg["host"], port, timeout=15) as s:
+            if port != 465:
+                try:
+                    s.starttls()
+                except smtplib.SMTPException:
+                    pass  # server without STARTTLS (e.g. local relay)
+            if cfg.get("user"):
+                s.login(cfg["user"], cfg.get("pass") or "")
+            s.send_message(msg)
+        return True
+    except Exception as e:
+        logger.error(f"Email send failed: {e}")
+        return False