From 2da8a285b393fe68e8ba166c3ac38f711adda25c Mon Sep 17 00:00:00 2001 From: Light1YT Date: Sun, 7 Jun 2026 18:07:56 +0500 Subject: [PATCH] =?UTF-8?q?feat(macro):=20=D0=A0=D0=BE=D1=81=D1=81=D1=82?= =?UTF-8?q?=D0=B0=D1=82=20open-data=20macro=20scraper=20->=20macro=5Findic?= =?UTF-8?q?ator=20(#946)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit EPIC2 macro-ingest: региональный (Свердловская обл. ОКТМО 65 / ЕКБ) скрейпер Росстата в унифицированную macro_indicator (м.123). Зеркалит форму cbr_macro: pure-парсеры + тонкий httpx, Celery-таска с SAVEPOINT-per-row upsert (CAST not ::, ON CONFLICT по PK indicator_type/region/obs_date). ЕМИСС/fedstat.ru (SDMX) за WAF (hard-403 на всех путях из dev-IP) -> ИПЦ/доходы/ СМР-цены пока недоступны (документировано в коде). rosstat.gov.ru/opendata (стандарт 4.0) WAF-free -> приземляет демографию population_total (sverdl 4.32M / ekb 1.41M, §7.11). - services/scrapers/rosstat_emiss.py — fetch+parse open-data (registry->meta->data CSV) - workers/tasks/rosstat_macro_sync.py — Celery upsert (source=rosstat) - beat: rosstat-macro-sync-monthly; include в celery_app - site_finder/macro.py: population_total -> region-aware default sverdl (additive) - tests: 18 offline-тестов парсера + контракта таски Без миграции (вписано в существующую схему), без новых зависимостей. Refs #946. --- .../app/services/scrapers/rosstat_emiss.py | 501 ++++++++++++++++++ backend/app/services/site_finder/macro.py | 22 +- backend/app/workers/beat_schedule.py | 14 + backend/app/workers/celery_app.py | 1 + .../app/workers/tasks/rosstat_macro_sync.py | 122 +++++ backend/tests/scrapers/test_rosstat_emiss.py | 173 ++++++ .../tests/workers/test_rosstat_macro_sync.py | 126 +++++ 7 files changed, 955 insertions(+), 4 deletions(-) create mode 100644 backend/app/services/scrapers/rosstat_emiss.py create mode 100644 backend/app/workers/tasks/rosstat_macro_sync.py create mode 100644 backend/tests/scrapers/test_rosstat_emiss.py create mode 100644 backend/tests/workers/test_rosstat_macro_sync.py 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-байтный ``

Forbidden

`` БЕЗ ``Server``-заголовка +(блок на фронтящем WAF/CDN до приложения — IP-reputation, не JS-challenge, который мог +бы решить браузер). ``sdmx.gks.ru`` — DNS мёртв, ``showdata.gks.ru`` — 502. То есть +EMISS-only ряды (ИПЦ, среднедушевые доходы, индекс цен на СМР) автоматически НЕдоступны. +Это независимо зафиксировано прежней разведкой (комментарий миграции 126 §7.11, vault +data-sources). + +``rosstat.gov.ru/opendata`` — БЕЗ WAF, машиночитаем (российский стандарт открытых +данных 4.0): + • реестр ``/opendata/list.csv`` — CSV (property,title,value,format), + value = URL ``…//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 555595b8..46d03d04 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 7eb01089..ef5e81f2 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 -- 2.45.3