From de92c54540843cba432c76c43173840ff1484b21 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Sun, 31 May 2026 12:44:56 +0000 Subject: [PATCH] =?UTF-8?q?feat(tradein):=20=D0=A1=D0=B1=D0=B5=D1=80=D0=98?= =?UTF-8?q?=D0=BD=D0=B4=D0=B5=D0=BA=D1=81=20monthly=20city-level=20price?= =?UTF-8?q?=20index=20pull=20(#887,=20data-layer)=20(#890)=20Co-authored-b?= =?UTF-8?q?y:=20bot-backend=20=20Co-committed-b?= =?UTF-8?q?y:=20bot-backend=20?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../backend/app/services/sber_index.py | 405 ++++++++++++++++++ tradein-mvp/backend/app/services/scheduler.py | 42 ++ .../backend/app/tasks/sber_index_pull.py | 94 ++++ .../backend/data/sql/092_sber_price_index.sql | 70 +++ ..._scrape_schedules_seed_sber_index_pull.sql | 73 ++++ tradein-mvp/backend/tests/test_sber_index.py | 308 +++++++++++++ 6 files changed, 992 insertions(+) create mode 100644 tradein-mvp/backend/app/services/sber_index.py create mode 100644 tradein-mvp/backend/app/tasks/sber_index_pull.py create mode 100644 tradein-mvp/backend/data/sql/092_sber_price_index.sql create mode 100644 tradein-mvp/backend/data/sql/093_scrape_schedules_seed_sber_index_pull.sql create mode 100644 tradein-mvp/backend/tests/test_sber_index.py diff --git a/tradein-mvp/backend/app/services/sber_index.py b/tradein-mvp/backend/app/services/sber_index.py new file mode 100644 index 00000000..f64cf30f --- /dev/null +++ b/tradein-mvp/backend/app/services/sber_index.py @@ -0,0 +1,405 @@ +"""СберИндекс city-level secondary-housing price index pull (#887). + +Fetches monthly time-series data from sberindex.ru/api/sowa (reverse-engineered +in vault research note 2026-05-31-0845-resolver-794) across three dashboards: + - real_estate_deals — hedonic sold-price index + - residential_real_estate_prices — repeat-sales hedonic (cleanest signal) + - dinamika-tsen-obyavlenii — asking-price benchmark + +API summary: + POST https://sberindex.ru/api/sowa + Body: {"SOWA": {"method": "GET", "route": "", "data": {"type": "object", "value": []}}} + route = base64( /dataset/v1/?filter=&limit=1000&offset=0 ) + filter = {"REF_AREA": "", "FREQ": "M", "SOURCE": "SI", "REALTY": "2"} + Required headers: content-type, rquid (random 32-hex), x-language, user-agent. + +Response fields (each cell is __string__ or __number__): + indicator_id, kpi_id, period, value, obs_status, source, ref_area, realty, + freq, unit_measure, unit_mult + +Data-quality benchmark: after upsert of the dinamika-tsen-obyavlenii (asking) series, +the latest asking index per city is logged at INFO level as a data-quality reference. +Full asking-vs-our-median reconciliation is a separate follow-up (#887 step 5+). + +NOTE — estimator time-adjust wire (step 4) is OUT OF SCOPE for this PR. +It is a client-visible change that requires a backtest gate. Defer to a separate issue. + +REF_AREA codes: + 643 = РФ (confirmed from live session, vault research note 2026-05-31-0845). + # TODO (#887 follow-up): resolve EKB (Свердловская обл) + Moscow REF_AREA codes. + # The dynpage config endpoint (GET /api/dynpage/ru/dashboards/) returns + # a base64-encoded React config with region options, but requires a working TLS chain + # from the deployment environment. Candidate codes: Свердловская обл ≈ 66, + # Москва ≈ 77 (ОКАТО-derived). Confirm by inspecting the dynpage form field + # «REF_AREA» values in a browser session or after fixing TLS in prod. + # Once confirmed, add them to SBER_REF_AREAS below and remove this TODO. +""" + +from __future__ import annotations + +import base64 +import json +import logging +import uuid +from collections.abc import AsyncIterator +from datetime import UTC, date, datetime +from urllib.parse import quote + +import httpx +from sqlalchemy import text +from sqlalchemy.orm import Session + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Config — REF_AREA codes +# --------------------------------------------------------------------------- + +# REF_AREA code → city label used as the primary key in sber_price_index. +# 643 = РФ is confirmed (live reverse-engineering, vault 2026-05-31-0845). +# EKB and Moscow codes are pending TLS resolution — see TODO in module docstring. +SBER_REF_AREAS: dict[str, str] = { + "643": "Россия", + # TODO (#887 follow-up): uncomment + verify EKB / Moscow REF_AREA codes. + # "66": "Свердловская область", # candidate — confirm via dynpage form + # "77": "Москва", # candidate — confirm via dynpage form +} + +# Dashboards to pull: slug → human label (used as 'segment' placeholder) +SBER_DASHBOARDS: list[str] = [ + "real_estate_deals", + "residential_real_estate_prices", + "dinamika-tsen-obyavlenii", +] + +SBER_API_URL = "https://sberindex.ru/api/sowa" +SBER_HTTP_TIMEOUT = 30.0 +_CHROME_UA = ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/124.0.0.0 Safari/537.36" +) + +# --------------------------------------------------------------------------- +# Route builder +# --------------------------------------------------------------------------- + + +def build_sber_route(dashboard: str, ref_area: str) -> str: + """Build the base64-encoded route for the /api/sowa payload. + + Encodes /dataset/v1/?filter=&limit=1000&offset=0 + where filter = {"REF_AREA": ref_area, "FREQ": "M", "SOURCE": "SI", "REALTY": "2"}. + + All filter values are strings per the live-captured payload (vault research note). + """ + filter_dict = { + "REF_AREA": str(ref_area), + "FREQ": "M", + "SOURCE": "SI", + "REALTY": "2", + } + # urlencode the JSON, then build the path + filter_str = quote(json.dumps(filter_dict, separators=(",", ":")), safe="") + path = f"/dataset/v1/{dashboard}?filter={filter_str}&limit=1000&offset=0" + return base64.b64encode(path.encode("utf-8")).decode("ascii") + + +# --------------------------------------------------------------------------- +# Response decoder +# --------------------------------------------------------------------------- + +# Field order returned by the API (confirmed from live session) +_FIELD_ORDER = [ + "indicator_id", + "kpi_id", + "period", + "value", + "obs_status", + "source", + "ref_area", + "realty", + "freq", + "unit_measure", + "unit_mult", +] + + +def _decode_cell(cell: str) -> str | float: + """Decode a single response cell. + + Cells arrive as one of: + '__string__' → decode UTF-8 string + '__number__' → parse as float + """ + if cell.startswith("__string__"): + encoded = cell[len("__string__") :] + return base64.b64decode(encoded).decode("utf-8") + if cell.startswith("__number__"): + return float(cell[len("__number__") :]) + # Fallback: try numeric, else return raw + try: + return float(cell) + except (ValueError, TypeError): + return str(cell) + + +def decode_sber_response( + response_data: dict, # type: ignore[type-arg] +) -> list[dict[str, str | float]]: + """Decode the SOWA response payload into a list of row dicts. + + Raises ValueError if the response structure is unexpected. + """ + try: + sowa = response_data["SOWA"] + values = sowa["data"]["value"] + except (KeyError, TypeError) as exc: + raise ValueError(f"Unexpected SOWA response structure: {exc}") from exc + + fields: list[str] | None = None + rows_raw: list[list[str]] = [] + + for entry in values: + key = entry.get("key") + val = entry.get("value") + if key == "fields": + fields = [_decode_cell(c) for c in val] # type: ignore[arg-type] + elif key == "data": + rows_raw = val + + if fields is None: + raise ValueError("No 'fields' key found in SOWA response") + + result: list[dict[str, str | float]] = [] + for raw_row in rows_raw: + if len(raw_row) != len(fields): + logger.warning( + "sber_index: row length mismatch (expected %d, got %d) — skip", + len(fields), + len(raw_row), + ) + continue + row_dict: dict[str, str | float] = {} + for col_name, cell in zip(fields, raw_row, strict=True): + row_dict[col_name] = _decode_cell(cell) # type: ignore[arg-type] + result.append(row_dict) + + return result + + +# --------------------------------------------------------------------------- +# Period normalisation +# --------------------------------------------------------------------------- + + +def _parse_period_month(period_str: str) -> date: + """Parse the API period string into the first day of the month. + + API returns ISO-8601 with UTC offset, e.g. '2017-01-30T21:00:00Z'. + We normalise to date(year, month, 1) — the first day of the reported month. + The day/time portion is an artefact of timezone shift and is discarded. + """ + dt = datetime.fromisoformat(period_str.replace("Z", "+00:00")) + # Shift to UTC date, then take first day of that month + dt_utc = dt.astimezone(UTC) + return date(dt_utc.year, dt_utc.month, 1) + + +# --------------------------------------------------------------------------- +# Async fetch — single dashboard × city +# --------------------------------------------------------------------------- + + +async def fetch_sber_index( + client: httpx.AsyncClient, + *, + dashboard: str, + ref_area: str, +) -> AsyncIterator[tuple[str, date, str, str, float]]: + """Fetch one dashboard × REF_AREA combination from sberindex.ru/api/sowa. + + Yields tuples of: + (city_label, period_month, segment_label, dashboard, index_value_rub_m2) + + city_label and segment_label come from the API response fields 'ref_area' and 'realty'. + + Raises httpx.HTTPStatusError on non-2xx response (caller handles per-series). + """ + route = build_sber_route(dashboard, ref_area) + rquid = uuid.uuid4().hex # 32-hex random, required header + + payload = { + "SOWA": { + "method": "GET", + "route": route, + "data": {"type": "object", "value": []}, + } + } + headers = { + "content-type": "application/json", + "rquid": rquid, + "x-language": "ru", + "accept": "application/json, text/plain, */*", + "user-agent": _CHROME_UA, + } + + resp = await client.post(SBER_API_URL, json=payload, headers=headers) + resp.raise_for_status() + + rows = decode_sber_response(resp.json()) + + for row in rows: + period_raw = row.get("period", "") + value_raw = row.get("value") + city_label = str(row.get("ref_area", ref_area)) + segment_label = str(row.get("realty", "")) + + if not period_raw or value_raw is None: + continue + + try: + period_month = _parse_period_month(str(period_raw)) + index_value = float(value_raw) + except (ValueError, TypeError) as exc: + logger.warning( + "sber_index: skip row — parse error for dashboard=%s ref_area=%s: %s", + dashboard, + ref_area, + exc, + ) + continue + + yield (city_label, period_month, segment_label, dashboard, index_value) + + +# --------------------------------------------------------------------------- +# Bulk pull + upsert +# --------------------------------------------------------------------------- + + +async def pull_sber_indices( + db: Session, + *, + cities: dict[str, str] | None = None, + dashboards: list[str] | None = None, +) -> dict[str, int]: + """Fetch all configured dashboards × cities and upsert into sber_price_index. + + Args: + db: SQLAlchemy Session (tradein DB). + cities: REF_AREA code → city label mapping. Defaults to SBER_REF_AREAS. + dashboards: list of dashboard slugs to pull. Defaults to SBER_DASHBOARDS. + + Returns dict of counters: {upserted: int, skipped: int, errors: int}. + + Per-series try/except: one failing series logs + continues (does not abort others). + After upserting 'dinamika-tsen-obyavlenii', logs the latest asking index per city + as a data-quality benchmark reference. + """ + if cities is None: + cities = SBER_REF_AREAS + if dashboards is None: + dashboards = SBER_DASHBOARDS + + counters: dict[str, int] = {"upserted": 0, "skipped": 0, "errors": 0} + + async with httpx.AsyncClient(timeout=SBER_HTTP_TIMEOUT) as client: + for ref_area, _city_hint in cities.items(): + for dashboard in dashboards: + try: + rows_to_upsert: list[tuple[str, date, str, str, float]] = [] + async for row in fetch_sber_index( + client, dashboard=dashboard, ref_area=ref_area + ): + rows_to_upsert.append(row) + + if not rows_to_upsert: + logger.info( + "sber_index: no rows returned for dashboard=%s ref_area=%s", + dashboard, + ref_area, + ) + counters["skipped"] += 1 + continue + + # Idempotent upsert — ON CONFLICT(city, period_month, dashboard) + for city_label, period_month, segment, dash, value in rows_to_upsert: + db.execute( + text( + """ + INSERT INTO sber_price_index + (city, period_month, segment, dashboard, + index_value_rub_m2, source, fetched_at) + VALUES ( + CAST(:city AS text), + CAST(:period_month AS date), + CAST(:segment AS text), + CAST(:dashboard AS text), + CAST(:index_value AS double precision), + 'sberindex', + now() + ) + ON CONFLICT (city, period_month, dashboard) + DO UPDATE SET + index_value_rub_m2 = EXCLUDED.index_value_rub_m2, + segment = EXCLUDED.segment, + fetched_at = now() + """ + ), + { + "city": city_label, + "period_month": period_month.isoformat(), + "segment": segment, + "dashboard": dash, + "index_value": value, + }, + ) + counters["upserted"] += 1 + + db.commit() + logger.info( + "sber_index: upserted %d rows for dashboard=%s ref_area=%s", + len(rows_to_upsert), + dashboard, + ref_area, + ) + + # Data-quality benchmark log for asking-price series + if dashboard == "dinamika-tsen-obyavlenii" and rows_to_upsert: + latest_row = max(rows_to_upsert, key=lambda r: r[1]) + city_lbl, latest_month, _, _, latest_val = latest_row + logger.info( + "sber_index benchmark [asking]: city=%r latest_month=%s " + "index_value_rub_m2=%.0f " + "(data-quality ref; full asking-vs-our-median is follow-up)", + city_lbl, + latest_month.isoformat(), + latest_val, + ) + + except httpx.HTTPStatusError as exc: + logger.error( + "sber_index: HTTP %d for dashboard=%s ref_area=%s — skip series", + exc.response.status_code, + dashboard, + ref_area, + ) + counters["errors"] += 1 + except httpx.RequestError as exc: + logger.error( + "sber_index: network error for dashboard=%s ref_area=%s: %s — skip", + dashboard, + ref_area, + exc, + ) + counters["errors"] += 1 + except Exception: + logger.exception( + "sber_index: unexpected error for dashboard=%s ref_area=%s — skip", + dashboard, + ref_area, + ) + counters["errors"] += 1 + + return counters diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py index 1bbbadab..c719d6c7 100644 --- a/tradein-mvp/backend/app/services/scheduler.py +++ b/tradein-mvp/backend/app/services/scheduler.py @@ -28,6 +28,10 @@ Sources: (tasks/deactivate_stale_avito.py, #759; marks avito listings with last_seen_at > TTL days as is_active=false — no DELETE, no external calls, enabled by default; window after avito sweep) + - sber_index_pull → run_sber_index_pull + (tasks/sber_index_pull.py, #887; monthly pull of СберИндекс + city-level secondary-housing price index — pure HTTP+DB, no + anti-bot/proxy; window 05:00-06:00 UTC, ~28-day cadence) """ from __future__ import annotations @@ -596,6 +600,42 @@ async def trigger_deactivate_stale_avito_run( return run_id +async def trigger_sber_index_pull_run(db: Session, schedule_row: dict[str, Any]) -> int | None: + """Создать scrape_runs + launch run_sber_index_pull в asyncio.create_task. + + Monthly pull (#887): fetches city-level secondary-housing price index from + sberindex.ru/api/sowa across three dashboards (real_estate_deals, + residential_real_estate_prices, dinamika-tsen-obyavlenii) and upserts into + sber_price_index. Pure HTTP+DB, no scraper proxy, no anti-bot. + + Mirrors trigger_yandex_address_backfill_run: claim run → create_task → mark_done/failed + delegated to tasks/sber_index_pull.py. + + Returns run_id или None (skip — already running). + """ + run_id = _claim_run(db, schedule_row) + if run_id is None: + return None + + params = schedule_row.get("default_params") or {} + + async def _run() -> None: + run_db = SessionLocal() + try: + from app.tasks.sber_index_pull import run_sber_index_pull + + await run_sber_index_pull(run_db, run_id=run_id, params=params) + except Exception: + logger.exception("scheduler: run_sber_index_pull crashed run_id=%d", run_id) + finally: + run_db.close() + + task = asyncio.create_task(_run()) + task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + logger.info("scheduler: triggered sber_index_pull run_id=%d", run_id) + return run_id + + async def trigger_asking_to_sold_ratio_run(db: Session, schedule_row: dict[str, Any]) -> int | None: """Создать scrape_runs + launch recompute_asking_to_sold_ratios в executor (sync DB-only task). @@ -916,6 +956,8 @@ async def scheduler_loop() -> None: await trigger_yandex_address_backfill_run(db, sch) elif source == "deactivate_stale_avito": await trigger_deactivate_stale_avito_run(db, sch) + elif source == "sber_index_pull": + await trigger_sber_index_pull_run(db, sch) else: logger.warning("scheduler: unknown source=%s, skip", source) finally: diff --git a/tradein-mvp/backend/app/tasks/sber_index_pull.py b/tradein-mvp/backend/app/tasks/sber_index_pull.py new file mode 100644 index 00000000..1e3aa6d5 --- /dev/null +++ b/tradein-mvp/backend/app/tasks/sber_index_pull.py @@ -0,0 +1,94 @@ +"""Scheduled task: pull СберИндекс city-level secondary-housing price index (#887). + +Delegates HTTP fetch + upsert to app.services.sber_index.pull_sber_indices(). +Wired into the in-app scheduler as source='sber_index_pull'. Triggered monthly via +the scrape_schedules seed (migration 093). + +Three dashboards fetched per configured city: + - real_estate_deals (hedonic sold-price index) + - residential_real_estate_prices (repeat-sales hedonic) + - dinamika-tsen-obyavlenii (asking-price benchmark) + +NOTE: estimator time-adjust wire (step 4, #887) is OUT OF SCOPE for this module. +That is a client-visible change requiring a backtest gate — see issue #887 step 4 +and the follow-up note in app/services/sber_index.py. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, field + +from sqlalchemy.orm import Session + +from app.services import scrape_runs as runs_mod +from app.services.sber_index import pull_sber_indices + +logger = logging.getLogger(__name__) + +__all__ = [ + "SberIndexPullResult", + "run_sber_index_pull", +] + + +@dataclass +class SberIndexPullResult: + """Result counters from a scheduled СберИндекс pull run.""" + + upserted: int = 0 + skipped: int = 0 + errors: int = 0 + duration_sec: float = field(default=0.0) + + +async def run_sber_index_pull( + db: Session, + *, + run_id: int, + params: dict, # type: ignore[type-arg] +) -> SberIndexPullResult: + """Execute СберИндекс pull with run lifecycle management. + + Wraps pull_sber_indices(), emitting heartbeat before the fetch and marking + the scrape_run as done/failed on completion. + + Params (from default_params jsonb in scrape_schedules): + interval_days: int — informational only (scheduler uses next_run_at arithmetic). + """ + import time + + counters: dict[str, int] = { + "upserted": 0, + "skipped": 0, + "errors": 0, + } + started_at = time.monotonic() + + try: + runs_mod.update_heartbeat(db, run_id, counters) + + result_counters = await pull_sber_indices(db) + counters = result_counters + + duration_sec = time.monotonic() - started_at + runs_mod.mark_done(db, run_id, {**counters, "duration_sec": int(duration_sec)}) + logger.info( + "sber_index_pull run_id=%d done — upserted=%d skipped=%d errors=%d %.1fs", + run_id, + counters["upserted"], + counters["skipped"], + counters["errors"], + duration_sec, + ) + return SberIndexPullResult( + upserted=counters["upserted"], + skipped=counters["skipped"], + errors=counters["errors"], + duration_sec=duration_sec, + ) + + except Exception as exc: + logger.exception("sber_index_pull run_id=%d failed", run_id) + runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters) + raise diff --git a/tradein-mvp/backend/data/sql/092_sber_price_index.sql b/tradein-mvp/backend/data/sql/092_sber_price_index.sql new file mode 100644 index 00000000..b08b4ac0 --- /dev/null +++ b/tradein-mvp/backend/data/sql/092_sber_price_index.sql @@ -0,0 +1,70 @@ +-- 092_sber_price_index.sql +-- Issue #887 — city-level secondary-housing price index from СберИндекс (sberindex.ru). +-- +-- Context: +-- СберИндекс publishes monthly city/region-level secondary-housing price statistics +-- across three dashboards: +-- * real_estate_deals — hedonic sold-price index (руб/м²) +-- * residential_real_estate_prices — repeat-sales / asking-adjusted hedonic +-- * dinamika-tsen-obyavlenii — asking-price benchmark (объявления, руб/м²) +-- +-- The pull service (app/services/sber_index.py) fetches all three series for configured +-- cities (initially РФ-643, with EKB/MSK codes TBD — see TODO in sber_index.py) via the +-- /api/sowa POST endpoint (reverse-engineered in #794 research note) and upserts rows here. +-- +-- The table feeds the estimator time-adjust coefficient (step 4, #887 out-of-scope — +-- backtest-gated, separate follow-up). +-- +-- Schema notes: +-- * city — REF_AREA label as returned by the API (e.g. «Россия», «Екатеринбург»). +-- * period_month — first day of the reported month (normalised from API timestamp). +-- * segment — realty type label from API (e.g. «Вторичный»). +-- * dashboard — slug used in the route (e.g. 'real_estate_deals'). +-- * PRIMARY KEY (city, period_month, dashboard) — one value per city×month×series; +-- upsert on conflict keeps the freshest fetched_at. +-- +-- Idempotent: CREATE TABLE IF NOT EXISTS — safe to re-apply. ADDITIVE only, no drops. +-- +-- Dependencies: none (standalone table). +-- +-- Deploy order: apply before deploying app/services/sber_index.py + +-- app/tasks/sber_index_pull.py + scheduler.py changes. + +BEGIN; + +CREATE TABLE IF NOT EXISTS sber_price_index ( + city text NOT NULL, + period_month date NOT NULL, + segment text NOT NULL DEFAULT '', + dashboard text NOT NULL, + index_value_rub_m2 double precision NOT NULL, + source text NOT NULL DEFAULT 'sberindex', + fetched_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (city, period_month, dashboard) +); + +COMMENT ON TABLE sber_price_index IS + 'Monthly city-level secondary-housing price index from СберИндекс (sberindex.ru). ' + 'Three series: real_estate_deals (hedonic sold), residential_real_estate_prices ' + '(repeat-sales hedonic), dinamika-tsen-obyavlenii (asking benchmark). ' + 'Populated by app/tasks/sber_index_pull.py via in-app scheduler (#887). ' + 'Estimator time-adjust wire is a separate backtest-gated follow-up (step 4, #887).'; + +COMMENT ON COLUMN sber_price_index.city IS + 'REF_AREA label from the API response (e.g. «Россия», «Екатеринбург»).'; +COMMENT ON COLUMN sber_price_index.period_month IS + 'First day of the reported month (normalised from API ISO timestamp).'; +COMMENT ON COLUMN sber_price_index.segment IS + 'Realty type label from API (e.g. «Вторичный»). Empty string when absent.'; +COMMENT ON COLUMN sber_price_index.dashboard IS + 'Dataset slug used in the /dataset/v1/ route.'; +COMMENT ON COLUMN sber_price_index.index_value_rub_m2 IS + 'Price index in rubles per square metre.'; +COMMENT ON COLUMN sber_price_index.fetched_at IS + 'Wall-clock timestamp when this row was last written/updated.'; + +-- Index for time-series queries (latest N months per city × dashboard) +CREATE INDEX IF NOT EXISTS idx_sber_price_index_city_dashboard_period + ON sber_price_index (city, dashboard, period_month DESC); + +COMMIT; diff --git a/tradein-mvp/backend/data/sql/093_scrape_schedules_seed_sber_index_pull.sql b/tradein-mvp/backend/data/sql/093_scrape_schedules_seed_sber_index_pull.sql new file mode 100644 index 00000000..e01073b7 --- /dev/null +++ b/tradein-mvp/backend/data/sql/093_scrape_schedules_seed_sber_index_pull.sql @@ -0,0 +1,73 @@ +-- 093_scrape_schedules_seed_sber_index_pull.sql +-- Issue #887 — seed scrape_schedules row for monthly СберИндекс price-index pull. +-- +-- Context: +-- app/services/sber_index.py fetches city-level secondary-housing price statistics +-- from sberindex.ru/api/sowa (three dashboards: real_estate_deals, +-- residential_real_estate_prices, dinamika-tsen-obyavlenii) and upserts rows into +-- sber_price_index (created in migration 092). +-- +-- The pull is a pure HTTP + DB operation — no scraper proxy, no anti-bot, no cookies. +-- The /api/sowa endpoint requires only the rquid header (32-hex UUID) and a +-- content-type: application/json header. Rate is low (3 series × 3 cities = 9 requests +-- per run, monthly data since 2017 — ~100 rows per series per city at most). +-- +-- Fix: +-- 1. services/sber_index.py — fetch_sber_index() + pull_sber_indices() service. +-- 2. tasks/sber_index_pull.py — run_sber_index_pull() task wrapper. +-- 3. scheduler.py — trigger_sber_index_pull_run() + dispatch branch +-- `elif source == "sber_index_pull"`. +-- 4. This migration — seeds the scrape_schedules row (enabled=true, monthly window). +-- +-- Schedule window 05:00-06:00 UTC: +-- Monthly cadence — runs roughly once a month. Placed after rosreestr_dkp_import +-- (04:00-06:00 UTC window) so both windows overlap in the early-morning batch slot. +-- The index is updated by СберИндекс once a month; daily pulls would be wasteful. +-- default_params.interval_days=28 mirrors the avito / yandex monthly approach +-- (scheduler re-computes next_run_at after each fire; 28 days ≈ monthly). +-- +-- next_run_at bootstrapped to tomorrow 05:00 UTC — scheduler will not fire immediately +-- on deploy (same pattern as 078/079/082/088/091). +-- +-- Idempotent: ON CONFLICT (source) DO NOTHING — safe to re-apply. +-- +-- Dependencies: +-- 052_scrape_schedules.sql (table + UNIQUE(source)). +-- 092_sber_price_index.sql (sber_price_index table must exist). +-- scheduler.py change (trigger_sber_index_pull_run must be deployed). +-- services/sber_index.py + tasks/sber_index_pull.py must be deployed. +-- +-- Deploy order: +-- Apply after deploying the scheduler.py + task + service changes so the scheduler +-- can dispatch 'sber_index_pull' correctly on first fire. + +BEGIN; + +INSERT INTO scrape_schedules ( + source, + enabled, + window_start_hour, + window_end_hour, + next_run_at, + default_params +) +VALUES +( + 'sber_index_pull', + true, -- SAFE: pure HTTP+DB, no anti-bot, low rate + 5, -- window 05:00-06:00 UTC + 6, + ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 5)) AT TIME ZONE 'UTC', + '{"interval_days": 28}'::jsonb -- monthly cadence (~28 days between runs) +) +ON CONFLICT (source) DO NOTHING; + +COMMENT ON TABLE scrape_schedules IS + 'In-app scheduler config (заменяет cron-script setup). ' + 'Sources: avito_city_sweep, yandex_city_sweep (dormant, #561), ' + 'cian_history_backfill, rosreestr_dkp_import, listing_source_snapshot (#570), ' + 'asking_to_sold_ratio_refresh (#648), refresh_search_matview (#769), ' + 'yandex_address_backfill (#855, EKB pilot), ' + 'sber_index_pull (#887, monthly СберИндекс city-level price index).'; + +COMMIT; diff --git a/tradein-mvp/backend/tests/test_sber_index.py b/tradein-mvp/backend/tests/test_sber_index.py new file mode 100644 index 00000000..f658ca40 --- /dev/null +++ b/tradein-mvp/backend/tests/test_sber_index.py @@ -0,0 +1,308 @@ +"""Tests for sber_index service — pure unit tests, no network/DB. + +Coverage: + (a) build_sber_route: base64 route builder produces correct encoded string. + (b) decode_sber_response: decodes a sample base64 payload into correct rows. + (c) pull_sber_indices upsert SQL is idempotent (ON CONFLICT shape assertion via + fake-db roundtrip — verifies upsert is called with correct SQL and params). +""" + +from __future__ import annotations + +import base64 +import json +import os +import sys +from datetime import date +from unittest.mock import MagicMock, patch +from urllib.parse import unquote + +# DATABASE_URL required by config before any app import. +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +# WeasyPrint stub — not installed in CI without GTK. +_wp_mock = MagicMock() +sys.modules.setdefault("weasyprint", _wp_mock) + +import pytest # noqa: E402 + +from app.services.sber_index import ( # noqa: E402 + _parse_period_month, + build_sber_route, + decode_sber_response, +) + +# --------------------------------------------------------------------------- +# (a) Route builder +# --------------------------------------------------------------------------- + + +def test_build_sber_route_rф_real_estate_deals() -> None: + """Route for РФ (643) + real_estate_deals should decode to expected path.""" + encoded = build_sber_route("real_estate_deals", "643") + decoded = base64.b64decode(encoded).decode("utf-8") + + assert decoded.startswith("/dataset/v1/real_estate_deals?filter=") + assert "limit=1000" in decoded + assert "offset=0" in decoded + + # Extract filter JSON from decoded path + filter_part = decoded.split("filter=")[1].split("&")[0] + filter_json = json.loads(unquote(filter_part)) + + assert filter_json["REF_AREA"] == "643" + assert filter_json["FREQ"] == "M" + assert filter_json["SOURCE"] == "SI" + assert filter_json["REALTY"] == "2" + + +def test_build_sber_route_different_dashboard() -> None: + """Route slug is correctly embedded in the decoded path.""" + encoded = build_sber_route("dinamika-tsen-obyavlenii", "643") + decoded = base64.b64decode(encoded).decode("utf-8") + assert "/dataset/v1/dinamika-tsen-obyavlenii?" in decoded + + +def test_build_sber_route_is_valid_base64() -> None: + """Encoded route must be decodeable ASCII base64.""" + encoded = build_sber_route("residential_real_estate_prices", "77") + # Should not raise + decoded_bytes = base64.b64decode(encoded) + assert len(decoded_bytes) > 0 + + +def test_build_sber_route_ref_area_embedded() -> None: + """REF_AREA value is preserved exactly in the filter JSON.""" + encoded = build_sber_route("real_estate_deals", "66") + decoded = base64.b64decode(encoded).decode("utf-8") + filter_part = decoded.split("filter=")[1].split("&")[0] + filter_json = json.loads(unquote(filter_part)) + assert filter_json["REF_AREA"] == "66" + + +# --------------------------------------------------------------------------- +# (b) Response decoder +# --------------------------------------------------------------------------- + + +def _make_b64_cell(value: str) -> str: + return "__string__" + base64.b64encode(value.encode("utf-8")).decode("ascii") + + +def _make_num_cell(value: float) -> str: + return f"__number__{value}" + + +def _make_sowa_response(rows: list[list[str]]) -> dict: # type: ignore[type-arg] + """Build a minimal SOWA response dict with a fixed field list.""" + fields = [ + "indicator_id", + "kpi_id", + "period", + "value", + "obs_status", + "source", + "ref_area", + "realty", + "freq", + "unit_measure", + "unit_mult", + ] + encoded_fields = [_make_b64_cell(f) for f in fields] + return { + "SOWA": { + "method": "GET", + "route": "test_route", + "data": { + "type": "object", + "value": [ + {"key": "fields", "value": encoded_fields}, + {"key": "data", "value": rows}, + ], + }, + } + } + + +def test_decode_sber_response_basic() -> None: + """Decode a two-row response and check period + value extraction.""" + row1 = [ + _make_b64_cell("1"), # indicator_id + _make_num_cell(643032001), # kpi_id + _make_b64_cell("2017-01-30T21:00:00Z"), # period + _make_num_cell(48549), # value + _make_b64_cell("A"), # obs_status + _make_b64_cell("Данные СберИндекса"), # source + _make_b64_cell("Россия"), # ref_area + _make_b64_cell("Вторичный"), # realty + _make_b64_cell("Месяц"), # freq + _make_b64_cell("руб"), # unit_measure + _make_num_cell(0), # unit_mult + ] + row2 = [ + _make_b64_cell("1"), + _make_num_cell(643032001), + _make_b64_cell("2024-06-30T21:00:00Z"), + _make_num_cell(120000), + _make_b64_cell("A"), + _make_b64_cell("Данные СберИндекса"), + _make_b64_cell("Россия"), + _make_b64_cell("Вторичный"), + _make_b64_cell("Месяц"), + _make_b64_cell("руб"), + _make_num_cell(0), + ] + payload = _make_sowa_response([row1, row2]) + decoded = decode_sber_response(payload) + + assert len(decoded) == 2 + assert decoded[0]["ref_area"] == "Россия" + assert decoded[0]["value"] == 48549.0 + assert decoded[0]["period"] == "2017-01-30T21:00:00Z" + assert decoded[1]["value"] == 120000.0 + + +def test_decode_sber_response_period_normalises_to_first_of_month() -> None: + """_parse_period_month always returns the first day of the month.""" + result = _parse_period_month("2017-01-30T21:00:00Z") + assert result == date(2017, 1, 1) + + result2 = _parse_period_month("2024-06-30T21:00:00Z") + assert result2 == date(2024, 6, 1) + + +def test_decode_sber_response_empty_data() -> None: + """Empty data section returns empty list.""" + payload = _make_sowa_response([]) + decoded = decode_sber_response(payload) + assert decoded == [] + + +def test_decode_sber_response_missing_fields_key_raises() -> None: + """Response without 'fields' key raises ValueError.""" + bad_payload = { + "SOWA": { + "method": "GET", + "route": "x", + "data": {"type": "object", "value": [{"key": "data", "value": []}]}, + } + } + with pytest.raises(ValueError, match="fields"): + decode_sber_response(bad_payload) + + +# --------------------------------------------------------------------------- +# (c) Upsert SQL idempotency — fake-db roundtrip +# --------------------------------------------------------------------------- + + +class _ExecuteCapture: + """Captures execute() calls and their params; simulate no-op DB.""" + + def __init__(self) -> None: + self.calls: list[tuple[str, dict]] = [] # type: ignore[type-arg] + + def execute(self, stmt: object, params: dict | None = None) -> MagicMock: # type: ignore[return-value] + sql_str = str(stmt) + self.calls.append((sql_str, params or {})) + return MagicMock(fetchone=lambda: None, fetchall=lambda: []) + + def commit(self) -> None: + pass + + +@pytest.mark.asyncio +async def test_pull_sber_indices_upsert_on_conflict_idempotent() -> None: + """pull_sber_indices emits ON CONFLICT ... DO UPDATE upsert SQL (idempotent).""" + from app.services.sber_index import pull_sber_indices + + fake_db = _ExecuteCapture() + fake_row = ("Россия", date(2024, 1, 1), "Вторичный", "real_estate_deals", 95000.0) + + # Patch fetch_sber_index to yield one deterministic row per call + async def _mock_fetch(client: object, *, dashboard: str, ref_area: str): # type: ignore[override] + yield fake_row + + with patch("app.services.sber_index.fetch_sber_index", side_effect=_mock_fetch): + # Only pull one dashboard × one city to keep assertion simple + result = await pull_sber_indices( + fake_db, # type: ignore[arg-type] + cities={"643": "Россия"}, + dashboards=["real_estate_deals"], + ) + + assert result["upserted"] == 1 + assert result["errors"] == 0 + + # Find the INSERT call + insert_calls = [sql for sql, _ in fake_db.calls if "INSERT INTO sber_price_index" in sql] + assert insert_calls, "Expected at least one INSERT INTO sber_price_index call" + + upsert_sql = insert_calls[0] + # Must contain ON CONFLICT clause with DO UPDATE SET + assert "ON CONFLICT" in upsert_sql + assert "DO UPDATE SET" in upsert_sql + assert "index_value_rub_m2" in upsert_sql + assert "fetched_at" in upsert_sql + + # Must NOT contain :: type casts (psycopg v3 rule) + import re + + assert not re.search( + r":[a-z_]+::[a-z]", upsert_sql + ), "SQL must not contain ::type casts — use CAST(... AS type) instead" + + +@pytest.mark.asyncio +async def test_pull_sber_indices_error_per_series_continues() -> None: + """A failing series increments errors but does not abort other series.""" + from app.services.sber_index import pull_sber_indices + + fake_db = _ExecuteCapture() + call_count = 0 + + async def _mock_fetch_with_error(client: object, *, dashboard: str, ref_area: str): # type: ignore[override] + nonlocal call_count + call_count += 1 + if dashboard == "real_estate_deals": + raise Exception("Simulated network failure") + yield ("Россия", date(2024, 1, 1), "Вторичный", dashboard, 90000.0) + + with patch("app.services.sber_index.fetch_sber_index", side_effect=_mock_fetch_with_error): + result = await pull_sber_indices( + fake_db, # type: ignore[arg-type] + cities={"643": "Россия"}, + dashboards=["real_estate_deals", "residential_real_estate_prices"], + ) + + # First series failed, second succeeded + assert result["errors"] == 1 + assert result["upserted"] == 1 + + +@pytest.mark.asyncio +async def test_pull_sber_indices_asking_benchmark_logged(caplog: pytest.LogCaptureFixture) -> None: + """After upserting dinamika-tsen-obyavlenii, a benchmark INFO line is logged.""" + import logging + + from app.services.sber_index import pull_sber_indices + + fake_db = _ExecuteCapture() + + async def _mock_fetch(client: object, *, dashboard: str, ref_area: str): # type: ignore[override] + yield ("Россия", date(2024, 6, 1), "Вторичный", dashboard, 115000.0) + + with patch("app.services.sber_index.fetch_sber_index", side_effect=_mock_fetch): + with caplog.at_level(logging.INFO, logger="app.services.sber_index"): + await pull_sber_indices( + fake_db, # type: ignore[arg-type] + cities={"643": "Россия"}, + dashboards=["dinamika-tsen-obyavlenii"], + ) + + assert any( + "benchmark" in record.message for record in caplog.records + ), "Expected a benchmark log line after upserting dinamika-tsen-obyavlenii" + assert any( + "115000" in record.message for record in caplog.records + ), "Benchmark log should include the latest asking index value"