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).

При сетевой ошибке / HTTP 5xx / таймауте — логируем warning, возвращаем
available=False. Отсутствие папки/файла квартала → available=False (штатный
случай до публикации квартала, до начала следующего месяца после конца квартала).
"""

from __future__ import annotations

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

import httpx
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:
            logger.warning(
                "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:
            logger.warning(
                "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

        logger.info(
            "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 as exc:
        logger.warning(
            "rosreestr_poll: unexpected error checking Q%d %d: %s — treating as unavailable",
            quarter,
            year,
            exc,
        )
        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,
        )

    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,
    }