feat(macro): Росстат open-data scraper → macro_indicator (#946, partial) #1143
7 changed files with 955 additions and 4 deletions
501
backend/app/services/scrapers/rosstat_emiss.py
Normal file
501
backend/app/services/scrapers/rosstat_emiss.py
Normal file
|
|
@ -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-байтный ``<h1>Forbidden</h1>`` БЕЗ ``Server``-заголовка
|
||||
(блок на фронтящем WAF/CDN до приложения — IP-reputation, не JS-challenge, который мог
|
||||
бы решить браузер). ``sdmx.gks.ru`` — DNS мёртв, ``showdata.gks.ru`` — 502. То есть
|
||||
EMISS-only ряды (ИПЦ, среднедушевые доходы, индекс цен на СМР) автоматически НЕдоступны.
|
||||
Это независимо зафиксировано прежней разведкой (комментарий миграции 126 §7.11, vault
|
||||
data-sources).
|
||||
|
||||
``rosstat.gov.ru/opendata`` — БЕЗ WAF, машиночитаем (российский стандарт открытых
|
||||
данных 4.0):
|
||||
• реестр ``/opendata/list.csv`` — CSV (property,title,value,format),
|
||||
value = URL ``…/<code>/meta.csv``;
|
||||
• метаданные ``/opendata/<code>/meta.csv`` — CP1251 CSV (property,value); строка
|
||||
``data-<ts>-structure-<ts>.csv`` →
|
||||
URL фактического файла данных;
|
||||
• данные ``/opendata/<code>/data-<ts>-…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/<code>/…``).
|
||||
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-<ts>-structure-<ts>.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()
|
||||
|
|
@ -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 ──────────────────────────────────────────────────────────────────────
|
||||
|
|
|
|||
|
|
@ -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 идемпотентен в рамках
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
122
backend/app/workers/tasks/rosstat_macro_sync.py
Normal file
122
backend/app/workers/tasks/rosstat_macro_sync.py
Normal file
|
|
@ -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()
|
||||
173
backend/tests/scrapers/test_rosstat_emiss.py
Normal file
173
backend/tests/scrapers/test_rosstat_emiss.py
Normal file
|
|
@ -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-<ts>-structure-<ts>.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
|
||||
126
backend/tests/workers/test_rosstat_macro_sync.py
Normal file
126
backend/tests/workers/test_rosstat_macro_sync.py
Normal file
|
|
@ -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
|
||||
Loading…
Add table
Reference in a new issue