From 2dd2d8e57e9baba5b95258c870a4a4e85903f3d1 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Sun, 31 May 2026 07:31:10 +0000 Subject: [PATCH] feat(tradein): schedule yandex address backfill task (#855) (#866) Co-authored-by: bot-backend Co-committed-by: bot-backend --- tradein-mvp/backend/app/services/scheduler.py | 45 ++++ .../app/tasks/yandex_address_backfill.py | 116 +++++++++++ ...schedules_seed_yandex_address_backfill.sql | 71 +++++++ .../tasks/test_yandex_address_backfill.py | 195 ++++++++++++++++++ 4 files changed, 427 insertions(+) create mode 100644 tradein-mvp/backend/app/tasks/yandex_address_backfill.py create mode 100644 tradein-mvp/backend/data/sql/091_scrape_schedules_seed_yandex_address_backfill.sql create mode 100644 tradein-mvp/backend/tests/tasks/test_yandex_address_backfill.py diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py index 8433ed74..413232a7 100644 --- a/tradein-mvp/backend/app/services/scheduler.py +++ b/tradein-mvp/backend/app/services/scheduler.py @@ -19,6 +19,11 @@ Sources: (tasks/asking_to_sold_ratio.py, #648 Stage 4; pure internal DB re-derivation of the asking→sold ratios — no external calls, enabled by default; window after rosreestr_dkp_import) + - yandex_address_backfill → run_yandex_address_backfill + (tasks/yandex_address_backfill.py, #855; EKB-pilot: fetches + yandex detail pages, extracts full address from , + updates listings.address for street-only addresses; + nightly 02:00-03:00 UTC, before refresh_search_matview) - deactivate_stale_avito → deactivate_stale_avito_listings (tasks/deactivate_stale_avito.py, #759; marks avito listings with last_seen_at > TTL days as is_active=false — no DELETE, @@ -412,6 +417,44 @@ async def _execute_cian_backfill( raise +async def trigger_yandex_address_backfill_run( + db: Session, schedule_row: dict[str, Any] +) -> int | None: + """Создать scrape_runs + launch run_yandex_address_backfill в asyncio.create_task. + + EKB-only pilot (#855): backfills listings.address for yandex listings that have + street-only addresses (no house number). Fetches detail pages via curl_cffi chrome120, + extracts full address from <title>, updates listings WHERE address differs. + + Mirrors trigger_cian_backfill_run: claim run → create_task → mark_done/failed. + executor logic lives in tasks/yandex_address_backfill.py which delegates to + services/yandex_address_backfill.py (backfill_yandex_addresses). + + 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.yandex_address_backfill import run_yandex_address_backfill + + await run_yandex_address_backfill(run_db, run_id=run_id, params=params) + except Exception: + logger.exception("scheduler: run_yandex_address_backfill 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 yandex_address_backfill run_id=%d", run_id) + return run_id + + async def trigger_rosreestr_dkp_run(db: Session, schedule_row: dict[str, Any]) -> int | None: """Создать scrape_runs + launch import_rosreestr_dkp в executor (sync I/O-bound task).""" run_id = _claim_run(db, schedule_row) @@ -868,6 +911,8 @@ async def scheduler_loop() -> None: await trigger_asking_to_sold_ratio_run(db, sch) elif source == "refresh_search_matview": await trigger_refresh_search_matview_run(db, sch) + elif source == "yandex_address_backfill": + await trigger_yandex_address_backfill_run(db, sch) elif source == "deactivate_stale_avito": await trigger_deactivate_stale_avito_run(db, sch) else: diff --git a/tradein-mvp/backend/app/tasks/yandex_address_backfill.py b/tradein-mvp/backend/app/tasks/yandex_address_backfill.py new file mode 100644 index 00000000..66e37f6d --- /dev/null +++ b/tradein-mvp/backend/app/tasks/yandex_address_backfill.py @@ -0,0 +1,116 @@ +"""Scheduled task: backfill listings.address for EKB yandex listings with street-only addresses. + +Pilot scope: Екатеринбург only (region_code=66, city hardcoded). +Yandex SERP cards include only the street name (e.g. «улица Горького»); the offer detail +page <title> contains the full address including house number: + «Продажа квартиры — Екатеринбург, улица Горького, 36 — id 7654321 на Яндекс.Недвижимости» + +This task selects active listings that match the predicate (source='yandex', region_code=66, +address IS NOT NULL, address has no house number, source_url IS NOT NULL), fetches each +detail page via curl_cffi(chrome120), extracts the full address from the <title>, and +UPDATE listings.address ONLY when the extracted value differs from the current one. + +Wired into the in-app scheduler as source='yandex_address_backfill'. Triggered nightly via +the scrape_schedules seed (migration 091). Manual trigger also available via POST +/api/v1/admin/scrape/yandex-address-backfill. + +Delegates actual HTTP + DB work to app.services.yandex_address_backfill.backfill_yandex_addresses +(the proven service module that already implements RE_TITLE_ADDRESS, _extract_address_from_title, +and _eligible_listings). +""" + +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.yandex_address_backfill import ( + YandexAddressBackfillResult, + backfill_yandex_addresses, +) + +logger = logging.getLogger(__name__) + +__all__ = [ + "YandexAddressBackfillTaskResult", + "run_yandex_address_backfill", +] + + +@dataclass +class YandexAddressBackfillTaskResult: + """Result counters from a scheduled backfill run.""" + + checked: int = 0 + saved: int = 0 + skipped: int = 0 + errors: int = 0 + duration_sec: float = field(default=0.0) + + +async def run_yandex_address_backfill( + db: Session, + *, + run_id: int, + params: dict, +) -> YandexAddressBackfillTaskResult: + """Execute Yandex address backfill with run lifecycle management. + + Wraps backfill_yandex_addresses(), emitting heartbeat before the batch call + and marking the scrape_run as done/failed on completion. + + Params (from default_params jsonb in scrape_schedules): + batch_size: int — max listings to process (default 200). + request_delay_sec: float — seconds between requests (default 3.0). + """ + batch_size = int(params.get("batch_size", 200)) + request_delay_sec = float(params.get("request_delay_sec", 3.0)) + + counters: dict[str, int] = { + "checked": 0, + "saved": 0, + "skipped": 0, + "errors": 0, + } + + try: + runs_mod.update_heartbeat(db, run_id, counters) + + result: YandexAddressBackfillResult = await backfill_yandex_addresses( + db, + limit=batch_size, + request_delay_sec=request_delay_sec, + ) + + counters = { + "checked": result.checked, + "saved": result.saved, + "skipped": result.skipped, + "errors": result.errors, + "duration_sec": int(result.duration_sec), + } + runs_mod.mark_done(db, run_id, counters) + logger.info( + "scheduler: yandex_address_backfill run_id=%d done — " + "checked=%d saved=%d skipped=%d errors=%d %.1fs", + run_id, + result.checked, + result.saved, + result.skipped, + result.errors, + result.duration_sec, + ) + return YandexAddressBackfillTaskResult( + checked=result.checked, + saved=result.saved, + skipped=result.skipped, + errors=result.errors, + duration_sec=result.duration_sec, + ) + except Exception as exc: + logger.exception("scheduler: yandex_address_backfill 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/091_scrape_schedules_seed_yandex_address_backfill.sql b/tradein-mvp/backend/data/sql/091_scrape_schedules_seed_yandex_address_backfill.sql new file mode 100644 index 00000000..3123e446 --- /dev/null +++ b/tradein-mvp/backend/data/sql/091_scrape_schedules_seed_yandex_address_backfill.sql @@ -0,0 +1,71 @@ +-- 091_scrape_schedules_seed_yandex_address_backfill.sql +-- Issue #855 — seed scrape_schedules row for nightly yandex address backfill. +-- +-- Context: +-- Active EKB yandex listings often have street-only addresses (e.g. «улица Горького») +-- because Yandex SERP cards omit the house number. The detail page <title> contains +-- the full address including house number: +-- «Продажа квартиры — Екатеринбург, улица Горького, 36 — id 7654321». +-- Without house numbers, geocoding falls back to street-centroid coordinates and +-- valuation analogs have lower match quality. +-- +-- Fix: +-- 1. services/yandex_address_backfill.py — backfill_yandex_addresses() service function +-- (curl_cffi chrome120, RE_TITLE_ADDRESS regex over <title>, UPDATE only-if-changed). +-- 2. tasks/yandex_address_backfill.py — run_yandex_address_backfill() task wrapper +-- (run lifecycle: heartbeat + mark_done/mark_failed). +-- 3. scheduler.py — trigger_yandex_address_backfill_run() + dispatch branch +-- `elif source == "yandex_address_backfill"` in scheduler_loop(). +-- 4. This migration — seeds the scrape_schedules row (enabled=true, 02:00-03:00 UTC). +-- +-- Schedule window 02:00-03:00 UTC: +-- Nightly, before refresh_search_matview (03:00-04:00 UTC). Runs after the yandex +-- city sweep (dormant/manual window 01:00-02:00 UTC) so backfill sees the freshest +-- is_active=TRUE rows. Expected duration: ~30 min for 200 listings at 3s delay. +-- +-- next_run_at bootstrapped to tomorrow 02:00 UTC so the scheduler does not fire +-- immediately on deploy (same pattern as 078/079/082/088). +-- +-- Idempotent: ON CONFLICT (source) DO NOTHING — safe to re-apply. +-- +-- Dependencies: +-- 052_scrape_schedules.sql (table + UNIQUE(source)). +-- scheduler.py change (trigger_yandex_address_backfill_run must be deployed). +-- services/yandex_address_backfill.py (backfill_yandex_addresses must be deployed). +-- +-- Deploy order: +-- Apply this migration after deploying the scheduler.py + task changes so the +-- scheduler can dispatch the new source name correctly on first fire. +-- +-- EKB pilot scope: only listings with region_code=66 are processed. Expanding to +-- other cities requires parametrizing the city/region_code in backfill_yandex_addresses. + +BEGIN; + +INSERT INTO scrape_schedules ( + source, + enabled, + window_start_hour, + window_end_hour, + next_run_at, + default_params +) +VALUES +( + 'yandex_address_backfill', + true, -- EKB pilot; HTTP calls (curl_cffi chrome120) + 2, -- window 02:00-03:00 UTC + 3, + ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 2)) AT TIME ZONE 'UTC', + '{"batch_size": 200, "request_delay_sec": 3.0}'::jsonb +) +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).'; + +COMMIT; diff --git a/tradein-mvp/backend/tests/tasks/test_yandex_address_backfill.py b/tradein-mvp/backend/tests/tasks/test_yandex_address_backfill.py new file mode 100644 index 00000000..3340fa86 --- /dev/null +++ b/tradein-mvp/backend/tests/tasks/test_yandex_address_backfill.py @@ -0,0 +1,195 @@ +"""Tests for yandex_address_backfill — pure regex extraction, no network/DB.""" + +from __future__ import annotations + +import os +import re +import sys +from unittest.mock import AsyncMock, MagicMock, patch + +# 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.yandex_address_backfill import ( # noqa: E402 + YandexAddressBackfillResult, + _extract_address_from_title, +) + +# Regex used to assert that an extracted address contains a house number. +_RE_HAS_HOUSE_NUMBER = re.compile(r",\s*\d+") + + +# --------------------------------------------------------------------------- +# Pure regex / extraction tests (no network, no DB) +# --------------------------------------------------------------------------- + + +def test_extract_address_standard_title() -> None: + """Standard title: «Продажа квартиры — Екатеринбург, улица Горького, 36 — id 7654321».""" + html = ( + "<title>Продажа квартиры — Екатеринбург, улица Горького, 36 " + "— id 7654321 на Яндекс.Недвижимости" + ) + addr = _extract_address_from_title(html) + assert addr is not None, "Should extract address from standard title" + assert _RE_HAS_HOUSE_NUMBER.search( + addr + ), f"Extracted address should contain house number: {addr!r}" + assert "Екатеринбург" in addr + assert "Горького" in addr + assert "36" in addr + + +def test_extract_address_with_zhk_prefix() -> None: + """Title with optional ЖК prefix before city: «— ЖК Каменный Ручей, Екатеринбург, …».""" + html = ( + "Продажа квартиры — ЖК Каменный Ручей, " + "Екатеринбург, проспект Космонавтов, 110А — id 99001 " + ) + addr = _extract_address_from_title(html) + assert addr is not None, "Should extract address when ЖК prefix is present" + assert _RE_HAS_HOUSE_NUMBER.search( + addr + ), f"Extracted address should contain house number: {addr!r}" + assert "Екатеринбург" in addr + + +def test_extract_address_nbsp_replaced() -> None: + """\\xa0 (non-breaking space) in title should be normalised before regex match.""" + html = ( + "Продажа\xa0квартиры\xa0—\xa0Екатеринбург,\xa0улица\xa0Ленина,\xa012" + "\xa0—\xa0id\xa0123456" + ) + addr = _extract_address_from_title(html) + assert addr is not None, "\\xa0 should be replaced before regex" + assert _RE_HAS_HOUSE_NUMBER.search( + addr + ), f"Extracted address should contain house number: {addr!r}" + + +def test_extract_address_no_title_returns_none() -> None: + """HTML without a matching returns None.""" + html = "<html><body>Объявление снято</body></html>" + addr = _extract_address_from_title(html) + assert addr is None + + +def test_extract_address_title_without_id_returns_none() -> None: + """Title that matches city but has no '— id NNN' sentinel returns None.""" + html = "<title>Екатеринбург, улица Горького, 36" + addr = _extract_address_from_title(html) + assert addr is None + + +def test_extract_address_street_only_still_extracted() -> None: + """Title with street-only (no house number) is extracted but fails the house-number check. + + The extractor itself does not gate on house number presence — the caller + (backfill_yandex_addresses) checks _RE_HAS_HOUSE_NUMBER separately and skips. + """ + html = "Продажа квартиры — Екатеринбург, улица Горького — id 7654321" + addr = _extract_address_from_title(html) + # May or may not match depending on whether «улица Горького» satisfies the regex — + # the key assertion is that IF it does, it does NOT contain a house number. + if addr is not None: + assert not _RE_HAS_HOUSE_NUMBER.search( + addr + ), "Street-only extracted address should NOT contain a house number" + + +# --------------------------------------------------------------------------- +# Task-level smoke tests (no real DB, no real HTTP) +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_run_yandex_address_backfill_no_eligible_listings() -> None: + """No eligible listings → run marked done with zeroed counters, no HTTP calls.""" + from app.tasks.yandex_address_backfill import run_yandex_address_backfill + + db = MagicMock() + + empty_result = YandexAddressBackfillResult( + checked=0, saved=0, skipped=0, errors=0, duration_sec=0.01 + ) + + with ( + patch( + "app.tasks.yandex_address_backfill.backfill_yandex_addresses", + new_callable=AsyncMock, + return_value=empty_result, + ) as mock_backfill, + patch("app.tasks.yandex_address_backfill.runs_mod") as mock_runs, + ): + result = await run_yandex_address_backfill(db, run_id=42, params={}) + + mock_backfill.assert_awaited_once() + mock_runs.update_heartbeat.assert_called_once_with( + db, 42, {"checked": 0, "saved": 0, "skipped": 0, "errors": 0} + ) + mock_runs.mark_done.assert_called_once() + mock_runs.mark_failed.assert_not_called() + + assert result.checked == 0 + assert result.saved == 0 + + +@pytest.mark.asyncio +async def test_run_yandex_address_backfill_saves_enriched_count() -> None: + """5 checked, 3 saved → result reflects correct counters.""" + from app.tasks.yandex_address_backfill import run_yandex_address_backfill + + db = MagicMock() + + backfill_result = YandexAddressBackfillResult( + checked=5, saved=3, skipped=1, errors=1, duration_sec=15.0 + ) + + with ( + patch( + "app.tasks.yandex_address_backfill.backfill_yandex_addresses", + new_callable=AsyncMock, + return_value=backfill_result, + ), + patch("app.tasks.yandex_address_backfill.runs_mod") as mock_runs, + ): + result = await run_yandex_address_backfill( + db, run_id=99, params={"batch_size": 100, "request_delay_sec": 2.0} + ) + + mock_runs.mark_done.assert_called_once() + done_counters = mock_runs.mark_done.call_args[0][2] # positional: (db, run_id, counters) + assert done_counters["checked"] == 5 + assert done_counters["saved"] == 3 + assert done_counters["errors"] == 1 + + assert result.saved == 3 + assert result.errors == 1 + + +@pytest.mark.asyncio +async def test_run_yandex_address_backfill_marks_failed_on_exception() -> None: + """backfill_yandex_addresses raises → mark_failed called, exception re-raised.""" + from app.tasks.yandex_address_backfill import run_yandex_address_backfill + + db = MagicMock() + + with ( + patch( + "app.tasks.yandex_address_backfill.backfill_yandex_addresses", + new_callable=AsyncMock, + side_effect=RuntimeError("network timeout"), + ), + patch("app.tasks.yandex_address_backfill.runs_mod") as mock_runs, + ): + with pytest.raises(RuntimeError, match="network timeout"): + await run_yandex_address_backfill(db, run_id=7, params={}) + + mock_runs.mark_failed.assert_called_once() + mock_runs.mark_done.assert_not_called()