gendesign/tradein-mvp/backend/app/services/rosreestr_poll.py

"""Scheduled poll: detect when a new Rosreestr quarter becomes available (#888).

Граница ответственности модуля:
  - DETECT: определить, появился ли новый квартал (next after latest loaded) в
    открытых данных Росреестра.
  - ALERT: залогировать actionable-сообщение, если квартал доступен.
  - НЕ скачивает много-гигабайтный ZIP и НЕ вызывает shell-loaders — это ручной
    ops-шаг. После алерта оператор запускает (оба скрипта живут вне этого
    репозитория — при смене базового пути их тоже нужно поправить вручную):
      data/sql/02_load_all_quarters.sh
      tradein-mvp/deploy/import-rosreestr.sh

ИСТОРИЯ (важно для будущих правок): раньше архив открытых данных Росреестра
(сделки ДКП/ДДУ) публиковался на https://rosreestr.gov.ru/opendata/ с индексом
/opendata/f.json. Этот портал был переделан на Bitrix — прямые ссылки на ZIP
теперь отвечают HTTP 200 с Content-Type: text/html (soft-404 "заглушка"), а
/opendata/f.json больше не отдаёт JSON. Старая версия поллера трактовала любой
HTTP 200 как "квартал доступен" и из-за soft-404 стала ложно репортить КАЖДЫЙ
квартал как доступный — это и есть баг #issue, который чинит этот модуль.

АКТУАЛЬНОЕ (2026-07) расположение датасетов — открытый Apache autoindex без
авторизации:
  https://rosreestr.gov.ru/data-sets/
Структура:
  - Папки кварталов названы в кодировке Windows-1251 (percent-encoded href —
    именно cp1251, НЕ utf-8), например "1 квартал 2026г." →
    href="1%20%EA%E2%E0%F0%F2%E0%EB%202026%E3./". Плюс папка
    "Архив до 2023г. включительно/" для старых периодов.
  - Внутри папки квартала лежит dataset_СДЕЛКИ_r-r_01-92_y_{YYYY}_q_{N}.csv.zip
    (тоже cp1251-кодированный href, например
    "dataset_%D1%C4%C5%CB%CA%C8_r-r_01-92_y_2026_q_1.csv.zip"), рядом —
    dataset_КАДАСТРСТОИМОСТЬ_* файлы (не наши).

Вместо того чтобы вручную высчитывать percent-encoding (хрупко: имена папок
расставляют руками, возможны отличия в пробелах/пунктуации между кварталами) —
поллер запрашивает autoindex HTML и парсит href'ы, декодируя каждый
percent-encoded href как cp1251 (urllib.parse.unquote(href, encoding="cp1251")),
затем ищет папку/файл по совпадению с ожидаемым (год, квартал) в декодированном
имени. См. check_new_quarter_available().

КЛЮЧЕВАЯ проверка доступности (защита от soft-404, см. "ИСТОРИЯ" выше): найденный
файл датасета считается ДОСТУПНЫМ только если HEAD-ответ имеет
Content-Type: application/zip И Content-Length больше ~100 KB. Голый HTTP 200
недостаточен.

rosreestr.gov.ru отдаёт сертификат от российского TLS root CA, которому нет
доверия в стандартном trust store — соединение делаем с verify=False, как
sber_index.py для sberindex.ru (см. #922, тот же паттерн: публичные
неавторизованные открытые данные, TLS verify отключаем осознанно). Сервер также
отвечает HTTP 403 без браузерного User-Agent — шлём Chrome UA (тот же паттерн,
что DEFAULT_UA в zhkh_flats_loader.py).

УРОВНИ СИГНАЛОВ (#2674 — в контейнере скрапера событием GlitchTip становится только
запись ERROR, см. scheduler_main.py LoggingIntegration(event_level=ERROR)):
  - Портал ответил не-200 на листинг каталога/папки → ERROR. Каталог — единственная
    опора поллера; портал УЖЕ один раз переехал (см. "ИСТОРИЯ"), и тогда поллер молча
    врал целыми кварталами. Такое обязано быть событием.
  - Файл датасета НАЙДЕН в листинге, но HEAD не отдал zip / размер ниже порога →
    ERROR. Тот же класс: это ровно поведение старой Bitrix-заглушки (200 + text/html).
    Ветка может сработать легитимно (файл выложили в листинг раньше, чем докачали),
    но цена асимметрична — ложное срабатывание стоит одного события в месяц (такт
    28 дней), пропуск стоит квартала молчания.
  - Таймаут / сетевая ошибка → WARNING, как раньше. Это транспортный блип раз в месяц
    (такт поллера), сам пройдёт; а «квартал так и не приехал» ловит отдельный
    deals_freshness_monitor ERROR-ом по max(deal_date).
  - Папки/файла квартала нет → INFO. Штатное состояние до публикации: квартал выходит
    4 раза в год, поллер ходит 12 — большинство прогонов ЗАКОННО пустые.
  - Квартал вышел → INFO + ЯВНОЕ событие capture_message(level="info"), см.
    poll_rosreestr_new_quarter.
"""

from __future__ import annotations

import logging
import re
from typing import Any
from urllib.parse import quote, unquote, urljoin

import httpx
import sentry_sdk
from sqlalchemy import text
from sqlalchemy.orm import Session

logger = logging.getLogger(__name__)

# Открытый Apache autoindex открытых данных Росреестра (см. докстринг модуля).
_DATA_SETS_BASE_URL = "https://rosreestr.gov.ru/data-sets/"

# Браузерный UA — без него rosreestr.gov.ru отвечает HTTP 403.
# Тот же паттерн, что DEFAULT_UA в app/services/zhkh_flats_loader.py.
_BROWSER_USER_AGENT = (
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
)

# Порог отличия реального ZIP-архива от soft-404 заглушки/пустого файла.
_MIN_DATASET_SIZE_BYTES = 100_000  # 100 KB

# Таймаут запросов. Росреестр может быть медленным — 15s достаточно.
_HTTP_TIMEOUT = 15.0

_HREF_RE = re.compile(r'href="([^"]+)"', re.IGNORECASE)


def _next_quarter(year: int, quarter: int) -> tuple[int, int]:
    """Вернуть (year, quarter) для следующего квартала.

    >>> _next_quarter(2026, 1)
    (2026, 2)
    >>> _next_quarter(2026, 4)
    (2027, 1)
    """
    if quarter < 4:
        return year, quarter + 1
    return year + 1, 1


def _period_start_to_quarter(period_start_date: Any) -> tuple[int, int]:
    """Конвертировать дату начала периода в (year, quarter).

    period_start_date — дата первого числа квартала (e.g. 2026-01-01 → Q1 2026).
    Принимает datetime.date, datetime.datetime или ISO-строку.
    """
    if hasattr(period_start_date, "year"):
        year = period_start_date.year
        month = period_start_date.month
    else:
        # ISO string fallback
        import datetime

        d = datetime.date.fromisoformat(str(period_start_date)[:10])
        year = d.year
        month = d.month
    quarter = (month - 1) // 3 + 1
    return year, quarter


def latest_loaded_quarter(db: Session) -> tuple[int, int] | None:
    """Вернуть (year, quarter) последнего загруженного квартала Росреестра.

    Запрашивает MAX(period_start_date) FROM deals WHERE source='rosreestr'.
    Возвращает None если таблица пуста или данных от rosreestr нет.

    Использует синхронный Session (SQLAlchemy, psycopg v3 через SQLAlchemy dialect).
    """
    row = db.execute(
        text(
            """
            SELECT MAX(deal_date) AS max_date
            FROM deals
            WHERE source = 'rosreestr'
              AND deal_date IS NOT NULL
            """
        )
    ).fetchone()

    if row is None or row[0] is None:
        logger.info(
            "rosreestr_poll: no rosreestr deals found in DB — cannot determine latest quarter"
        )
        return None

    year, quarter = _period_start_to_quarter(row[0])
    logger.info(
        "rosreestr_poll: latest loaded quarter = Q%d %d (derived from max deal_date=%s)",
        quarter,
        year,
        row[0],
    )
    return year, quarter


def rosreestr_dataset_url(year: int, quarter: int) -> str:
    """Сформировать best-effort прямую ссылку на ZIP датасета (для alert-сообщения оператору).

    ВНИМАНИЕ: это ДЕТЕРМИНИРОВАННАЯ догадка по известному шаблону именования —
    папка "{quarter} квартал {year}г.", файл
    "dataset_СДЕЛКИ_r-r_01-92_y_{year}_q_{quarter}.csv.zip" — закодированная как
    cp1251 percent-encoding (подтверждено вручную против реального URL портала).
    Она НЕ используется для самой проверки доступности: check_new_quarter_available()
    парсит реальный autoindex и не полагается на угаданный URL, потому что
    Росреестр расставляет имена папок вручную и они могут отличаться в мелочах
    (лишний пробел, пунктуация) от квартала к кварталу.

    Если угаданная ссылка не откроется — ориентир для оператора: _DATA_SETS_BASE_URL
    (https://rosreestr.gov.ru/data-sets/), там нужно найти папку глазами.
    """
    folder_name = f"{quarter} квартал {year}г."
    file_name = f"dataset_СДЕЛКИ_r-r_01-92_y_{year}_q_{quarter}.csv.zip"
    folder_enc = quote(folder_name.encode("cp1251"), safe="")
    file_enc = quote(file_name.encode("cp1251"), safe="")
    return f"{_DATA_SETS_BASE_URL}{folder_enc}/{file_enc}"


def _decode_cp1251_href(href: str) -> str:
    """Декодировать percent-encoded href как cp1251 (портал кодирует кириллицу так, НЕ utf-8)."""
    try:
        return unquote(href, encoding="cp1251", errors="strict")
    except (UnicodeDecodeError, LookupError):
        # href без кириллицы (например "../") или неожиданная кодировка — fallback.
        return unquote(href)


def _normalize_ws(s: str) -> str:
    """Схлопнуть повторяющиеся пробелы — Росреестр расставляет имена папок вручную."""
    return " ".join(s.split())


def _extract_hrefs(html: str) -> list[str]:
    """Достать все href="..." из HTML autoindex-страницы (простой regex, не наш HTML)."""
    return _HREF_RE.findall(html)


def _find_quarter_folder_href(html: str, year: int, quarter: int) -> str | None:
    """Найти (не декодированный) href папки квартала (year, quarter) в HTML /data-sets/.

    Сравнение — по декодированному (cp1251) и нормализованному по пробелам имени
    папки, ожидаемый паттерн "{quarter} квартал {year}г.".
    """
    expected_name = f"{quarter} квартал {year}г."
    for href in _extract_hrefs(html):
        decoded = _normalize_ws(_decode_cp1251_href(href).rstrip("/"))
        if decoded == expected_name:
            return href
    return None


def _find_dataset_file_href(html: str, year: int, quarter: int) -> str | None:
    """Найти href файла dataset_СДЕЛКИ_..._y_{year}_q_{quarter}.csv.zip в HTML папки квартала."""
    expected_suffix = f"_y_{year}_q_{quarter}.csv.zip"
    for href in _extract_hrefs(html):
        decoded = _decode_cp1251_href(href)
        if decoded.startswith("dataset_СДЕЛКИ") and decoded.endswith(expected_suffix):
            return href
    return None


async def check_new_quarter_available(
    client: httpx.AsyncClient,
    year: int,
    quarter: int,
) -> bool:
    """Проверить, опубликован ли датасет СДЕЛКИ Росреестра для (year, quarter).

    Шаги (см. докстринг модуля за подробностями):
      1. GET autoindex _DATA_SETS_BASE_URL, найти папку квартала по декодированному
         (cp1251) имени "{quarter} квартал {year}г.".
      2. GET найденную папку, найти файл dataset_СДЕЛКИ_..._y_{year}_q_{quarter}.csv.zip.
      3. HEAD найденный файл — считаем ДОСТУПНЫМ, только если Content-Type
         начинается с application/zip И Content-Length > _MIN_DATASET_SIZE_BYTES.
         Голый HTTP 200 НЕ считается доступностью — это была причина бага
         (Bitrix-заглушка /opendata/ тоже отвечала 200).

    Отсутствие папки/файла квартала → available=False (нормально до публикации).
    Сетевая ошибка / таймаут / неожиданное исключение → warning, available=False.
    Никогда не поднимает исключения в вызывающий код (scheduler-safe).
    """
    try:
        index_resp = await client.get(_DATA_SETS_BASE_URL, follow_redirects=True)
        if index_resp.status_code != 200:
            # ERROR (#2674): без каталога поллер слеп — см. "УРОВНИ СИГНАЛОВ".
            logger.error(
                "rosreestr_poll: unexpected HTTP %d listing %s — treating Q%d %d as unavailable",
                index_resp.status_code,
                _DATA_SETS_BASE_URL,
                quarter,
                year,
            )
            return False

        folder_href = _find_quarter_folder_href(index_resp.text, year, quarter)
        if folder_href is None:
            logger.info(
                "rosreestr_poll: quarter folder for Q%d %d not found under %s — not yet published",
                quarter,
                year,
                _DATA_SETS_BASE_URL,
            )
            return False

        folder_url = urljoin(_DATA_SETS_BASE_URL, folder_href)
        folder_resp = await client.get(folder_url, follow_redirects=True)
        if folder_resp.status_code != 200:
            # ERROR (#2674): папка квартала НАЙДЕНА в каталоге, но не открывается —
            # это уже не «ещё не опубликовали», а поломка портала.
            logger.error(
                "rosreestr_poll: unexpected HTTP %d listing folder %s — "
                "treating Q%d %d as unavailable",
                folder_resp.status_code,
                folder_url,
                quarter,
                year,
            )
            return False

        file_href = _find_dataset_file_href(folder_resp.text, year, quarter)
        if file_href is None:
            logger.info(
                "rosreestr_poll: dataset_СДЕЛКИ file for Q%d %d not found in folder %s",
                quarter,
                year,
                folder_url,
            )
            return False

        file_url = urljoin(folder_url, file_href)
        file_resp = await client.head(file_url, follow_redirects=True)

        content_type = file_resp.headers.get("content-type", "")
        try:
            content_length = int(file_resp.headers.get("content-length") or "0")
        except ValueError:
            content_length = 0

        if (
            file_resp.status_code == 200
            and content_type.startswith("application/zip")
            and content_length > _MIN_DATASET_SIZE_BYTES
        ):
            logger.info(
                "rosreestr_poll: Q%d %d available at %s (HTTP 200, %s, %d bytes)",
                quarter,
                year,
                file_url,
                content_type,
                content_length,
            )
            return True

        # ERROR (#2674, ревью PR #2681): файл ЕСТЬ в листинге, но HEAD отдал не zip
        # либо размер ниже порога — это буквально тот сбой, из-за которого поллер уже
        # врал (Bitrix-заглушка отвечала 200 с text/html вместо архива, см. "ИСТОРИЯ").
        # Ветка может сработать и легитимно — файл появился в листинге раньше, чем
        # докачался, — но цена асимметрична: такт 28 дней, значит ложное срабатывание
        # стоит максимум одного события в месяц, а пропуск стоит квартала молчания.
        logger.error(
            "rosreestr_poll: Q%d %d file found (%s) but failed availability check "
            "(HTTP %d, Content-Type=%r, Content-Length=%d) — soft-404 guard, "
            "treating as unavailable",
            quarter,
            year,
            file_url,
            file_resp.status_code,
            content_type,
            content_length,
        )
        return False

    except httpx.TimeoutException:
        logger.warning(
            "rosreestr_poll: timeout checking Q%d %d under %s — treating as unavailable",
            quarter,
            year,
            _DATA_SETS_BASE_URL,
        )
        return False
    except httpx.RequestError as exc:
        logger.warning(
            "rosreestr_poll: network error checking Q%d %d: %s — treating as unavailable",
            quarter,
            year,
            exc,
        )
        return False
    except Exception:
        # ERROR + traceback (#2674): сюда попадает НАШ баг (сменилась разметка, упал
        # парсер href'ов), а не сбой сети. Под WARNING он молча превращался в
        # «квартала нет» — ровно тот сценарий, из-за которого поллер врал кварталами.
        logger.exception(
            "rosreestr_poll: unexpected error checking Q%d %d — treating as unavailable",
            quarter,
            year,
        )
        return False


async def poll_rosreestr_new_quarter(db: Session) -> dict[str, Any]:
    """Orchestrator: определить следующий квартал после загруженного, проверить наличие.

    Шаги:
      1. Найти MAX(deal_date) WHERE source='rosreestr' → (loaded_year, loaded_quarter).
      2. Вычислить next = _next_quarter(loaded_year, loaded_quarter).
      3. Проверить наличие датасета на rosreestr.gov.ru/data-sets/ (autoindex + Content-Type).
      4. Если доступен — логировать actionable INFO с инструкцией по запуску ingest.
      5. Вернуть dict с результатом.

    Если в БД нет данных rosreestr → предполагаем 2025Q4, проверяем 2026Q1.
    Это fallback — в штатном режиме данные всегда есть после initial load.

    Возвращает:
        {"available": bool, "year": int, "quarter": int,
         "latest_loaded_year": int | None, "latest_loaded_quarter": int | None}
    """
    # 1. Текущий максимальный загруженный квартал
    latest = latest_loaded_quarter(db)
    if latest is None:
        # Fallback: нет данных — проверяем 2026Q1 как первую точку
        loaded_year, loaded_quarter = 2025, 4
        logger.warning(
            "rosreestr_poll: no deals in DB — using fallback baseline 2025Q4, "
            "checking 2026Q1 availability"
        )
    else:
        loaded_year, loaded_quarter = latest

    # 2. Следующий квартал
    next_year, next_quarter = _next_quarter(loaded_year, loaded_quarter)
    logger.info(
        "rosreestr_poll: checking Q%d %d (next after loaded Q%d %d)",
        next_quarter,
        next_year,
        loaded_quarter,
        loaded_year,
    )

    # 3. Проверка наличия (verify=False: см. докстринг модуля — RU root CA не в trust store;
    # браузерный UA: без него rosreestr.gov.ru отвечает 403)
    async with httpx.AsyncClient(
        timeout=_HTTP_TIMEOUT,
        verify=False,
        headers={"User-Agent": _BROWSER_USER_AGENT},
    ) as client:
        available = await check_new_quarter_available(client, next_year, next_quarter)

    # 4. Алерт если доступен
    if available:
        logger.info(
            "rosreestr_poll: NEW QUARTER AVAILABLE — Q%d %d (%s). "
            "Run ingest to load: data/sql/02_load_all_quarters.sh + "
            "tradein-mvp/deploy/import-rosreestr.sh "
            "(both live outside this repo — if they still point at the old "
            "rosreestr.gov.ru/opendata/ base, update them to %s first)",
            next_quarter,
            next_year,
            rosreestr_dataset_url(next_year, next_quarter),
            _DATA_SETS_BASE_URL,
        )
        # #2674: это ХОРОШАЯ новость, но она требует ручного шага оператора (импорт
        # много-гигабайтного ZIP), а INFO-строка живёт только в docker-логах и
        # теряется на редеплое. Отсюда явный capture_message вместо logger.error:
        # событие в GlitchTip будет, а error-rate и стрик-алерты не соврут «сбой».
        # Шума не создаёт: такт поллера — раз в 28 дней, квартал выходит 4 раза в
        # год, а повтор до самого импорта — это и есть нужное напоминание (#2670).
        try:
            sentry_sdk.capture_message(
                f"Rosreestr: доступен новый квартал Q{next_quarter} {next_year} — "
                "нужен ручной импорт (02_load_all_quarters.sh + import-rosreestr.sh)",
                level="info",
            )
        except Exception:
            # Алертинг best-effort: падение отправки события не должно валить поллер.
            logger.warning("rosreestr_poll: capture_message failed", exc_info=True)

    return {
        "available": available,
        "year": next_year,
        "quarter": next_quarter,
        "latest_loaded_year": loaded_year if latest is not None else None,
        "latest_loaded_quarter": loaded_quarter if latest is not None else None,
    }