rates/backend/jobs/check_scrape_health.py

"""
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()