"""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 @celery_app.task( bind=True, name="tasks.cbr_macro_sync.cbr_macro_sync", max_retries=2, ) def cbr_macro_sync( self: Any, 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()