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

"""Монитор ОТСТАВАНИЯ ЗАГРУЗКИ СберИндекса (не календарного возраста периода).

ЧТО БЫЛО НЕ ТАК (замер на проде 2026-08-12, #2846).

Монитор мерил `now() - max(period_month)` и алертил при возрасте > 60 суток
(sber_index_max_age_days 35 + lag_allowance 25). Такой возраст НЕДОСТИЖИМО МАЛ по
построению: `period_month` — метка ПЕРВОГО числа месяца, поэтому на закрытии месяца
возрасту уже ≥30; плюс собственный лаг публикации источника. За 31 сутки прямых
наблюдений монитора (07-13 … 08-12, scrape_runs.counters) возраст лежал в 46..76 и
НИ РАЗУ не опускался ниже 46. Порог 35 у оценщика был истинным 100% времени — ноль бит.

Порог 60 у монитора не лучше: он лежит ВНУТРИ рабочего диапазона, поэтому монитор
мерил не источник, а нашу же пилу. Миграция 212 (такт 28 → 7) обещала потолок
возраста ≈46+7=53 < 60. Прод это ОПРОВЕРГ: 2026-08-12 возраст 72 при ПОЛНОМ прогоне
загрузки шестидневной давности (08-06, errors=0, upserted=639) — источник просто не
опубликовал июль. Двенадцать суток подряд (08-01 … 08-12) монитор писал ERROR при
исправной загрузке. Потолок 53 держится, только если источник публикует строго
помесячно; он не публикует.

ЧТО МЕРИМ ТЕПЕРЬ. Загрузчик тянет ВСЮ серию (limit=1000&offset=0, отсечки по периоду
нет), поэтому после прогона с errors=0 AND upserted>0 наш max(period_month) РАВЕН
максимуму источника ПО ПОСТРОЕНИЮ. Значит вопрос «отстали ли мы» — это вопрос
«давно ли был последний ЗАВЕДОМО ПОЛНЫЙ прогон», и он не зависит от возраста периода:

  последний полный прогон свежий  → наш max == max источника → источник не публиковал,
                                     молчание ПРАВИЛЬНОЕ (возраст = лаг источника);
  последний полный прогон старый  → мы не забрали → тревога про ЗАГРУЗЧИК.

ЛОВУШКА: `status='done'` НЕ означает успех — прогон id=37 (2026-05-31) имеет
{errors: 9, upserted: 0} и статус done. Успех = errors=0 AND upserted>0 (все 9 серий
3 табло × 3 региона прошли: errors — счётчик по всему прогону).

ПОРОГ — не круглое число, а такт самой загрузки: `scrape_schedules.default_params
.interval_days` для sber_index_pull, ЧИТАЕТСЯ ИЗ ТОЙ ЖЕ СТРОКИ, по которой планировщик
запускает прогон (orchestration/scheduler.py::_defer_next_run_at). Разъехаться с
тактом порог не может: поменяли такт — порог поехал следом. Тревога после
MISSED_PULL_CYCLES=2 пропущенных тактов: один пропуск (сдвиг окна, разовый сбой сети)
поглощается, два подряд означают, что загрузка встала. При нынешнем такте 7 это 14
суток; на прод-истории такое состояние ДОСТИЖИМО — разрывы между полными прогонами
были 14.8 и 20 суток (05-31→06-15 и 07-17→08-06).

ПО ТАБЛО, А НЕ ПО max() ВСЕЙ ТАБЛИЦЫ. Оценщик берёт ПЕРВОЕ НЕПУСТОЕ табло из
estimator.SBER_COEFF_DASHBOARDS; у real_estate_deals latest=2026-06, у
dinamika-tsen-obyavlenii — 2026-05 (на 2026-08-12). max() по таблице маскирует
отставшее табло, поэтому монитор идёт тем же порядком, что и оценщик, и берёт ту же
серию — список импортируется из estimator, дублировать его тут нельзя.

ЧЕГО ЭТОТ МОНИТОР НЕ ЛОВИТ (осознанно, #2846). Если источник ЗАМОЛЧИТ НАВСЕГДА, а
загрузка останется исправной — монитор промолчит: по нашим данным «источник не
публиковал 2 месяца» неотличимо от «источник публикует раз в 2 месяца». Такт
публикации источника ретроспективно невосстановим — его затёр апсерт
(sber_index.py ставил fetched_at=now() всем строкам серии). С этого PR fetched_at
не переписывается при конфликте и означает «когда мы ВПЕРВЫЕ увидели этот период»,
т.е. такт публикации станет измеримым; вернуться к вопросу порога «источник встал»
имеет смысл после 3 наблюдённых публикаций (ориентир — ноябрь 2026).

РЯДОВ ТЕПЕРЬ НЕСКОЛЬКО (#3051). sber_price_index ключуется текстовой колонкой city,
и с #3051 оценщик выбирает ряд по region_code сделки: 66 → «Свердловская область»,
77 → «Москва», остальное → «Россия» (estimator.SBER_MONITORED_REGIONS). Монитор,
следивший ровно за свердловским рядом, пропустил бы пропажу московского — а под
ним 212 937 сделок региона 77. Теперь опрашиваются ВСЕ ряды из того же кортежа.

КОМПРОМИСС, честно. Второго независимого вердикта тут нет и быть не может: stale
считается по такту ЗАГРУЗКИ (sber_index_pull тянет все 3 табло × 3 региона одним
прогоном, errors — счётчик по всему прогону), поэтому для всех рядов он ОДИН И ТОТ ЖЕ
по построению. Многорядность ловит другое — ПРОПАЖУ ряда. Ранний выход с mark_failed
остался РОВНО за прежним случаем: нет ряда региона по умолчанию (свердловского) —
вердикт считать не из чего. Пропажа ЛЮБОГО другого ряда его больше не подавляет:
иначе переименование «Москва» → «г. Москва» отключало бы мониторинг Екатеринбурга
(latest_*=0, age_days=0, alert=0 — свежесть 66 не считалась вовсе), да ещё и с ложным
текстом «sber_price_index empty», хотя таблица непуста. Теперь пропажа обязательного
ряда 77 — свой ERROR с ИМЕНЕМ ряда при done-прогоне; нет фолбэчного «Россия» →
WARNING (по нему сегодня не считается ни одна сделка).

ЧЕМ ИМЕННО ЗДЕСЬ АЛЕРТЯТ (честно, не путать со счётчиком). Канал тревоги в проекте
ровно один и тот же у всех соседей — ERROR-запись логгера, которую LoggingIntegration
(event_level=ERROR) превращает в событие GlitchTip; это и проверяется в
tests/test_alerts_become_events.py — по ФАКТУ СОБЫТИЯ, а не по levelno. Счётчиков
прогона (scrape_runs.counters) не читает ни одно правило алертинга: единственный их
потребитель, стрик-алерт, смотрит на status прогона, а не на ключи counters. Поэтому
`alert_regions_missing` — НАБЛЮДЕНИЕ для ретроспективы по scrape_runs (как
regions_missing и age_days_max), а НЕ канал тревоги; обещание «свой алерт по счётчику»
было неправдой и убрано. Тревога по пропавшему ряду держится на ERROR выше, и именно
это проверяется тестом через тот же харнесс событий, что у соседей.
Расхождение latest-периодов между регионами кладётся в counters
(age_days_max) и в лог как НАБЛЮДЕНИЕ, но алертом не становится: источник вправе
публиковать регионы вразнобой, а частоту таких расхождений мы не мерили — заводить
порог без замера значит повторить дефект #2846 (порог внутри рабочего диапазона).
Семантика вердикта и ключи counters свердловского ряда не изменились.

Изоляция чужих рядов доведена до конца: не только пустая выборка, но и ИСКЛЮЧЕНИЕ
на запросе чужого ряда (таймаут, обрыв соединения посреди обхода) больше не уходит
во внешний except с mark_failed — каждый чужой ряд опрашивается в своём try, сбой
попадает в лог и в regions_missing. Наружу поднимается только сбой на ряде региона
по умолчанию: вердикт всё равно не из чего считать. Свой try без ОТКАТА эту изоляцию
не давал: ошибка драйвера деактивирует транзакцию Session, и следующий же запрос
(за интервалом загрузки) падает с PendingRollbackError — исключение не
распространялось, зато сессия оставалась испорченной, и вердикт по свердловскому ряду
терялся ровно как раньше. Поэтому в per-region except стоит db.rollback().

ERROR, а не WARNING (#2674): в контейнере скрапера GlitchTip поднят с
LoggingIntegration(event_level=ERROR), WARNING событием не становится вообще.

Задача синхронная (DB-only) — запускается kit-scheduler'ом через
product_handlers._job_sber_freshness_monitor в run_in_executor. Вердикт считает
ЧИСТАЯ функция evaluate_sber_freshness() (frozen-now, тестируется без БД).

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

from __future__ import annotations

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

from sqlalchemy import text
from sqlalchemy.orm import Session

from app.services import scrape_runs as runs_mod
from app.services.estimator import (
    SBER_COEFF_DASHBOARDS,
    SBER_MONITORED_REGIONS,
    SBER_REQUIRED_REGIONS,
    SBER_TIME_ADJUST_REGION,
)

logger = logging.getLogger(__name__)

__all__ = [
    "DEFAULT_PULL_INTERVAL_DAYS",
    "MISSED_PULL_CYCLES",
    "SBER_FRESHNESS_PULL_SOURCE",
    "SberFreshnessVerdict",
    "check_sber_freshness",
    "evaluate_sber_freshness",
]

# Джоба-загрузчик, чей такт и успешность мы и мониторим.
SBER_FRESHNESS_PULL_SOURCE = "sber_index_pull"

# Сколько тактов загрузки подряд можно пропустить до тревоги. 1 = разовый сбой/сдвиг
# окна (поглощаем), 2 = загрузка встала (алерт).
MISSED_PULL_CYCLES = 2

# Фолбэк, если в scrape_schedules нет строки/ключа interval_days (миграция 212 ставит 7).
DEFAULT_PULL_INTERVAL_DAYS = 7

_LATEST_PERIOD_SQL = text("""
    SELECT max(period_month) AS latest
    FROM sber_price_index
    WHERE city = CAST(:city AS text)
      AND dashboard = CAST(:dash AS text)
      -- #R2-H1: только вторичный рынок (эстиматор — вторичка); первичка
      -- (новостройки) = направленно неверная коррекция. Зеркалит фильтр
      -- estimator._load_sber_index_series.
      AND (segment IS NULL OR segment ILIKE '%вторичн%')
""")

# Последний ЗАВЕДОМО ПОЛНЫЙ прогон загрузчика. status='done' сюда не входит намеренно:
# прогон id=37 имеет done при {errors: 9, upserted: 0}. Сравнения — jsonb-ные, без
# CAST(... AS int): counters других источников планировщик может отфильтровать позже
# каста, а не раньше, и нечисловое значение уронило бы запрос. Для jsonb-чисел
# оператор > численный.
_LAST_COMPLETE_PULL_SQL = text("""
    SELECT max(finished_at) AS last_pull
    FROM scrape_runs
    WHERE source = CAST(:src AS text)
      AND counters @> CAST('{"errors": 0}' AS jsonb)
      AND counters -> 'upserted' > CAST('0' AS jsonb)
""")

# Такт загрузки — из той же строки, по которой планировщик считает next_run_at.
_PULL_INTERVAL_SQL = text("""
    SELECT default_params ->> 'interval_days' AS interval_days
    FROM scrape_schedules
    WHERE source = CAST(:src AS text)
""")


@dataclass(frozen=True)
class SberFreshnessVerdict:
    """Вердикт: отстала ли ЗАГРУЗКА СберИндекса от собственного такта."""

    latest_period: date
    age_days: int  # наблюдение (лаг публикации источника), НЕ критерий тревоги
    pull_lag_days: int  # суток с последнего полного прогона; -1 = полных прогонов не было
    max_pull_lag_days: int  # порог = MISSED_PULL_CYCLES × такт загрузки
    stale: bool


def evaluate_sber_freshness(
    latest_period: date,
    now: datetime,
    *,
    last_complete_pull_at: datetime | None,
    pull_interval_days: int,
) -> SberFreshnessVerdict:
    """Чистая логика: отстала ли загрузка от собственного такта.

    stale = полных прогонов не было ВООБЩЕ, либо последний старше
    MISSED_PULL_CYCLES × pull_interval_days. Возраст периода считается и кладётся в
    вердикт как НАБЛЮДЕНИЕ, но на вердикт не влияет: после полного прогона наш
    max(period_month) равен максимуму источника по построению, и его возраст — это
    лаг ПУБЛИКАЦИИ, на который мы повлиять не можем.
    """
    age_days = (now.date() - latest_period).days
    max_pull_lag_days = MISSED_PULL_CYCLES * pull_interval_days
    if last_complete_pull_at is None:
        return SberFreshnessVerdict(latest_period, age_days, -1, max_pull_lag_days, True)
    pull_lag_days = (now - last_complete_pull_at).days
    return SberFreshnessVerdict(
        latest_period=latest_period,
        age_days=age_days,
        pull_lag_days=pull_lag_days,
        max_pull_lag_days=max_pull_lag_days,
        stale=pull_lag_days > max_pull_lag_days,
    )


def _load_estimator_dashboard(
    db: Session, city: str = SBER_TIME_ADJUST_REGION
) -> tuple[str, date] | None:
    """Табло, которое возьмёт оценщик для ряда `city`, и его latest период.

    Тот же порядок, что и estimator._load_sber_index_series: первое НЕПУСТОЕ табло
    из SBER_COEFF_DASHBOARDS. max() по всей таблице маскировал бы отставшее табло.

    #3051: `city` — имя ряда (sber_price_index.city), дефолт — свердловский, чтобы
    вызов без аргумента остался прежним.
    """
    for dash in SBER_COEFF_DASHBOARDS:
        row = db.execute(_LATEST_PERIOD_SQL, {"city": city, "dash": dash}).first()
        latest = row.latest if row is not None else None
        if latest is not None:
            return dash, latest
    return None


def _pull_interval_days(db: Session) -> int:
    """Такт загрузчика из scrape_schedules (фолбэк DEFAULT_PULL_INTERVAL_DAYS)."""
    row = db.execute(_PULL_INTERVAL_SQL, {"src": SBER_FRESHNESS_PULL_SOURCE}).first()
    raw = row.interval_days if row is not None else None
    try:
        return int(raw) if raw is not None else DEFAULT_PULL_INTERVAL_DAYS
    except (TypeError, ValueError):
        logger.warning(
            "sber freshness: interval_days=%r в scrape_schedules нечисловой — беру %d",
            raw,
            DEFAULT_PULL_INTERVAL_DAYS,
        )
        return DEFAULT_PULL_INTERVAL_DAYS


def check_sber_freshness(
    db: Session,
    run_id: int,
    params: dict | None = None,  # type: ignore[type-arg]
    now: datetime | None = None,
) -> dict[str, int]:
    """Проверить, не отстала ли загрузка СберИндекса, и алертить при отставании.

    Sync (вызывается scheduler-триггером в executor, как check_deals_freshness).
    Читает: latest период табло оценщика, время последнего ПОЛНОГО прогона
    sber_index_pull, такт загрузки из scrape_schedules. Вердикт — чистой функцией.

    `params` больше ничего не настраивает: порог берётся из такта самой загрузки
    (унаследованный default_params.lag_allowance_days=25 монитора игнорируется —
    он кодировал мёртвый календарный порог). `now` инъектируется в тестах.

    Returns counters {latest_year, latest_month, age_days, pull_lag_days,
    max_pull_lag_days, alert}.
    mark_failed только если у оценщика нет серии вообще (нечего оценивать);
    при алерте прогон помечается done (это монитор, не сбой джобы).
    """
    now = now or datetime.now(UTC)
    counters: dict[str, int] = {
        "latest_year": 0,
        "latest_month": 0,
        "age_days": 0,
        "pull_lag_days": -1,
        "max_pull_lag_days": 0,
        "alert": 0,
        # #3051: сколько рядов из SBER_MONITORED_REGIONS не нашлось и каков худший
        # возраст среди найденных (наблюдение, не критерий тревоги).
        "regions_missing": 0,
        "age_days_max": 0,
        # Обязательный ряд не наблюдался этим прогоном: пропал из таблицы либо запрос
        # по нему сорвался. В раннем выходе (нет свердловского) счётчик тоже заполнен.
        # НАБЛЮДЕНИЕ, а не канал тревоги (см. «ЧЕМ ИМЕННО ЗДЕСЬ АЛЕРТЯТ» в шапке):
        # counters не читает ни одно правило алертинга, тревогу поднимает ERROR-лог.
        "alert_regions_missing": 0,
    }
    try:
        runs_mod.update_heartbeat(db, run_id, counters)

        # #3051: опрашиваем ВСЕ ряды, которые способен прочитать оценщик, а не один.
        # ДЕДУПЛИКАЦИЯ обязательна: SBER_MONITORED_REGIONS — кортеж ИМЁН рядов, а карта
        # оценщика вправе свести два region_code на одно имя (заведём регион, чей ряд
        # совпал с фолбэчной «Россией» — кортеж станет длиннее на элемент, а РАЗЛИЧНЫХ
        # рядов останется столько же). Считать пропажи по длине кортежа значило бы
        # залипнуть на regions_missing=1 навсегда при всех живых рядах.
        monitored_regions = tuple(dict.fromkeys(SBER_MONITORED_REGIONS))

        found_by_region: dict[str, tuple[str, date]] = {}
        probe_failed: list[str] = []  # ряд не удалось СПРОСИТЬ (не то же, что «нет ряда»)
        for region in monitored_regions:
            # КАЖДЫЙ ЧУЖОЙ РЯД — В СВОЁМ try. Раньше весь обход шёл под общим except:
            # таймаут или обрыв соединения на запросе московского ряда улетал наружу,
            # давал mark_failed и повторный подъём — и вердикт по свердловскому ряду
            # снова не считался, хотя сам ряд на месте. Это тот же дефект, что чинили
            # ранним выходом по ПУСТОЙ выборке, только по ветке ИСКЛЮЧЕНИЯ.
            try:
                got = _load_estimator_dashboard(db, region)
            except Exception:
                if region == SBER_TIME_ADJUST_REGION:
                    # Ряд региона по умолчанию — единственный источник вердикта:
                    # его сбой подавлять нечего и незачем, отдаём во внешний except
                    # (там и откат, и mark_failed).
                    raise
                # ОТКАТ, А НЕ ПРОСТО continue. Прошлый круг изолировал РАСПРОСТРАНЕНИЕ
                # исключения, но не ПОРЧУ СЕССИИ — это разные вещи, и второго мало.
                # Настоящая ошибка драйвера (таймаут инструкции, обрыв соединения)
                # ДЕАКТИВИРУЕТ транзакцию Session: следующий запрос падает с
                # PendingRollbackError, даже не дойдя до БД. Без отката изоляция была
                # мнимой — цикл шёл дальше, но первый же запрос ЗА ИНТЕРВАЛОМ ЗАГРУЗКИ
                # падал на испорченной сессии, улетал во внешний except, давал
                # mark_failed и подъём: вердикт по свердловскому ряду опять не доезжал.
                # Пустая выборка этого не воспроизводит вовсе — тест обязан имитировать
                # именно ошибку драйвера (tests/test_sber_region_series_3051.py).
                try:
                    db.rollback()
                except Exception:
                    # Откат не прошёл — мертво соединение целиком, а не один запрос:
                    # вердикт всё равно считать не из чего, отдаём наружу.
                    logger.exception(
                        "sber freshness: откат сессии после сбоя на ряде %s не прошёл — "
                        "соединение непригодно, вердикт по %s не считаем",
                        region,
                        SBER_TIME_ADJUST_REGION,
                    )
                    raise
                probe_failed.append(region)
                logger.exception(
                    "sber freshness: запрос ряда СберИндекса %s сорвался — сессия "
                    "откачена, ряд помечен ненаблюдённым, вердикт по %s считаем дальше",
                    region,
                    SBER_TIME_ADJUST_REGION,
                )
                continue
            if got is not None:
                found_by_region[region] = got
        counters["regions_missing"] = len(monitored_regions) - len(found_by_region)

        # Обязательный ряд может быть не наблюдён по двум разным причинам: его нет в
        # таблице (переименование в источнике) или запрос по нему сорвался. Причину
        # разделяет ЛОГ; в counters она не ветвится — новых ключей не заводим, а для
        # тревоги оба случая равнозначны: обязательного ряда за этот прогон нет.
        missing_required = [
            r for r in SBER_REQUIRED_REGIONS if r not in found_by_region and r not in probe_failed
        ]
        # НЕ то же самое, что missing_required: сюда попадает и ряд, который не удалось
        # СПРОСИТЬ. Для тревоги эти случаи равнозначны («обязательного ряда за этот
        # прогон нет»), поэтому счётчик считается по required_unseen, а не по
        # missing_required; причину разделяет лог (ERROR ниже vs logger.exception в цикле).
        required_unseen = [r for r in SBER_REQUIRED_REGIONS if r not in found_by_region]

        # ДО РАННЕГО ВЫХОДА, а не после. Раньше этот блок стоял ниже возврата, то есть
        # в ветке раннего выхода не выполнялся НИКОГДА: при одновременной пропаже
        # свердловского и московского рядов в мониторинг уходило сообщение только про
        # свердловский, хотя под московским 212 937 сделок региона 77 и это отдельный
        # дефект с отдельной починкой. Теперь про КАЖДЫЙ пропавший обязательный ряд
        # сообщение уходит всегда, а ранний выход остаётся только вопросом вердикта.
        #
        # Почему ERROR при done-прогоне, а не mark_failed (когда свердловский на месте):
        #  (а) вердикт по 66 уже посчитан и обязан доехать до дашборда, а counters
        #      упавшего прогона там не читаются — ровно эта подмена и превращала
        #      пропажу Москвы в отключение мониторинга Екатеринбурга;
        #  (б) mark_failed виден только стрик-алерту, т.е. на третьи сутки, а ERROR
        #      уходит в GlitchTip тем же прогоном (#2674);
        #  (в) отдельный ключ counters не перегружает `alert`, который значит «загрузка
        #      отстала»: это другой дефект, чинится в другом месте (имя ряда).
        # Ряд региона ПО УМОЛЧАНИЮ разбирается отдельной веткой ниже (ранний выход),
        # и у неё свой ERROR. Без этого условия одновременная пропажа обоих рядов
        # давала ДВА события об одном факте — лишняя issue в GlitchTip, не сигнал.
        default_found = found_by_region.get(SBER_TIME_ADJUST_REGION)
        if missing_required and default_found is not None:
            logger.error(
                "sber freshness: пропал обязательный ряд СберИндекса %s — %s. "
                "По этим регионам есть сделки, а time-поправку взять неоткуда: "
                "проверь имя ряда в источнике (переименование city) и карту "
                "estimator._SBER_REGION_SERIES",
                missing_required,
                # «Таблица НЕ пуста» — утверждение о факте, поэтому только когда хоть
                # один ряд действительно прочитан: прежний безусловный текст врал.
                f"таблица НЕ пуста, остальные ряды на месте ({sorted(found_by_region)})"
                if found_by_region
                else "ни одного ряда прочитать не удалось",
            )

        missing_optional = [
            r
            for r in monitored_regions
            if r not in found_by_region and r not in SBER_REQUIRED_REGIONS and r not in probe_failed
        ]
        if missing_optional:
            # WARNING (не ERROR): это фолбэчный ряд для региона вне карты оценщика —
            # сегодня по нему не считается ни одна сделка, ронять монитор незачем.
            logger.warning(
                "sber freshness: нет фолбэчной серии %s — регион вне "
                "estimator._SBER_REGION_SERIES останется без time-поправки",
                missing_optional,
            )

        # Ранний выход — ТОЛЬКО ради того случая, ради которого он и заводился:
        # у оценщика нет серии по региону ПО УМОЛЧАНИЮ, считать вердикт не из чего.
        # Пропажа любого другого ряда его больше не подавляет (см. шапку).
        if default_found is None:
            # ERROR (#2674): монитор не может выполнить свою работу вовсе — это сбой,
            # а не наблюдение. mark_failed ниже виден только стрик-алерту (3 подряд),
            # а монитор ходит раз в сутки — три дня молчания на пустом бенчмарке.
            # Единственное событие этой ветки: называет и ряд по умолчанию, и все
            # прочие пропавшие обязательные ряды — блок missing_required выше
            # здесь намеренно молчит, чтобы не дублировать issue.
            logger.error(
                "sber freshness: у оценщика нет серии — ни одно табло %s не даёт строк "
                "для region=%s (вторичка); оценить нечего. Пропавшие обязательные "
                "ряды целиком: %s",
                list(SBER_COEFF_DASHBOARDS),
                SBER_TIME_ADJUST_REGION,
                missing_required or [SBER_TIME_ADJUST_REGION],
            )
            # Новые счётчики заполняем и ЗДЕСЬ. Нули по ним делали ранний выход слепым:
            # одновременная пропажа свердловского и московского рядов выглядела ровно
            # как пропажа одного свердловского (alert_regions_missing=0, age_days_max=0),
            # хотя второй дефект — отдельный и по нему 212 937 сделок региона 77.
            counters["alert_regions_missing"] = int(bool(required_unseen))
            counters["age_days_max"] = max(
                ((now.date() - d).days for _, d in found_by_region.values()), default=0
            )
            runs_mod.mark_failed(
                db,
                run_id,
                # Текст называет КОНКРЕТНЫЙ ряд: «таблица пуста» было ложью — в ней
                # могут лежать все остальные регионы.
                f"sber_price_index: нет серии '{SBER_TIME_ADJUST_REGION}' "
                f"в табло {list(SBER_COEFF_DASHBOARDS)}",
                counters,
            )
            return counters

        # Вердикт — по свердловскому ряду, как и до #3051 (см. КОМПРОМИСС в шапке:
        # такт загрузки общий для всех рядов, второго независимого вердикта нет).
        dashboard, latest = default_found
        last_pull_row = db.execute(
            _LAST_COMPLETE_PULL_SQL, {"src": SBER_FRESHNESS_PULL_SOURCE}
        ).first()
        last_complete_pull_at = last_pull_row.last_pull if last_pull_row is not None else None
        verdict = evaluate_sber_freshness(
            latest,
            now,
            last_complete_pull_at=last_complete_pull_at,
            pull_interval_days=_pull_interval_days(db),
        )

        counters = {
            "latest_year": latest.year,
            "latest_month": latest.month,
            "age_days": verdict.age_days,
            "pull_lag_days": verdict.pull_lag_days,
            "max_pull_lag_days": verdict.max_pull_lag_days,
            "alert": int(verdict.stale),
            "regions_missing": counters["regions_missing"],
            "age_days_max": max((now.date() - d).days for _, d in found_by_region.values()),
            "alert_regions_missing": int(bool(required_unseen)),
        }

        # #3051: наблюдение по всем рядам — расхождение latest между регионами видно
        # в логе, но алертом не становится (порог без замера = дефект #2846).
        logger.info(
            "sber freshness: ряды оценщика — %s",
            "; ".join(
                f"{r}: {found_by_region[r][1]} ({found_by_region[r][0]})"
                if r in found_by_region
                else (f"{r}: СБОЙ ЗАПРОСА" if r in probe_failed else f"{r}: НЕТ")
                for r in monitored_regions
            ),
        )

        if verdict.stale:
            # ERROR (#2674): WARNING не долетает до GlitchTip (event_level=ERROR).
            logger.error(
                "sber freshness: загрузка СберИндекса отстала — последний ПОЛНЫЙ прогон "
                "%s (%s суток назад, порог %d = %d такта × %d суток; "
                "status='done' с errors>0 за успех НЕ считается). "
                "Наш max(period_month)=%s (табло %s) мог разойтись с источником — "
                "проверь sber_index_pull: планировщик, сеть, /api/sowa 404",
                last_complete_pull_at.isoformat() if last_complete_pull_at else "НИ РАЗУ",
                verdict.pull_lag_days if verdict.pull_lag_days >= 0 else "∞",
                verdict.max_pull_lag_days,
                MISSED_PULL_CYCLES,
                verdict.max_pull_lag_days // MISSED_PULL_CYCLES,
                latest,
                dashboard,
            )
        else:
            logger.info(
                "sber freshness: загрузка в такте — последний полный прогон %d суток назад "
                "(≤ порога %d). max(period_month)=%s (табло %s, возраст %d суток) равен "
                "максимуму источника по построению: возраст = лаг ПУБЛИКАЦИИ источника, "
                "не наше отставание — алерта нет",
                verdict.pull_lag_days,
                verdict.max_pull_lag_days,
                latest,
                dashboard,
                verdict.age_days,
            )

        runs_mod.mark_done(db, run_id, counters)
        logger.info(
            "check_sber_freshness run_id=%d done: latest=%s dash=%s alert=%d "
            "pull_lag_days=%d age_days=%d",
            run_id,
            latest,
            dashboard,
            counters["alert"],
            counters["pull_lag_days"],
            counters["age_days"],
        )
        return counters
    except Exception as exc:
        logger.exception("check_sber_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