diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py
index fbff867c..cf83e05c 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)
"""
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:
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()