feat(tradein): schedule yandex address backfill task (#855)

The backfill service (app/services/yandex_address_backfill.py) + its manual
admin endpoint already exist, but it was never wired into the in-app scheduler
— so residual street-only yandex listings (no house number) were only fixed
on manual trigger. Add a recurring scheduler task: trigger_yandex_address_backfill_run
mirrors trigger_cian_backfill_run (claim run + mark done/failed), new task
wrapper app/tasks/yandex_address_backfill.py delegates to the existing
backfill_yandex_addresses(), dispatch branch + seed migration 091 (window
02:00-03:00 UTC, ON CONFLICT(source) idempotent). EKB-only (region_code=66).

Refs #855, #726

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 <title>,
+                              updates listings.address for street-only addresses;
+                              nightly 02:00-03:00 UTC, before refresh_search_matview)
 """
 
 from __future__ import annotations
@@ -408,6 +413,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)
@@ -829,6 +872,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)
                     else:
                         logger.warning("scheduler: unknown source=%s, skip", source)
             finally:

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

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;

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 на Яндекс.Недвижимости</title>"
+    )
+    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 = (
+        "<title>Продажа квартиры — ЖК Каменный Ручей, "
+        "Екатеринбург, проспект Космонавтов, 110А — id 99001 </title>"
+    )
+    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 = (
+        "<title>Продажа\xa0квартиры\xa0—\xa0Екатеринбург,\xa0улица\xa0Ленина,\xa012"
+        "\xa0—\xa0id\xa0123456</title>"
+    )
+    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 <title> 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</title>"
+    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 = "<title>Продажа квартиры — Екатеринбург, улица Горького — id 7654321</title>"
+    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()