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

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

Тянет региональные ряды Росстата (open-data, см. ``rosstat_emiss.fetch_all_datasets``)
для Свердловской обл. (ОКТМО 65) / ЕКБ и апсертит каждую обсервацию в
``macro_indicator`` под ``source='rosstat'``. Сейчас приземляется демография
(``population_total``); ИПЦ/доходы/СМР-цены — EMISS-only за WAF (документировано в
``rosstat_emiss``), добавятся новым DatasetSpec, когда появится рабочий доступ.

Детерминированно, без 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 MacroRow, fetch_all_datasets
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 123):
#   (indicator_type, region, obs_date, value, source, frequency, unit, comment,
#    updated_at) PK (indicator_type, region, obs_date). source='rosstat',
#   frequency='yearly' (текущие ряды Росстата — годовые: obs_date = 1 января года).
UPSERT_ROSSTAT_SQL = text(
    """
    INSERT INTO macro_indicator (
        indicator_type, region, obs_date, value,
        source, frequency, unit, comment
    ) VALUES (
        CAST(:itype AS text), CAST(:region AS text), CAST(:d AS date),
        CAST(:v AS numeric),
        'rosstat', 'yearly', CAST(:unit AS text), CAST(:comment AS text)
    )
    ON CONFLICT (indicator_type, region, obs_date) 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


@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]:
    """Загрузить региональные ряды Росстата и апсертить в macro_indicator.

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

    Raises:
        Пробрасывает любую ошибку fetch'а (RosstatScraperError и пр.) — чтобы фейл
        был виден в Celery / GlitchTip, а не молча проглочен. Частичный сбой
        отдельного датасета НЕ роняет таску (fetch_all_datasets graceful per-dataset).
    """
    db = SessionLocal()
    try:
        rows = fetch_all_datasets()
        fetched = len(rows)
        logger.info("rosstat_macro_sync: fetched %d rosstat rows", fetched)

        upserted = _upsert_rows(db, rows)
        logger.info("rosstat_macro_sync: upserted %d/%d rosstat rows", upserted, fetched)

        return {"fetched": fetched, "upserted": upserted}
    except Exception as e:
        logger.exception("rosstat_macro_sync failed: %s", e)
        raise
    finally:
        db.close()