diff --git a/backend/app/services/scrapers/rosstat_emiss.py b/backend/app/services/scrapers/rosstat_emiss.py new file mode 100644 index 00000000..7be04b96 --- /dev/null +++ b/backend/app/services/scrapers/rosstat_emiss.py @@ -0,0 +1,501 @@ +"""Росстат / ЕМИСС macro-indicator scraper → ``macro_indicator`` (#946, EPIC2). + +Заполняет региональные (Свердловская обл., ОКТМО 65 / ЕКБ) макро-ряды Росстата в +унифицированной таблице ``macro_indicator`` (миграция ``data/sql/123_macro_indicator.sql``). +Детерминированно, без LLM. Зеркалит форму ``cbr_macro.py``: pure-парсеры (offline- +тестируемые на фикстурах) + тонкая httpx-обвязка (timeout, UA, retry); upsert делает +Celery-таска ``rosstat_macro_sync`` через SAVEPOINT-per-row. + +──────────────────────────────────────────────────────────────────────────────── +ИСТОЧНИК ДАННЫХ — почему open-data, а НЕ ЕМИСС-SDMX (разведано live 2026-06-07) +──────────────────────────────────────────────────────────────────────────────── +ЕМИСС / fedstat.ru (SDMX + ``/indicator/data.do``) — за WAF: ЛЮБОЙ путь/метод/набор +заголовков отдаёт статичный 80-байтный ``
/meta.csv``;
+ • метаданные ``/opendata//meta.csv`` — CP1251 CSV (property,value); строка
+ ``data--structure-.csv`` →
+ URL фактического файла данных;
+ • данные ``/opendata//data--…csv`` — табличный CSV; регионы ключуются
+ ОКТМО (Свердловская обл. = префикс
+ ``65``; ЕКБ-горокруг = ``6570100000``).
+
+Что РЕАЛЬНО приземляется отсюда для региона 65 (см. ниже DATASETS):
+ • ``population_total`` — Численность постоянного населения (датасет ``…-population``,
+ муниципальный срез; берём субъект-агрегат ОКТМО ``6500000000`` + г. Екатеринбург
+ ``657010000011000``). Это § 7.11 «демография (численность)».
+
+Что НЕ приземляется (документировано, частичная поставка допустима ТЗ):
+ • среднедушевые доходы, ИПЦ, индекс цен на СМР — EMISS-only → WAF (см. выше). В
+ open-data их нет (проверено по реестру: 0 совпадений). Закрывается, когда появится
+ рабочий доступ к ЕМИСС (прокси / ручная выгрузка) — тогда добавится новый
+ ``DatasetSpec`` с тем же upsert-контрактом.
+
+Standalone smoke (ручная проверка против живого rosstat.gov.ru):
+ python -m app.services.scrapers.rosstat_emiss
+
+Mirror conventions: ``httpx.Client`` + явный User-Agent (как cbr_macro / ekburg_permits),
+``logger`` вместо print, типизированные сигнатуры, без хардкода секретов.
+"""
+
+from __future__ import annotations
+
+import csv
+import io
+import logging
+from dataclasses import dataclass
+from datetime import date
+from decimal import Decimal, InvalidOperation
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+# ── constants ────────────────────────────────────────────────────────────────
+
+ROSSTAT_OPENDATA_BASE = "https://rosstat.gov.ru/opendata"
+
+# Браузероподобный UA: rosstat.gov.ru отдаёт open-data без WAF, но «голый» httpx-UA
+# иногда ловит 403 на edge-кэше — даём обычный desktop-Chrome UA (как nspd_client).
+USER_AGENT = (
+ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
+ "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
+)
+DEFAULT_TIMEOUT_S = 30.0
+DEFAULT_RETRIES = 3
+
+# Кодировки CSV Росстата в порядке проб: data-файлы бывают и utf-8(-sig), и CP1251;
+# meta.csv почти всегда CP1251. Декодируем «первой подошедшей» (см. _decode_csv).
+_CSV_ENCODINGS: tuple[str, ...] = ("utf-8-sig", "cp1251", "utf-8")
+
+# rosstat.gov.ru отдаёт TLS-цепочку российского гос-УЦ (Russian Trusted / Минцифры),
+# корневой CA которой НЕ в trust-store наших контейнеров — verify=True падает с
+# CERTIFICATE_VERIFY_FAILED (подтверждено live 2026-06-07). Отключаем верификацию:
+# данные ПУБЛИЧНЫЕ open-data, секреты не передаются → MITM-риск незначим. Прецедент:
+# ekb_geoportal_client._VERIFY_TLS / nspd_lite._SSL_CTX для тех же росс-гос-эндпоинтов.
+_VERIFY_TLS = False
+
+# ОКТМО-коды интересующих территорий (Свердловская обл. / ЕКБ). ТЗ §7.11 «region 65».
+# • субъект «Свердловская область» — ОКТМО-агрегат '6500000000' (region='sverdl');
+# • г. Екатеринбург (горокруг) — ОКТМО '657010000011000' (region='ekb').
+# Ключуем по ТОЧНОМУ ОКТМО (а не по имени) — имена в датасетах варьируются
+# («Муниципальные образования Свердловской области» и т.п.), ОКТМО стабилен.
+OKTMO_SVERDLOVSK_OBLAST = "6500000000"
+OKTMO_EKATERINBURG = "657010000011000"
+
+
+class RosstatScraperError(RuntimeError):
+ """Сетевая / протокольная ошибка обращения к rosstat.gov.ru/opendata."""
+
+
+# ── dataset specs (что и откуда тянем) ──────────────────────────────────────────
+
+
+@dataclass(frozen=True)
+class RegionTarget:
+ """Целевая территория ряда: ОКТМО-ключ → (region-метка macro_indicator)."""
+
+ oktmo: str
+ region: str # метка region в macro_indicator: 'sverdl' | 'ekb'
+
+
+@dataclass(frozen=True)
+class DatasetSpec:
+ """Описание одного приземляемого open-data датасета Росстата.
+
+ Атрибуты:
+ code: код датасета в реестре (часть URL ``/opendata//…``).
+ indicator_type: тип ряда в macro_indicator (напр. 'population_total').
+ unit: единица измерения (справочно, в PK не участвует).
+ value_col_header: заголовок колонки со значением в data-CSV
+ (для population — 'total'; матчим без учёта регистра/пробелов).
+ oktmo_col_header: заголовок колонки с ОКТМО-кодом (для population — 'kode').
+ year_col_header: заголовок колонки с годом наблюдения, если он есть в данных
+ (для population — 'year'); None → год берётся из meta ``modified``.
+ targets: список территорий (ОКТМО → region-метка), которые извлекаем.
+ comment: пояснение для colонки comment macro_indicator.
+ """
+
+ code: str
+ indicator_type: str
+ unit: str
+ value_col_header: str
+ oktmo_col_header: str
+ year_col_header: str | None
+ targets: tuple[RegionTarget, ...]
+ comment: str = ""
+
+
+# Демография (§7.11 «численность населения»). Датасет муниципального уровня; берём
+# субъект-агрегат Свердловской обл. + г. Екатеринбург. year — в самих данных ('year').
+POPULATION_SPEC = DatasetSpec(
+ code="7708234640-population",
+ indicator_type="population_total",
+ unit="чел",
+ value_col_header="total",
+ oktmo_col_header="kode",
+ year_col_header="year",
+ targets=(
+ RegionTarget(oktmo=OKTMO_SVERDLOVSK_OBLAST, region="sverdl"),
+ RegionTarget(oktmo=OKTMO_EKATERINBURG, region="ekb"),
+ ),
+ comment="Росстат open-data 'Численность постоянного населения' (#946 §7.11)",
+)
+
+# Активные спеки скрейпера. Расширяется добавлением DatasetSpec (напр. когда появится
+# доступ к EMISS-рядам ИПЦ/доходов — у них будет свой Spec с тем же upsert-контрактом).
+DATASETS: tuple[DatasetSpec, ...] = (POPULATION_SPEC,)
+
+
+# ── parsed-row contract ─────────────────────────────────────────────────────────
+
+
+@dataclass(frozen=True)
+class MacroRow:
+ """Одна готовая к upsert строка macro_indicator (источник 'rosstat')."""
+
+ indicator_type: str
+ region: str
+ obs_date: date
+ value: Decimal
+ unit: str
+ comment: str
+
+
+# ── pure helpers (unit-testable offline) ────────────────────────────────────────
+
+
+def _decode_csv(raw: bytes) -> str:
+ """Декодировать байты CSV Росстата «первой подошедшей» кодировкой (_CSV_ENCODINGS).
+
+ Росстат смешивает utf-8(-sig) и CP1251 между файлами. Возвращает str; если ни
+ одна кодировка не подошла строго — падаем на utf-8 с errors='replace' (лучше
+ частично-битый текст, чем исключение на одном файле).
+ """
+ for enc in _CSV_ENCODINGS:
+ try:
+ return raw.decode(enc)
+ except UnicodeDecodeError:
+ continue
+ logger.warning("Rosstat CSV: ни одна кодировка не подошла строго — utf-8/replace")
+ return raw.decode("utf-8", errors="replace")
+
+
+def _sniff_rows(text: str) -> list[list[str]]:
+ """Разобрать CSV-текст в список строк, авто-детект разделителя (',' либо ';').
+
+ Берём тот разделитель, что даёт > 1 колонки в первой непустой строке. Росстат
+ использует и запятую (population), и точку-с-запятой (часть рядов) как delimiter.
+ """
+ for delim in (",", ";"):
+ rows = list(csv.reader(io.StringIO(text), delimiter=delim))
+ first = next((r for r in rows if any(c.strip() for c in r)), [])
+ if len(first) > 1:
+ return rows
+ # фоллбэк — как запятая (даже если одна колонка)
+ return list(csv.reader(io.StringIO(text), delimiter=","))
+
+
+def _norm(s: str) -> str:
+ """Нормализовать заголовок колонки для матча: lower + strip пробелов/BOM."""
+ return (s or "").strip().lstrip("").lower()
+
+
+def parse_meta_data_url(meta_text: str) -> str | None:
+ """Извлечь URL файла данных из meta.csv (open-data standard 4.0).
+
+ meta.csv — две колонки (property,value). Строка данных имеет property вида
+ ``data--structure-.csv`` и value = абсолютный URL CSV-данных. Если
+ таких строк несколько — возвращаем ПОСЛЕДНЮЮ (самый свежий снапшот). None —
+ если строки данных нет.
+ """
+ rows = _sniff_rows(meta_text)
+ data_url: str | None = None
+ for row in rows:
+ if len(row) < 2:
+ continue
+ prop = row[0].strip()
+ if prop.startswith("data-") and row[1].strip().lower().startswith("http"):
+ data_url = row[1].strip()
+ return data_url
+
+
+def parse_meta_modified_year(meta_text: str) -> int | None:
+ """Извлечь год из meta.csv ``modified`` (формат 'YYYYMMDD') — фоллбэк для obs_date.
+
+ Используется, когда в самих данных нет колонки года (year_col_header is None).
+ None — если поле отсутствует / не парсится.
+ """
+ rows = _sniff_rows(meta_text)
+ for row in rows:
+ if len(row) >= 2 and row[0].strip() == "modified":
+ digits = row[1].strip()[:4]
+ if digits.isdigit():
+ return int(digits)
+ return None
+
+
+def _parse_value(raw: str) -> Decimal | None:
+ """Распарсить числовое значение Росстата → Decimal.
+
+ Принимает запятую как десятичный разделитель (Росстат: '1,5'), вырезает
+ пробелы-разделители тысяч (включая узкий неразрывный / ). Пустая
+ строка / '-' / нечисло → None (строка пропускается caller'ом).
+ """
+ raw = (raw or "").strip()
+ if not raw or raw in {"-", "—", "…", "..."}:
+ return None
+ cleaned = (
+ raw.replace(" ", "").replace(" ", "").replace(" ", "").replace(",", ".")
+ )
+ try:
+ return Decimal(cleaned)
+ except (InvalidOperation, ValueError):
+ logger.warning("Rosstat: не удалось распарсить значение %r", raw)
+ return None
+
+
+def _obs_date_for_year(year: int) -> date:
+ """Годовой ряд → obs_date нормализуем к 1 января года (как monthly-ряды к 1-му числу)."""
+ return date(year, 1, 1)
+
+
+def parse_dataset_rows(
+ data_text: str,
+ spec: DatasetSpec,
+ *,
+ fallback_year: int | None,
+) -> list[MacroRow]:
+ """Pure-парсер data-CSV датасета → список MacroRow для целевых ОКТМО.
+
+ Алгоритм:
+ 1. Находит индекс колонок ОКТМО / значения / года по их заголовкам
+ (spec.*_col_header, матч через _norm). Нет нужной колонки → [] (+warning).
+ 2. Для каждой строки: если ОКТМО ∈ spec.targets — парсит значение и год
+ (из year-колонки либо fallback_year), эмитит MacroRow с region-меткой цели.
+ 3. Дубли по (indicator_type, region, obs_date) схлопываются (последний выигрывает) —
+ защита от повторов строк в исходнике; PK macro_indicator всё равно один.
+
+ Никогда не бросает на данных: битые/пустые ячейки → строка пропускается.
+ """
+ rows = _sniff_rows(data_text)
+ if not rows:
+ return []
+
+ # Заголовок — первая строка, где встречается искомый заголовок ОКТМО-колонки.
+ want_oktmo = _norm(spec.oktmo_col_header)
+ want_value = _norm(spec.value_col_header)
+ want_year = _norm(spec.year_col_header) if spec.year_col_header else None
+
+ header_idx: int | None = None
+ for i, row in enumerate(rows[:5]): # заголовок Росстата в первых нескольких строках
+ norm = [_norm(c) for c in row]
+ if want_oktmo in norm and want_value in norm:
+ header_idx = i
+ break
+ if header_idx is None:
+ logger.warning(
+ "Rosstat %s: не найдены колонки %r/%r в заголовке",
+ spec.code,
+ spec.oktmo_col_header,
+ spec.value_col_header,
+ )
+ return []
+
+ header = [_norm(c) for c in rows[header_idx]]
+ col_oktmo = header.index(want_oktmo)
+ col_value = header.index(want_value)
+ col_year = header.index(want_year) if want_year and want_year in header else None
+
+ target_by_oktmo = {t.oktmo: t for t in spec.targets}
+ by_key: dict[tuple[str, date], MacroRow] = {}
+
+ for row in rows[header_idx + 1 :]:
+ if len(row) <= max(col_oktmo, col_value):
+ continue
+ oktmo = (row[col_oktmo] or "").strip()
+ target = target_by_oktmo.get(oktmo)
+ if target is None:
+ continue
+ value = _parse_value(row[col_value])
+ if value is None:
+ continue
+ year: int | None = None
+ if col_year is not None and len(row) > col_year:
+ yr = (row[col_year] or "").strip()[:4]
+ if yr.isdigit():
+ year = int(yr)
+ if year is None:
+ year = fallback_year
+ if year is None:
+ logger.warning("Rosstat %s: нет года для ОКТМО %s — пропуск", spec.code, oktmo)
+ continue
+ obs_date = _obs_date_for_year(year)
+ by_key[(target.region, obs_date)] = MacroRow(
+ indicator_type=spec.indicator_type,
+ region=target.region,
+ obs_date=obs_date,
+ value=value,
+ unit=spec.unit,
+ comment=spec.comment,
+ )
+
+ return [by_key[k] for k in sorted(by_key, key=lambda k: (k[0], k[1]))]
+
+
+# ── HTTP fetch (thin) ───────────────────────────────────────────────────────────
+
+
+def _http_get(
+ client: httpx.Client, url: str, *, retries: int = DEFAULT_RETRIES
+) -> bytes:
+ """GET с ретраями (сеть / 5xx) → байты тела. Module-level для monkeypatch в тестах.
+
+ Raises:
+ RosstatScraperError: на не-200 / сетевой ошибке после всех ретраев.
+ """
+ import time
+
+ last_exc: Exception | None = None
+ for attempt in range(retries + 1):
+ try:
+ resp = client.get(url)
+ except httpx.HTTPError as e:
+ last_exc = e
+ logger.warning(
+ "Rosstat GET 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"Rosstat GET HTTP {resp.status_code} {url}: {resp.text[:200]}"
+ )
+ if resp.status_code < 500:
+ break # 4xx — ретрай бессмыслен
+ logger.warning(
+ "Rosstat GET 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"Rosstat GET failed after {retries + 1} attempts {url}: {last_exc}")
+
+
+def fetch_dataset(
+ spec: DatasetSpec,
+ *,
+ timeout_s: float = DEFAULT_TIMEOUT_S,
+ retries: int = DEFAULT_RETRIES,
+ client: httpx.Client | None = None,
+) -> list[MacroRow]:
+ """Загрузить и распарсить один датасет Росстата → список MacroRow.
+
+ Два запроса: meta.csv (→ URL данных + fallback-год) и сам data-CSV. Парсинг —
+ в pure-функциях выше (offline-тестируемых). Возвращает строки для целевых ОКТМО.
+
+ Args:
+ spec: описание датасета (код, колонки, целевые ОКТМО).
+ timeout_s: httpx timeout на запрос.
+ retries: число ретраев на сетевых / 5xx ошибках.
+ client: переиспользуемый httpx.Client (для батча). None → создаём свой.
+
+ Raises:
+ RosstatScraperError: на сетевой/HTTP-ошибке либо если meta не содержит URL данных.
+ """
+ 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:
+ meta_url = f"{ROSSTAT_OPENDATA_BASE}/{spec.code}/meta.csv"
+ meta_text = _decode_csv(_http_get(client, meta_url, retries=retries))
+ data_url = parse_meta_data_url(meta_text)
+ if not data_url:
+ raise RosstatScraperError(f"Rosstat {spec.code}: meta.csv без URL данных")
+ fallback_year = parse_meta_modified_year(meta_text)
+
+ data_text = _decode_csv(_http_get(client, data_url, retries=retries))
+ rows = parse_dataset_rows(data_text, spec, fallback_year=fallback_year)
+ logger.info(
+ "Rosstat %s (%s): распарсено %d строк для %d целевых ОКТМО",
+ spec.code,
+ spec.indicator_type,
+ len(rows),
+ len(spec.targets),
+ )
+ return rows
+ finally:
+ if own_client:
+ client.close()
+
+
+def fetch_all_datasets(
+ specs: tuple[DatasetSpec, ...] = DATASETS,
+ *,
+ timeout_s: float = DEFAULT_TIMEOUT_S,
+ retries: int = DEFAULT_RETRIES,
+) -> list[MacroRow]:
+ """Загрузить все активные датасеты Росстата → плоский список MacroRow.
+
+ Один общий httpx.Client на батч. Сбой одного датасета логируется и НЕ валит
+ остальные (graceful — частичная поставка допустима ТЗ).
+ """
+ out: list[MacroRow] = []
+ 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_dataset(spec, retries=retries, client=client))
+ except RosstatScraperError as e:
+ logger.warning("Rosstat: датасет %s пропущен: %s", spec.code, e)
+ return out
+
+
+# ── standalone smoke ────────────────────────────────────────────────────────────
+
+
+def _main() -> None:
+ """``python -m app.services.scrapers.rosstat_emiss`` — печать нескольких строк
+ для ручной верификации против живого rosstat.gov.ru."""
+ logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")
+ rows = fetch_all_datasets()
+ logger.info("Rosstat rows fetched: %d", len(rows))
+ for r in rows:
+ logger.info(
+ " %s / %s @ %s = %s %s",
+ r.indicator_type,
+ r.region,
+ r.obs_date.isoformat(),
+ r.value,
+ r.unit,
+ )
+ if not rows:
+ logger.warning("No rows — проверь доступность rosstat.gov.ru/opendata.")
+
+
+if __name__ == "__main__":
+ _main()
diff --git a/backend/app/services/site_finder/macro.py b/backend/app/services/site_finder/macro.py
index 96b96ec0..3ab104bf 100644
--- a/backend/app/services/site_finder/macro.py
+++ b/backend/app/services/site_finder/macro.py
@@ -73,15 +73,29 @@ _MORTGAGE_INDICATOR_TYPES: frozenset[str] = frozenset(
itype for itype, _unit in _CBR_SERIES_MAP.values()
)
+# Региональные ряды Росстата (#946, source='rosstat') кладутся под region='sverdl'
+# (субъект-агрегат, ОКТМО 65) и 'ekb' (г. Екатеринбург, ОКТМО 6570100000), НЕ под 'rf'.
+# Дефолтный region таких типов = 'sverdl' (субъект — основной уровень потребления
+# downstream affordability/forecast). Явный region='ekb' читатель задаёт сам.
+# Источник правды по типам — DatasetSpec'и в rosstat_emiss (синхронить при добавлении).
+_ROSSTAT_SVERDL_INDICATOR_TYPES: frozenset[str] = frozenset({"population_total"})
+
def _canonical_region(indicator_type: str) -> str:
"""Канонический region для indicator_type, когда вызыватель не задал явный.
- Детерминированно, без БД. 6 CBR-ипотечных типов (mortgage_*) → 'sverdl'
- (так они лежат в macro_indicator на prod); всё остальное (key_rate и пр.) → 'rf'.
- Источник истины по ипотечному множеству — _CBR_SERIES_MAP (зеркало миграции).
+ Детерминированно, без БД:
+ • 6 CBR-ипотечных типов (mortgage_*) → 'sverdl' (так лежат на prod);
+ • региональные ряды Росстата (population_total, #946) → 'sverdl' (субъект-уровень);
+ • всё остальное (key_rate и пр.) → 'rf'.
+ Источники истины — _CBR_SERIES_MAP (зеркало миграции 123) и _ROSSTAT_SVERDL_INDICATOR_TYPES
+ (зеркало DatasetSpec'ов rosstat_emiss).
"""
- return "sverdl" if indicator_type in _MORTGAGE_INDICATOR_TYPES else "rf"
+ if indicator_type in _MORTGAGE_INDICATOR_TYPES:
+ return "sverdl"
+ if indicator_type in _ROSSTAT_SVERDL_INDICATOR_TYPES:
+ return "sverdl"
+ return "rf"
# ── Reader ──────────────────────────────────────────────────────────────────────
diff --git a/backend/app/workers/beat_schedule.py b/backend/app/workers/beat_schedule.py
index 6bc7e43f..d5e4992b 100644
--- a/backend/app/workers/beat_schedule.py
+++ b/backend/app/workers/beat_schedule.py
@@ -321,6 +321,20 @@ def build_beat_schedule() -> dict:
"options": {"queue": "celery"},
}
+ # Росстат макро-показатели (демография региона 65 / ЕКБ) → macro_indicator
+ # (#946 EPIC2). Ряды Росстата (open-data) обновляются редко (годовые) — ежемесячного
+ # прогона с большим запасом достаточно. ON CONFLICT делает прогон идемпотентным
+ # (полный ре-fetch каждый раз; дёшево — пара CSV-файлов). 1-е число месяца 05:45 МСК
+ # (Celery conf.timezone=Europe/Moscow → crontab в МСК); по нагрузке не пересекается с
+ # cbr-macro (пн 05:30). ИПЦ/доходы/СМР-цены — EMISS-only за WAF, пока не тянутся
+ # (см. rosstat_emiss docstring). Задача техническая, не управляется через job_settings —
+ # добавлена сюда (как cbr/poi/refresh). Оператор может перенести расписание правкой crontab.
+ schedule["rosstat-macro-sync-monthly"] = {
+ "task": "tasks.rosstat_macro_sync.rosstat_macro_sync",
+ "schedule": _parse_cron("45 5 1 * *"),
+ "options": {"queue": "celery"},
+ }
+
# Supply-layers пересчёт (#950 EPIC 6 Step5, GG-форсайт) → supply_layers (м.125).
# Детерминированно UPSERT-ит L1(open)/L2(hidden)/L3(future) по всем районам Свердл.
# обл. через supply_layers.compute_all_layers. ON CONFLICT идемпотентен в рамках
diff --git a/backend/app/workers/celery_app.py b/backend/app/workers/celery_app.py
index 6097a1d9..658b0c7f 100644
--- a/backend/app/workers/celery_app.py
+++ b/backend/app/workers/celery_app.py
@@ -61,6 +61,7 @@ celery_app = Celery(
"app.workers.tasks.scrape_cadastre",
"app.workers.tasks.ekburg_permits_sync",
"app.workers.tasks.cbr_macro_sync",
+ "app.workers.tasks.rosstat_macro_sync",
"app.workers.tasks.supply_layers_refresh",
"app.workers.tasks.forecast",
"app.workers.tasks.ird_harvest",
diff --git a/backend/app/workers/tasks/rosstat_macro_sync.py b/backend/app/workers/tasks/rosstat_macro_sync.py
new file mode 100644
index 00000000..4d39f715
--- /dev/null
+++ b/backend/app/workers/tasks/rosstat_macro_sync.py
@@ -0,0 +1,122 @@
+"""Celery task: синхронизация макро-показателей Росстата в ``macro_indicator``
+(#946, EPIC2 «macro-ingest», GG-форсайт / Site Finder v2).
+
+Тянет региональные ряды Росстата (open-data, см. ``rosstat_emiss.fetch_all_datasets``)
+для Свердловской обл. (ОКТМО 65) / ЕКБ и апсертит каждую обсервацию в
+``macro_indicator`` под ``source='rosstat'``. Сейчас приземляется демография
+(``population_total``); ИПЦ/доходы/СМР-цены — EMISS-only за WAF (документировано в
+``rosstat_emiss``), добавятся новым DatasetSpec, когда появится рабочий доступ.
+
+Детерминированно, без LLM. ON CONFLICT делает повторные прогоны идемпотентными
+(re-run обновляет value + updated_at). Расписание — ежемесячно (эти ряды
+обновляются редко), регистрируется в beat_schedule.py.
+
+Mirror conventions (как cbr_macro_sync): ``SessionLocal()`` + try/finally close,
+``logger`` (не print), SAVEPOINT per-row в цикле upsert (backend.md),
+``CAST(:x AS type)`` — никогда ``:x::type`` (psycopg v3). На fetch-фейле —
+логируем и пробрасываем (surfaces в Celery/GlitchTip), не глотаем.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from app.core.db import SessionLocal
+from app.services.scrapers.rosstat_emiss import MacroRow, fetch_all_datasets
+from app.workers.celery_app import celery_app
+
+logger = logging.getLogger(__name__)
+
+# psycopg v3: CAST(:x AS type) — НИКОГДА :x::type (SQLAlchemy+psycopg3 роняет
+# синтаксис на ::). Контракт колонок совпадает с macro_indicator (migration 123):
+# (indicator_type, region, obs_date, value, source, frequency, unit, comment,
+# updated_at) PK (indicator_type, region, obs_date). source='rosstat',
+# frequency='yearly' (текущие ряды Росстата — годовые: obs_date = 1 января года).
+UPSERT_ROSSTAT_SQL = text(
+ """
+ INSERT INTO macro_indicator (
+ indicator_type, region, obs_date, value,
+ source, frequency, unit, comment
+ ) VALUES (
+ CAST(:itype AS text), CAST(:region AS text), CAST(:d AS date),
+ CAST(:v AS numeric),
+ 'rosstat', 'yearly', CAST(:unit AS text), CAST(:comment AS text)
+ )
+ ON CONFLICT (indicator_type, region, obs_date) DO UPDATE SET
+ value = EXCLUDED.value,
+ source = EXCLUDED.source,
+ frequency = EXCLUDED.frequency,
+ unit = EXCLUDED.unit,
+ comment = EXCLUDED.comment,
+ updated_at = now()
+ """
+)
+
+
+def _upsert_rows(db: Session, rows: list[MacroRow]) -> int:
+ """Апсертит MacroRow в macro_indicator. SAVEPOINT per-row, чтобы один битый
+ ряд не откатывал всю транзакцию. Возвращает число успешных upsert'ов."""
+ upserted = 0
+ for r in rows:
+ try:
+ with db.begin_nested():
+ db.execute(
+ UPSERT_ROSSTAT_SQL,
+ {
+ "itype": r.indicator_type,
+ "region": r.region,
+ "d": r.obs_date,
+ "v": r.value,
+ "unit": r.unit,
+ "comment": r.comment,
+ },
+ )
+ upserted += 1
+ except Exception as e:
+ logger.warning(
+ "upsert rosstat %s/%s@%s=%s failed: %s",
+ r.indicator_type,
+ r.region,
+ r.obs_date,
+ r.value,
+ e,
+ )
+ db.commit()
+ return upserted
+
+
+@celery_app.task(
+ bind=True,
+ name="tasks.rosstat_macro_sync.rosstat_macro_sync",
+ max_retries=2,
+)
+def rosstat_macro_sync(self: Any) -> dict[str, Any]:
+ """Загрузить региональные ряды Росстата и апсертить в macro_indicator.
+
+ Returns:
+ Счётчики ``{"fetched": N, "upserted": M}``.
+
+ Raises:
+ Пробрасывает любую ошибку fetch'а (RosstatScraperError и пр.) — чтобы фейл
+ был виден в Celery / GlitchTip, а не молча проглочен. Частичный сбой
+ отдельного датасета НЕ роняет таску (fetch_all_datasets graceful per-dataset).
+ """
+ db = SessionLocal()
+ try:
+ rows = fetch_all_datasets()
+ fetched = len(rows)
+ logger.info("rosstat_macro_sync: fetched %d rosstat rows", fetched)
+
+ upserted = _upsert_rows(db, rows)
+ logger.info("rosstat_macro_sync: upserted %d/%d rosstat rows", upserted, fetched)
+
+ return {"fetched": fetched, "upserted": upserted}
+ except Exception as e:
+ logger.exception("rosstat_macro_sync failed: %s", e)
+ raise
+ finally:
+ db.close()
diff --git a/backend/tests/scrapers/test_rosstat_emiss.py b/backend/tests/scrapers/test_rosstat_emiss.py
new file mode 100644
index 00000000..d3316a69
--- /dev/null
+++ b/backend/tests/scrapers/test_rosstat_emiss.py
@@ -0,0 +1,173 @@
+"""Тесты pure-парсеров Росстат-скрейпера (offline, без живой сети).
+
+Покрывают: разбор meta.csv (URL данных + год), парсинг значений (запятая-десятичный,
+разделители тысяч), извлечение целевых ОКТМО-строк из data-CSV (Свердловская обл. /
+ЕКБ), фоллбэк года из meta, дедуп по (region, obs_date), graceful на пустом /
+без-нужных-колонок CSV. Фикстуры — из реального формата open-data Росстата 4.0
+(verified live 2026-06-07: датасет 7708234640-population).
+"""
+
+from __future__ import annotations
+
+from datetime import date
+from decimal import Decimal
+
+from app.services.scrapers.rosstat_emiss import (
+ OKTMO_EKATERINBURG,
+ OKTMO_SVERDLOVSK_OBLAST,
+ POPULATION_SPEC,
+ DatasetSpec,
+ RegionTarget,
+ _parse_value,
+ parse_dataset_rows,
+ parse_meta_data_url,
+ parse_meta_modified_year,
+)
+
+# Реальный shape meta.csv Росстата (open-data 4.0): property,value; строка данных —
+# property 'data--structure-.csv' → абсолютный URL CSV-данных.
+FIXTURE_META = (
+ "property,value\n"
+ "standardversion,https://rosstat.gov.ru/storage/.../versiya_4.0.pdf\n"
+ 'identifier,7708234640-population\n'
+ 'title,"Численность постоянного населения"\n'
+ "created,20180905\n"
+ "modified,20140919\n"
+ "format,CSV\n"
+ "data-20140919T1709-structure-20141013T1710.csv,"
+ "https://rosstat.gov.ru/opendata/7708234640-population/"
+ "data-20140919T1709-structure-20141013T1710.csv\n"
+ "structure-20141013T1710.csv,"
+ "https://rosstat.gov.ru/opendata/7708234640-population/structure-20141013T1710.csv\n"
+)
+
+# Реальный shape data-CSV (population): header + муниципальные строки, ОКТМО в 'kode',
+# значение в 'total', год в 'year'. Запятая — разделитель полей; значения — целые.
+FIXTURE_POPULATION_DATA = (
+ "number,year,kode,region,municipalities,total,urban,rural\n"
+ '1,2014,7900000000,Республика Адыгея,Муниципальные образования Республики Адыгеи,'
+ "446406,209929,236477\n"
+ "2,2014,6500000000,Свердловская область,Муниципальные образования Свердловской области,"
+ "4320677,3500000,820677\n"
+ "3,2014,657010000011000,Свердловская область,г. Екатеринбург,1412346,1412346,0\n"
+ "4,2014,657013640011000,Свердловская область,Верх-Исетский район,201817,201817,0\n"
+)
+
+
+def test_parse_meta_data_url() -> None:
+ """meta.csv → URL последнего data-файла."""
+ url = parse_meta_data_url(FIXTURE_META)
+ assert url == (
+ "https://rosstat.gov.ru/opendata/7708234640-population/"
+ "data-20140919T1709-structure-20141013T1710.csv"
+ )
+
+
+def test_parse_meta_modified_year() -> None:
+ """meta.csv 'modified' (YYYYMMDD) → год."""
+ assert parse_meta_modified_year(FIXTURE_META) == 2014
+
+
+def test_parse_meta_missing_data_url() -> None:
+ """meta без строки данных → None (не бросает)."""
+ assert parse_meta_data_url("property,value\nidentifier,foo\n") is None
+
+
+def test_parse_value_comma_decimal() -> None:
+ """Запятая как десятичный разделитель: '1,5' → Decimal('1.5')."""
+ assert _parse_value("1,5") == Decimal("1.5")
+
+
+def test_parse_value_thousands_separator() -> None:
+ """Пробелы-разделители тысяч вырезаются: '4 320 677' → Decimal('4320677')."""
+ assert _parse_value("4 320 677") == Decimal("4320677")
+ # узкий неразрывный пробел (Росстат иногда так)
+ assert _parse_value("1 412 346") == Decimal("1412346")
+
+
+def test_parse_value_empty_and_dash() -> None:
+ """Пустое / прочерк / многоточие → None (строка пропускается)."""
+ assert _parse_value("") is None
+ assert _parse_value("-") is None
+ assert _parse_value("…") is None
+ assert _parse_value("не число") is None
+
+
+def test_parse_dataset_targets_only() -> None:
+ """Из data-CSV извлекаются ТОЛЬКО целевые ОКТМО (субъект + ЕКБ), не районы/др.регионы."""
+ rows = parse_dataset_rows(FIXTURE_POPULATION_DATA, POPULATION_SPEC, fallback_year=None)
+ by_region = {r.region: r for r in rows}
+ assert set(by_region) == {"sverdl", "ekb"}
+ # Адыгея и Верх-Исетский район — НЕ среди целей
+ assert all(r.region in {"sverdl", "ekb"} for r in rows)
+
+
+def test_parse_dataset_values_and_dates() -> None:
+ """Значения и obs_date (1 января года из колонки 'year') корректны."""
+ rows = parse_dataset_rows(FIXTURE_POPULATION_DATA, POPULATION_SPEC, fallback_year=None)
+ by_region = {r.region: r for r in rows}
+ assert by_region["sverdl"].value == Decimal("4320677")
+ assert by_region["sverdl"].obs_date == date(2014, 1, 1)
+ assert by_region["ekb"].value == Decimal("1412346")
+ assert by_region["ekb"].obs_date == date(2014, 1, 1)
+ # тип / единица из spec
+ assert by_region["sverdl"].indicator_type == "population_total"
+ assert by_region["sverdl"].unit == "чел"
+
+
+def test_parse_dataset_fallback_year_when_no_year_col() -> None:
+ """Если в spec year_col_header=None — год берётся из fallback_year (meta modified)."""
+ spec_no_year = DatasetSpec(
+ code=POPULATION_SPEC.code,
+ indicator_type="population_total",
+ unit="чел",
+ value_col_header="total",
+ oktmo_col_header="kode",
+ year_col_header=None, # год не из данных
+ targets=(RegionTarget(oktmo=OKTMO_SVERDLOVSK_OBLAST, region="sverdl"),),
+ )
+ rows = parse_dataset_rows(FIXTURE_POPULATION_DATA, spec_no_year, fallback_year=2014)
+ assert len(rows) == 1
+ assert rows[0].obs_date == date(2014, 1, 1)
+
+
+def test_parse_dataset_dedup_last_wins() -> None:
+ """Дубль строки по (region, obs_date) схлопывается — последняя выигрывает."""
+ data = (
+ "number,year,kode,region,municipalities,total,urban,rural\n"
+ "1,2014,6500000000,Свердловская область,МО,4320677,0,0\n"
+ "2,2014,6500000000,Свердловская область,МО,4400000,0,0\n" # дубль ОКТМО+год
+ )
+ rows = parse_dataset_rows(data, POPULATION_SPEC, fallback_year=None)
+ assert len(rows) == 1
+ assert rows[0].value == Decimal("4400000")
+
+
+def test_parse_dataset_skips_unparsable_value() -> None:
+ """Строка с непарсящимся значением пропускается, валидные остаются."""
+ data = (
+ "number,year,kode,region,municipalities,total,urban,rural\n"
+ "1,2014,6500000000,Свердловская область,МО,-,0,0\n" # value '-'
+ "2,2014,657010000011000,Свердловская область,г. Екатеринбург,1412346,0,0\n"
+ )
+ rows = parse_dataset_rows(data, POPULATION_SPEC, fallback_year=None)
+ assert len(rows) == 1
+ assert rows[0].region == "ekb"
+
+
+def test_parse_dataset_missing_columns_graceful() -> None:
+ """CSV без нужных колонок (kode/total) → [] (+warning), не бросает."""
+ data = "foo,bar\n1,2\n3,4\n"
+ assert parse_dataset_rows(data, POPULATION_SPEC, fallback_year=2014) == []
+
+
+def test_parse_dataset_empty_graceful() -> None:
+ """Пустой CSV → []."""
+ assert parse_dataset_rows("", POPULATION_SPEC, fallback_year=2014) == []
+
+
+def test_oktmo_constants_used_in_spec() -> None:
+ """Спека population таргетит именно субъект-ОКТМО и ЕКБ-ОКТМО (контракт ТЗ §7.11)."""
+ oktmos = {t.oktmo for t in POPULATION_SPEC.targets}
+ assert OKTMO_SVERDLOVSK_OBLAST in oktmos
+ assert OKTMO_EKATERINBURG in oktmos
diff --git a/backend/tests/workers/test_rosstat_macro_sync.py b/backend/tests/workers/test_rosstat_macro_sync.py
new file mode 100644
index 00000000..cf7d9856
--- /dev/null
+++ b/backend/tests/workers/test_rosstat_macro_sync.py
@@ -0,0 +1,126 @@
+"""Тесты Celery-таски rosstat_macro_sync (mock session + mock fetch, без живой сети).
+
+Покрывают: контракт upsert-SQL (source='rosstat', ON CONFLICT, CAST not ::),
+счётчики fetched/upserted, проброс ошибки fetch'а, регистрацию в beat (ежемесячно).
+"""
+
+from __future__ import annotations
+
+from datetime import date
+from decimal import Decimal
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+
+def _make_mock_db() -> tuple[MagicMock, list[dict[str, Any]]]:
+ """Mock DB session с поддержкой db.begin_nested() context manager (SAVEPOINT).
+
+ Возвращает (db, captured_params) — captured содержит params каждого
+ db.execute(..., params).
+ """
+ captured: list[dict[str, Any]] = []
+
+ def execute_router(stmt: Any, params: Any = None) -> MagicMock:
+ if params is not None:
+ captured.append(params)
+ return MagicMock()
+
+ db = MagicMock()
+ db.execute = execute_router
+ db.begin_nested.return_value.__enter__ = MagicMock(return_value=None)
+ db.begin_nested.return_value.__exit__ = MagicMock(return_value=False)
+ return db, captured
+
+
+def test_upsert_sql_contract() -> None:
+ """UPSERT_ROSSTAT_SQL содержит обязательные литералы контракта macro_indicator
+ и НЕ использует :x::type (psycopg v3)."""
+ from app.workers.tasks.rosstat_macro_sync import UPSERT_ROSSTAT_SQL
+
+ sql = str(UPSERT_ROSSTAT_SQL)
+ upper = sql.upper()
+
+ assert "INTO MACRO_INDICATOR" in upper
+ assert "'rosstat'" in sql
+ assert "'yearly'" in sql
+ assert "ON CONFLICT (INDICATOR_TYPE, REGION, OBS_DATE) DO UPDATE" in upper
+ assert "VALUE = EXCLUDED.VALUE" in upper
+ assert "UPDATED_AT = NOW()" in upper
+ # psycopg v3: CAST(:x AS type), НИКОГДА :x::type
+ assert "CAST(:ITYPE AS TEXT)" in upper
+ assert "CAST(:D AS DATE)" in upper
+ assert "CAST(:V AS NUMERIC)" in upper
+ assert "::" not in sql
+
+
+def test_task_upserts_each_row_with_correct_params() -> None:
+ """Каждая MacroRow → один execute с bind-параметрами itype/region/d/v/unit/comment."""
+ from app.services.scrapers.rosstat_emiss import MacroRow
+ from app.workers.tasks import rosstat_macro_sync as task_mod
+
+ db, captured = _make_mock_db()
+ fake_rows = [
+ MacroRow("population_total", "sverdl", date(2014, 1, 1), Decimal("4320677"), "чел", "c1"),
+ MacroRow("population_total", "ekb", date(2014, 1, 1), Decimal("1412346"), "чел", "c2"),
+ ]
+
+ with (
+ patch.object(task_mod, "SessionLocal", return_value=db),
+ patch.object(task_mod, "fetch_all_datasets", return_value=fake_rows) as mock_fetch,
+ ):
+ result = task_mod.rosstat_macro_sync.run()
+
+ mock_fetch.assert_called_once()
+ assert result["fetched"] == 2
+ assert result["upserted"] == 2
+ assert captured == [
+ {
+ "itype": "population_total",
+ "region": "sverdl",
+ "d": date(2014, 1, 1),
+ "v": Decimal("4320677"),
+ "unit": "чел",
+ "comment": "c1",
+ },
+ {
+ "itype": "population_total",
+ "region": "ekb",
+ "d": date(2014, 1, 1),
+ "v": Decimal("1412346"),
+ "unit": "чел",
+ "comment": "c2",
+ },
+ ]
+ db.commit.assert_called()
+ db.close.assert_called_once()
+
+
+def test_task_reraises_fetch_failure() -> None:
+ """Ошибка fetch'а пробрасывается (surfaces в Celery/GlitchTip), DB закрывается."""
+ from app.workers.tasks import rosstat_macro_sync as task_mod
+
+ db, _ = _make_mock_db()
+
+ with (
+ patch.object(task_mod, "SessionLocal", return_value=db),
+ patch.object(task_mod, "fetch_all_datasets", side_effect=RuntimeError("rosstat down")),
+ ):
+ try:
+ task_mod.rosstat_macro_sync.run()
+ raise AssertionError("expected RuntimeError to propagate")
+ except RuntimeError as e:
+ assert "rosstat down" in str(e)
+
+ db.close.assert_called_once()
+
+
+def test_task_registered_in_beat_monthly() -> None:
+ """Задача зарегистрирована в beat как ежемесячная (1-е число)."""
+ from app.workers.beat_schedule import build_beat_schedule
+
+ schedule = build_beat_schedule()
+ assert "rosstat-macro-sync-monthly" in schedule
+ entry = schedule["rosstat-macro-sync-monthly"]
+ assert entry["task"] == "tasks.rosstat_macro_sync.rosstat_macro_sync"
+ # crontab day_of_month = 1 (1-е число месяца)
+ assert 1 in entry["schedule"].day_of_month