Light1YT
f7c4d7a8c5
EPIC2 macro-ingest: региональный (Свердловская обл. ОКТМО 65 / ЕКБ) скрейпер Росстата в унифицированную macro_indicator (м.123). Зеркалит форму cbr_macro: pure-парсеры + тонкий httpx, Celery-таска с SAVEPOINT-per-row upsert (CAST not ::, ON CONFLICT по PK indicator_type/region/obs_date). ЕМИСС/fedstat.ru (SDMX) за WAF (hard-403 на всех путях из dev-IP) -> ИПЦ/доходы/ СМР-цены пока недоступны (документировано в коде). rosstat.gov.ru/opendata (стандарт 4.0) WAF-free -> приземляет демографию population_total (sverdl 4.32M / ekb 1.41M, §7.11). - services/scrapers/rosstat_emiss.py — fetch+parse open-data (registry->meta->data CSV) - workers/tasks/rosstat_macro_sync.py — Celery upsert (source=rosstat) - beat: rosstat-macro-sync-monthly; include в celery_app - site_finder/macro.py: population_total -> region-aware default sverdl (additive) - tests: 18 offline-тестов парсера + контракта таски Без миграции (вписано в существующую схему), без новых зависимостей. Refs #946.
122 lines
5.1 KiB
Python
122 lines
5.1 KiB
Python
"""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()
|