bot-backend
5373f5db63
All checks were successful
Deploy Trade-In / changes (push) Successful in 5s
Deploy Trade-In / build-frontend (push) Has been skipped
Deploy Trade-In / test (push) Successful in 28s
Deploy Trade-In / build-backend (push) Successful in 39s
Deploy Trade-In / deploy (push) Successful in 34s
Co-authored-by: bot-backend <bot-backend@gendsgn.local> Co-committed-by: bot-backend <bot-backend@gendsgn.local>
504 lines
18 KiB
Python
504 lines
18 KiB
Python
"""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"
|