feat(macro): ЕМИСС income_per_capita → macro_indicator (#946 pt2) #1144
7 changed files with 1016 additions and 42 deletions
|
|
@ -7,15 +7,32 @@
|
|||
Celery-таска ``rosstat_macro_sync`` через SAVEPOINT-per-row.
|
||||
|
||||
────────────────────────────────────────────────────────────────────────────────
|
||||
ИСТОЧНИК ДАННЫХ — почему open-data, а НЕ ЕМИСС-SDMX (разведано live 2026-06-07)
|
||||
ИСТОЧНИК ДАННЫХ — open-data + ЕМИСС-SDMX (разведано live 2026-06-07)
|
||||
────────────────────────────────────────────────────────────────────────────────
|
||||
ЕМИСС / fedstat.ru (SDMX + ``/indicator/data.do``) — за WAF: ЛЮБОЙ путь/метод/набор
|
||||
заголовков отдаёт статичный 80-байтный ``<h1>Forbidden</h1>`` БЕЗ ``Server``-заголовка
|
||||
(блок на фронтящем WAF/CDN до приложения — IP-reputation, не JS-challenge, который мог
|
||||
бы решить браузер). ``sdmx.gks.ru`` — DNS мёртв, ``showdata.gks.ru`` — 502. То есть
|
||||
EMISS-only ряды (ИПЦ, среднедушевые доходы, индекс цен на СМР) автоматически НЕдоступны.
|
||||
Это независимо зафиксировано прежней разведкой (комментарий миграции 126 §7.11, vault
|
||||
data-sources).
|
||||
ЕМИСС / fedstat.ru — за WAF с DEV-машины (hard-403), но ДОСТУПЕН С ПРОДА (verified
|
||||
2026-06-07: ``POST https://www.fedstat.ru/indicator/data.do`` body ``id=<id>&format=sdmx``
|
||||
→ 200 + SDMX-ML v1.0 GenericData). Поэтому ЕМИСС-fetch выполняется ТОЛЬКО на проде
|
||||
(Celery); парсер SDMX — pure и dev-тестируем на фикстурах ``tests/fixtures/emiss_*.xml``.
|
||||
|
||||
ЧТО ПРИЗЕМЛЯЕТСЯ из ЕМИСС (см. EMISS_INDICATORS ниже):
|
||||
• income_per_capita — «Среднедушевые денежные доходы населения», fedstat id=57039,
|
||||
размерность ``s_OKATO`` (Свердл = ОКАТО 65000000000), квартальный (рубль/мес).
|
||||
ВАЖНО: дефолтный SDMX-экспорт этого индикатора СОДЕРЖИТ все субъекты, в т.ч. 65.
|
||||
|
||||
ЧТО НЕ ПРИЗЕМЛЯЕТСЯ из ЕМИСС (документировано, частичная поставка допустима ТЗ §7.11):
|
||||
• cpi (ИПЦ) — fedstat id=31074 «Индексы потребительских цен на товары и услуги»
|
||||
(% , монотип, есть месячные и квартальные периоды). НЕ landed: индикатор много-
|
||||
мерный (поля 57831=ОКАТО, 58273=виды товаров/услуг, 57937=виды показателя,
|
||||
33560=период, 3=год), и его ДЕФОЛТНЫЙ data.do-экспорт отдаёт лишь узкий срез
|
||||
(ОКАТО 030/14/15 — ЦФО/Белгород/Брянск), БЕЗ Свердл-65. Чтобы выбрать регион,
|
||||
fedstat требует POST ``dataGrid.do`` c ``selectedFilterIds=<fieldId>_<internalId>``,
|
||||
перечисляющим выбранные коды ПО ВСЕМ размерностям, где ``internalId`` — внутренний
|
||||
id кода (НЕ значение ОКАТО), берётся отдельным filter-tree AJAX; плюс на странице
|
||||
индикатора reCAPTCHA. Это глубокий RE — отложено (закрывается тем же EmissIndicatorSpec
|
||||
+ параметризацией data.do-выборки, когда добудем filter_id-маппинг).
|
||||
• construction_price_index (индекс цен на СМР) — id не подтверждён: кандидаты
|
||||
(31448 «Средние потреб. цены», 37426 «Еженедельные цены») — много-мерные и БЕЗ
|
||||
Свердл в дефолт-экспорте (та же проблема, что у ИПЦ). НЕ landed.
|
||||
|
||||
``rosstat.gov.ru/opendata`` — БЕЗ WAF, машиночитаем (российский стандарт открытых
|
||||
данных 4.0):
|
||||
|
|
@ -33,13 +50,11 @@ data-sources).
|
|||
муниципальный срез; берём субъект-агрегат ОКТМО ``6500000000`` + г. Екатеринбург
|
||||
``657010000011000``). Это § 7.11 «демография (численность)».
|
||||
|
||||
Что НЕ приземляется (документировано, частичная поставка допустима ТЗ):
|
||||
• среднедушевые доходы, ИПЦ, индекс цен на СМР — EMISS-only → WAF (см. выше). В
|
||||
open-data их нет (проверено по реестру: 0 совпадений). Закрывается, когда появится
|
||||
рабочий доступ к ЕМИСС (прокси / ручная выгрузка) — тогда добавится новый
|
||||
``DatasetSpec`` с тем же upsert-контрактом.
|
||||
Что НЕ приземляется из open-data (но среднедушевые доходы теперь идут через ЕМИСС-прод):
|
||||
• ИПЦ, индекс цен на СМР — EMISS-only и в open-data их нет (0 совпадений по реестру);
|
||||
из ЕМИСС пока не landed (много-мерные, см. EMISS_INDICATORS-блок выше).
|
||||
|
||||
Standalone smoke (ручная проверка против живого rosstat.gov.ru):
|
||||
Standalone smoke (ручная проверка; open-data — c dev, ЕМИСС-fetch — только с прода):
|
||||
python -m app.services.scrapers.rosstat_emiss
|
||||
|
||||
Mirror conventions: ``httpx.Client`` + явный User-Agent (как cbr_macro / ekburg_permits),
|
||||
|
|
@ -51,6 +66,7 @@ from __future__ import annotations
|
|||
import csv
|
||||
import io
|
||||
import logging
|
||||
import xml.etree.ElementTree as ET
|
||||
from dataclasses import dataclass
|
||||
from datetime import date
|
||||
from decimal import Decimal, InvalidOperation
|
||||
|
|
@ -63,6 +79,10 @@ logger = logging.getLogger(__name__)
|
|||
|
||||
ROSSTAT_OPENDATA_BASE = "https://rosstat.gov.ru/opendata"
|
||||
|
||||
# ЕМИСС / fedstat.ru — SDMX-экспорт индикатора (POST id=<id>&format=sdmx). За WAF с
|
||||
# dev (403), доступен с прода (см. шапку модуля). verify off — тот же российский гос-УЦ.
|
||||
FEDSTAT_DATA_URL = "https://www.fedstat.ru/indicator/data.do"
|
||||
|
||||
# Браузероподобный UA: rosstat.gov.ru отдаёт open-data без WAF, но «голый» httpx-UA
|
||||
# иногда ловит 403 на edge-кэше — даём обычный desktop-Chrome UA (как nspd_client).
|
||||
USER_AGENT = (
|
||||
|
|
@ -397,6 +417,56 @@ def _http_get(
|
|||
raise RosstatScraperError(f"Rosstat GET failed after {retries + 1} attempts {url}: {last_exc}")
|
||||
|
||||
|
||||
def _http_post(
|
||||
client: httpx.Client,
|
||||
url: str,
|
||||
*,
|
||||
data: dict[str, str],
|
||||
retries: int = DEFAULT_RETRIES,
|
||||
) -> bytes:
|
||||
"""POST form-data с ретраями (сеть / 5xx) → байты тела. Для ЕМИСС data.do.
|
||||
|
||||
Module-level для monkeypatch в тестах. Раздельно от _http_get: fedstat принимает
|
||||
данные ТОЛЬКО POST'ом (form-urlencoded id+format), GET отдаёт SPA-оболочку.
|
||||
|
||||
Raises:
|
||||
RosstatScraperError: на не-200 / сетевой ошибке после всех ретраев.
|
||||
"""
|
||||
import time
|
||||
|
||||
last_exc: Exception | None = None
|
||||
for attempt in range(retries + 1):
|
||||
try:
|
||||
resp = client.post(url, data=data)
|
||||
except httpx.HTTPError as e:
|
||||
last_exc = e
|
||||
logger.warning(
|
||||
"EMISS POST network error (attempt %d/%d) %s: %s",
|
||||
attempt + 1,
|
||||
retries + 1,
|
||||
url,
|
||||
e,
|
||||
)
|
||||
else:
|
||||
if resp.status_code == 200:
|
||||
return resp.content
|
||||
last_exc = RosstatScraperError(
|
||||
f"EMISS POST HTTP {resp.status_code} {url}: {resp.text[:200]}"
|
||||
)
|
||||
if resp.status_code < 500:
|
||||
break # 4xx (в т.ч. WAF-403 с dev) — ретрай бессмыслен
|
||||
logger.warning(
|
||||
"EMISS POST HTTP %d (attempt %d/%d) %s",
|
||||
resp.status_code,
|
||||
attempt + 1,
|
||||
retries + 1,
|
||||
url,
|
||||
)
|
||||
if attempt < retries:
|
||||
time.sleep(min(2**attempt, 10))
|
||||
raise RosstatScraperError(f"EMISS POST failed after {retries + 1} attempts {url}: {last_exc}")
|
||||
|
||||
|
||||
def fetch_dataset(
|
||||
spec: DatasetSpec,
|
||||
*,
|
||||
|
|
@ -475,6 +545,307 @@ def fetch_all_datasets(
|
|||
return out
|
||||
|
||||
|
||||
# ══════════════════════════════════════════════════════════════════════════════════
|
||||
# ЕМИСС / fedstat.ru — SDMX-ML v1.0 GenericData (EMISS-only ряды, fetch только с прода)
|
||||
# ══════════════════════════════════════════════════════════════════════════════════
|
||||
# Источник: POST https://www.fedstat.ru/indicator/data.do body: id=<id>&format=sdmx
|
||||
# → SDMX-ML v1.0 GenericData (см. tests/fixtures/emiss_income_57039.xml). Структура:
|
||||
# <GenericData xmlns=...message xmlns:generic=...generic xmlns:structure=...>
|
||||
# <CodeLists><structure:CodeList id="s_OKATO">…коды регионов…</structure:CodeList>
|
||||
# <Description><Indicator id="57039" name="…"/></Description> (fedstat-расширение)
|
||||
# <DataSet>
|
||||
# <generic:Series>
|
||||
# <generic:SeriesKey><generic:Value concept="s_OKATO" value="65000000000"/></…>
|
||||
# <generic:Attributes>
|
||||
# <generic:Value concept="EI" value="рубль"/> (единица измерения)
|
||||
# <generic:Value concept="PERIOD" value="I квартал"/> (под-период года)
|
||||
# </generic:Attributes>
|
||||
# <generic:Obs><generic:Time>2024</generic:Time>
|
||||
# <generic:ObsValue value="54006"/></generic:Obs>
|
||||
# </generic:Series> …
|
||||
# Один <Series> = один (регион, PERIOD) ряд; год наблюдения — в <Obs><Time> (напр. 2024),
|
||||
# фактический момент = PERIOD + Time-год (I квартал 2024 → 2024-01-01). Регион берём из
|
||||
# SeriesKey concept="s_OKATO" (ОКАТО-код), фильтруем по целевым кодам (Свердл = 65…).
|
||||
# Парсер pure → dev-тестируем на фикстурах; сам fetch (httpx POST) исполняется на проде.
|
||||
|
||||
_SDMX_NS = {
|
||||
"msg": "http://www.SDMX.org/resources/SDMXML/schemas/v1_0/message",
|
||||
"g": "http://www.SDMX.org/resources/SDMXML/schemas/v1_0/generic",
|
||||
}
|
||||
|
||||
# PERIOD-значения fedstat → стартовый месяц периода (obs_date нормализуем к 1-му числу
|
||||
# стартового месяца под-периода, как monthly-ряды к 1-му числу). Покрывает кварталы и
|
||||
# месяцы (ИПЦ публикуется и помесячно). Незнакомый PERIOD → None (ряд пропускается).
|
||||
_QUARTER_START_MONTH: dict[str, int] = {
|
||||
"i квартал": 1,
|
||||
"ii квартал": 4,
|
||||
"iii квартал": 7,
|
||||
"iv квартал": 10,
|
||||
}
|
||||
_MONTH_NAME_TO_NUM: dict[str, int] = {
|
||||
"январь": 1,
|
||||
"февраль": 2,
|
||||
"март": 3,
|
||||
"апрель": 4,
|
||||
"май": 5,
|
||||
"июнь": 6,
|
||||
"июль": 7,
|
||||
"август": 8,
|
||||
"сентябрь": 9,
|
||||
"октябрь": 10,
|
||||
"ноябрь": 11,
|
||||
"декабрь": 12,
|
||||
}
|
||||
# Годовой под-период (некоторые индикаторы): «значение показателя за год», «год».
|
||||
_YEAR_PERIODS: frozenset[str] = frozenset(
|
||||
{"год", "значение показателя за год", "за год"}
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class EmissIndicatorSpec:
|
||||
"""Описание одного приземляемого ЕМИСС-индикатора (fedstat data.do, SDMX).
|
||||
|
||||
Атрибуты:
|
||||
indicator_id: числовой fedstat ``id`` (param ``id=`` в data.do).
|
||||
indicator_type: тип ряда в macro_indicator (напр. 'income_per_capita').
|
||||
unit: единица измерения (справочно; в PK не участвует).
|
||||
frequency: 'monthly' | 'quarterly' | 'yearly' — для колонки frequency.
|
||||
targets: целевые территории (ОКАТО-код → region-метка macro_indicator).
|
||||
comment: пояснение для колонки comment macro_indicator.
|
||||
"""
|
||||
|
||||
indicator_id: str
|
||||
indicator_type: str
|
||||
unit: str
|
||||
frequency: str
|
||||
targets: tuple[RegionTarget, ...]
|
||||
comment: str = ""
|
||||
|
||||
|
||||
# Свердловская обл. = ОКАТО '65000000000' (region='sverdl'). ЕКБ-горокруг в ОКАТО-
|
||||
# размерности ЕМИСС-индикаторов отсутствует (только субъектовый срез), поэтому здесь
|
||||
# целимся ТОЛЬКО в субъект-65 (в отличие от open-data population, где есть и ЕКБ).
|
||||
OKATO_SVERDLOVSK_OBLAST = "65000000000"
|
||||
|
||||
INCOME_PER_CAPITA_SPEC = EmissIndicatorSpec(
|
||||
indicator_id="57039",
|
||||
indicator_type="income_per_capita",
|
||||
unit="руб",
|
||||
frequency="quarterly",
|
||||
targets=(RegionTarget(oktmo=OKATO_SVERDLOVSK_OBLAST, region="sverdl"),),
|
||||
comment="ЕМИСС id=57039 'Среднедушевые денежные доходы населения' (#946 §7.11)",
|
||||
)
|
||||
|
||||
# Активные ЕМИСС-индикаторы. ИПЦ (id=31074) / индекс цен на СМР пока НЕ здесь —
|
||||
# многомерные, дефолт-экспорт data.do без Свердл, нужна dataGrid-выборка (см. шапку).
|
||||
EMISS_INDICATORS: tuple[EmissIndicatorSpec, ...] = (INCOME_PER_CAPITA_SPEC,)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class EmissRow:
|
||||
"""Одна готовая к upsert строка macro_indicator из ЕМИСС (source='emiss').
|
||||
|
||||
Отдельно от ``MacroRow`` (open-data, source='rosstat', yearly): ЕМИСС-ряды несут
|
||||
свою frequency (quarterly/monthly) и source — контракт upsert'а у них иной.
|
||||
"""
|
||||
|
||||
indicator_type: str
|
||||
region: str
|
||||
obs_date: date
|
||||
value: Decimal
|
||||
unit: str
|
||||
frequency: str
|
||||
comment: str
|
||||
|
||||
|
||||
def _emiss_period_to_month(period: str) -> int | None:
|
||||
"""PERIOD-строку fedstat → стартовый месяц под-периода (1..12) или None.
|
||||
|
||||
Квартал ('I квартал'…'IV квартал') → первый месяц квартала; месяц ('январь'…) →
|
||||
его номер; годовой под-период ('год') → 1 (январь). Незнакомое → None (caller
|
||||
пропустит ряд + warning) — защита от неожиданных под-периодов (недели и т.п.).
|
||||
"""
|
||||
p = (period or "").strip().lower()
|
||||
if not p:
|
||||
return None
|
||||
if p in _QUARTER_START_MONTH:
|
||||
return _QUARTER_START_MONTH[p]
|
||||
if p in _MONTH_NAME_TO_NUM:
|
||||
return _MONTH_NAME_TO_NUM[p]
|
||||
if p in _YEAR_PERIODS:
|
||||
return 1
|
||||
return None
|
||||
|
||||
|
||||
def _series_concepts(series: ET.Element) -> tuple[dict[str, str], dict[str, str]]:
|
||||
"""Извлечь (SeriesKey, Attributes) concept→value словари из <generic:Series>.
|
||||
|
||||
SeriesKey несёт размерности (s_OKATO и пр.); Attributes — EI (единица) и PERIOD.
|
||||
Отсутствующий блок → пустой словарь (не падаем).
|
||||
"""
|
||||
|
||||
def _vals(parent: ET.Element | None) -> dict[str, str]:
|
||||
out: dict[str, str] = {}
|
||||
if parent is None:
|
||||
return out
|
||||
for v in parent.findall("g:Value", _SDMX_NS):
|
||||
concept = v.get("concept")
|
||||
value = v.get("value")
|
||||
if concept is not None and value is not None:
|
||||
out[concept] = value
|
||||
return out
|
||||
|
||||
key = _vals(series.find("g:SeriesKey", _SDMX_NS))
|
||||
attrs = _vals(series.find("g:Attributes", _SDMX_NS))
|
||||
return key, attrs
|
||||
|
||||
|
||||
def parse_emiss_sdmx(raw: bytes | str, spec: EmissIndicatorSpec) -> list[EmissRow]:
|
||||
"""Pure-парсер SDMX-ML v1.0 GenericData fedstat → список EmissRow целевых регионов.
|
||||
|
||||
Алгоритм:
|
||||
1. Парсит XML (ET). Битый/непарсящийся XML → [] (+warning), не бросает.
|
||||
2. Для каждого <generic:Series>: регион = SeriesKey concept='s_OKATO'. Если ОКАТО
|
||||
не среди spec.targets — ряд пропускается. PERIOD (Attributes) → стартовый месяц
|
||||
под-периода; год — из <Obs><Time>; obs_date = date(year, month, 1). Значение —
|
||||
из <Obs><ObsValue value=...> через _parse_value (запятая-десятичный, разделители).
|
||||
3. Дубли по (region, obs_date) схлопываются (последний выигрывает) — PK всё равно один.
|
||||
|
||||
Никогда не бросает на данных: битые ячейки/периоды/годы → наблюдение пропускается.
|
||||
"""
|
||||
if isinstance(raw, bytes):
|
||||
try:
|
||||
text_xml = raw.decode("utf-8")
|
||||
except UnicodeDecodeError:
|
||||
text_xml = raw.decode("utf-8", errors="replace")
|
||||
else:
|
||||
text_xml = raw
|
||||
try:
|
||||
root = ET.fromstring(text_xml)
|
||||
except ET.ParseError as e:
|
||||
logger.warning("EMISS %s: SDMX XML не распарсился: %s", spec.indicator_id, e)
|
||||
return []
|
||||
|
||||
dataset = root.find("msg:DataSet", _SDMX_NS)
|
||||
if dataset is None:
|
||||
logger.warning("EMISS %s: нет <DataSet> в SDMX", spec.indicator_id)
|
||||
return []
|
||||
|
||||
target_by_okato = {t.oktmo: t for t in spec.targets}
|
||||
by_key: dict[tuple[str, date], EmissRow] = {}
|
||||
|
||||
for series in dataset.findall("g:Series", _SDMX_NS):
|
||||
key, attrs = _series_concepts(series)
|
||||
okato = key.get("s_OKATO", "").strip()
|
||||
target = target_by_okato.get(okato)
|
||||
if target is None:
|
||||
continue
|
||||
month = _emiss_period_to_month(attrs.get("PERIOD", ""))
|
||||
if month is None:
|
||||
logger.warning(
|
||||
"EMISS %s: неизвестный PERIOD %r (ОКАТО %s) — пропуск",
|
||||
spec.indicator_id,
|
||||
attrs.get("PERIOD"),
|
||||
okato,
|
||||
)
|
||||
continue
|
||||
for obs in series.findall("g:Obs", _SDMX_NS):
|
||||
time_el = obs.find("g:Time", _SDMX_NS)
|
||||
val_el = obs.find("g:ObsValue", _SDMX_NS)
|
||||
if time_el is None or time_el.text is None or val_el is None:
|
||||
continue
|
||||
yr = time_el.text.strip()[:4]
|
||||
if not yr.isdigit():
|
||||
continue
|
||||
value = _parse_value(val_el.get("value", ""))
|
||||
if value is None:
|
||||
continue
|
||||
obs_date = date(int(yr), month, 1)
|
||||
by_key[(target.region, obs_date)] = EmissRow(
|
||||
indicator_type=spec.indicator_type,
|
||||
region=target.region,
|
||||
obs_date=obs_date,
|
||||
value=value,
|
||||
unit=spec.unit,
|
||||
frequency=spec.frequency,
|
||||
comment=spec.comment,
|
||||
)
|
||||
|
||||
return [by_key[k] for k in sorted(by_key, key=lambda k: (k[0], k[1]))]
|
||||
|
||||
|
||||
def fetch_emiss_indicator(
|
||||
spec: EmissIndicatorSpec,
|
||||
*,
|
||||
timeout_s: float = DEFAULT_TIMEOUT_S,
|
||||
retries: int = DEFAULT_RETRIES,
|
||||
client: httpx.Client | None = None,
|
||||
) -> list[EmissRow]:
|
||||
"""Загрузить и распарсить один ЕМИСС-индикатор с fedstat → список EmissRow.
|
||||
|
||||
ВНИМАНИЕ: fedstat.ru за WAF с dev-машины (403) — этот fetch рассчитан на исполнение
|
||||
с ПРОДА (Celery). POST ``data.do`` body ``id=<id>&format=sdmx``; verify off (российский
|
||||
гос-УЦ, как и open-data выше). Парсинг — в pure ``parse_emiss_sdmx`` (offline-тестируем).
|
||||
|
||||
Raises:
|
||||
RosstatScraperError: на сетевой/HTTP-ошибке после ретраев.
|
||||
"""
|
||||
own_client = client is None
|
||||
if client is None:
|
||||
client = httpx.Client(
|
||||
timeout=timeout_s,
|
||||
headers={"User-Agent": USER_AGENT, "Accept-Language": "ru-RU,ru;q=0.9"},
|
||||
follow_redirects=True,
|
||||
verify=_VERIFY_TLS,
|
||||
)
|
||||
try:
|
||||
raw = _http_post(
|
||||
client,
|
||||
f"{FEDSTAT_DATA_URL}",
|
||||
data={"id": spec.indicator_id, "format": "sdmx"},
|
||||
retries=retries,
|
||||
)
|
||||
rows = parse_emiss_sdmx(raw, spec)
|
||||
logger.info(
|
||||
"EMISS %s (%s): распарсено %d строк для %d целевых ОКАТО",
|
||||
spec.indicator_id,
|
||||
spec.indicator_type,
|
||||
len(rows),
|
||||
len(spec.targets),
|
||||
)
|
||||
return rows
|
||||
finally:
|
||||
if own_client:
|
||||
client.close()
|
||||
|
||||
|
||||
def fetch_all_emiss(
|
||||
specs: tuple[EmissIndicatorSpec, ...] = EMISS_INDICATORS,
|
||||
*,
|
||||
timeout_s: float = DEFAULT_TIMEOUT_S,
|
||||
retries: int = DEFAULT_RETRIES,
|
||||
) -> list[EmissRow]:
|
||||
"""Загрузить все активные ЕМИСС-индикаторы → плоский список EmissRow.
|
||||
|
||||
Один общий httpx.Client на батч. Сбой одного индикатора логируется и НЕ валит
|
||||
остальные (graceful — частичная поставка допустима ТЗ). Исполняется на проде.
|
||||
"""
|
||||
out: list[EmissRow] = []
|
||||
with httpx.Client(
|
||||
timeout=timeout_s,
|
||||
headers={"User-Agent": USER_AGENT, "Accept-Language": "ru-RU,ru;q=0.9"},
|
||||
follow_redirects=True,
|
||||
verify=_VERIFY_TLS,
|
||||
) as client:
|
||||
for spec in specs:
|
||||
try:
|
||||
out.extend(fetch_emiss_indicator(spec, retries=retries, client=client))
|
||||
except RosstatScraperError as e:
|
||||
logger.warning("EMISS: индикатор %s пропущен: %s", spec.indicator_id, e)
|
||||
return out
|
||||
|
||||
|
||||
# ── standalone smoke ────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -73,12 +73,17 @@ _MORTGAGE_INDICATOR_TYPES: frozenset[str] = frozenset(
|
|||
itype for itype, _unit in _CBR_SERIES_MAP.values()
|
||||
)
|
||||
|
||||
# Региональные ряды Росстата (#946, source='rosstat') кладутся под region='sverdl'
|
||||
# (субъект-агрегат, ОКТМО 65) и 'ekb' (г. Екатеринбург, ОКТМО 6570100000), НЕ под 'rf'.
|
||||
# Региональные ряды Росстата (#946) кладутся под region='sverdl' (субъект-агрегат) и
|
||||
# 'ekb' (г. Екатеринбург), НЕ под 'rf'. Два источника:
|
||||
# • open-data (source='rosstat'): population_total (ОКТМО 65 / ЕКБ);
|
||||
# • ЕМИСС/fedstat (source='emiss'): income_per_capita (среднедушевые доходы, ОКАТО 65 —
|
||||
# только субъект, ЕКБ-среза в ОКАТО-размерности ЕМИСС нет).
|
||||
# Дефолтный region таких типов = 'sverdl' (субъект — основной уровень потребления
|
||||
# downstream affordability/forecast). Явный region='ekb' читатель задаёт сам.
|
||||
# Источник правды по типам — DatasetSpec'и в rosstat_emiss (синхронить при добавлении).
|
||||
_ROSSTAT_SVERDL_INDICATOR_TYPES: frozenset[str] = frozenset({"population_total"})
|
||||
# Источник правды по типам — DatasetSpec/EmissIndicatorSpec в rosstat_emiss (синхронить).
|
||||
_ROSSTAT_SVERDL_INDICATOR_TYPES: frozenset[str] = frozenset(
|
||||
{"population_total", "income_per_capita"}
|
||||
)
|
||||
|
||||
|
||||
def _canonical_region(indicator_type: str) -> str:
|
||||
|
|
@ -86,7 +91,7 @@ def _canonical_region(indicator_type: str) -> str:
|
|||
|
||||
Детерминированно, без БД:
|
||||
• 6 CBR-ипотечных типов (mortgage_*) → 'sverdl' (так лежат на prod);
|
||||
• региональные ряды Росстата (population_total, #946) → 'sverdl' (субъект-уровень);
|
||||
• региональные ряды Росстата (population_total, income_per_capita, #946) → 'sverdl';
|
||||
• всё остальное (key_rate и пр.) → 'rf'.
|
||||
Источники истины — _CBR_SERIES_MAP (зеркало миграции 123) и _ROSSTAT_SVERDL_INDICATOR_TYPES
|
||||
(зеркало DatasetSpec'ов rosstat_emiss).
|
||||
|
|
|
|||
|
|
@ -1,11 +1,15 @@
|
|||
"""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, когда появится рабочий доступ.
|
||||
Тянет региональные ряды Росстата для Свердловской обл. из ДВУХ источников и апсертит
|
||||
каждую обсервацию в ``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'.
|
||||
Каждый источник в своём try/except (per-source guard): сбой одного НЕ роняет другой и
|
||||
не валит таску — частичная поставка допустима (ТЗ §7.11). ЕМИСС-fetch доступен только
|
||||
с прода (fedstat за WAF с dev); ИПЦ/индекс цен на СМР пока не landed (см. rosstat_emiss).
|
||||
|
||||
Детерминированно, без LLM. ON CONFLICT делает повторные прогоны идемпотентными
|
||||
(re-run обновляет value + updated_at). Расписание — ежемесячно (эти ряды
|
||||
|
|
@ -26,7 +30,12 @@ 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.services.scrapers.rosstat_emiss import (
|
||||
EmissRow,
|
||||
MacroRow,
|
||||
fetch_all_datasets,
|
||||
fetch_all_emiss,
|
||||
)
|
||||
from app.workers.celery_app import celery_app
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
|
@ -56,6 +65,29 @@ UPSERT_ROSSTAT_SQL = text(
|
|||
"""
|
||||
)
|
||||
|
||||
# ЕМИСС-ряды (source='emiss'): frequency параметризована (quarterly для доходов,
|
||||
# monthly для ИПЦ когда добавится) — в отличие от open-data, где она фикс 'yearly'.
|
||||
# Тот же PK и ON CONFLICT-контракт. CAST(:x AS type) — НИКОГДА :x::type (psycopg v3).
|
||||
UPSERT_EMISS_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),
|
||||
'emiss', CAST(:freq AS text), 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, чтобы один битый
|
||||
|
|
@ -89,34 +121,102 @@ def _upsert_rows(db: Session, rows: list[MacroRow]) -> int:
|
|||
return upserted
|
||||
|
||||
|
||||
def _upsert_emiss_rows(db: Session, rows: list[EmissRow]) -> int:
|
||||
"""Апсертит EmissRow в macro_indicator (source='emiss', frequency per-row).
|
||||
SAVEPOINT per-row, чтобы один битый ряд не откатывал всю транзакцию. Возвращает
|
||||
число успешных upsert'ов."""
|
||||
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,
|
||||
"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 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.
|
||||
"""Загрузить региональные ряды Росстата (open-data + ЕМИСС) и апсертить в macro_indicator.
|
||||
|
||||
Источники выполняются НЕЗАВИСИМО (per-source try/except): сбой одного источника
|
||||
логируется, апсертит здоровый источник, и в конце ошибка ПРОБРАСЫВАЕТСЯ (чтобы фейл
|
||||
был виден в Celery / GlitchTip, а не молча проглочен). Re-run идемпотентен (ON CONFLICT),
|
||||
поэтому Celery-retry безопасно повторит оба источника.
|
||||
|
||||
Returns:
|
||||
Счётчики ``{"fetched": N, "upserted": M}``.
|
||||
Счётчики ``{"fetched": N, "upserted": M, "rosstat": ..., "emiss": ...}``.
|
||||
|
||||
Raises:
|
||||
Пробрасывает любую ошибку fetch'а (RosstatScraperError и пр.) — чтобы фейл
|
||||
был виден в Celery / GlitchTip, а не молча проглочен. Частичный сбой
|
||||
отдельного датасета НЕ роняет таску (fetch_all_datasets graceful per-dataset).
|
||||
Первую пойманную ошибку любого источника — ПОСЛЕ попытки обоих (surfaces в
|
||||
Celery/GlitchTip). Частичный сбой отдельного датасета/индикатора внутри источника
|
||||
НЕ роняет источник (fetch_all_* graceful per-item).
|
||||
"""
|
||||
db = SessionLocal()
|
||||
first_error: Exception | None = None
|
||||
rosstat_fetched = rosstat_upserted = 0
|
||||
emiss_fetched = emiss_upserted = 0
|
||||
try:
|
||||
rows = fetch_all_datasets()
|
||||
fetched = len(rows)
|
||||
logger.info("rosstat_macro_sync: fetched %d rosstat rows", fetched)
|
||||
# ── Источник 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)
|
||||
|
||||
upserted = _upsert_rows(db, rows)
|
||||
logger.info("rosstat_macro_sync: upserted %d/%d rosstat rows", upserted, fetched)
|
||||
# ── Источник 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)
|
||||
|
||||
return {"fetched": fetched, "upserted": upserted}
|
||||
except Exception as e:
|
||||
logger.exception("rosstat_macro_sync failed: %s", e)
|
||||
raise
|
||||
if first_error is not None:
|
||||
raise first_error
|
||||
|
||||
return {
|
||||
"fetched": rosstat_fetched + emiss_fetched,
|
||||
"upserted": rosstat_upserted + emiss_upserted,
|
||||
"rosstat": {"fetched": rosstat_fetched, "upserted": rosstat_upserted},
|
||||
"emiss": {"fetched": emiss_fetched, "upserted": emiss_upserted},
|
||||
}
|
||||
finally:
|
||||
db.close()
|
||||
|
|
|
|||
16
backend/tests/fixtures/emiss_empty.xml
vendored
Normal file
16
backend/tests/fixtures/emiss_empty.xml
vendored
Normal file
|
|
@ -0,0 +1,16 @@
|
|||
<?xml version="1.0" encoding="utf-8"?>
|
||||
<GenericData
|
||||
xmlns="http://www.SDMX.org/resources/SDMXML/schemas/v1_0/message"
|
||||
xmlns:generic="http://www.SDMX.org/resources/SDMXML/schemas/v1_0/generic"
|
||||
xmlns:structure="http://www.SDMX.org/resources/SDMXML/schemas/v1_0/structure">
|
||||
<!--
|
||||
Валидный SDMX-конверт БЕЗ данных (пустой <DataSet>). Имитирует ответ fedstat,
|
||||
когда индикатор/выборка пуста. Парсер должен вернуть [] (не падать).
|
||||
-->
|
||||
<Header Id="singHead">
|
||||
<ID>0</ID>
|
||||
<Test>false</Test>
|
||||
</Header>
|
||||
<DataSet>
|
||||
</DataSet>
|
||||
</GenericData>
|
||||
176
backend/tests/fixtures/emiss_income_57039.xml
vendored
Normal file
176
backend/tests/fixtures/emiss_income_57039.xml
vendored
Normal file
|
|
@ -0,0 +1,176 @@
|
|||
<?xml version="1.0" encoding="utf-8"?>
|
||||
<GenericData
|
||||
xmlns="http://www.SDMX.org/resources/SDMXML/schemas/v1_0/message"
|
||||
xmlns:common="http://www.SDMX.org/resources/SDMXML/schemas/v1_0/common"
|
||||
xmlns:compact="http://www.SDMX.org/resources/SDMXML/schemas/v1_0/compact"
|
||||
xmlns:cross="http://www.SDMX.org/resources/SDMXML/schemas/v1_0/cross"
|
||||
xmlns:generic="http://www.SDMX.org/resources/SDMXML/schemas/v1_0/generic"
|
||||
xmlns:query="http://www.SDMX.org/resources/SDMXML/schemas/v1_0/query"
|
||||
xmlns:structure="http://www.SDMX.org/resources/SDMXML/schemas/v1_0/structure"
|
||||
xmlns:utility="http://www.SDMX.org/resources/SDMXML/schemas/v1_0/utility"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://www.SDMX.org/resources/SDMXML/schemas/v1_0/message SDMXMessage.xsd">
|
||||
<!--
|
||||
Реальная (обрезанная) выборка SDMX-ML v1.0 GenericData c fedstat.ru:
|
||||
POST https://www.fedstat.ru/indicator/data.do body: id=57039&format=sdmx
|
||||
Индикатор 57039 = «Среднедушевые денежные доходы населения» (рубль/мес,
|
||||
квартальный), размерность s_OKATO. Свердловская обл. = ОКАТО 65000000000.
|
||||
Снято с PROD 2026-06-07 (dev WAF-403, fetch только с прода). Значения РЕАЛЬНЫЕ.
|
||||
Оставлены: субъект 65 (Свердл, 4 квартала 2024 + I кв.2025), субъект 79 (Адыгея —
|
||||
для проверки region-фильтра) и федокруг 030 (агрегат-нерегион — тоже исключается).
|
||||
-->
|
||||
<Header Id="singHead">
|
||||
<ID>-1573386947</ID>
|
||||
<Test>false</Test>
|
||||
<Truncated>false</Truncated>
|
||||
<Prepared>2026-06-07T16:14:12</Prepared>
|
||||
<Extracted>2026-06-07T16:14:12</Extracted>
|
||||
<Sender id="UIS">
|
||||
<Name xml:lang="en">Official statistics</Name>
|
||||
<Name xml:lang="ru">Государственная статистика</Name>
|
||||
</Sender>
|
||||
<DataSetAgency>1</DataSetAgency>
|
||||
<DataSetID>9460067</DataSetID>
|
||||
</Header>
|
||||
<CodeLists>
|
||||
<structure:CodeList id="s_OKATO">
|
||||
<structure:Name xml:lang="ru">Классификатор объектов административно-территориального деления (ОКАТО)</structure:Name>
|
||||
<structure:Code value="643">
|
||||
<structure:Description xml:lang="ru">Российская Федерация</structure:Description>
|
||||
</structure:Code>
|
||||
<structure:Code value="030">
|
||||
<structure:Description xml:lang="ru">Центральный федеральный округ</structure:Description>
|
||||
</structure:Code>
|
||||
<structure:Code value="79000000000">
|
||||
<structure:Description xml:lang="ru">Республика Адыгея (Адыгея)</structure:Description>
|
||||
</structure:Code>
|
||||
<structure:Code value="65000000000">
|
||||
<structure:Description xml:lang="ru">Свердловская область</structure:Description>
|
||||
</structure:Code>
|
||||
</structure:CodeList>
|
||||
</CodeLists>
|
||||
<Description>
|
||||
<Indicator id="57039" name="Среднедушевые денежные доходы населения">
|
||||
<Units>
|
||||
<Unit value="рубль"/>
|
||||
</Units>
|
||||
<Periodicities>
|
||||
<Periodicity value="Квартальная - не охарактеризована" releases="на 19 - 23-й рабочий день после квартала, следующего за отчетным" next-release="05.06.2027"/>
|
||||
</Periodicities>
|
||||
<DataRange start="1995" end="2026"/>
|
||||
<LastUpdate value="2026-06-05T00:05:23"/>
|
||||
<Dimensions>
|
||||
<Dimension value="s_OKATO" agencyID="1">
|
||||
<Name xml:lang="ru">Классификатор объектов административно-территориального деления (ОКАТО)</Name>
|
||||
</Dimension>
|
||||
</Dimensions>
|
||||
<Organization value="Федеральная служба государственной статистики"/>
|
||||
</Indicator>
|
||||
</Description>
|
||||
<DataSet>
|
||||
<generic:Series>
|
||||
<generic:SeriesKey>
|
||||
<generic:Value concept="s_OKATO" value="643"/>
|
||||
</generic:SeriesKey>
|
||||
<generic:Attributes>
|
||||
<generic:Value concept="EI" value="рубль"/>
|
||||
<generic:Value concept="PERIOD" value="I квартал"/>
|
||||
</generic:Attributes>
|
||||
<generic:Obs>
|
||||
<generic:Time>2024</generic:Time>
|
||||
<generic:ObsValue value="53221"/>
|
||||
</generic:Obs>
|
||||
</generic:Series>
|
||||
<generic:Series>
|
||||
<generic:SeriesKey>
|
||||
<generic:Value concept="s_OKATO" value="030"/>
|
||||
</generic:SeriesKey>
|
||||
<generic:Attributes>
|
||||
<generic:Value concept="EI" value="рубль"/>
|
||||
<generic:Value concept="PERIOD" value="I квартал"/>
|
||||
</generic:Attributes>
|
||||
<generic:Obs>
|
||||
<generic:Time>2024</generic:Time>
|
||||
<generic:ObsValue value="68110"/>
|
||||
</generic:Obs>
|
||||
</generic:Series>
|
||||
<generic:Series>
|
||||
<generic:SeriesKey>
|
||||
<generic:Value concept="s_OKATO" value="79000000000"/>
|
||||
</generic:SeriesKey>
|
||||
<generic:Attributes>
|
||||
<generic:Value concept="EI" value="рубль"/>
|
||||
<generic:Value concept="PERIOD" value="I квартал"/>
|
||||
</generic:Attributes>
|
||||
<generic:Obs>
|
||||
<generic:Time>2024</generic:Time>
|
||||
<generic:ObsValue value="40190"/>
|
||||
</generic:Obs>
|
||||
</generic:Series>
|
||||
<generic:Series>
|
||||
<generic:SeriesKey>
|
||||
<generic:Value concept="s_OKATO" value="65000000000"/>
|
||||
</generic:SeriesKey>
|
||||
<generic:Attributes>
|
||||
<generic:Value concept="EI" value="рубль"/>
|
||||
<generic:Value concept="PERIOD" value="I квартал"/>
|
||||
</generic:Attributes>
|
||||
<generic:Obs>
|
||||
<generic:Time>2024</generic:Time>
|
||||
<generic:ObsValue value="54006"/>
|
||||
</generic:Obs>
|
||||
</generic:Series>
|
||||
<generic:Series>
|
||||
<generic:SeriesKey>
|
||||
<generic:Value concept="s_OKATO" value="65000000000"/>
|
||||
</generic:SeriesKey>
|
||||
<generic:Attributes>
|
||||
<generic:Value concept="EI" value="рубль"/>
|
||||
<generic:Value concept="PERIOD" value="II квартал"/>
|
||||
</generic:Attributes>
|
||||
<generic:Obs>
|
||||
<generic:Time>2024</generic:Time>
|
||||
<generic:ObsValue value="61136"/>
|
||||
</generic:Obs>
|
||||
</generic:Series>
|
||||
<generic:Series>
|
||||
<generic:SeriesKey>
|
||||
<generic:Value concept="s_OKATO" value="65000000000"/>
|
||||
</generic:SeriesKey>
|
||||
<generic:Attributes>
|
||||
<generic:Value concept="EI" value="рубль"/>
|
||||
<generic:Value concept="PERIOD" value="III квартал"/>
|
||||
</generic:Attributes>
|
||||
<generic:Obs>
|
||||
<generic:Time>2024</generic:Time>
|
||||
<generic:ObsValue value="66849"/>
|
||||
</generic:Obs>
|
||||
</generic:Series>
|
||||
<generic:Series>
|
||||
<generic:SeriesKey>
|
||||
<generic:Value concept="s_OKATO" value="65000000000"/>
|
||||
</generic:SeriesKey>
|
||||
<generic:Attributes>
|
||||
<generic:Value concept="EI" value="рубль"/>
|
||||
<generic:Value concept="PERIOD" value="IV квартал"/>
|
||||
</generic:Attributes>
|
||||
<generic:Obs>
|
||||
<generic:Time>2024</generic:Time>
|
||||
<generic:ObsValue value="76701"/>
|
||||
</generic:Obs>
|
||||
</generic:Series>
|
||||
<generic:Series>
|
||||
<generic:SeriesKey>
|
||||
<generic:Value concept="s_OKATO" value="65000000000"/>
|
||||
</generic:SeriesKey>
|
||||
<generic:Attributes>
|
||||
<generic:Value concept="EI" value="рубль"/>
|
||||
<generic:Value concept="PERIOD" value="I квартал"/>
|
||||
</generic:Attributes>
|
||||
<generic:Obs>
|
||||
<generic:Time>2025</generic:Time>
|
||||
<generic:ObsValue value="64361"/>
|
||||
</generic:Obs>
|
||||
</generic:Series>
|
||||
</DataSet>
|
||||
</GenericData>
|
||||
222
backend/tests/scrapers/test_emiss_sdmx.py
Normal file
222
backend/tests/scrapers/test_emiss_sdmx.py
Normal file
|
|
@ -0,0 +1,222 @@
|
|||
"""Тесты SDMX-парсера ЕМИСС/fedstat (offline, на фикстурах — без живой сети).
|
||||
|
||||
Парсер ``parse_emiss_sdmx`` — pure (SDMX-байты → list[EmissRow]); fetch с fedstat
|
||||
исполняется только на проде (за WAF с dev), поэтому здесь тестируем РАЗБОР на реальных
|
||||
(обрезанных) SDMX-выгрузках, снятых с прода 2026-06-07:
|
||||
• emiss_income_57039.xml — «Среднедушевые денежные доходы населения» (id=57039),
|
||||
размерность s_OKATO, квартальный (рубль/мес). Свердл = ОКАТО 65000000000.
|
||||
• emiss_empty.xml — валидный конверт с пустым <DataSet>.
|
||||
|
||||
Покрывают: извлечение конкретных строк (Свердл, реальные период+значение),
|
||||
region-фильтрацию (др. регионы/федокруга исключены), маппинг квартала→obs_date,
|
||||
дедуп, graceful на пустом/битом/без-DataSet SDMX → [].
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import date
|
||||
from decimal import Decimal
|
||||
from pathlib import Path
|
||||
|
||||
from app.services.scrapers.rosstat_emiss import (
|
||||
INCOME_PER_CAPITA_SPEC,
|
||||
OKATO_SVERDLOVSK_OBLAST,
|
||||
EmissIndicatorSpec,
|
||||
RegionTarget,
|
||||
_emiss_period_to_month,
|
||||
parse_emiss_sdmx,
|
||||
)
|
||||
|
||||
_FIXTURES = Path(__file__).resolve().parent.parent / "fixtures"
|
||||
|
||||
|
||||
def _load(name: str) -> bytes:
|
||||
return (_FIXTURES / name).read_bytes()
|
||||
|
||||
|
||||
# ── period mapping ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_period_quarter_to_start_month() -> None:
|
||||
"""Квартал → стартовый месяц квартала (регистр/пробелы игнор)."""
|
||||
assert _emiss_period_to_month("I квартал") == 1
|
||||
assert _emiss_period_to_month("II квартал") == 4
|
||||
assert _emiss_period_to_month("III квартал") == 7
|
||||
assert _emiss_period_to_month(" iv КВАРТАЛ ") == 10
|
||||
|
||||
|
||||
def test_period_month_name_to_num() -> None:
|
||||
"""Русское название месяца → номер месяца."""
|
||||
assert _emiss_period_to_month("январь") == 1
|
||||
assert _emiss_period_to_month("Декабрь") == 12
|
||||
|
||||
|
||||
def test_period_year_and_unknown() -> None:
|
||||
"""Годовой под-период → январь; незнакомый/пустой → None."""
|
||||
assert _emiss_period_to_month("год") == 1
|
||||
assert _emiss_period_to_month("значение показателя за год") == 1
|
||||
assert _emiss_period_to_month("3 неделя") is None
|
||||
assert _emiss_period_to_month("") is None
|
||||
|
||||
|
||||
# ── real-fixture extraction (income id=57039) ─────────────────────────────────────
|
||||
|
||||
|
||||
def test_income_extracts_sverdlovsk_only() -> None:
|
||||
"""Из реальной SDMX-выгрузки извлекается ТОЛЬКО Свердл (ОКАТО 65) — РФ/ЦФО/Адыгея нет."""
|
||||
rows = parse_emiss_sdmx(_load("emiss_income_57039.xml"), INCOME_PER_CAPITA_SPEC)
|
||||
assert rows, "ожидались строки по Свердл"
|
||||
assert {r.region for r in rows} == {"sverdl"}
|
||||
assert all(r.indicator_type == "income_per_capita" for r in rows)
|
||||
assert all(r.unit == "руб" for r in rows)
|
||||
assert all(r.frequency == "quarterly" for r in rows)
|
||||
|
||||
|
||||
def test_income_concrete_values_and_dates() -> None:
|
||||
"""Конкретные реальные значения Свердл + obs_date = 1-е число стартового месяца квартала.
|
||||
|
||||
Реальные данные fedstat (Свердл, 2024): I=54006, II=61136, III=66849, IV=76701 ₽/мес;
|
||||
I кв.2025 = 64361. Квартал → стартовый месяц (I→01, II→04, III→07, IV→10).
|
||||
"""
|
||||
rows = parse_emiss_sdmx(_load("emiss_income_57039.xml"), INCOME_PER_CAPITA_SPEC)
|
||||
by_date = {r.obs_date: r.value for r in rows}
|
||||
|
||||
assert by_date[date(2024, 1, 1)] == Decimal("54006")
|
||||
assert by_date[date(2024, 4, 1)] == Decimal("61136")
|
||||
assert by_date[date(2024, 7, 1)] == Decimal("66849")
|
||||
assert by_date[date(2024, 10, 1)] == Decimal("76701")
|
||||
assert by_date[date(2025, 1, 1)] == Decimal("64361")
|
||||
# ровно 5 свердловских наблюдений в фикстуре (4 квартала 2024 + I кв.2025)
|
||||
assert len(rows) == 5
|
||||
# отсортировано по дате ASC
|
||||
assert [r.obs_date for r in rows] == sorted(by_date)
|
||||
|
||||
|
||||
def test_income_other_regions_excluded() -> None:
|
||||
"""ОКАТО других субъектов/агрегатов (Адыгея 79, ЦФО 030, РФ 643) НЕ попадают.
|
||||
|
||||
Значение Адыгеи (40190) и ЦФО (68110) присутствуют в фикстуре, но НЕ должны
|
||||
появиться среди извлечённых (фильтр строго по целевому ОКАТО 65000000000).
|
||||
"""
|
||||
rows = parse_emiss_sdmx(_load("emiss_income_57039.xml"), INCOME_PER_CAPITA_SPEC)
|
||||
values = {r.value for r in rows}
|
||||
assert Decimal("40190") not in values # Адыгея
|
||||
assert Decimal("68110") not in values # ЦФО
|
||||
assert Decimal("53221") not in values # РФ
|
||||
|
||||
|
||||
def test_income_region_filter_targets_adygea() -> None:
|
||||
"""Подменив целевой ОКАТО на Адыгею — извлекается ИМЕННО Адыгея (доказывает фильтр)."""
|
||||
spec = EmissIndicatorSpec(
|
||||
indicator_id="57039",
|
||||
indicator_type="income_per_capita",
|
||||
unit="руб",
|
||||
frequency="quarterly",
|
||||
targets=(RegionTarget(oktmo="79000000000", region="adygea"),),
|
||||
)
|
||||
rows = parse_emiss_sdmx(_load("emiss_income_57039.xml"), spec)
|
||||
assert len(rows) == 1
|
||||
assert rows[0].region == "adygea"
|
||||
assert rows[0].value == Decimal("40190")
|
||||
assert rows[0].obs_date == date(2024, 1, 1)
|
||||
|
||||
|
||||
# ── graceful degradation ──────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_empty_dataset_returns_empty() -> None:
|
||||
"""Валидный SDMX с пустым <DataSet> → [] (не падает)."""
|
||||
assert parse_emiss_sdmx(_load("emiss_empty.xml"), INCOME_PER_CAPITA_SPEC) == []
|
||||
|
||||
|
||||
def test_malformed_xml_returns_empty() -> None:
|
||||
"""Битый XML → [] (+warning), не бросает."""
|
||||
assert parse_emiss_sdmx(b"<GenericData><DataSet><not closed", INCOME_PER_CAPITA_SPEC) == []
|
||||
assert parse_emiss_sdmx("не xml вовсе", INCOME_PER_CAPITA_SPEC) == []
|
||||
|
||||
|
||||
def test_no_dataset_element_returns_empty() -> None:
|
||||
"""SDMX-конверт без <DataSet> → [] (не падает)."""
|
||||
xml = (
|
||||
'<GenericData xmlns="http://www.SDMX.org/resources/SDMXML/schemas/v1_0/message">'
|
||||
"<Header/></GenericData>"
|
||||
)
|
||||
assert parse_emiss_sdmx(xml, INCOME_PER_CAPITA_SPEC) == []
|
||||
|
||||
|
||||
def test_unknown_period_series_skipped() -> None:
|
||||
"""Серия с незнакомым PERIOD (неделя) пропускается, валидная остаётся."""
|
||||
xml = (
|
||||
'<GenericData xmlns="http://www.SDMX.org/resources/SDMXML/schemas/v1_0/message"'
|
||||
' xmlns:generic="http://www.SDMX.org/resources/SDMXML/schemas/v1_0/generic">'
|
||||
"<DataSet>"
|
||||
"<generic:Series><generic:SeriesKey>"
|
||||
'<generic:Value concept="s_OKATO" value="65000000000"/></generic:SeriesKey>'
|
||||
'<generic:Attributes><generic:Value concept="PERIOD" value="3 неделя"/>'
|
||||
"</generic:Attributes>"
|
||||
'<generic:Obs><generic:Time>2024</generic:Time>'
|
||||
'<generic:ObsValue value="999"/></generic:Obs></generic:Series>'
|
||||
"<generic:Series><generic:SeriesKey>"
|
||||
'<generic:Value concept="s_OKATO" value="65000000000"/></generic:SeriesKey>'
|
||||
'<generic:Attributes><generic:Value concept="PERIOD" value="I квартал"/>'
|
||||
"</generic:Attributes>"
|
||||
'<generic:Obs><generic:Time>2024</generic:Time>'
|
||||
'<generic:ObsValue value="50000"/></generic:Obs></generic:Series>'
|
||||
"</DataSet></GenericData>"
|
||||
)
|
||||
rows = parse_emiss_sdmx(xml, INCOME_PER_CAPITA_SPEC)
|
||||
assert len(rows) == 1
|
||||
assert rows[0].value == Decimal("50000")
|
||||
assert rows[0].obs_date == date(2024, 1, 1)
|
||||
|
||||
|
||||
def test_comma_decimal_value_parsed() -> None:
|
||||
"""Значение с запятой-десятичным (ИПЦ-вариант '113,2') → Decimal('113.2')."""
|
||||
xml = (
|
||||
'<GenericData xmlns="http://www.SDMX.org/resources/SDMXML/schemas/v1_0/message"'
|
||||
' xmlns:generic="http://www.SDMX.org/resources/SDMXML/schemas/v1_0/generic">'
|
||||
"<DataSet><generic:Series><generic:SeriesKey>"
|
||||
'<generic:Value concept="s_OKATO" value="65000000000"/></generic:SeriesKey>'
|
||||
'<generic:Attributes><generic:Value concept="PERIOD" value="январь"/>'
|
||||
"</generic:Attributes>"
|
||||
'<generic:Obs><generic:Time>2024</generic:Time>'
|
||||
'<generic:ObsValue value="113,2"/></generic:Obs></generic:Series></DataSet>'
|
||||
"</GenericData>"
|
||||
)
|
||||
rows = parse_emiss_sdmx(xml, INCOME_PER_CAPITA_SPEC)
|
||||
assert len(rows) == 1
|
||||
assert rows[0].value == Decimal("113.2")
|
||||
assert rows[0].obs_date == date(2024, 1, 1)
|
||||
|
||||
|
||||
def test_dedup_last_wins() -> None:
|
||||
"""Дубль (region, obs_date) схлопывается — последний выигрывает (PK всё равно один)."""
|
||||
xml = (
|
||||
'<GenericData xmlns="http://www.SDMX.org/resources/SDMXML/schemas/v1_0/message"'
|
||||
' xmlns:generic="http://www.SDMX.org/resources/SDMXML/schemas/v1_0/generic">'
|
||||
"<DataSet>"
|
||||
"<generic:Series><generic:SeriesKey>"
|
||||
'<generic:Value concept="s_OKATO" value="65000000000"/></generic:SeriesKey>'
|
||||
'<generic:Attributes><generic:Value concept="PERIOD" value="I квартал"/>'
|
||||
"</generic:Attributes>"
|
||||
'<generic:Obs><generic:Time>2024</generic:Time>'
|
||||
'<generic:ObsValue value="100"/></generic:Obs></generic:Series>'
|
||||
"<generic:Series><generic:SeriesKey>"
|
||||
'<generic:Value concept="s_OKATO" value="65000000000"/></generic:SeriesKey>'
|
||||
'<generic:Attributes><generic:Value concept="PERIOD" value="I квартал"/>'
|
||||
"</generic:Attributes>"
|
||||
'<generic:Obs><generic:Time>2024</generic:Time>'
|
||||
'<generic:ObsValue value="200"/></generic:Obs></generic:Series>'
|
||||
"</DataSet></GenericData>"
|
||||
)
|
||||
rows = parse_emiss_sdmx(xml, INCOME_PER_CAPITA_SPEC)
|
||||
assert len(rows) == 1
|
||||
assert rows[0].value == Decimal("200")
|
||||
|
||||
|
||||
def test_spec_targets_sverdlovsk_okato() -> None:
|
||||
"""Активная спека доходов целит ИМЕННО ОКАТО Свердл (контракт §7.11)."""
|
||||
assert OKATO_SVERDLOVSK_OBLAST == "65000000000"
|
||||
assert any(t.oktmo == OKATO_SVERDLOVSK_OBLAST for t in INCOME_PER_CAPITA_SPEC.targets)
|
||||
assert INCOME_PER_CAPITA_SPEC.indicator_id == "57039"
|
||||
assert INCOME_PER_CAPITA_SPEC.frequency == "quarterly"
|
||||
|
|
@ -1,7 +1,8 @@
|
|||
"""Тесты Celery-таски rosstat_macro_sync (mock session + mock fetch, без живой сети).
|
||||
|
||||
Покрывают: контракт upsert-SQL (source='rosstat', ON CONFLICT, CAST not ::),
|
||||
счётчики fetched/upserted, проброс ошибки fetch'а, регистрацию в beat (ежемесячно).
|
||||
Покрывают: контракт upsert-SQL обоих источников (source='rosstat'/'emiss', ON CONFLICT,
|
||||
CAST not ::), счётчики fetched/upserted (open-data + ЕМИСС), per-source guard (сбой
|
||||
одного источника НЕ мешает другому, но пробрасывается), регистрацию в beat (ежемесячно).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
|
@ -67,12 +68,16 @@ def test_task_upserts_each_row_with_correct_params() -> None:
|
|||
with (
|
||||
patch.object(task_mod, "SessionLocal", return_value=db),
|
||||
patch.object(task_mod, "fetch_all_datasets", return_value=fake_rows) as mock_fetch,
|
||||
patch.object(task_mod, "fetch_all_emiss", return_value=[]) as mock_emiss,
|
||||
):
|
||||
result = task_mod.rosstat_macro_sync.run()
|
||||
|
||||
mock_fetch.assert_called_once()
|
||||
mock_emiss.assert_called_once()
|
||||
assert result["fetched"] == 2
|
||||
assert result["upserted"] == 2
|
||||
assert result["rosstat"] == {"fetched": 2, "upserted": 2}
|
||||
assert result["emiss"] == {"fetched": 0, "upserted": 0}
|
||||
assert captured == [
|
||||
{
|
||||
"itype": "population_total",
|
||||
|
|
@ -96,7 +101,10 @@ def test_task_upserts_each_row_with_correct_params() -> None:
|
|||
|
||||
|
||||
def test_task_reraises_fetch_failure() -> None:
|
||||
"""Ошибка fetch'а пробрасывается (surfaces в Celery/GlitchTip), DB закрывается."""
|
||||
"""Ошибка open-data fetch'а пробрасывается (surfaces в Celery/GlitchTip), DB закрывается.
|
||||
|
||||
Even with the per-source guard, ошибка ПОСЛЕ попытки обоих источников re-raise'ится.
|
||||
"""
|
||||
from app.workers.tasks import rosstat_macro_sync as task_mod
|
||||
|
||||
db, _ = _make_mock_db()
|
||||
|
|
@ -104,6 +112,7 @@ def test_task_reraises_fetch_failure() -> None:
|
|||
with (
|
||||
patch.object(task_mod, "SessionLocal", return_value=db),
|
||||
patch.object(task_mod, "fetch_all_datasets", side_effect=RuntimeError("rosstat down")),
|
||||
patch.object(task_mod, "fetch_all_emiss", return_value=[]),
|
||||
):
|
||||
try:
|
||||
task_mod.rosstat_macro_sync.run()
|
||||
|
|
@ -114,6 +123,81 @@ def test_task_reraises_fetch_failure() -> None:
|
|||
db.close.assert_called_once()
|
||||
|
||||
|
||||
def test_task_emiss_failure_does_not_block_opendata() -> None:
|
||||
"""Per-source guard: сбой ЕМИСС НЕ мешает open-data апсертнуться, но ошибка пробрасывается."""
|
||||
from app.services.scrapers.rosstat_emiss import MacroRow
|
||||
from app.workers.tasks import rosstat_macro_sync as task_mod
|
||||
|
||||
db, captured = _make_mock_db()
|
||||
rosstat_rows = [
|
||||
MacroRow("population_total", "sverdl", date(2014, 1, 1), Decimal("4320677"), "чел", "c"),
|
||||
]
|
||||
|
||||
with (
|
||||
patch.object(task_mod, "SessionLocal", return_value=db),
|
||||
patch.object(task_mod, "fetch_all_datasets", return_value=rosstat_rows),
|
||||
patch.object(task_mod, "fetch_all_emiss", side_effect=RuntimeError("emiss down")),
|
||||
):
|
||||
try:
|
||||
task_mod.rosstat_macro_sync.run()
|
||||
raise AssertionError("expected RuntimeError to propagate")
|
||||
except RuntimeError as e:
|
||||
assert "emiss down" in str(e)
|
||||
|
||||
# open-data всё равно апсертнулся ДО проброса emiss-ошибки
|
||||
assert any(p.get("itype") == "population_total" for p in captured)
|
||||
db.close.assert_called_once()
|
||||
|
||||
|
||||
def test_task_upserts_emiss_rows_with_correct_params() -> None:
|
||||
"""Каждая EmissRow → execute UPSERT_EMISS_SQL с itype/region/d/v/freq/unit/comment."""
|
||||
from app.services.scrapers.rosstat_emiss import EmissRow
|
||||
from app.workers.tasks import rosstat_macro_sync as task_mod
|
||||
|
||||
db, captured = _make_mock_db()
|
||||
emiss_rows = [
|
||||
EmissRow(
|
||||
"income_per_capita", "sverdl", date(2024, 1, 1),
|
||||
Decimal("54006"), "руб", "quarterly", "c",
|
||||
),
|
||||
]
|
||||
|
||||
with (
|
||||
patch.object(task_mod, "SessionLocal", return_value=db),
|
||||
patch.object(task_mod, "fetch_all_datasets", return_value=[]),
|
||||
patch.object(task_mod, "fetch_all_emiss", return_value=emiss_rows),
|
||||
):
|
||||
result = task_mod.rosstat_macro_sync.run()
|
||||
|
||||
assert result["emiss"] == {"fetched": 1, "upserted": 1}
|
||||
assert captured == [
|
||||
{
|
||||
"itype": "income_per_capita",
|
||||
"region": "sverdl",
|
||||
"d": date(2024, 1, 1),
|
||||
"v": Decimal("54006"),
|
||||
"freq": "quarterly",
|
||||
"unit": "руб",
|
||||
"comment": "c",
|
||||
},
|
||||
]
|
||||
|
||||
|
||||
def test_upsert_emiss_sql_contract() -> None:
|
||||
"""UPSERT_EMISS_SQL: source='emiss', frequency параметризован, ON CONFLICT, CAST not ::."""
|
||||
from app.workers.tasks.rosstat_macro_sync import UPSERT_EMISS_SQL
|
||||
|
||||
sql = str(UPSERT_EMISS_SQL)
|
||||
upper = sql.upper()
|
||||
assert "INTO MACRO_INDICATOR" in upper
|
||||
assert "'emiss'" in sql
|
||||
assert "CAST(:FREQ AS TEXT)" in upper
|
||||
assert "ON CONFLICT (INDICATOR_TYPE, REGION, OBS_DATE) DO UPDATE" in upper
|
||||
assert "CAST(:ITYPE AS TEXT)" in upper
|
||||
assert "CAST(:V AS NUMERIC)" in upper
|
||||
assert "::" not in sql
|
||||
|
||||
|
||||
def test_task_registered_in_beat_monthly() -> None:
|
||||
"""Задача зарегистрирована в beat как ежемесячная (1-е число)."""
|
||||
from app.workers.beat_schedule import build_beat_schedule
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue