feat(tradein): admin backfill endpoints — cian price-history (T8a) + yandex address (T10) + house IMV (T7) #856

Merged
bot-reviewer merged 1 commit from feat/tradein-scrapers-wave2-backfills into main 2026-05-31 06:37:11 +00:00
5 changed files with 1623 additions and 0 deletions
Showing only changes of commit e51722b3f1 - Show all commits

View file

@ -1186,3 +1186,188 @@ async def scrape_cian_backfill_history(
dry_run=dry_run,
)
return CianBackfillResp(ok=True, **result.__dict__)
# ── T8a: Cian price-history backfill ─────────────────────────────────────────
class CianPriceHistoryRequest(BaseModel):
batch_size: int = Field(default=50, ge=1, le=500)
listing_id: int | None = Field(
default=None,
description="Если указан — обработать один конкретный листинг.",
)
class BackfillCountersResp(BaseModel):
ok: bool
checked: int
saved: int
skipped: int
errors: int
duration_sec: float
@router.post("/scrape/cian-price-history", response_model=BackfillCountersResp)
async def scrape_cian_price_history(
payload: CianPriceHistoryRequest,
db: Session = Depends(get_db), # noqa: B008
) -> BackfillCountersResp:
"""T8a: Backfill offer_price_history для Cian-листингов.
Fetches Cian detail pages (curl_cffi, без Playwright) для листингов без строк
в offer_price_history, парсит priceChanges из _cianConfig defaultState.
batch_size: сколько листингов обработать за запуск (default 50).
listing_id: обработать один конкретный листинг (debug / точечная перезаливка).
Rate-limit: задержка из scraper_settings['cian'] (default 5s) между страницами.
Idempotent: повторный запуск безопасен (ON CONFLICT DO NOTHING).
"""
from app.services.cian_price_history import (
CianPriceHistoryResult,
backfill_cian_price_history,
)
result: CianPriceHistoryResult = await backfill_cian_price_history(
db,
batch_size=payload.batch_size,
listing_id=payload.listing_id,
)
return BackfillCountersResp(
ok=True,
checked=result.checked,
saved=result.saved,
skipped=result.skipped,
errors=result.errors,
duration_sec=round(result.duration_sec, 2),
)
# ── T10: Yandex address backfill ─────────────────────────────────────────────
class YandexAddressBackfillRequest(BaseModel):
limit: int = Field(default=200, ge=1, le=2000)
request_delay_sec: float = Field(default=3.0, ge=1.0, le=15.0)
@router.post("/scrape/yandex-address-backfill", response_model=BackfillCountersResp)
async def scrape_yandex_address_backfill(
payload: YandexAddressBackfillRequest,
background_tasks: BackgroundTasks,
db: Session = Depends(get_db), # noqa: B008
) -> BackfillCountersResp:
"""T10: Backfill listings.address для Yandex-листингов без номера дома.
Yandex SERP-карточки содержат только улицу («улица Горького»). Этот endpoint
фетчит detail-страницы через curl_cffi (chrome120), извлекает полный адрес
из <title> («Екатеринбург, улица Горького, 36 id »), обновляет
listings.address и сбрасывает geocode_tried_at для перегеокодирования.
limit: сколько листингов обработать за запуск (default 200).
request_delay_sec: пауза между запросами (default 3.0s).
Запускает задачу в foreground (не в BackgroundTasks), т.к. batch небольшой.
Для > 500 листингов рекомендуется вызывать несколько раз.
"""
from app.services.yandex_address_backfill import (
YandexAddressBackfillResult,
backfill_yandex_addresses,
)
result: YandexAddressBackfillResult = await backfill_yandex_addresses(
db,
limit=payload.limit,
request_delay_sec=payload.request_delay_sec,
)
return BackfillCountersResp(
ok=True,
checked=result.checked,
saved=result.saved,
skipped=result.skipped,
errors=result.errors,
duration_sec=round(result.duration_sec, 2),
)
# ── T7: House IMV bulk backfill ───────────────────────────────────────────────
class HouseIMVBackfillRequest(BaseModel):
batch_size: int = Field(default=50, ge=1, le=500)
request_delay_sec: float = Field(
default=5.0,
ge=1.0,
le=30.0,
description="Пауза между Avito IMV вызовами (default 5s — anti-captcha).",
)
only_status: str = Field(
default="pending",
description="Обрабатывать дома с этим imv_status. 'transient_error' — retry.",
)
house_id: int | None = Field(
default=None,
description="Обработать один конкретный дом (debug).",
)
class HouseIMVBackfillResp(BaseModel):
ok: bool
checked: int
saved: int
skipped: int
errors: int
duration_sec: float
status_counts: dict[str, int]
@router.post("/scrape/house-imv-backfill", response_model=HouseIMVBackfillResp)
async def scrape_house_imv_backfill(
payload: HouseIMVBackfillRequest,
background_tasks: BackgroundTasks,
db: Session = Depends(get_db), # noqa: B008
) -> HouseIMVBackfillResp:
"""T7: Avito IMV bulk backfill на уровне домов (house_imv_evaluations).
Итерирует дома с imv_status='pending' (или 'transient_error'), вычисляет
медианные параметры квартиры из связанных листингов, вызывает Avito IMV API,
сохраняет результаты в house_imv_evaluations + house_placement_history +
house_suggestions.
Resumable: дома помечаются imv_status='ok'/'not_found'/'error' после обработки.
Idempotent: повторный запуск не дублирует данные (UPSERT on house_id).
batch_size: сколько домов обработать за запуск (default 50).
request_delay_sec: пауза между IMV-вызовами (default 5s). ВАЖНО: Avito IMV
реагирует на частые запросы с datacenter-IP. Не снижать < 3s.
only_status: по умолчанию 'pending'. Для retry failed 'transient_error'.
house_id: обработать один дом (debug).
Примечание по прокси: Avito IMV использует собственную curl_cffi-сессию.
settings.scraper_proxy_url прокидывается в неё автоматически через avito_imv.py
(evaluate_via_imv создаёт сессию без явного proxy-параметра сессия использует
env HTTP_PROXY/HTTPS_PROXY если задан. При необходимости добавить явный
proxy-параметр в evaluate_via_imv / backfill_house_imv на уровне сервиса).
"""
from app.services.house_imv_backfill import (
HouseIMVBackfillResult,
backfill_house_imv,
)
result: HouseIMVBackfillResult = await backfill_house_imv(
db,
batch_size=payload.batch_size,
request_delay_sec=payload.request_delay_sec,
only_status=payload.only_status,
house_id=payload.house_id,
)
return HouseIMVBackfillResp(
ok=True,
checked=result.checked,
saved=result.saved,
skipped=result.skipped,
errors=result.errors,
duration_sec=round(result.duration_sec, 2),
status_counts=result.status_counts,
)

View file

@ -0,0 +1,153 @@
"""T8a: Cian offer price-history backfill service.
Fetches Cian detail pages (curl_cffi, без Playwright) for listings that have
no rows in offer_price_history, extracts priceChanges from
_cianConfig['frontend-offer-card'] defaultState, writes to offer_price_history.
Deliberately NOT wired into scheduler rate-limit risk with datacenter IPs.
Triggered manually via POST /api/v1/admin/scrape/cian-price-history.
"""
from __future__ import annotations
import asyncio
import logging
import time
from dataclasses import dataclass, field
from sqlalchemy import text
from sqlalchemy.orm import Session
from app.services.scraper_settings import get_scraper_delay
from app.services.scrapers.cian_detail import fetch_detail, save_detail_enrichment
logger = logging.getLogger(__name__)
@dataclass
class CianPriceHistoryResult:
checked: int = 0
saved: int = 0
skipped: int = 0
errors: int = 0
duration_sec: float = field(default=0.0)
async def backfill_cian_price_history(
db: Session,
*,
batch_size: int = 50,
listing_id: int | None = None,
) -> CianPriceHistoryResult:
"""Fetch Cian detail pages and write missing price-history rows.
Args:
db: SQLAlchemy session (caller-owned; commits internally per listing).
batch_size: max listings to process when listing_id is None.
listing_id: process a single specific listing (ignores batch_size).
Selection query picks cian listings with no existing offer_price_history rows.
Idempotent: re-run safe via ON CONFLICT DO NOTHING in save_detail_enrichment.
"""
result = CianPriceHistoryResult()
t0 = time.time()
delay = get_scraper_delay("cian") # default 5.0s
if listing_id is not None:
rows = (
db.execute(
text("""
SELECT l.id, l.source_url
FROM listings l
WHERE l.id = CAST(:lid AS bigint)
AND l.source = 'cian'
AND l.source_url IS NOT NULL
"""),
{"lid": listing_id},
)
.mappings()
.all()
)
else:
rows = (
db.execute(
text("""
SELECT l.id, l.source_url
FROM listings l
LEFT JOIN offer_price_history oph ON oph.listing_id = l.id
WHERE l.source = 'cian'
AND l.source_url IS NOT NULL
AND oph.listing_id IS NULL
ORDER BY l.id
LIMIT :lim
"""),
{"lim": batch_size},
)
.mappings()
.all()
)
result.checked = len(rows)
logger.info(
"cian_price_history backfill: checked=%d delay=%.1fs",
result.checked,
delay,
)
for i, row in enumerate(rows):
lid: int = row["id"]
url: str = row["source_url"]
try:
enrichment = await fetch_detail(url)
except Exception as exc:
logger.warning(
"cian_price_history: fetch failed listing_id=%s url=%s: %s",
lid,
url,
exc,
)
result.errors += 1
await asyncio.sleep(delay)
continue
if enrichment is None:
logger.warning(
"cian_price_history: fetch returned None listing_id=%s url=%s",
lid,
url,
)
result.errors += 1
await asyncio.sleep(delay)
continue
if not enrichment.price_changes:
logger.debug("cian_price_history: no price_changes listing_id=%s", lid)
result.skipped += 1
else:
try:
save_detail_enrichment(db, lid, enrichment)
result.saved += len(enrichment.price_changes)
except Exception as exc:
logger.warning("cian_price_history: save failed listing_id=%s: %s", lid, exc)
result.errors += 1
try:
db.rollback()
except Exception:
pass
await asyncio.sleep(delay)
continue
if i < len(rows) - 1:
await asyncio.sleep(delay)
result.duration_sec = time.time() - t0
logger.info(
"cian_price_history done: checked=%d saved=%d skipped=%d errors=%d %.1fs",
result.checked,
result.saved,
result.skipped,
result.errors,
result.duration_sec,
)
return result

View file

@ -0,0 +1,504 @@
"""T7: House-level IMV backfill service.
Iterates houses with imv_status='pending' (or 'transient_error') that have
lat/lon coordinates and at least one linked listing with rooms+area data.
For each house:
1. Pick median lot-params from linked listings (rooms, area_m2, floor,
total_floors, house_type).
2. Enrich address with region prefix (prevents Avito geocoder ambiguity).
3. Call avito_imv.evaluate_via_imv().
4. Persist house_imv_evaluations + house_placement_history + house_suggestions.
5. Mark houses.imv_status accordingly (ok/not_found/no_params/error/transient_error).
Resumable: re-running picks up from where the previous run stopped.
Deliberately NOT wired into scheduler triggered manually via admin API.
"""
from __future__ import annotations
import asyncio
import json
import logging
import time
from dataclasses import dataclass, field
from typing import Literal
from sqlalchemy import text
from sqlalchemy.orm import Session
from app.services.scrapers.avito_imv import (
IMVAddressNotFoundError,
IMVAuthError,
IMVEvaluation,
IMVTransientError,
evaluate_via_imv,
)
logger = logging.getLogger(__name__)
# ── house_type normalisation ─────────────────────────────────────────────────
_HOUSE_TYPE_TO_IMV: dict[str, str] = {
"panel": "panel",
"brick": "brick",
"monolith": "monolithic",
"monolithic": "monolithic",
"monolith_brick": "monolithic", # Avito API не принимает гибриды
"block": "block",
"wood": "wood",
}
_HOUSE_TYPE_DEFAULT = "panel" # самый распространённый в ЕКБ
def _map_house_type(raw: str | None) -> str:
if not raw:
return _HOUSE_TYPE_DEFAULT
return _HOUSE_TYPE_TO_IMV.get(raw.lower().strip(), _HOUSE_TYPE_DEFAULT)
# ── Region bbox prefix для Avito geocoder ────────────────────────────────────
_REGION_BBOX: list[tuple[str, float, float, float, float]] = [
# (region_name, lat_min, lat_max, lon_min, lon_max)
("Свердловская область, Екатеринбург", 56.5, 57.1, 59.9, 61.0),
("Свердловская область", 56.0, 61.0, 58.0, 65.0),
("Кировская область, Киров", 58.4, 58.8, 49.4, 49.9),
("Ставропольский край, Ставрополь", 44.9, 45.2, 41.7, 42.2),
("Тюменская область, Тюмень", 57.0, 57.3, 65.3, 65.8),
("Челябинская область, Челябинск", 55.0, 55.4, 61.2, 61.7),
]
_REGION_DEFAULT = "Свердловская область, Екатеринбург"
def _detect_region_prefix(lat: float | None, lon: float | None) -> str:
if lat is None or lon is None:
return _REGION_DEFAULT
for name, lat_min, lat_max, lon_min, lon_max in _REGION_BBOX:
if lat_min <= lat <= lat_max and lon_min <= lon <= lon_max:
return name
return _REGION_DEFAULT
def _enrich_address_for_imv(raw: str, lat: float | None, lon: float | None) -> str:
"""Prepend region prefix if address lacks a region marker."""
addr = (raw or "").strip()
addr_lower = addr.lower()
if any(
m in addr_lower for m in ("область,", " обл.,", " обл,", "край,", "респ.,", "республика,")
):
return addr
prefix = _detect_region_prefix(lat, lon)
return f"{prefix}, {addr}"
# ── DB helpers ────────────────────────────────────────────────────────────────
def pick_lot_params(db: Session, house_id: int) -> dict:
"""Pick representative lot-params — median values from linked listings."""
row = (
db.execute(
text("""
SELECT
CAST(percentile_cont(0.5) WITHIN GROUP (ORDER BY rooms)
AS integer) AS rooms,
CAST(percentile_cont(0.5) WITHIN GROUP (ORDER BY area_m2)
AS numeric(8,2)) AS area_m2,
CAST(percentile_cont(0.5) WITHIN GROUP (ORDER BY floor)
AS integer) AS floor,
CAST(percentile_cont(0.5) WITHIN GROUP (ORDER BY total_floors)
AS integer) AS total_floors,
mode() WITHIN GROUP (ORDER BY house_type) AS house_type
FROM listings
WHERE house_id_fk = :hid
AND rooms IS NOT NULL
AND area_m2 IS NOT NULL
"""),
{"hid": house_id},
)
.mappings()
.first()
)
if row is None or row["rooms"] is None:
return {}
house = (
db.execute(
text("SELECT house_type, total_floors FROM houses WHERE id = :hid"),
{"hid": house_id},
)
.mappings()
.first()
)
rooms_raw = int(row["rooms"])
return {
"rooms": max(rooms_raw, 1), # Avito IMV отклоняет rooms=0 (студия)
"area_m2": float(row["area_m2"]),
"floor": int(row["floor"] or 1),
"floor_at_home": int(row["total_floors"] or (house and house["total_floors"]) or 9),
"house_type": _map_house_type(row["house_type"] or (house and house["house_type"])),
"renovation_type": "cosmetic",
"has_balcony": True,
"has_loggia": False,
}
def save_imv_result(db: Session, house_id: int, params: dict, result: IMVEvaluation) -> None:
"""Persist evaluation to house_imv_evaluations, house_placement_history, house_suggestions."""
# 1. House-level evaluation (UPSERT on house_id)
db.execute(
text("""
INSERT INTO house_imv_evaluations (
house_id, cache_key, recommended_price, lower_price, higher_price,
market_count, raw_response, fetched_at,
rooms, area_m2, floor, floor_at_home, house_type,
renovation_type, has_balcony, has_loggia
) VALUES (
:hid, :ck, :rec, :lo, :hi,
:mc, CAST(:raw AS jsonb), NOW(),
:rooms, CAST(:area AS numeric), :floor, :fah, :htype,
:rtype, :balcony, :loggia
)
ON CONFLICT (house_id) DO UPDATE SET
cache_key = EXCLUDED.cache_key,
recommended_price = EXCLUDED.recommended_price,
lower_price = EXCLUDED.lower_price,
higher_price = EXCLUDED.higher_price,
market_count = EXCLUDED.market_count,
raw_response = EXCLUDED.raw_response,
fetched_at = NOW(),
rooms = EXCLUDED.rooms,
area_m2 = EXCLUDED.area_m2,
floor = EXCLUDED.floor,
floor_at_home = EXCLUDED.floor_at_home,
house_type = EXCLUDED.house_type,
renovation_type = EXCLUDED.renovation_type,
has_balcony = EXCLUDED.has_balcony,
has_loggia = EXCLUDED.has_loggia
"""),
{
"hid": house_id,
"ck": result.cache_key,
"rec": result.recommended_price,
"lo": result.lower_price,
"hi": result.higher_price,
"mc": result.market_count,
"raw": (
json.dumps(result.raw_response, ensure_ascii=False) if result.raw_response else None
),
"rooms": params["rooms"],
"area": params["area_m2"],
"floor": params["floor"],
"fah": params["floor_at_home"],
"htype": params["house_type"],
"rtype": params["renovation_type"],
"balcony": params["has_balcony"],
"loggia": params["has_loggia"],
},
)
# 2. Placement history items
for item in result.placement_history:
db.execute(
text("""
INSERT INTO house_placement_history (
source, house_id, ext_item_id, title, rooms, area_m2,
floor, total_floors, start_price, start_price_date,
last_price, last_price_date, removed_date, exposure_days,
raw_payload, scraped_at
) VALUES (
'avito_imv', :hid, :ext, :title, :rooms, CAST(:area AS numeric),
:floor, :total_floors, :sp, :spd, :lp, :lpd, :rmd, :exp,
CAST(:raw AS jsonb), NOW()
)
ON CONFLICT (source, ext_item_id) DO UPDATE SET
house_id = EXCLUDED.house_id,
title = EXCLUDED.title,
rooms = EXCLUDED.rooms,
area_m2 = EXCLUDED.area_m2,
floor = EXCLUDED.floor,
total_floors = EXCLUDED.total_floors,
start_price = EXCLUDED.start_price,
start_price_date = EXCLUDED.start_price_date,
last_price = EXCLUDED.last_price,
last_price_date = EXCLUDED.last_price_date,
removed_date = EXCLUDED.removed_date,
exposure_days = EXCLUDED.exposure_days,
raw_payload = EXCLUDED.raw_payload,
scraped_at = NOW()
"""),
{
"hid": house_id,
"ext": item.ext_item_id,
"title": item.title,
"rooms": item.rooms,
"area": item.area_m2,
"floor": item.floor,
"total_floors": item.total_floors,
"sp": item.start_price,
"spd": item.start_price_date,
"lp": item.last_price,
"lpd": item.last_price_date,
"rmd": item.removed_date,
"exp": item.exposure_days,
"raw": (
json.dumps(item.raw_payload, ensure_ascii=False) if item.raw_payload else None
),
},
)
# 3. Suggestions
for sug in result.suggestions:
db.execute(
text("""
INSERT INTO house_suggestions (
house_id, ext_item_id, title, address, price_rub,
exposure_days, publish_date,
item_link, metro_name, metro_distance,
has_good_price_badge, raw_payload, fetched_at
) VALUES (
:hid, :ext, :title, :addr, :price,
:exp, :pdate,
:link, :mname, :mdist,
:gpb, CAST(:raw AS jsonb), NOW()
)
ON CONFLICT (house_id, ext_item_id) DO UPDATE SET
title = EXCLUDED.title,
price_rub = EXCLUDED.price_rub,
exposure_days = EXCLUDED.exposure_days,
publish_date = EXCLUDED.publish_date,
item_link = EXCLUDED.item_link,
metro_name = EXCLUDED.metro_name,
metro_distance = EXCLUDED.metro_distance,
has_good_price_badge = EXCLUDED.has_good_price_badge,
raw_payload = EXCLUDED.raw_payload,
fetched_at = NOW()
"""),
{
"hid": house_id,
"ext": sug.ext_item_id,
"title": sug.title,
"addr": sug.address,
"price": sug.price_rub,
"exp": sug.exposure_days,
"pdate": sug.publish_date,
"link": sug.item_url,
"mname": sug.metro_name,
"mdist": sug.metro_distance,
"gpb": sug.has_good_price_badge,
"raw": (
json.dumps(sug.raw_payload, ensure_ascii=False) if sug.raw_payload else None
),
},
)
# 4. Mark house success
db.execute(
text("""
UPDATE houses
SET imv_status = 'ok',
last_imv_attempt_at = NOW(),
imv_error_reason = NULL
WHERE id = :hid
"""),
{"hid": house_id},
)
def _mark_status(
db: Session,
house_id: int,
status: str,
reason: str | None = None,
) -> None:
db.execute(
text("""
UPDATE houses
SET imv_status = :s,
last_imv_attempt_at = NOW(),
imv_error_reason = :r
WHERE id = :hid
"""),
{"hid": house_id, "s": status, "r": reason},
)
db.commit()
# ── Top-level orchestrator ────────────────────────────────────────────────────
_IMVStatus = Literal[
"ok", "no_params", "no_address", "not_found", "auth_error", "transient", "error"
]
@dataclass
class HouseIMVBackfillResult:
checked: int = 0
saved: int = 0
skipped: int = 0
errors: int = 0
duration_sec: float = field(default=0.0)
status_counts: dict[str, int] = field(default_factory=dict)
async def backfill_house_imv(
db: Session,
*,
batch_size: int = 50,
request_delay_sec: float = 5.0,
only_status: str = "pending",
house_id: int | None = None,
) -> HouseIMVBackfillResult:
"""Run Avito IMV evaluation for each house in scope, save results.
Args:
db: SQLAlchemy session.
batch_size: max houses to process (ignored when house_id given).
request_delay_sec: sleep between Avito API calls (default 5s anti-bot).
only_status: process houses with this imv_status (default 'pending').
Use 'transient_error' to retry failures.
house_id: process a single specific house (debug).
Returns:
HouseIMVBackfillResult with aggregate counters.
"""
result = HouseIMVBackfillResult()
t0 = time.time()
if house_id is not None:
rows = (
db.execute(
text("""
SELECT id, address, full_address, lat, lon
FROM houses
WHERE id = CAST(:hid AS bigint)
AND lat IS NOT NULL
AND lon IS NOT NULL
"""),
{"hid": house_id},
)
.mappings()
.all()
)
else:
rows = (
db.execute(
text("""
SELECT id, address, full_address, lat, lon
FROM houses
WHERE imv_status = :status
AND lat IS NOT NULL
AND lon IS NOT NULL
AND address IS NOT NULL
ORDER BY last_imv_attempt_at NULLS FIRST, id
LIMIT :batch
"""),
{"status": only_status, "batch": batch_size},
)
.mappings()
.all()
)
result.checked = len(rows)
if not rows:
logger.info("house_imv_backfill: nothing to process (status=%r)", only_status)
result.duration_sec = time.time() - t0
return result
logger.info(
"house_imv_backfill: %d houses (status=%r delay=%.1fs)",
result.checked,
only_status,
request_delay_sec,
)
for i, house in enumerate(rows):
hid: int = house["id"]
status_str = await _process_one_house(db, dict(house))
result.status_counts[status_str] = result.status_counts.get(status_str, 0) + 1
if status_str == "ok":
result.saved += 1
elif status_str in ("no_params", "no_address"):
result.skipped += 1
else:
result.errors += 1
logger.info(
"[%d/%d] house_id=%d status=%s",
i + 1,
result.checked,
hid,
status_str,
)
if i < len(rows) - 1:
if status_str == "auth_error":
logger.warning("IMV auth error — extra delay 60s before next house")
await asyncio.sleep(60)
else:
await asyncio.sleep(request_delay_sec)
result.duration_sec = time.time() - t0
logger.info(
"house_imv_backfill done: checked=%d saved=%d skipped=%d errors=%d %.1fs %s",
result.checked,
result.saved,
result.skipped,
result.errors,
result.duration_sec,
result.status_counts,
)
return result
async def _process_one_house(db: Session, house: dict) -> str:
"""Process a single house. Returns final status string."""
hid: int = house["id"]
params = pick_lot_params(db, hid)
if not params:
_mark_status(db, hid, "no_params", "no listings with rooms+area")
return "no_params"
address = house.get("address") or house.get("full_address")
if not address:
_mark_status(db, hid, "no_address", "house.address is NULL")
return "no_address"
enriched = _enrich_address_for_imv(address, house.get("lat"), house.get("lon"))
if enriched != address:
logger.debug("house_imv: enriched address house=%d %r -> %r", hid, address, enriched)
try:
eval_result = await evaluate_via_imv(address=enriched, **params)
except IMVAddressNotFoundError as exc:
_mark_status(db, hid, "not_found", str(exc)[:200])
return "not_found"
except IMVAuthError as exc:
_mark_status(db, hid, "transient_error", f"auth: {exc!s}"[:200])
return "auth_error"
except IMVTransientError as exc:
_mark_status(db, hid, "transient_error", str(exc)[:200])
return "transient"
except Exception as exc:
_mark_status(db, hid, "error", repr(exc)[:200])
logger.error("house_imv: unexpected error house=%d: %r", hid, exc)
return "error"
try:
save_imv_result(db, hid, params, eval_result)
db.commit()
except Exception as exc:
logger.error("house_imv: save failed house=%d: %r", hid, exc)
try:
db.rollback()
except Exception:
pass
_mark_status(db, hid, "error", f"save_failed: {exc!s}"[:200])
return "error"
return "ok"

View file

@ -0,0 +1,218 @@
"""T10: Yandex listing address backfill from detail-page <title>.
Yandex SERP cards contain only street-level addresses (e.g. «улица Горького»).
The offer detail page <title> has the full address including house number:
«Екатеринбург, улица Горького, 36 id 12345».
This service finds active yandex listings without a house number in `address`,
fetches each detail page via curl_cffi(chrome120), extracts the full address
from the <title> regex, and UPDATE listings.address.
Deliberately NOT wired into scheduler manual trigger only.
Triggered via POST /api/v1/admin/scrape/yandex-address-backfill.
"""
from __future__ import annotations
import asyncio
import logging
import re
import time
from dataclasses import dataclass, field
from curl_cffi.requests import AsyncSession
from sqlalchemy import text
from sqlalchemy.orm import Session
logger = logging.getLogger(__name__)
# Extract 'Екатеринбург, <full address>' from <title> — matches:
# «Продажа квартиры — Екатеринбург, улица Горького, 36 — id 7654321 на Яндекс.Недвижимости»
# Group 1 = «Екатеринбург, улица Горького, 36» (stripped of trailing punctuation/spaces)
_RE_TITLE_ADDRESS = re.compile(
r"<title>[^<]*?—\s*"
r"(?:[^—,]*?,\s*)?" # optional ЖК/жилой комплекс prefix
r"(Екатеринбург,\s*[^<—]+?)"
r"\s*—\s*id\s*\d+[^<]*</title>", # allow trailing text after id NNN (e.g. 'на Яндекс...')
re.IGNORECASE,
)
# Listings without a house number: address contains no digit after comma-space
# (e.g. «улица Горького» but not «улица Горького, 36»).
# This regex matches a number right after «, » (house number pattern).
_RE_HAS_HOUSE_NUMBER = re.compile(r",\s*\d+")
@dataclass
class YandexAddressBackfillResult:
checked: int = 0
saved: int = 0
skipped: int = 0
errors: int = 0
duration_sec: float = field(default=0.0)
def _extract_address_from_title(html: str) -> str | None:
"""Extract full address from Yandex offer detail page <title>."""
html_clean = html.replace("\xa0", " ")
m = _RE_TITLE_ADDRESS.search(html_clean)
if not m:
return None
addr = m.group(1).strip().rstrip(",").strip()
return addr if addr else None
def _eligible_listings(db: Session, limit: int) -> list[dict]:
"""Active EKB yandex listings without house number in address."""
rows = (
db.execute(
text("""
SELECT id, source_id, address, source_url
FROM listings
WHERE source = 'yandex'
AND region_code = 66
AND is_active = TRUE
AND address IS NOT NULL
AND NOT (address ~ ',\\s*\\d+')
AND source_url IS NOT NULL
ORDER BY scraped_at DESC
LIMIT :lim
"""),
{"lim": limit},
)
.mappings()
.all()
)
return [dict(r) for r in rows]
async def backfill_yandex_addresses(
db: Session,
*,
limit: int = 200,
request_delay_sec: float = 3.0,
) -> YandexAddressBackfillResult:
"""Fetch Yandex detail pages and update address with full street+house number.
Args:
db: SQLAlchemy session.
limit: max listings to process.
request_delay_sec: sleep between requests (anti-bot; default 3.0s).
Returns:
YandexAddressBackfillResult with checked/saved/skipped/errors counters.
"""
from app.core.config import settings
result = YandexAddressBackfillResult()
t0 = time.time()
items = _eligible_listings(db, limit)
result.checked = len(items)
if not items:
logger.info("yandex_address_backfill: no eligible listings")
result.duration_sec = time.time() - t0
return result
logger.info(
"yandex_address_backfill: %d listings, delay=%.1fs",
result.checked,
request_delay_sec,
)
_proxy_url = settings.scraper_proxy_url
_proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None
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 i, item in enumerate(items):
lid: int = item["id"]
old_addr: str = item["address"]
url: str = item["source_url"]
try:
resp = await session.get(url, allow_redirects=True)
except Exception as exc:
logger.warning("yandex_address_backfill: fetch failed listing_id=%s: %s", lid, exc)
result.errors += 1
await asyncio.sleep(request_delay_sec)
continue
if resp.status_code != 200:
logger.warning(
"yandex_address_backfill: HTTP %d listing_id=%s", resp.status_code, lid
)
result.errors += 1
await asyncio.sleep(request_delay_sec)
continue
new_addr = _extract_address_from_title(resp.text)
if new_addr is None:
logger.debug("yandex_address_backfill: no address extracted listing_id=%s", lid)
result.skipped += 1
elif new_addr == old_addr:
logger.debug(
"yandex_address_backfill: unchanged listing_id=%s addr=%r", lid, old_addr
)
result.skipped += 1
elif not _RE_HAS_HOUSE_NUMBER.search(new_addr):
# Extracted address still has no house number — skip to avoid overwrite
# with another street-only value.
logger.debug(
"yandex_address_backfill: still no house number listing_id=%s addr=%r",
lid,
new_addr,
)
result.skipped += 1
else:
logger.info(
"yandex_address_backfill: updating listing_id=%s %r -> %r",
lid,
old_addr,
new_addr,
)
try:
db.execute(
text("""
UPDATE listings
SET address = :addr,
geocode_tried_at = NULL
WHERE id = CAST(:id AS bigint)
"""),
{"addr": new_addr, "id": lid},
)
db.commit()
result.saved += 1
except Exception as exc:
logger.warning(
"yandex_address_backfill: DB update failed listing_id=%s: %s",
lid,
exc,
)
result.errors += 1
try:
db.rollback()
except Exception:
pass
if i < len(items) - 1:
await asyncio.sleep(request_delay_sec)
result.duration_sec = time.time() - t0
logger.info(
"yandex_address_backfill done: checked=%d saved=%d skipped=%d errors=%d %.1fs",
result.checked,
result.saved,
result.skipped,
result.errors,
result.duration_sec,
)
return result

View file

@ -0,0 +1,563 @@
"""Unit tests for T7/T8a/T10 backfill services (wave-2).
All tests are offline no DB, no network required.
"""
from __future__ import annotations
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
# ============================================================================
# T10 — yandex_address_backfill: _extract_address_from_title
# ============================================================================
class TestYandexTitleExtract:
"""Tests for _extract_address_from_title helper."""
def _extract(self, html: str) -> str | None:
from app.services.yandex_address_backfill import _extract_address_from_title
return _extract_address_from_title(html)
def test_extracts_full_address_standard(self):
html = (
"<title>Продажа квартиры — Екатеринбург, улица Горького, 36 — id 7654321 "
"на Яндекс.Недвижимости</title>"
)
result = self._extract(html)
assert result == "Екатеринбург, улица Горького, 36"
def test_extracts_address_with_corpus(self):
html = (
"<title>Продажа квартиры — Екатеринбург, улица Ленина, 52/1 — id 9999 "
"на Яндекс.Недвижимости</title>"
)
result = self._extract(html)
assert result == "Екатеринбург, улица Ленина, 52/1"
def test_returns_none_when_no_id_marker(self):
html = "<title>Продажа квартиры — Екатеринбург, улица Ленина, 52/1</title>"
result = self._extract(html)
assert result is None
def test_returns_none_when_no_title(self):
html = "<html><body>No title here</body></html>"
result = self._extract(html)
assert result is None
def test_strips_trailing_punctuation(self):
html = (
"<title>Купить квартиру — Екатеринбург, улица Гагарина, 11, — id 111 "
"на Яндекс.Недвижимости</title>"
)
result = self._extract(html)
# should strip trailing comma
assert result is not None
assert not result.endswith(",")
def test_nbsp_replaced(self):
html = (
"<title>— Екатеринбург,\xa0улица Горького, 36 — id 12 " "на Яндекс.Недвижимости</title>"
)
result = self._extract(html)
assert result is not None
assert "\xa0" not in result
def test_case_insensitive_city(self):
html = (
"<title>— ЕКАТЕРИНБУРГ, улица Горького, 36 — id 777 " "на Яндекс.Недвижимости</title>"
)
# regex is case-insensitive; city name retains original case
result = self._extract(html)
assert result is not None
assert "Горького, 36" in result
class TestYandexHouseNumberRegex:
def test_no_house_number_pattern(self):
from app.services.yandex_address_backfill import _RE_HAS_HOUSE_NUMBER
assert not _RE_HAS_HOUSE_NUMBER.search("улица Горького")
assert not _RE_HAS_HOUSE_NUMBER.search("Екатеринбург, улица Горького")
def test_has_house_number_pattern(self):
from app.services.yandex_address_backfill import _RE_HAS_HOUSE_NUMBER
assert _RE_HAS_HOUSE_NUMBER.search("Екатеринбург, улица Горького, 36")
assert _RE_HAS_HOUSE_NUMBER.search("улица Ленина, 52/1")
# ============================================================================
# T10 — backfill_yandex_addresses: integration (mocked)
# ============================================================================
@pytest.mark.asyncio
async def test_backfill_yandex_addresses_saves_enriched():
"""Happy path: fetch returns full address → saved=1."""
from app.services.yandex_address_backfill import backfill_yandex_addresses
html_with_addr = (
"<title>Продажа — Екатеринбург, улица Горького, 36 — id 11 "
"на Яндекс.Недвижимости</title>"
)
mock_resp = MagicMock()
mock_resp.status_code = 200
mock_resp.text = html_with_addr
mock_session = AsyncMock()
mock_session.get = AsyncMock(return_value=mock_resp)
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
mock_session.__aexit__ = AsyncMock(return_value=False)
mock_db = MagicMock()
mock_db.execute = MagicMock()
mock_db.commit = MagicMock()
items = [
{
"id": 42,
"source_id": "yandex-42",
"address": "улица Горького",
"source_url": "https://realty.yandex.ru/offer/11/",
}
]
with (
patch(
"app.services.yandex_address_backfill._eligible_listings",
return_value=items,
),
patch(
"app.services.yandex_address_backfill.AsyncSession",
return_value=mock_session,
),
):
result = await backfill_yandex_addresses(mock_db, limit=10)
assert result.checked == 1
assert result.saved == 1
assert result.errors == 0
assert result.skipped == 0
@pytest.mark.asyncio
async def test_backfill_yandex_addresses_skips_no_house_number():
"""Address extracted but still has no house number → skipped."""
from app.services.yandex_address_backfill import backfill_yandex_addresses
# Title without house number in extracted part
html = "<title>— Екатеринбург, улица Горького — id 22 на Яндекс.Недвижимости</title>"
mock_resp = MagicMock()
mock_resp.status_code = 200
mock_resp.text = html
mock_session = AsyncMock()
mock_session.get = AsyncMock(return_value=mock_resp)
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
mock_session.__aexit__ = AsyncMock(return_value=False)
items = [
{
"id": 99,
"source_id": "y-99",
"address": "улица Горького",
"source_url": "https://realty.yandex.ru/offer/22/",
}
]
with (
patch(
"app.services.yandex_address_backfill._eligible_listings",
return_value=items,
),
patch(
"app.services.yandex_address_backfill.AsyncSession",
return_value=mock_session,
),
):
result = await backfill_yandex_addresses(MagicMock(), limit=10)
assert result.skipped == 1
assert result.saved == 0
@pytest.mark.asyncio
async def test_backfill_yandex_addresses_error_on_non_200():
"""Non-200 response → errors counter incremented, no crash."""
from app.services.yandex_address_backfill import backfill_yandex_addresses
mock_resp = MagicMock()
mock_resp.status_code = 403
mock_resp.text = ""
mock_session = AsyncMock()
mock_session.get = AsyncMock(return_value=mock_resp)
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
mock_session.__aexit__ = AsyncMock(return_value=False)
items = [
{
"id": 55,
"source_id": "y-55",
"address": "улица Горького",
"source_url": "https://realty.yandex.ru/offer/55/",
}
]
with (
patch(
"app.services.yandex_address_backfill._eligible_listings",
return_value=items,
),
patch(
"app.services.yandex_address_backfill.AsyncSession",
return_value=mock_session,
),
):
result = await backfill_yandex_addresses(MagicMock(), limit=10)
assert result.errors == 1
assert result.saved == 0
# ============================================================================
# T8a — cian_price_history: backfill_cian_price_history
# ============================================================================
@pytest.mark.asyncio
async def test_cian_price_history_saves_changes():
"""Happy path: fetch returns DetailEnrichment with price_changes → saved."""
from app.services.cian_price_history import backfill_cian_price_history
from app.services.scrapers.cian_detail import DetailEnrichment
enrichment = DetailEnrichment(
cian_id=123456,
price_changes=[
{"change_time": "2026-04-01T00:00:00Z", "price_rub": 5000000, "diff_percent": None},
{"change_time": "2026-04-15T00:00:00Z", "price_rub": 4900000, "diff_percent": -2.0},
],
)
mock_db = MagicMock()
# Simulate listings rows returned by SELECT
rows = [{"id": 10, "source_url": "https://ekb.cian.ru/sale/flat/123456/"}]
with (
patch(
"app.services.cian_price_history.fetch_detail",
new_callable=AsyncMock,
return_value=enrichment,
) as mock_fetch,
patch("app.services.cian_price_history.save_detail_enrichment") as mock_save,
patch("app.services.cian_price_history.get_scraper_delay", return_value=0.0),
):
mock_mappings = MagicMock()
mock_mappings.all.return_value = rows
mock_db.execute.return_value.mappings.return_value = mock_mappings
result = await backfill_cian_price_history(mock_db, batch_size=10)
assert result.checked == 1
assert result.saved == 2 # 2 price_changes
assert result.errors == 0
mock_fetch.assert_called_once()
mock_save.assert_called_once()
@pytest.mark.asyncio
async def test_cian_price_history_skips_no_changes():
"""Fetch succeeds but no price_changes → skipped (not error)."""
from app.services.cian_price_history import backfill_cian_price_history
from app.services.scrapers.cian_detail import DetailEnrichment
enrichment = DetailEnrichment(cian_id=111, price_changes=[])
mock_db = MagicMock()
rows = [{"id": 20, "source_url": "https://ekb.cian.ru/sale/flat/111/"}]
with (
patch(
"app.services.cian_price_history.fetch_detail",
new_callable=AsyncMock,
return_value=enrichment,
),
patch("app.services.cian_price_history.save_detail_enrichment") as mock_save,
patch("app.services.cian_price_history.get_scraper_delay", return_value=0.0),
):
mock_mappings = MagicMock()
mock_mappings.all.return_value = rows
mock_db.execute.return_value.mappings.return_value = mock_mappings
result = await backfill_cian_price_history(mock_db, batch_size=10)
assert result.skipped == 1
assert result.saved == 0
mock_save.assert_not_called()
@pytest.mark.asyncio
async def test_cian_price_history_handles_fetch_none():
"""fetch_detail returns None → errors+1, no crash."""
from app.services.cian_price_history import backfill_cian_price_history
mock_db = MagicMock()
rows = [{"id": 30, "source_url": "https://ekb.cian.ru/sale/flat/999/"}]
with (
patch(
"app.services.cian_price_history.fetch_detail",
new_callable=AsyncMock,
return_value=None,
),
patch("app.services.cian_price_history.get_scraper_delay", return_value=0.0),
):
mock_mappings = MagicMock()
mock_mappings.all.return_value = rows
mock_db.execute.return_value.mappings.return_value = mock_mappings
result = await backfill_cian_price_history(mock_db, batch_size=10)
assert result.errors == 1
assert result.saved == 0
# ============================================================================
# T7 — house_imv_backfill: helpers
# ============================================================================
class TestHouseTypeMap:
def test_known_types(self):
from app.services.house_imv_backfill import _map_house_type
assert _map_house_type("panel") == "panel"
assert _map_house_type("brick") == "brick"
assert _map_house_type("monolith") == "monolithic"
assert _map_house_type("monolith_brick") == "monolithic"
assert _map_house_type("monolithic") == "monolithic"
assert _map_house_type("block") == "block"
assert _map_house_type("wood") == "wood"
def test_unknown_falls_back_to_panel(self):
from app.services.house_imv_backfill import _map_house_type
assert _map_house_type("unknown_type") == "panel"
assert _map_house_type(None) == "panel"
assert _map_house_type("") == "panel"
def test_case_insensitive(self):
from app.services.house_imv_backfill import _map_house_type
assert _map_house_type("PANEL") == "panel"
assert _map_house_type("Brick") == "brick"
assert _map_house_type("MONOLITH_BRICK") == "monolithic"
class TestRegionPrefix:
def test_ekb_coords(self):
from app.services.house_imv_backfill import _detect_region_prefix
prefix = _detect_region_prefix(56.8296, 60.6001)
assert prefix == "Свердловская область, Екатеринбург"
def test_none_coords_default(self):
from app.services.house_imv_backfill import _detect_region_prefix
prefix = _detect_region_prefix(None, None)
assert prefix == "Свердловская область, Екатеринбург"
def test_out_of_any_region_default(self):
from app.services.house_imv_backfill import _detect_region_prefix
# Moscow coords — not in any bbox → default EKB
prefix = _detect_region_prefix(55.75, 37.6)
assert prefix == "Свердловская область, Екатеринбург"
class TestEnrichAddress:
def test_prepends_region_when_no_marker(self):
from app.services.house_imv_backfill import _enrich_address_for_imv
result = _enrich_address_for_imv("улица Горького, 36", 56.83, 60.60)
assert result.startswith("Свердловская область, Екатеринбург,")
assert "улица Горького, 36" in result
def test_does_not_prepend_when_region_already_present(self):
from app.services.house_imv_backfill import _enrich_address_for_imv
addr = "Свердловская область, Екатеринбург, улица Горького, 36"
result = _enrich_address_for_imv(addr, 56.83, 60.60)
assert result == addr
def test_does_not_prepend_when_kraj_present(self):
from app.services.house_imv_backfill import _enrich_address_for_imv
addr = "Ставропольский край, Ставрополь, улица Ленина, 1"
result = _enrich_address_for_imv(addr, 45.05, 41.97)
assert result == addr
# ============================================================================
# T7 — backfill_house_imv: integration (mocked)
# ============================================================================
@pytest.mark.asyncio
async def test_backfill_house_imv_ok_path():
"""Happy path: picks params, calls IMV, saves → saved=1."""
from app.services.house_imv_backfill import backfill_house_imv
from app.services.scrapers.avito_imv import IMVEvaluation, IMVGeo
fake_geo = IMVGeo(geo_hash="tok", lat=56.83, lon=60.60)
fake_eval = IMVEvaluation(
cache_key="abc123",
address="ЕКБ ул. Горького, 36",
rooms=2,
area_m2=52.0,
floor=3,
floor_at_home=10,
house_type="panel",
renovation_type="cosmetic",
has_balcony=True,
has_loggia=False,
geo=fake_geo,
recommended_price=5_000_000,
lower_price=4_500_000,
higher_price=5_500_000,
market_count=24,
placement_history=[],
suggestions=[],
)
fake_params = {
"rooms": 2,
"area_m2": 52.0,
"floor": 3,
"floor_at_home": 10,
"house_type": "panel",
"renovation_type": "cosmetic",
"has_balcony": True,
"has_loggia": False,
}
mock_db = MagicMock()
mock_db.commit = MagicMock()
houses = [
{"id": 7, "address": "улица Горького, 36", "full_address": None, "lat": 56.83, "lon": 60.60}
]
with (
patch(
"app.services.house_imv_backfill.pick_lot_params",
return_value=fake_params,
),
patch(
"app.services.house_imv_backfill.evaluate_via_imv",
new_callable=AsyncMock,
return_value=fake_eval,
),
patch("app.services.house_imv_backfill.save_imv_result") as mock_save,
):
mock_mappings = MagicMock()
mock_mappings.all.return_value = houses
mock_db.execute.return_value.mappings.return_value = mock_mappings
result = await backfill_house_imv(mock_db, batch_size=10, request_delay_sec=0.0)
assert result.checked == 1
assert result.saved == 1
assert result.errors == 0
assert result.status_counts.get("ok") == 1
mock_save.assert_called_once()
@pytest.mark.asyncio
async def test_backfill_house_imv_no_params():
"""pick_lot_params returns {} → skipped (no_params status)."""
from app.services.house_imv_backfill import backfill_house_imv
mock_db = MagicMock()
mock_db.execute = MagicMock()
mock_db.commit = MagicMock()
houses = [
{"id": 8, "address": "улица Ленина, 1", "full_address": None, "lat": 56.83, "lon": 60.60}
]
with (
patch(
"app.services.house_imv_backfill.pick_lot_params",
return_value={},
),
patch("app.services.house_imv_backfill._mark_status") as mock_mark,
):
mock_mappings = MagicMock()
mock_mappings.all.return_value = houses
mock_db.execute.return_value.mappings.return_value = mock_mappings
result = await backfill_house_imv(mock_db, batch_size=10, request_delay_sec=0.0)
assert result.skipped == 1
assert result.saved == 0
mock_mark.assert_called_once_with(mock_db, 8, "no_params", "no listings with rooms+area")
@pytest.mark.asyncio
async def test_backfill_house_imv_not_found():
"""IMVAddressNotFoundError → status not_found, no crash."""
from app.services.house_imv_backfill import backfill_house_imv
from app.services.scrapers.avito_imv import IMVAddressNotFoundError
fake_params = {
"rooms": 2,
"area_m2": 52.0,
"floor": 3,
"floor_at_home": 10,
"house_type": "panel",
"renovation_type": "cosmetic",
"has_balcony": True,
"has_loggia": False,
}
mock_db = MagicMock()
mock_db.commit = MagicMock()
houses = [
{
"id": 9,
"address": "несуществующий адрес",
"full_address": None,
"lat": 56.83,
"lon": 60.60,
}
]
with (
patch(
"app.services.house_imv_backfill.pick_lot_params",
return_value=fake_params,
),
patch(
"app.services.house_imv_backfill.evaluate_via_imv",
new_callable=AsyncMock,
side_effect=IMVAddressNotFoundError("not found"),
),
patch("app.services.house_imv_backfill._mark_status") as mock_mark,
):
mock_mappings = MagicMock()
mock_mappings.all.return_value = houses
mock_db.execute.return_value.mappings.return_value = mock_mappings
result = await backfill_house_imv(mock_db, batch_size=10, request_delay_sec=0.0)
assert result.errors == 1
assert result.saved == 0
mock_mark.assert_called_once_with(mock_db, 9, "not_found", "not found")