gendesign/tradein-mvp/backend/app/tasks/deals_freshness_monitor.py

"""Мониторинг свежести ДАННЫХ deals (не статуса джобы) — #2212.

Проблема аудита: rosreestr_quarter_poll / rosreestr_dkp_import честно рапортуют
`done` (rows_fetched>0, inserted=0 == «всё уже импортировано»), но это МАСКИРУЕТ
staleness самих данных. `deals.deal_date` — лейбл начала квартала (2026-01-01 = 2026Q1);
Росреестр публикует квартальными дампами, скачивание дампа — РУЧНОЙ операторский шаг
(poll лишь алертит о доступности). Если оператор пропустил импорт нового квартала,
джоба-статус остаётся зелёным, а данные тихо устаревают.

Этот монитор смотрит на `max(deal_date)` и поднимает ERROR-алерт (→ GlitchTip через
sentry logging integration), когда следующий квартал ПРОСРОЧЕН публикацией.

Арифметика просрочки (проверена на проде 2026-07-02, max=2026Q1, q_end=2026-03-31):
  latest_q_start    = max(deal_date)                         (2026-01-01)
  latest_q_end      = latest_q_start + 3 months              (2026-04-01, старт Q2)
  overdue_threshold = latest_q_end + 3 months + LAG_DAYS     (2026-07-01 + 45д = 2026-08-15)
  overdue           = now() > overdue_threshold
Смысл: следующий квартал (Q2) заканчивается через 3 мес после конца имеющегося; даём
LAG_ALLOWANCE_DAYS (по умолчанию 45) на публикацию дампа Росреестром. Сейчас (2026-07-02)
до порога 2026-08-15 ещё далеко → алерта НЕТ. Если к 2026-08-16 Q2 не импортирован → алерт.

Задача синхронная (DB-only, один SELECT max(deal_date)) — запускается kit-scheduler'ом
через product_handlers._job_deals_freshness_monitor в run_in_executor, по образцу
listing_source_snapshot / rosreestr_dkp_import. Вердикт вычисляет ЧИСТАЯ функция
evaluate_deals_freshness() (frozen-now тестируется без БД).

Прогон НЕ помечается failed при алерте (это МОНИТОР, а не сбой джобы) — ERROR-лог
достаточен. mark_failed только если deals недоступна/пуста (нечего оценивать).
"""

from __future__ import annotations

import logging
from dataclasses import dataclass
from datetime import UTC, date, datetime, timedelta

from sqlalchemy import text
from sqlalchemy.orm import Session

from app.services import scrape_runs as runs_mod

logger = logging.getLogger(__name__)

__all__ = [
    "DEFAULT_LAG_ALLOWANCE_DAYS",
    "FreshnessVerdict",
    "check_deals_freshness",
    "evaluate_deals_freshness",
]

# Лаг публикации дампа Росреестром: сколько дней после конца следующего квартала
# считаем допустимой задержкой до алерта. Читается из default_params.lag_allowance_days.
DEFAULT_LAG_ALLOWANCE_DAYS = 45

_LATEST_DEAL_DATE_SQL = text("SELECT max(deal_date) AS latest FROM deals")


@dataclass(frozen=True)
class FreshnessVerdict:
    """Вердикт свежести deals по max(deal_date)."""

    latest_q_start: date
    latest_q_end: date
    overdue_threshold: date
    overdue: bool
    days_overdue: int


def _add_months(d: date, months: int) -> date:
    """Прибавить `months` к дате. deal_date всегда старт квартала (день=1), но
    хелпер устойчив к произвольному дню — при переполнении зажимает на конец месяца.
    """
    total = (d.month - 1) + months
    year = d.year + total // 12
    month = total % 12 + 1
    # Зажим дня на последний день целевого месяца (для дней 29-31; для day=1 no-op).
    if month == 12:
        next_month_first = date(year + 1, 1, 1)
    else:
        next_month_first = date(year, month + 1, 1)
    last_day = (next_month_first - timedelta(days=1)).day
    return date(year, month, min(d.day, last_day))


def evaluate_deals_freshness(
    latest_q_start: date,
    now: datetime,
    lag_allowance_days: int = DEFAULT_LAG_ALLOWANCE_DAYS,
) -> FreshnessVerdict:
    """Чистая логика: просрочен ли следующий квартал относительно max(deal_date).

    overdue = now() > (latest_q_start + 6 месяцев + lag_allowance_days).
    days_overdue = max(0, дни сверх порога) — 0 когда не просрочено.

    Тестируется с frozen `now` без БД.
    """
    latest_q_end = _add_months(latest_q_start, 3)
    overdue_threshold = _add_months(latest_q_start, 6) + timedelta(days=lag_allowance_days)
    now_date = now.date()
    overdue = now_date > overdue_threshold
    days_overdue = max(0, (now_date - overdue_threshold).days)
    return FreshnessVerdict(
        latest_q_start=latest_q_start,
        latest_q_end=latest_q_end,
        overdue_threshold=overdue_threshold,
        overdue=overdue,
        days_overdue=days_overdue,
    )


def check_deals_freshness(
    db: Session,
    run_id: int,
    params: dict | None = None,  # type: ignore[type-arg]
    now: datetime | None = None,
) -> dict[str, int]:
    """Проверить свежесть deals по max(deal_date) и поднять алерт при просрочке.

    Sync (вызывается scheduler-триггером в executor, как snapshot_listing_sources).
    Читает один SELECT max(deal_date), считает вердикт чистой функцией, логирует
    ERROR при overdue (→ GlitchTip) и финализирует run.

    Params (default_params jsonb):
        lag_allowance_days: int — допустимый лаг публикации дампа (default 45).
    `now` инъектируется в тестах (frozen); в проде — None → datetime.now(UTC).

    Returns counters {latest_year, latest_quarter, days_overdue, alert}.
    mark_failed только если deals пуста/недоступна (нечего оценивать); при алерте
    прогон помечается done (это монитор, не сбой джобы).
    """
    params = params or {}
    now = now or datetime.now(UTC)
    counters: dict[str, int] = {
        "latest_year": 0,
        "latest_quarter": 0,
        "days_overdue": 0,
        "alert": 0,
    }
    try:
        runs_mod.update_heartbeat(db, run_id, counters)

        row = db.execute(_LATEST_DEAL_DATE_SQL).first()
        latest: date | None = row.latest if row is not None else None
        if latest is None:
            # ERROR (#2674): монитор не может выполнить работу — сбой, а не наблюдение.
            # Соседняя ветка (overdue) писала ERROR с самого начала; эта расходилась.
            logger.error(
                "deals freshness: таблица deals пуста/недоступна — оценить свежесть нельзя"
            )
            runs_mod.mark_failed(db, run_id, "deals empty or unavailable", counters)
            return counters

        lag_days = int(params.get("lag_allowance_days", DEFAULT_LAG_ALLOWANCE_DAYS))
        verdict = evaluate_deals_freshness(latest, now, lag_days)

        quarter = (latest.month - 1) // 3 + 1
        counters = {
            "latest_year": latest.year,
            "latest_quarter": quarter,
            "days_overdue": verdict.days_overdue,
            "alert": int(verdict.overdue),
        }

        if verdict.overdue:
            logger.error(
                "deals freshness: max(deal_date)=%s — новый квартал просрочен на %d дней; "
                "проверь rosreestr_quarter_poll + операторский импорт "
                "(data/sql/02_load_all_quarters.sh)",
                latest,
                verdict.days_overdue,
            )
        else:
            logger.info(
                "deals freshness: max(deal_date)=%s (Q%d %d) свежий — порог просрочки %s "
                "ещё не наступил, алерта нет",
                latest,
                quarter,
                latest.year,
                verdict.overdue_threshold,
            )

        runs_mod.mark_done(db, run_id, counters)
        logger.info(
            "check_deals_freshness run_id=%d done: latest=%s alert=%d days_overdue=%d",
            run_id,
            latest,
            counters["alert"],
            counters["days_overdue"],
        )
        return counters
    except Exception as exc:
        logger.exception("check_deals_freshness run_id=%d failed", run_id)
        try:
            db.rollback()
        except Exception:
            pass
        runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters)
        raise