diff --git a/backend/app/services/scrapers/rosstat_emiss.py b/backend/app/services/scrapers/rosstat_emiss.py
index 7be04b96..239d70e7 100644
--- a/backend/app/services/scrapers/rosstat_emiss.py
+++ b/backend/app/services/scrapers/rosstat_emiss.py
@@ -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-байтный ``
Forbidden `` БЕЗ ``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=&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=_``,
+ перечисляющим выбранные коды ПО ВСЕМ размерностям, где ``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=&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=&format=sdmx
+# → SDMX-ML v1.0 GenericData (см. tests/fixtures/emiss_income_57039.xml). Структура:
+#
+# …коды регионов…
+# (fedstat-расширение)
+#
+#
+# …>
+#
+# (единица измерения)
+# (под-период года)
+#
+# 2024
+#
+# …
+# Один = один (регион, PERIOD) ряд; год наблюдения — в (напр. 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 словари из .
+
+ 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. Для каждого : регион = SeriesKey concept='s_OKATO'. Если ОКАТО
+ не среди spec.targets — ряд пропускается. PERIOD (Attributes) → стартовый месяц
+ под-периода; год — из ; obs_date = date(year, month, 1). Значение —
+ из через _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: нет в 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=&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 ────────────────────────────────────────────────────────────
diff --git a/backend/app/services/site_finder/macro.py b/backend/app/services/site_finder/macro.py
index 3ab104bf..c4443882 100644
--- a/backend/app/services/site_finder/macro.py
+++ b/backend/app/services/site_finder/macro.py
@@ -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).
diff --git a/backend/app/workers/tasks/rosstat_macro_sync.py b/backend/app/workers/tasks/rosstat_macro_sync.py
index 4d39f715..81dcc0ab 100644
--- a/backend/app/workers/tasks/rosstat_macro_sync.py
+++ b/backend/app/workers/tasks/rosstat_macro_sync.py
@@ -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()
diff --git a/backend/tests/fixtures/emiss_empty.xml b/backend/tests/fixtures/emiss_empty.xml
new file mode 100644
index 00000000..7346878e
--- /dev/null
+++ b/backend/tests/fixtures/emiss_empty.xml
@@ -0,0 +1,16 @@
+
+
+
+
+
+
+
diff --git a/backend/tests/fixtures/emiss_income_57039.xml b/backend/tests/fixtures/emiss_income_57039.xml
new file mode 100644
index 00000000..bb642d3e
--- /dev/null
+++ b/backend/tests/fixtures/emiss_income_57039.xml
@@ -0,0 +1,176 @@
+
+
+
+
+ -1573386947
+ false
+ false
+ 2026-06-07T16:14:12
+ 2026-06-07T16:14:12
+
+ Official statistics
+ Государственная статистика
+
+ 1
+ 9460067
+
+
+
+ Классификатор объектов административно-территориального деления (ОКАТО)
+
+ Российская Федерация
+
+
+ Центральный федеральный округ
+
+
+ Республика Адыгея (Адыгея)
+
+
+ Свердловская область
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ Классификатор объектов административно-территориального деления (ОКАТО)
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ 2024
+
+
+
+
+
+
+
+
+
+
+
+
+ 2024
+
+
+
+
+
+
+
+
+
+
+
+
+ 2024
+
+
+
+
+
+
+
+
+
+
+
+
+ 2024
+
+
+
+
+
+
+
+
+
+
+
+
+ 2024
+
+
+
+
+
+
+
+
+
+
+
+
+ 2024
+
+
+
+
+
+
+
+
+
+
+
+
+ 2024
+
+
+
+
+
+
+
+
+
+
+
+
+ 2025
+
+
+
+
+
diff --git a/backend/tests/scrapers/test_emiss_sdmx.py b/backend/tests/scrapers/test_emiss_sdmx.py
new file mode 100644
index 00000000..cc8665ae
--- /dev/null
+++ b/backend/tests/scrapers/test_emiss_sdmx.py
@@ -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 — валидный конверт с пустым .
+
+Покрывают: извлечение конкретных строк (Свердл, реальные период+значение),
+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 с пустым → [] (не падает)."""
+ 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" None:
+ """SDMX-конверт без → [] (не падает)."""
+ xml = (
+ ''
+ " "
+ )
+ assert parse_emiss_sdmx(xml, INCOME_PER_CAPITA_SPEC) == []
+
+
+def test_unknown_period_series_skipped() -> None:
+ """Серия с незнакомым PERIOD (неделя) пропускается, валидная остаётся."""
+ xml = (
+ ''
+ ""
+ ""
+ ' '
+ ' '
+ " "
+ '2024 '
+ ' '
+ ""
+ ' '
+ ' '
+ " "
+ '2024 '
+ ' '
+ " "
+ )
+ 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 = (
+ ''
+ ""
+ ' '
+ ' '
+ " "
+ '2024 '
+ ' '
+ " "
+ )
+ 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 = (
+ ''
+ ""
+ ""
+ ' '
+ ' '
+ " "
+ '2024 '
+ ' '
+ ""
+ ' '
+ ' '
+ " "
+ '2024 '
+ ' '
+ " "
+ )
+ 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"
diff --git a/backend/tests/workers/test_rosstat_macro_sync.py b/backend/tests/workers/test_rosstat_macro_sync.py
index cf7d9856..3d10262e 100644
--- a/backend/tests/workers/test_rosstat_macro_sync.py
+++ b/backend/tests/workers/test_rosstat_macro_sync.py
@@ -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