feat(scrapers): yandex detail-enrichment — save + ночной backfill через прокси (part of #1553) #1554
5 changed files with 941 additions and 0 deletions
|
|
@ -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:
|
||||||
|
|
|
||||||
|
|
@ -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
|
||||||
|
|
|
||||||
301
tradein-mvp/backend/app/tasks/yandex_detail_backfill.py
Normal file
301
tradein-mvp/backend/app/tasks/yandex_detail_backfill.py
Normal 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 parse→None 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 parse→None 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). parse→None 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
|
||||||
49
tradein-mvp/backend/data/sql/113_yandex_detail_backfill.sql
Normal file
49
tradein-mvp/backend/data/sql/113_yandex_detail_backfill.sql
Normal 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;
|
||||||
440
tradein-mvp/backend/tests/tasks/test_yandex_detail_backfill.py
Normal file
440
tradein-mvp/backend/tests/tasks/test_yandex_detail_backfill.py
Normal 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
|
||||||
Loading…
Add table
Reference in a new issue