bot-backend
6982255fb3
All checks were successful
CI Trade-In / changes (pull_request) Successful in 7s
CI / changes (pull_request) Successful in 9s
CI Trade-In / backend-tests (pull_request) Has been skipped
CI Trade-In / browser-tests (pull_request) Has been skipped
CI Trade-In / frontend-checks (pull_request) Has been skipped
CI / frontend-tests (pull_request) Has been skipped
CI / openapi-codegen-check (pull_request) Successful in 1m54s
CI / backend-tests (pull_request) Successful in 17m0s
11 тасок объявляли `max_retries=2`, но ретраи не реализовывали: ни `autoretry_for` в декораторе, ни вызова `self.retry()` в теле. Celery в таком виде параметр не применяет — при исключении таска падает с первой попытки. Читающий код видит «до 3 попыток», а их одна. Убран `max_retries` у: cbr_macro_sync, rosstat_macro_sync, developer_registry_refresh, location_refresh, mv_sales_tracker_refresh, refresh_analytics, refresh_layout_velocity, refresh_quarter_price_index, scrape_objective.sync_objective_group, supply_layers_refresh, scrape_kn.scrape_kn_region. Заодно убран `bind=True` там, где `self` не использовался вовсе; в `scrape_kn_region` он оставлен — `self.request.id` пишется в kn_scrape_log. Не тронуты и не должны быть: `resume_kn_run` (max_retries=12 + настоящий self.retry()), `nspd_sync`/`scrape_cadastre` (autoretry_for), `nspd_geo`/`objective_etl` (max_retries=0 — честное «ретраев нет»). Гейт `test_2464_retry_config_is_real.py` разбирает AST всех модулей `app/workers/tasks/` и требует: если декоратор объявляет ненулевой max_retries, в нём есть autoretry_for либо в теле функции есть self.retry(). Три таски из одиннадцати гейт нашёл сверх списка эпика. Проверка гейта: с фиксом зелено, при возврате `max_retries=2` в supply_layers_refresh — красно с указанием на эту таску. Плюс два контроля: гейт видит ≥20 тасок (не молчит из-за пустой выборки) и признаёт обе законные формы ретраев. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
326 lines
15 KiB
Python
326 lines
15 KiB
Python
"""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
|
||
|
||
|
||
# Ретраев здесь НЕТ намеренно, и параметров, обещающих их, тоже быть не должно
|
||
# (#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.rosstat_macro_sync.rosstat_macro_sync",
|
||
)
|
||
def rosstat_macro_sync() -> 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()
|