gendesign/backend/app/workers/tasks/rosstat_macro_sync.py

"""Celery task: синхронизация макро-показателей Росстата в ``macro_indicator``
(#946, EPIC2 «macro-ingest», GG-форсайт / Site Finder v2).

Тянет региональные/федеральные ряды Росстата из ТРЁХ источников и апсертит
каждую обсервацию в ``macro_indicator``:
  • open-data (``rosstat_emiss.fetch_all_datasets``) — демография ``population_total``
    (ОКТМО 65 / ЕКБ), ``source='rosstat'``, frequency='yearly';
  • ЕМИСС / fedstat (``rosstat_emiss.fetch_all_emiss``) — ``income_per_capita``
    (среднедушевые доходы, ОКАТО 65), ``source='emiss'``, frequency='quarterly';
  • открытый xlsx Росстата (``rosstat_emiss.fetch_construction_index``) —
    ``construction_price_index`` (индекс цен на СМР, region='rf'), ``source='rosstat'``,
    frequency='monthly'. ОТКРЫТЫЙ источник (без dataGrid.do / reCAPTCHA), fetch с dev.
Каждый источник в своём try/except (per-source guard): сбой одного НЕ роняет другой и
не валит таску — частичная поставка допустима (ТЗ §7.11). ЕМИСС-fetch доступен только
с прода (fedstat за WAF с dev); индекс цен на СМР теперь landed через открытый
rosstat.gov.ru-xlsx (reCAPTCHA-gated dataGrid.do НАМЕРЕННО НЕ используется); ИПЦ пока
не landed (см. rosstat_emiss).

Детерминированно, без LLM. ON CONFLICT делает повторные прогоны идемпотентными
(re-run обновляет value + updated_at). Расписание — ежемесячно (эти ряды
обновляются редко), регистрируется в beat_schedule.py.

Mirror conventions (как cbr_macro_sync): ``SessionLocal()`` + try/finally close,
``logger`` (не print), SAVEPOINT per-row в цикле upsert (backend.md),
``CAST(:x AS type)`` — никогда ``:x::type`` (psycopg v3). На fetch-фейле —
логируем и пробрасываем (surfaces в Celery/GlitchTip), не глотаем.
"""

from __future__ import annotations

import logging
from typing import Any

from sqlalchemy import text
from sqlalchemy.orm import Session

from app.core.db import SessionLocal
from app.services.scrapers.rosstat_emiss import (
    EmissRow,
    MacroRow,
    fetch_all_datasets,
    fetch_all_emiss,
    fetch_construction_index,
)
from app.workers.celery_app import celery_app

logger = logging.getLogger(__name__)

# psycopg v3: CAST(:x AS type) — НИКОГДА :x::type (SQLAlchemy+psycopg3 роняет
# синтаксис на ::). Контракт колонок совпадает с macro_indicator (migration 163):
#   (indicator_type, region, obs_date, period_type, value, source, frequency, unit,
#    comment, updated_at). PK (indicator_type, region, obs_date, period_type).
# Не-ЕМИСС источники (open-data, СМР) используют period_type='unknown' (литерал) —
# они различаются по obs_date без sub-period granularity.
# source='rosstat', frequency='yearly' (текущие ряды Росстата — годовые).
UPSERT_ROSSTAT_SQL = text(
    """
    INSERT INTO macro_indicator (
        indicator_type, region, obs_date, period_type, value,
        source, frequency, unit, comment
    ) VALUES (
        CAST(:itype AS text), CAST(:region AS text), CAST(:d AS date),
        'unknown', CAST(:v AS numeric),
        'rosstat', 'yearly', CAST(:unit AS text), CAST(:comment AS text)
    )
    ON CONFLICT (indicator_type, region, obs_date, period_type) DO UPDATE SET
        value = EXCLUDED.value,
        source = EXCLUDED.source,
        frequency = EXCLUDED.frequency,
        unit = EXCLUDED.unit,
        comment = EXCLUDED.comment,
        updated_at = now()
    """
)

# ЕМИСС-ряды (source='emiss'): period_type параметризован — 'year' | 'quarter' |
# 'month' (из _emiss_period_granularity). Это ключевое отличие от не-ЕМИСС источников:
# период ЕМИСС-ряда несёт granularity, необходимую для разделения годового агрегата
# ('год' → 'year') и Q1 ('I квартал' → 'quarter'), оба с obs_date=YYYY-01-01 (#1606).
# frequency параметризована (quarterly для доходов, monthly для ИПЦ когда добавится).
# CAST(:x AS type) — НИКОГДА :x::type (psycopg v3).
UPSERT_EMISS_SQL = text(
    """
    INSERT INTO macro_indicator (
        indicator_type, region, obs_date, period_type, value,
        source, frequency, unit, comment
    ) VALUES (
        CAST(:itype AS text), CAST(:region AS text), CAST(:d AS date),
        CAST(:period_type AS text), CAST(:v AS numeric),
        'emiss', CAST(:freq AS text), CAST(:unit AS text), CAST(:comment AS text)
    )
    ON CONFLICT (indicator_type, region, obs_date, period_type) DO UPDATE SET
        value = EXCLUDED.value,
        source = EXCLUDED.source,
        frequency = EXCLUDED.frequency,
        unit = EXCLUDED.unit,
        comment = EXCLUDED.comment,
        updated_at = now()
    """
)

# Индекс цен на СМР — открытый xlsx Росстата: source='rosstat' (rosstat.gov.ru-файл),
# НО frequency='monthly' (в отличие от open-data демографии — там фикс 'yearly'),
# поэтому frequency параметризована. MacroRow не несёт frequency/period_type-полей;
# подставляем литералами. CAST(:x AS type) — psycopg v3.
UPSERT_ROSSTAT_MONTHLY_SQL = text(
    """
    INSERT INTO macro_indicator (
        indicator_type, region, obs_date, period_type, value,
        source, frequency, unit, comment
    ) VALUES (
        CAST(:itype AS text), CAST(:region AS text), CAST(:d AS date),
        'unknown', CAST(:v AS numeric),
        'rosstat', 'monthly', CAST(:unit AS text), CAST(:comment AS text)
    )
    ON CONFLICT (indicator_type, region, obs_date, period_type) DO UPDATE SET
        value = EXCLUDED.value,
        source = EXCLUDED.source,
        frequency = EXCLUDED.frequency,
        unit = EXCLUDED.unit,
        comment = EXCLUDED.comment,
        updated_at = now()
    """
)


def _upsert_rows(db: Session, rows: list[MacroRow]) -> int:
    """Апсертит MacroRow в macro_indicator. SAVEPOINT per-row, чтобы один битый
    ряд не откатывал всю транзакцию. Возвращает число успешных upsert'ов."""
    upserted = 0
    for r in rows:
        try:
            with db.begin_nested():
                db.execute(
                    UPSERT_ROSSTAT_SQL,
                    {
                        "itype": r.indicator_type,
                        "region": r.region,
                        "d": r.obs_date,
                        "v": r.value,
                        "unit": r.unit,
                        "comment": r.comment,
                    },
                )
            upserted += 1
        except Exception as e:
            logger.warning(
                "upsert rosstat %s/%s@%s=%s failed: %s",
                r.indicator_type,
                r.region,
                r.obs_date,
                r.value,
                e,
            )
    db.commit()
    return upserted


def _upsert_monthly_rows(db: Session, rows: list[MacroRow]) -> int:
    """Апсертит месячные MacroRow (индекс цен на СМР) — source='rosstat',
    frequency='monthly'. SAVEPOINT per-row, чтобы один битый ряд не откатывал всю
    транзакцию. Возвращает число успешных upsert'ов."""
    upserted = 0
    for r in rows:
        try:
            with db.begin_nested():
                db.execute(
                    UPSERT_ROSSTAT_MONTHLY_SQL,
                    {
                        "itype": r.indicator_type,
                        "region": r.region,
                        "d": r.obs_date,
                        "v": r.value,
                        "unit": r.unit,
                        "comment": r.comment,
                    },
                )
            upserted += 1
        except Exception as e:
            logger.warning(
                "upsert rosstat-monthly %s/%s@%s=%s failed: %s",
                r.indicator_type,
                r.region,
                r.obs_date,
                r.value,
                e,
            )
    db.commit()
    return upserted


def _upsert_emiss_rows(db: Session, rows: list[EmissRow]) -> int:
    """Апсертит EmissRow в macro_indicator (source='emiss', frequency+period_type per-row).

    SAVEPOINT per-row, чтобы один битый ряд не откатывал всю транзакцию. Возвращает
    число успешных upsert'ов.

    period_type (из r.period_type) — часть нового PK (migration 163): позволяет
    годовому агрегату ('year') и Q1 ('quarter') за один год коexist в таблице без
    взаимной перезаписи (#1606).
    """
    upserted = 0
    for r in rows:
        try:
            with db.begin_nested():
                db.execute(
                    UPSERT_EMISS_SQL,
                    {
                        "itype": r.indicator_type,
                        "region": r.region,
                        "d": r.obs_date,
                        "period_type": r.period_type,
                        "v": r.value,
                        "freq": r.frequency,
                        "unit": r.unit,
                        "comment": r.comment,
                    },
                )
            upserted += 1
        except Exception as e:
            logger.warning(
                "upsert emiss %s/%s@%s[%s]=%s failed: %s",
                r.indicator_type,
                r.region,
                r.obs_date,
                r.period_type,
                r.value,
                e,
            )
    db.commit()
    return upserted


@celery_app.task(
    bind=True,
    name="tasks.rosstat_macro_sync.rosstat_macro_sync",
    max_retries=2,
)
def rosstat_macro_sync(self: Any) -> dict[str, Any]:
    """Загрузить ряды Росстата (open-data + ЕМИСС + xlsx-СМР) и апсертить в macro_indicator.

    Источники выполняются НЕЗАВИСИМО (per-source try/except): сбой одного источника
    логируется, апсертит здоровый источник, и в конце ошибка ПРОБРАСЫВАЕТСЯ (чтобы фейл
    был виден в Celery / GlitchTip, а не молча проглочен). Re-run идемпотентен (ON CONFLICT),
    поэтому Celery-retry безопасно повторит все источники.

    Returns:
        Счётчики ``{"fetched": N, "upserted": M, "rosstat": ..., "emiss": ...,
        "construction": ...}``.

    Raises:
        Первую пойманную ошибку любого источника — ПОСЛЕ попытки всех трёх (surfaces в
        Celery/GlitchTip). Частичный сбой отдельного датасета/индикатора внутри источника
        НЕ роняет источник (fetch_all_* / fetch_construction_index graceful per-item).
    """
    db = SessionLocal()
    first_error: Exception | None = None
    rosstat_fetched = rosstat_upserted = 0
    emiss_fetched = emiss_upserted = 0
    constr_fetched = constr_upserted = 0
    try:
        # ── Источник 1: Росстат open-data (демография) ──────────────────────────
        try:
            rows = fetch_all_datasets()
            rosstat_fetched = len(rows)
            rosstat_upserted = _upsert_rows(db, rows)
            logger.info(
                "rosstat_macro_sync: open-data upserted %d/%d rows",
                rosstat_upserted,
                rosstat_fetched,
            )
        except Exception as e:
            first_error = first_error or e
            logger.exception("rosstat_macro_sync: open-data source failed: %s", e)

        # ── Источник 2: ЕМИСС / fedstat (доходы; fetch только с прода) ───────────
        try:
            emiss_rows = fetch_all_emiss()
            emiss_fetched = len(emiss_rows)
            emiss_upserted = _upsert_emiss_rows(db, emiss_rows)
            logger.info(
                "rosstat_macro_sync: emiss upserted %d/%d rows",
                emiss_upserted,
                emiss_fetched,
            )
        except Exception as e:
            first_error = first_error or e
            logger.exception("rosstat_macro_sync: emiss source failed: %s", e)

        # ── Источник 3: открытый xlsx Росстата (индекс цен на СМР, region='rf') ──
        # Открытый источник (без dataGrid.do/captcha); месячный → _upsert_monthly_rows.
        try:
            constr_rows = fetch_construction_index()
            constr_fetched = len(constr_rows)
            constr_upserted = _upsert_monthly_rows(db, constr_rows)
            logger.info(
                "rosstat_macro_sync: construction upserted %d/%d rows",
                constr_upserted,
                constr_fetched,
            )
        except Exception as e:
            first_error = first_error or e
            logger.exception("rosstat_macro_sync: construction source failed: %s", e)

        if first_error is not None:
            raise first_error

        return {
            "fetched": rosstat_fetched + emiss_fetched + constr_fetched,
            "upserted": rosstat_upserted + emiss_upserted + constr_upserted,
            "rosstat": {"fetched": rosstat_fetched, "upserted": rosstat_upserted},
            "emiss": {"fetched": emiss_fetched, "upserted": emiss_upserted},
            "construction": {"fetched": constr_fetched, "upserted": constr_upserted},
        }
    finally:
        db.close()