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

"""Celery task: синхронизация макро-показателей ЦБ (ключевая ставка + инфляция) в
``macro_indicator`` (#945 PR B / #946, GG-форсайт / Site Finder v2).

Тянет два ряда ЦБ (оба ``region='rf'``, ``source='cbr'``):
  • ``key_rate``      — ключевая ставка через ``fetch_key_rate`` (живой SOAP CBR
                        DailyInfoWebServ), ДНЕВНОЙ, ``frequency='daily'``.
  • ``inflation_yoy`` — инфляция «% г/г» (ИПЦ YoY) через ``fetch_inflation`` (CBR
                        UniDbQuery DownloadExcel, #946), МЕСЯЧНЫЙ,
                        ``frequency='monthly'``, ``obs_date`` = 1-е число месяца.

#946 / ИПЦ: путь fedstat ИПЦ заблокирован reCAPTCHA — но ЦБ публикует инфляцию
открыто (см. cbr_macro docstring). Это закрывает инфляционный вход макро-слоя.

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

Per-series guard: сбой fetch'а инфляции НЕ мешает ставке (и наоборот) — каждый ряд
апсертится независимо, но первая возникшая ошибка ПРОБРАСЫВАЕТСЯ в конце (surfaces
в Celery/GlitchTip), не глотается. Зеркалит per-source-дисциплину rosstat_macro_sync.

Mirror conventions: ``SessionLocal()`` + try/finally close, ``logger`` (не
print), SAVEPOINT per-row в цикле upsert (backend.md), ``CAST(:x AS type)`` —
никогда ``:x::type`` (psycopg v3).
"""

from __future__ import annotations

import logging
from datetime import date
from decimal import Decimal
from typing import Any

from sqlalchemy import text
from sqlalchemy.orm import Session

from app.core.db import SessionLocal
from app.services.scrapers.cbr_macro import fetch_inflation, fetch_key_rate
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).
# CBR-ряды используют period_type='unknown' (литерал) — они не несут sub-period
# granularity и различаются по obs_date.
UPSERT_KEY_RATE_SQL = text(
    """
    INSERT INTO macro_indicator (
        indicator_type, region, obs_date, period_type, value,
        source, frequency, unit, comment
    ) VALUES (
        'key_rate', 'rf', CAST(:d AS date), 'unknown', CAST(:v AS numeric),
        'cbr', 'daily', '%', 'CBR key rate'
    )
    ON CONFLICT (indicator_type, region, obs_date, period_type) DO UPDATE SET
        value = EXCLUDED.value,
        updated_at = now()
    """
)

# Инфляция «% г/г» (ИПЦ YoY): indicator_type='inflation_yoy', monthly, region='rf'.
# obs_date уже нормализован к 1-му числу месяца парсером (parse_inflation_xlsx).
# period_type='unknown' — месячный ряд без sub-period granularity.
UPSERT_INFLATION_SQL = text(
    """
    INSERT INTO macro_indicator (
        indicator_type, region, obs_date, period_type, value,
        source, frequency, unit, comment
    ) VALUES (
        'inflation_yoy', 'rf', CAST(:d AS date), 'unknown', CAST(:v AS numeric),
        'cbr', 'monthly', '%', 'CBR inflation YoY (ИПЦ, % г/г)'
    )
    ON CONFLICT (indicator_type, region, obs_date, period_type) DO UPDATE SET
        value = EXCLUDED.value,
        updated_at = now()
    """
)


def _upsert_key_rate(db: Session, rows: list[tuple[date, Decimal]]) -> int:
    """Апсертит (date, rate) в macro_indicator. SAVEPOINT per-row, чтобы один
    битый ряд не откатывал всю транзакцию. Возвращает число успешных upsert'ов."""
    upserted = 0
    for d, v in rows:
        try:
            with db.begin_nested():
                db.execute(UPSERT_KEY_RATE_SQL, {"d": d, "v": v})
            upserted += 1
        except Exception as e:
            logger.warning("upsert key_rate %s=%s failed: %s", d, v, e)
    db.commit()
    return upserted


def _upsert_inflation(db: Session, rows: list[tuple[date, Decimal]]) -> int:
    """Апсертит (month1st, inflation%) в macro_indicator. SAVEPOINT per-row.
    Возвращает число успешных upsert'ов."""
    upserted = 0
    for d, v in rows:
        try:
            with db.begin_nested():
                db.execute(UPSERT_INFLATION_SQL, {"d": d, "v": v})
            upserted += 1
        except Exception as e:
            logger.warning("upsert inflation_yoy %s=%s failed: %s", d, v, e)
    db.commit()
    return upserted


# Ретраев здесь НЕТ намеренно, и параметров, обещающих их, тоже быть не должно
# (#2464). Раньше стояло `bind=True, max_retries=2` — но self не использовался,
# self.retry() не вызывался и autoretry_for задан не был, поэтому конфигурация
# ретраев не имела ни малейшего эффекта: таска падала окончательно с первой ошибки,
# а параметр обещал до двух повторов. Соседи, где ретраи действительно нужны,
# задают их явно: autoretry_for в nspd_sync и scrape_cadastre, self.retry() в
# scrape_kn.
#
# Отсутствие ретраев — это и есть задуманное поведение, оно описано в докстринге
# ниже: «первая возникшая ошибка пробрасывается в конце (surfaces в
# Celery/GlitchTip), не глотается». Ряды тянутся по расписанию, следующий тик
# повторит попытку; молча ретраить внутри тика значило бы прятать отказ источника.
@celery_app.task(
    name="tasks.cbr_macro_sync.cbr_macro_sync",
)
def cbr_macro_sync(
    from_date: str | None = None,
    to_date: str | None = None,
) -> dict[str, Any]:
    """Загрузить ряды ЦБ (ключевая ставка + инфляция YoY) и апсертить в macro_indicator.

    Per-series guard (зеркало rosstat_macro_sync): сбой одного ряда не блокирует
    другой — оба пытаются fetch+upsert независимо; первая возникшая ошибка
    пробрасывается в конце (surfaces в Celery/GlitchTip), не глотается.

    Args:
        from_date: ISO-дата начала периода ('YYYY-MM-DD'). None → дефолт КАЖДОГО
            scraper'а (key_rate 2019-01-01, inflation 2013-01-01; полный бэкфилл).
        to_date: ISO-дата конца периода. None → сегодня.

    Returns:
        Счётчики. ``fetched``/``upserted`` на верхнем уровне — по ключевой ставке
        (обратная совместимость); ``key_rate`` и ``inflation`` — пер-ряд под-счётчики.

    Raises:
        Пробрасывает первую ошибку fetch'а (CBRScraperError и пр.) после попытки
        обоих рядов — чтобы фейл был виден в Celery / GlitchTip, а не молча проглочен.
    """
    fd = date.fromisoformat(from_date) if from_date else None
    td = date.fromisoformat(to_date) if to_date else None

    db = SessionLocal()
    try:
        first_error: Exception | None = None

        # ── key_rate (дневной, SOAP) ──────────────────────────────────────────
        kr_fetched = kr_upserted = 0
        try:
            kr_rows = fetch_key_rate(from_date=fd, to_date=td)
            kr_fetched = len(kr_rows)
            logger.info("cbr_macro_sync: fetched %d key_rate rows", kr_fetched)
            kr_upserted = _upsert_key_rate(db, kr_rows)
            logger.info("cbr_macro_sync: upserted %d/%d key_rate rows", kr_upserted, kr_fetched)
        except Exception as e:
            logger.exception("cbr_macro_sync: key_rate sync failed: %s", e)
            first_error = first_error or e

        # ── inflation_yoy (месячный, Excel #946) ──────────────────────────────
        infl_fetched = infl_upserted = 0
        try:
            infl_rows = fetch_inflation(from_date=fd, to_date=td)
            infl_fetched = len(infl_rows)
            logger.info("cbr_macro_sync: fetched %d inflation_yoy rows", infl_fetched)
            infl_upserted = _upsert_inflation(db, infl_rows)
            logger.info(
                "cbr_macro_sync: upserted %d/%d inflation_yoy rows", infl_upserted, infl_fetched
            )
        except Exception as e:
            logger.exception("cbr_macro_sync: inflation sync failed: %s", e)
            first_error = first_error or e

        if first_error is not None:
            raise first_error

        return {
            # обратная совместимость: верхнеуровневые счётчики = ключевая ставка
            "fetched": kr_fetched,
            "upserted": kr_upserted,
            "key_rate": {"fetched": kr_fetched, "upserted": kr_upserted},
            "inflation": {"fetched": infl_fetched, "upserted": infl_upserted},
            "from_date": fd.isoformat() if fd else None,
            "to_date": td.isoformat() if td else None,
        }
    finally:
        db.close()