feat(scrapers): yandex detail-enrichment — save + ночной backfill через прокси (part of #1553) #1554

Merged
bot-backend merged 3 commits from feat/yandex-detail-backfill into main 2026-06-16 11:18:19 +00:00
5 changed files with 941 additions and 0 deletions

View file

@ -56,6 +56,12 @@ Sources:
всех источников с NULL coords; address-dedup batch loop с всех источников с NULL coords; address-dedup batch loop с
wall-clock budget; window 06:00-09:00 UTC после city sweeps wall-clock budget; window 06:00-09:00 UTC после city sweeps
~03-04 UTC и deals OS-cron 05:30/05:45 UTC) ~03-04 UTC и deals OS-cron 05:30/05:45 UTC)
- yandex_detail_backfill run_yandex_detail_backfill
(tasks/yandex_detail_backfill.py, #1553; nightly backfill of
yandex detail_enriched_at for ~3952 listings never enriched by
city_sweep; fetch via curl_cffi chrome120 + scraper_proxy_url,
parse via YandexDetailScraper.parse;
window 12:00-15:00 UTC после avito_detail_backfill 09-12 UTC)
""" """
from __future__ import annotations from __future__ import annotations
@ -1207,6 +1213,43 @@ async def trigger_avito_detail_backfill_run(
return run_id return run_id
async def trigger_yandex_detail_backfill_run(
db: Session, schedule_row: dict[str, Any]
) -> int | None:
"""Create scrape_run + launch run_yandex_detail_backfill in asyncio.create_task.
Nightly backfill detail_enriched_at for legacy yandex listings (#1553).
~3952 yandex listings have detail_enriched_at IS NULL -- SERP-only data,
never enriched by YandexDetailScraper.
Mirrors trigger_avito_detail_backfill_run: claim run -> create_task ->
mark_done/failed delegated to tasks/yandex_detail_backfill.run_yandex_detail_backfill.
Returns run_id or 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_detail_backfill import run_yandex_detail_backfill
await run_yandex_detail_backfill(run_db, run_id=run_id, params=params)
except Exception:
logger.exception("scheduler: run_yandex_detail_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_detail_backfill run_id=%d", run_id)
return run_id
def get_due_schedules(db: Session) -> list[dict[str, Any]]: def get_due_schedules(db: Session) -> list[dict[str, Any]]:
"""SELECT scrape_schedules WHERE enabled AND (next_run_at IS NULL OR next_run_at <= NOW()).""" """SELECT scrape_schedules WHERE enabled AND (next_run_at IS NULL OR next_run_at <= NOW())."""
rows = ( rows = (
@ -1276,6 +1319,8 @@ async def scheduler_loop() -> None:
await trigger_geocode_missing_listings_run(db, sch) await trigger_geocode_missing_listings_run(db, sch)
elif source == "avito_detail_backfill": elif source == "avito_detail_backfill":
await trigger_avito_detail_backfill_run(db, sch) await trigger_avito_detail_backfill_run(db, sch)
elif source == "yandex_detail_backfill":
await trigger_yandex_detail_backfill_run(db, sch)
else: else:
logger.warning("scheduler: unknown source=%s, skip", source) logger.warning("scheduler: unknown source=%s, skip", source)
finally: finally:

View file

@ -7,6 +7,7 @@ a DetailEnrichment Pydantic model. Used by enrichment pipeline
from __future__ import annotations from __future__ import annotations
import json
import logging import logging
import re import re
from datetime import date from datetime import date
@ -14,6 +15,8 @@ from typing import Any
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
from selectolax.parser import HTMLParser, Node from selectolax.parser import HTMLParser, Node
from sqlalchemy import text
from sqlalchemy.orm import Session
from app.services.scraper_settings import get_scraper_delay from app.services.scraper_settings import get_scraper_delay
from app.services.scrapers.base import BaseScraper from app.services.scrapers.base import BaseScraper
@ -364,3 +367,106 @@ def _extract_seller_name(summary_text: str, agency_name: str | None) -> str | No
# last short token sequence (likely "Имя Фамилия") # last short token sequence (likely "Имя Фамилия")
m = re.findall(r"([А-ЯЁ][а-яё]+(?:\s+[А-ЯЁ][а-яё]+){1,2})", head) m = re.findall(r"([А-ЯЁ][а-яё]+(?:\s+[А-ЯЁ][а-яё]+){1,2})", head)
return m[-1] if m else None return m[-1] if m else None
# ── Save helper ───────────────────────────────────────────────────────────────
def save_detail_enrichment(db: Session, listing_id: int, e: DetailEnrichment) -> bool:
"""Persist Yandex DetailEnrichment to listings row.
UPDATE listings SET <col> = COALESCE(:val, <col>), ..., detail_enriched_at = NOW()
WHERE id = listing_id.
COALESCE semantics: keeps existing non-NULL value if new value is NULL (never
overwrites a populated column with NULL). area_m2 from detail is more accurate
than SERP, but COALESCE preserves SERP value if detail returns NULL acceptable.
Returns True if the UPDATE touched at least one row (listing_id found in DB).
"""
metro_json: str | None = None
if e.metro_stations:
metro_json = json.dumps(
[s.model_dump(exclude_none=True) for s in e.metro_stations],
ensure_ascii=False,
)
photo_json: str | None = None
if e.photo_urls:
photo_json = json.dumps(e.photo_urls, ensure_ascii=False)
result = db.execute(
text("""
UPDATE listings SET
rooms = COALESCE(CAST(:rooms AS int), rooms),
area_m2 = COALESCE(CAST(:area_m2 AS numeric), area_m2),
floor = COALESCE(CAST(:floor AS int), floor),
total_floors = COALESCE(CAST(:total_floors AS int), total_floors),
address = COALESCE(CAST(:address AS text), address),
description = COALESCE(CAST(:description AS text), description),
repair_state = COALESCE(CAST(:repair_state AS text),repair_state),
publish_date = COALESCE(CAST(:publish_date AS date),publish_date),
views_total_yandex = COALESCE(CAST(:views_total AS int), views_total_yandex),
publish_date_relative= COALESCE(
CAST(:pub_date_rel AS text),
publish_date_relative
),
agency_name = COALESCE(CAST(:agency_name AS text), agency_name),
agency_founded_year = COALESCE(
CAST(:agency_founded_year AS int),
agency_founded_year
),
agency_objects_count = COALESCE(
CAST(:agency_objects_count AS int),
agency_objects_count
),
metro_stations = COALESCE(
CAST(:metro_stations AS jsonb),
metro_stations
),
photo_urls = COALESCE(
CAST(:photo_urls AS jsonb),
photo_urls
),
newbuilding_url = COALESCE(
CAST(:newbuilding_url AS text),
newbuilding_url
),
newbuilding_id = COALESCE(
CAST(:newbuilding_id AS text),
newbuilding_id
),
detail_enriched_at = NOW()
WHERE id = CAST(:listing_id AS bigint)
"""),
{
"listing_id": listing_id,
"rooms": e.rooms,
"area_m2": e.area_m2,
"floor": e.floor,
"total_floors": e.total_floors,
"address": e.address,
"description": e.description,
"repair_state": e.repair_state,
"publish_date": e.publish_date,
"views_total": e.views_total,
"pub_date_rel": e.publish_date_relative,
"agency_name": e.agency_name,
"agency_founded_year": e.agency_founded_year,
"agency_objects_count": e.agency_objects_count,
"metro_stations": metro_json,
"photo_urls": photo_json,
"newbuilding_url": e.newbuilding_url,
"newbuilding_id": e.newbuilding_id,
},
)
db.commit()
saved = (result.rowcount or 0) > 0
logger.info(
"yandex detail enrichment saved listing_id=%s (metro=%d photos=%d saved=%s)",
listing_id,
len(e.metro_stations),
len(e.photo_urls),
saved,
)
return saved

View file

@ -0,0 +1,301 @@
"""Scheduled backfill: detail-enrichment for legacy yandex listings (#1553).
Nightly window 12:00-15:00 UTC (migration 113, source=yandex_detail_backfill).
Offset from avito_detail_backfill (09-12 UTC) to avoid parallel egress on shared IP.
Problem: ~3952 yandex listings have detail_enriched_at IS NULL.
Yandex city sweep does not call YandexDetailScraper SERP data only.
area_m2 coverage 23%, living/kitchen 0%, repair_state 1% on prod.
Solution: single snapshot SELECT at start (guarantees termination), fetch each
offer detail page via curl_cffi AsyncSession (chrome120 + proxy) mirrors
yandex_address_backfill.py which already gets full HTML from Yandex on prod.
Parse HTML via YandexDetailScraper.parse (pure, no network). Persist via
save_detail_enrichment. Track consecutive parseNone results; abort after
max_consecutive_blocks (mark_done, not mark_failed retry next night via
NULL detail_enriched_at).
Why curl_cffi and not YandexDetailScraper.fetch_detail:
fetch_detail uses BaseScraper._http_get (plain httpx, no proxy, no TLS
fingerprinting). On datacenter IPs Yandex returns captcha / shell-HTML
parse always returns None backfill would be 0% effective. The
curl_cffi path (chrome120 impersonation + mobile proxy) is already proven
by yandex_address_backfill, which fetches identical offer detail pages.
"""
from __future__ import annotations
import asyncio
import logging
import time
from dataclasses import dataclass, field
from curl_cffi.requests import AsyncSession
from sqlalchemy import text
from sqlalchemy.orm import Session
from app.core.config import settings
from app.services import scrape_runs as runs_mod
from app.services.scrapers.yandex_detail import YandexDetailScraper, save_detail_enrichment
logger = logging.getLogger(__name__)
__all__ = [
"YandexDetailBackfillResult",
"run_yandex_detail_backfill",
]
@dataclass
class YandexDetailBackfillResult:
"""Counters for one yandex detail backfill run."""
attempted: int = 0
enriched: int = 0
failed: int = 0
duration_sec: float = field(default=0.0)
def to_dict(self) -> dict[str, int]:
return {
"attempted": self.attempted,
"enriched": self.enriched,
"failed": self.failed,
"duration_sec": int(self.duration_sec),
}
async def run_yandex_detail_backfill(
db: Session,
*,
run_id: int,
params: dict,
) -> YandexDetailBackfillResult:
"""Backfill detail_enriched_at for legacy yandex listings via curl_cffi + parse.
Params (from default_params jsonb in scrape_schedules):
batch_size: int -- snapshot size (SELECT LIMIT), default 800.
budget_sec: float -- wall-clock budget per run, default 3600s.
request_delay_sec: float -- delay between listings, default 5s.
max_consecutive_blocks: int -- consecutive parseNone before abort, default 5.
Fetch mechanism:
curl_cffi AsyncSession(impersonate="chrome120") + scraper_proxy_url mirrors
yandex_address_backfill. On HTTP 200: pass resp.text to
YandexDetailScraper().parse(html, offer_url). parseNone counts as a fail
(possible captcha wall); consecutive None abort after max_consecutive_blocks.
Lifecycle: update_heartbeat -> snapshot -> loop with budget guard ->
mark_done (incl. partial / consecutive-None abort) / mark_failed (exception only).
"""
batch_size = int(params.get("batch_size", 800))
budget_sec = float(params.get("budget_sec", 3600))
request_delay_sec = float(params.get("request_delay_sec", 5.0))
max_consecutive_blocks = int(params.get("max_consecutive_blocks", 5))
counters = YandexDetailBackfillResult()
current_counters: dict[str, int] = counters.to_dict()
start = time.monotonic()
try:
runs_mod.update_heartbeat(db, run_id, current_counters)
# SNAPSHOT: single SELECT at start -- NOT re-selected in loop.
# Priority: is_active DESC (active first), scraped_at DESC (newest first).
snapshot = (
db.execute(
text(
"""
SELECT id, source_url
FROM listings
WHERE source = 'yandex'
AND detail_enriched_at IS NULL
AND source_url IS NOT NULL
ORDER BY is_active DESC NULLS LAST, scraped_at DESC NULLS LAST
LIMIT CAST(:batch_size AS int)
"""
),
{"batch_size": batch_size},
)
.mappings()
.all()
)
if not snapshot:
logger.info(
"yandex_detail_backfill: run_id=%d -- no pending listings "
"(detail_enriched_at IS NULL = 0), done",
run_id,
)
runs_mod.mark_done(db, run_id, current_counters)
return counters
logger.info(
"yandex_detail_backfill: run_id=%d snapshot=%d (budget=%.0fs "
"delay=%.1fs max_consecutive_none=%d)",
run_id,
len(snapshot),
budget_sec,
request_delay_sec,
max_consecutive_blocks,
)
# Build proxies dict once — mirrors yandex_address_backfill.py
_proxy = settings.scraper_proxy_url
_proxies = {"http": _proxy, "https": _proxy} if _proxy else None
consecutive_none = 0
do_sleep = False
scraper = YandexDetailScraper()
async with AsyncSession(
impersonate="chrome120",
timeout=30.0,
proxies=_proxies,
headers={
"Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8",
},
) as session:
for idx, row in enumerate(snapshot):
# Budget guard
elapsed = time.monotonic() - start
if elapsed > budget_sec:
logger.info(
"yandex_detail_backfill: run_id=%d -- budget %.0fs exhausted "
"(elapsed=%.1fs), stopping at #%d/%d",
run_id,
budget_sec,
elapsed,
idx,
len(snapshot),
)
break
# Delay before each request except the first
if do_sleep:
await asyncio.sleep(request_delay_sec)
do_sleep = True
source_url: str = row["source_url"]
listing_id: int = row["id"]
counters.attempted += 1
try:
try:
resp = await session.get(source_url, allow_redirects=True)
except Exception as fetch_exc:
consecutive_none += 1
counters.failed += 1
logger.warning(
"yandex_detail_backfill: run_id=%d listing_id=%d "
"fetch error (consecutive=%d): %s",
run_id,
listing_id,
consecutive_none,
fetch_exc,
)
if consecutive_none >= max_consecutive_blocks:
logger.error(
"yandex_detail_backfill: run_id=%d ABORT -- %d consecutive "
"errors. enriched=%d attempted=%d",
run_id,
consecutive_none,
counters.enriched,
counters.attempted,
)
break
continue
if resp.status_code != 200:
consecutive_none += 1
counters.failed += 1
logger.warning(
"yandex_detail_backfill: run_id=%d listing_id=%d "
"HTTP %d (consecutive=%d)",
run_id,
listing_id,
resp.status_code,
consecutive_none,
)
if consecutive_none >= max_consecutive_blocks:
logger.error(
"yandex_detail_backfill: run_id=%d ABORT -- %d consecutive "
"non-200 responses. enriched=%d attempted=%d",
run_id,
consecutive_none,
counters.enriched,
counters.attempted,
)
break
continue
enrichment = scraper.parse(resp.text, offer_url=source_url)
if enrichment is None:
# parse→None: captcha wall / shell-HTML / no JSON-LD.
# Do not mark listing as done — retry next night.
consecutive_none += 1
counters.failed += 1
logger.warning(
"yandex_detail_backfill: run_id=%d listing_id=%d source_url=%s "
"-> parse None (consecutive=%d)",
run_id,
listing_id,
source_url,
consecutive_none,
)
if consecutive_none >= max_consecutive_blocks:
logger.error(
"yandex_detail_backfill: run_id=%d ABORT -- %d consecutive "
"parse-None results (captcha wall?). enriched=%d attempted=%d",
run_id,
consecutive_none,
counters.enriched,
counters.attempted,
)
break
continue
consecutive_none = 0
if save_detail_enrichment(db, listing_id, enrichment):
counters.enriched += 1
except Exception as exc:
counters.failed += 1
logger.warning(
"yandex_detail_backfill: save/iteration error for listing_id=%d: %s",
listing_id,
exc,
)
try:
db.rollback()
except Exception:
pass
if counters.attempted % 25 == 0:
current_counters = counters.to_dict()
runs_mod.update_heartbeat(db, run_id, current_counters)
counters.duration_sec = time.monotonic() - start
current_counters = counters.to_dict()
runs_mod.mark_done(db, run_id, current_counters)
logger.info(
"yandex_detail_backfill: run_id=%d DONE -- attempted=%d enriched=%d "
"failed=%d duration=%.1fs",
run_id,
counters.attempted,
counters.enriched,
counters.failed,
counters.duration_sec,
)
return counters
except Exception as exc:
counters.duration_sec = time.monotonic() - start
logger.exception(
"yandex_detail_backfill: run_id=%d FAILED after %.1fs",
run_id,
counters.duration_sec,
)
runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters.to_dict())
raise

View file

@ -0,0 +1,49 @@
-- Migration 113: yandex detail backfill — новые колонки + seed расписания.
--
-- Добавляет в listings три колонки, которых нет ни в 002_core_tables.sql ни в более ранних
-- migrations, но которые парсит YandexDetailScraper.DetailEnrichment:
--
-- photo_urls jsonb -- список URL фото из Product JSON-LD image[]
-- newbuilding_url text -- ссылка на ЖК (/kupit/novostrojka/...) если есть
-- newbuilding_id text -- id ЖК из URL (числовая строка)
--
-- Уже существующие колонки (не дублируем):
-- rooms, area_m2, floor, total_floors, address, description, repair_state,
-- publish_date, metro_stations, detail_enriched_at -- 002/011_*.sql
-- views_total_yandex, publish_date_relative, agency_name,
-- agency_founded_year, agency_objects_count -- 033_listings_alter_yandex.sql
--
-- Seed scrape_schedules: source='yandex_detail_backfill', window 12-15 UTC,
-- разнесено с avito_detail_backfill (09-12 UTC) чтобы не делить трафик.
-- ON CONFLICT DO NOTHING — идемпотентно.
BEGIN;
-- ── listings — новые колонки для detail enrichment ────────────────────────────
ALTER TABLE listings
ADD COLUMN IF NOT EXISTS photo_urls jsonb, -- [url, ...] из Product JSON-LD image[]
ADD COLUMN IF NOT EXISTS newbuilding_url text, -- ссылка на ЖК (/kupit/novostrojka/...)
ADD COLUMN IF NOT EXISTS newbuilding_id text; -- id ЖК из URL (числовая строка)
-- ── scrape_schedules — seed yandex_detail_backfill ────────────────────────────
INSERT INTO scrape_schedules
(source, enabled, window_start_hour, window_end_hour, next_run_at, default_params)
VALUES (
'yandex_detail_backfill',
true,
12, -- 12:00 UTC = 15:00 МСК
15, -- 15:00 UTC = 18:00 МСК
(NOW() + INTERVAL '1 day')::date + TIME '12:00:00' + INTERVAL '0 second',
'{"batch_size": 800, "budget_sec": 3600, "request_delay_sec": 5, "max_consecutive_blocks": 5}'::jsonb
)
ON CONFLICT (source) DO NOTHING;
COMMENT ON TABLE scrape_schedules IS
'Периодические задачи: avito/yandex/n1/cian city sweep, rosreestr import, '
'listing snapshot, ratio refresh, geocode backfill, deactivate stale, '
'sber/rosreestr quarter poll, newbuilding enrich, yandex newbuilding sweep, '
'avito detail backfill, yandex detail backfill.';
COMMIT;

View file

@ -0,0 +1,440 @@
"""Tests for app.tasks.yandex_detail_backfill.
Fetch mechanism changed: curl_cffi AsyncSession(chrome120+proxy) + YandexDetailScraper.parse
(instead of old fetch_detail path). Mocks target session.get and scraper.parse.
"""
from __future__ import annotations
import json
import os
import sys
from unittest.mock import AsyncMock, MagicMock, patch
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
_wp_mock = MagicMock()
sys.modules.setdefault("weasyprint", _wp_mock)
import pytest # noqa: E402
from app.tasks.yandex_detail_backfill import ( # noqa: E402
YandexDetailBackfillResult,
run_yandex_detail_backfill,
)
# ---------------------------------------------------------------------------
# Path constants for patching
# ---------------------------------------------------------------------------
_ASYNC_SESSION = "app.tasks.yandex_detail_backfill.AsyncSession"
_PARSE = "app.tasks.yandex_detail_backfill.YandexDetailScraper.parse"
_SAVE = "app.tasks.yandex_detail_backfill.save_detail_enrichment"
_RUNS = "app.tasks.yandex_detail_backfill.runs_mod"
_SLEEP = "app.tasks.yandex_detail_backfill.asyncio.sleep"
_SETTINGS = "app.tasks.yandex_detail_backfill.settings"
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_snapshot(n: int) -> list[dict]:
return [
{"id": i + 1, "source_url": f"https://realty.yandex.ru/offer/{i + 1}/"} for i in range(n)
]
def _mock_db(snapshot: list[dict]) -> MagicMock:
"""Fake Session: first execute() returns snapshot via .mappings().all()."""
db = MagicMock()
sel = MagicMock()
sel.mappings.return_value.all.return_value = snapshot
db.execute.return_value = sel
return db
def _make_resp(status: int = 200, text: str = "<html>ok</html>") -> MagicMock:
"""Fake curl_cffi response."""
resp = MagicMock()
resp.status_code = status
resp.text = text
return resp
def _make_session_ctx(get_side_effect) -> MagicMock:
"""Build AsyncSession context-manager mock with session.get side_effect."""
session = AsyncMock()
session.get = AsyncMock(side_effect=get_side_effect)
ctx = MagicMock()
ctx.__aenter__ = AsyncMock(return_value=session)
ctx.__aexit__ = AsyncMock(return_value=None)
session_cls = MagicMock(return_value=ctx)
return session_cls, session
def _mock_settings(proxy: str | None = "http://proxy:3128") -> MagicMock:
s = MagicMock()
s.scraper_proxy_url = proxy
return s
# ---------------------------------------------------------------------------
# Tests: run_yandex_detail_backfill
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_backfill_empty_snapshot_marks_done() -> None:
"""Empty snapshot -> mark_done immediately, no session.get calls."""
db = _mock_db([])
runs = MagicMock()
session_cls, session = _make_session_ctx([])
with (
patch(_ASYNC_SESSION, session_cls),
patch(_RUNS, runs),
patch(_SETTINGS, _mock_settings()),
):
result = await run_yandex_detail_backfill(
db, run_id=1, params={"batch_size": 10, "budget_sec": 60}
)
assert isinstance(result, YandexDetailBackfillResult)
assert result.attempted == 0
assert result.enriched == 0
session.get.assert_not_called()
runs.mark_done.assert_called_once()
runs.mark_failed.assert_not_called()
@pytest.mark.asyncio
async def test_backfill_processes_snapshot_to_completion() -> None:
"""3 listings -> all fetched (200) + parse success -> enriched=3, mark_done."""
snapshot = _make_snapshot(3)
db = _mock_db(snapshot)
runs = MagicMock()
mock_enrichment = MagicMock()
responses = [_make_resp(200), _make_resp(200), _make_resp(200)]
session_cls, session = _make_session_ctx(responses)
with (
patch(_ASYNC_SESSION, session_cls),
patch(_PARSE, return_value=mock_enrichment),
patch(_RUNS, runs),
patch(_SAVE, return_value=True),
patch(_SLEEP, new_callable=AsyncMock),
patch(_SETTINGS, _mock_settings()),
):
result = await run_yandex_detail_backfill(
db, run_id=2, params={"batch_size": 10, "budget_sec": 3600}
)
assert result.attempted == 3
assert result.enriched == 3
assert result.failed == 0
assert session.get.call_count == 3
runs.mark_done.assert_called_once()
runs.mark_failed.assert_not_called()
@pytest.mark.asyncio
async def test_backfill_parse_none_abort_after_max_consecutive() -> None:
"""5 consecutive parse→None results -> abort, mark_done (NOT mark_failed)."""
snapshot = _make_snapshot(10)
db = _mock_db(snapshot)
runs = MagicMock()
responses = [_make_resp(200)] * 10
session_cls, _session = _make_session_ctx(responses)
with (
patch(_ASYNC_SESSION, session_cls),
patch(_PARSE, return_value=None),
patch(_RUNS, runs),
patch(_SLEEP, new_callable=AsyncMock),
patch(_SETTINGS, _mock_settings()),
):
result = await run_yandex_detail_backfill(
db,
run_id=3,
params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 5},
)
assert result.attempted == 5
assert result.failed == 5
assert result.enriched == 0
runs.mark_done.assert_called_once()
runs.mark_failed.assert_not_called()
@pytest.mark.asyncio
async def test_backfill_parse_none_resets_on_success() -> None:
"""1 parse→None + 1 success + 1 parse→None -> consecutive resets, no abort."""
snapshot = _make_snapshot(3)
db = _mock_db(snapshot)
runs = MagicMock()
mock_enrichment = MagicMock()
responses = [_make_resp(200)] * 3
session_cls, _session = _make_session_ctx(responses)
parse_results = [None, mock_enrichment, None]
with (
patch(_ASYNC_SESSION, session_cls),
patch(_PARSE, side_effect=parse_results),
patch(_RUNS, runs),
patch(_SAVE, return_value=True),
patch(_SLEEP, new_callable=AsyncMock),
patch(_SETTINGS, _mock_settings()),
):
result = await run_yandex_detail_backfill(
db,
run_id=4,
params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 5},
)
assert result.attempted == 3
assert result.enriched == 1
assert result.failed == 2
runs.mark_done.assert_called_once()
@pytest.mark.asyncio
async def test_backfill_non200_counts_as_fail_and_aborts() -> None:
"""5 consecutive HTTP 403 responses -> abort after max_consecutive_blocks."""
snapshot = _make_snapshot(10)
db = _mock_db(snapshot)
runs = MagicMock()
responses = [_make_resp(403)] * 10
session_cls, _session = _make_session_ctx(responses)
with (
patch(_ASYNC_SESSION, session_cls),
patch(_PARSE, return_value=MagicMock()),
patch(_RUNS, runs),
patch(_SLEEP, new_callable=AsyncMock),
patch(_SETTINGS, _mock_settings()),
):
result = await run_yandex_detail_backfill(
db,
run_id=5,
params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 5},
)
assert result.failed == 5
assert result.enriched == 0
runs.mark_done.assert_called_once()
runs.mark_failed.assert_not_called()
@pytest.mark.asyncio
async def test_backfill_budget_guard_stops_loop() -> None:
"""Budget expired before first listing -> session.get not called."""
snapshot = _make_snapshot(5)
db = _mock_db(snapshot)
runs = MagicMock()
responses = [_make_resp(200)] * 5
session_cls, session = _make_session_ctx(responses)
mono_values = iter([0.0, 999.0, 999.0])
with (
patch(_ASYNC_SESSION, session_cls),
patch(_RUNS, runs),
patch("app.tasks.yandex_detail_backfill.time.monotonic", side_effect=mono_values),
patch(_SETTINGS, _mock_settings()),
):
await run_yandex_detail_backfill(db, run_id=6, params={"batch_size": 5, "budget_sec": 1})
session.get.assert_not_called()
runs.mark_done.assert_called_once()
@pytest.mark.asyncio
async def test_backfill_top_level_exception_marks_failed() -> None:
"""db.execute raises -> mark_failed called, exception re-raised."""
db = MagicMock()
db.execute.side_effect = RuntimeError("DB connection lost")
runs = MagicMock()
with (
patch(_RUNS, runs),
patch(_SETTINGS, _mock_settings()),
):
with pytest.raises(RuntimeError, match="DB connection lost"):
await run_yandex_detail_backfill(
db, run_id=7, params={"batch_size": 5, "budget_sec": 60}
)
runs.mark_failed.assert_called_once()
runs.mark_done.assert_not_called()
@pytest.mark.asyncio
async def test_backfill_fetch_exception_continues() -> None:
"""session.get raises on first listing -> failed++, loop continues for second."""
snapshot = _make_snapshot(2)
db = _mock_db(snapshot)
runs = MagicMock()
mock_enrichment = MagicMock()
# First call raises, second returns 200 OK
responses = [RuntimeError("connection reset"), _make_resp(200)]
session_cls, _session = _make_session_ctx(responses)
with (
patch(_ASYNC_SESSION, session_cls),
patch(_PARSE, return_value=mock_enrichment),
patch(_RUNS, runs),
patch(_SAVE, return_value=True),
patch(_SLEEP, new_callable=AsyncMock),
patch(_SETTINGS, _mock_settings()),
):
result = await run_yandex_detail_backfill(
db, run_id=8, params={"batch_size": 10, "budget_sec": 3600}
)
assert result.failed == 1
assert result.enriched == 1
assert result.attempted == 2
# fetch exception is caught by inner try/except (not DB) — no rollback needed
db.rollback.assert_not_called()
runs.mark_done.assert_called_once()
@pytest.mark.asyncio
async def test_backfill_no_proxy_when_settings_none() -> None:
"""scraper_proxy_url=None -> AsyncSession called with proxies=None."""
snapshot = _make_snapshot(1)
db = _mock_db(snapshot)
runs = MagicMock()
mock_enrichment = MagicMock()
responses = [_make_resp(200)]
session_cls, _session = _make_session_ctx(responses)
with (
patch(_ASYNC_SESSION, session_cls),
patch(_PARSE, return_value=mock_enrichment),
patch(_RUNS, runs),
patch(_SAVE, return_value=True),
patch(_SLEEP, new_callable=AsyncMock),
patch(_SETTINGS, _mock_settings(proxy=None)),
):
result = await run_yandex_detail_backfill(
db, run_id=9, params={"batch_size": 10, "budget_sec": 3600}
)
assert result.enriched == 1
# Verify AsyncSession was constructed with proxies=None
call_kwargs = session_cls.call_args[1]
assert call_kwargs.get("proxies") is None
# ---------------------------------------------------------------------------
# Tests: save_detail_enrichment (unchanged — kept for coverage)
# ---------------------------------------------------------------------------
def test_save_detail_enrichment_maps_fields_to_update() -> None:
"""save_detail_enrichment calls db.execute with UPDATE and commits."""
from app.services.scrapers.yandex_detail import (
DetailEnrichment,
MetroStation,
save_detail_enrichment,
)
enrichment = DetailEnrichment(
offer_id="12345",
source_url="https://realty.yandex.ru/offer/12345/",
rooms=2,
area_m2=54.5,
floor=3,
total_floors=9,
address="Екатеринбург, ул. Ленина, 10",
description="Хорошая квартира",
repair_state="standard",
publish_date=None,
views_total=42,
publish_date_relative="вчера",
agency_name="Агентство «Тест»",
agency_founded_year=2005,
agency_objects_count=150,
metro_stations=[MetroStation(name="Чкаловская", walk_min=11)],
photo_urls=["https://example.com/1.jpg", "https://example.com/2.jpg"],
newbuilding_url="https://realty.yandex.ru/kupit/novostrojka/test-12345/",
newbuilding_id="12345",
)
db = MagicMock()
result_mock = MagicMock()
result_mock.rowcount = 1
db.execute.return_value = result_mock
saved = save_detail_enrichment(db, listing_id=99, e=enrichment)
assert saved is True
db.execute.assert_called_once()
db.commit.assert_called_once()
call_args = db.execute.call_args
params = call_args[0][1]
assert params["rooms"] == 2
assert params["area_m2"] == 54.5
assert params["agency_name"] == "Агентство «Тест»"
metro_val = params["metro_stations"]
assert metro_val is not None
parsed_metro = json.loads(metro_val)
assert parsed_metro[0]["name"] == "Чкаловская"
assert parsed_metro[0]["walk_min"] == 11
photo_val = params["photo_urls"]
assert photo_val is not None
parsed_photos = json.loads(photo_val)
assert len(parsed_photos) == 2
assert parsed_photos[0] == "https://example.com/1.jpg"
def test_save_detail_enrichment_empty_metro_and_photos() -> None:
"""Empty metro_stations and photo_urls -> NULL passed for both jsonb columns."""
from app.services.scrapers.yandex_detail import DetailEnrichment, save_detail_enrichment
enrichment = DetailEnrichment(
offer_id="99",
source_url="https://realty.yandex.ru/offer/99/",
)
db = MagicMock()
result_mock = MagicMock()
result_mock.rowcount = 1
db.execute.return_value = result_mock
save_detail_enrichment(db, listing_id=1, e=enrichment)
params = db.execute.call_args[0][1]
assert params["metro_stations"] is None
assert params["photo_urls"] is None
def test_save_detail_enrichment_rowcount_zero_returns_false() -> None:
"""rowcount=0 (listing_id not found) -> returns False."""
from app.services.scrapers.yandex_detail import DetailEnrichment, save_detail_enrichment
enrichment = DetailEnrichment(
offer_id="404",
source_url="https://realty.yandex.ru/offer/404/",
)
db = MagicMock()
result_mock = MagicMock()
result_mock.rowcount = 0
db.execute.return_value = result_mock
saved = save_detail_enrichment(db, listing_id=404, e=enrichment)
assert saved is False