1069 lines
39 KiB
Python
1069 lines
39 KiB
Python
"""Scraper for наш.дом.рф kn-API (objects + flats).
|
||
|
||
Endpoints discovered via chrome-devtools probe:
|
||
GET /сервисы/api/kn/object
|
||
params: offset, limit, place, objStatus, sortField=default, sortType=desc
|
||
place: "0-4" (city Екатеринбург) or "66" (region Свердл)
|
||
objStatus: 0=строящиеся, 1=архив-1, 2=сданные (sweep 0..2 for full coverage)
|
||
Returns {data: {list: [obj, ...], total: N}}.
|
||
|
||
GET /portal-kn/api/sales/portal/table?externalId={objId}
|
||
Per-object flat catalogue: nested entrances → floors → flats.
|
||
|
||
Auth header: `Authorization: Basic MTpxd2U=` (decodes to `1:qwe`, public debug
|
||
auth shipped in their frontend bundle).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
from datetime import date, datetime
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from sqlalchemy import text
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.db import SessionLocal
|
||
from app.services.scrapers.stealth import BASE_URL, BrowserSession
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
PATH_OBJECTS = "/сервисы/api/kn/object"
|
||
PATH_FLATS_TABLE = "/portal-kn/api/sales/portal/table"
|
||
PATH_SALE_GRAPH = "/сервисы/api/object/{obj_id}/sale_graph" # ?type=apartments|parking
|
||
PATH_SALES_AGG = "/сервисы/api/object/{obj_id}/sales_agg"
|
||
PATH_INFRA = "/сервисы/api/object/{obj_id}/infrastructure"
|
||
PATH_PHOTOS = "/сервисы/api/object/construction/progress/photo/{obj_id}"
|
||
RAW_SECTION = "kn_api"
|
||
|
||
SALE_GRAPH_TYPES = ("apartments", "parking")
|
||
PHOTOS_DIR_DEFAULT = Path("data/raw/domrf_photos")
|
||
|
||
# Three statuses observed in production (Apr 2026):
|
||
# 0 — строящиеся ЖК
|
||
# 1 — небольшая выборка ~2 объекта на регион (legacy/archive flag)
|
||
# 2 — сданные / завершённые (самый большой бакет)
|
||
DEFAULT_STATUSES = (0, 1, 2)
|
||
|
||
|
||
# ── helpers ──────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _g(d: dict[str, Any], *keys: str, default: Any = None) -> Any:
|
||
for k in keys:
|
||
if k in d and d[k] is not None:
|
||
return d[k]
|
||
return default
|
||
|
||
|
||
def _to_bool(v: Any) -> bool | None:
|
||
"""Coerce DOM.РФ 0/1 ints into Python bool. Pass through actual bools/None."""
|
||
if v is None:
|
||
return None
|
||
if isinstance(v, bool):
|
||
return v
|
||
if isinstance(v, int | float):
|
||
return bool(v)
|
||
if isinstance(v, str):
|
||
s = v.strip().lower()
|
||
if s in ("true", "1", "y", "yes"):
|
||
return True
|
||
if s in ("false", "0", "n", "no", ""):
|
||
return False
|
||
return None
|
||
|
||
|
||
def _to_date(v: Any) -> date | None:
|
||
"""Coerce date string to date. Accept 'YYYY-MM-DD', 'YYYY-MM-DD HH:MM:SS',
|
||
'DD-MM-YYYY HH:MM:SS', or quarter-string like 'IV кв. 2028'.
|
||
Quarter strings get mapped to the last day of the quarter.
|
||
"""
|
||
if v is None or v == "":
|
||
return None
|
||
if isinstance(v, date) and not isinstance(v, datetime):
|
||
return v
|
||
if isinstance(v, datetime):
|
||
return v.date()
|
||
if not isinstance(v, str):
|
||
return None
|
||
s = v.strip()
|
||
for fmt in ("%Y-%m-%d", "%Y-%m-%d %H:%M:%S", "%d-%m-%Y %H:%M:%S", "%d-%m-%Y", "%d.%m.%Y"):
|
||
try:
|
||
return datetime.strptime(s, fmt).date()
|
||
except ValueError:
|
||
pass
|
||
# Quarter-like 'IV кв. 2028' / '4 кв. 2028' / 'III квартал 2028'
|
||
quarter_end = {1: (3, 31), 2: (6, 30), 3: (9, 30), 4: (12, 31)}
|
||
roman = {"I": 1, "II": 2, "III": 3, "IV": 4}
|
||
parts = s.replace(".", " ").replace("кв", " ").replace("квартал", " ").split()
|
||
qq = year = None
|
||
for p in parts:
|
||
p = p.strip()
|
||
if p in roman and qq is None:
|
||
qq = roman[p]
|
||
elif p.isdigit():
|
||
n = int(p)
|
||
if 1 <= n <= 4 and qq is None:
|
||
qq = n
|
||
elif 2000 <= n <= 2050 and year is None:
|
||
year = n
|
||
if qq and year:
|
||
m, d = quarter_end[qq]
|
||
return date(year, m, d)
|
||
return None
|
||
|
||
|
||
def _problem_text(v: Any) -> str | None:
|
||
"""Coerce problem flag (sometimes int 0/1, sometimes text) to text or None."""
|
||
if v is None:
|
||
return None
|
||
if isinstance(v, bool):
|
||
return "1" if v else "0"
|
||
if isinstance(v, int | float):
|
||
return str(int(v))
|
||
if isinstance(v, str):
|
||
return v.strip() or None
|
||
return None
|
||
|
||
|
||
def _extract_total(payload: dict[str, Any]) -> int | None:
|
||
data = payload.get("data")
|
||
if isinstance(data, dict):
|
||
for key in ("total", "totalCount", "count"):
|
||
v = data.get(key)
|
||
if isinstance(v, int):
|
||
return v
|
||
return None
|
||
|
||
|
||
def _extract_list(payload: dict[str, Any]) -> list[dict[str, Any]]:
|
||
data = payload.get("data")
|
||
if isinstance(data, list):
|
||
return data
|
||
if isinstance(data, dict):
|
||
for key in ("list", "items", "objects", "content", "rows"):
|
||
v = data.get(key)
|
||
if isinstance(v, list):
|
||
return v
|
||
return []
|
||
|
||
|
||
# ── normalisation ────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _norm_object(row: dict[str, Any], region_cd: int | None = None) -> dict[str, Any]:
|
||
"""Map naш.дом.рф /kn/object row → domrf_kn_objects column dict.
|
||
|
||
Real API field names:
|
||
objId, hobjId, developer{devId, shortName, fullName, groupName, companyGroup, devInn},
|
||
rpdRegionCd, objAddr, shortAddr, objCommercNm, objFloorMin, objFloorMax,
|
||
objElemLivingCnt, objSquareLiving, objReady100PercDt, objClass, latitude, longitude,
|
||
objProblemFlg, problemFlag, siteStatus, objGreenHouseFlg, objGuarantyEscrowFlg,
|
||
objStatus, freeFlatsInfo{priceMin, numberFlats}.
|
||
"""
|
||
dev = row.get("developer") if isinstance(row.get("developer"), dict) else {}
|
||
company_group = _g(dev, "companyGroup") if dev else None
|
||
# Our DB convention: dev_id="<companyGroup>_0" matches v_developer_full_metrics.
|
||
dev_id = f"{company_group}_0" if company_group else None
|
||
return {
|
||
"obj_id": _g(row, "objId", "obj_id", "id"),
|
||
"hobj_id": _g(row, "hobjId", "hobj_id"),
|
||
"comm_name": _g(row, "objCommercNm"),
|
||
"addr": _g(row, "objAddr"),
|
||
"short_addr": _g(row, "shortAddr"),
|
||
"region_cd": _g(row, "rpdRegionCd") or region_cd,
|
||
"dev_id": dev_id,
|
||
"dev_name": _g(dev, "groupName") or _g(dev, "shortName") or _g(dev, "fullName"),
|
||
"dev_inn": _g(dev, "devInn"),
|
||
"floor_min": _g(row, "objFloorMin"),
|
||
"floor_max": _g(row, "objFloorMax"),
|
||
"flat_count": _g(row, "objElemLivingCnt"),
|
||
"square_living": _g(row, "objSquareLiving"),
|
||
"ready_dt": _to_date(_g(row, "objReady100PercDt")),
|
||
"problem_flag": _problem_text(_g(row, "problemFlag", "objProblemFlg")),
|
||
"site_status": _g(row, "siteStatus"),
|
||
"green_house": _to_bool(_g(row, "objGreenHouseFlg")),
|
||
"escrow": _to_bool(_g(row, "objGuarantyEscrowFlg")),
|
||
"obj_class": _g(row, "objClass"),
|
||
"wall_type": _g(row, "wallType"),
|
||
"energy_eff": _g(row, "energyEff"),
|
||
"latitude": _g(row, "latitude"),
|
||
"longitude": _g(row, "longitude"),
|
||
"obj_status": _g(row, "objStatus"),
|
||
}
|
||
|
||
|
||
def _flatten_table(table_payload: dict[str, Any]) -> list[dict[str, Any]]:
|
||
"""Iterate nested entrances → floors → flats[] from /sales/portal/table response."""
|
||
out: list[dict[str, Any]] = []
|
||
ext_id = table_payload.get("externalId")
|
||
for entrance in table_payload.get("entrances") or []:
|
||
for floor in entrance.get("floors") or []:
|
||
for flat in floor.get("flats") or []:
|
||
# Inject obj_id and floor for normalisation
|
||
flat = {
|
||
**flat,
|
||
"_objId": ext_id,
|
||
"_floor": floor.get("floorNumber"),
|
||
"_entrance": entrance.get("entranceNumber"),
|
||
}
|
||
out.append(flat)
|
||
return out
|
||
|
||
|
||
def _norm_flat(row: dict[str, Any], region_cd: int | None) -> dict[str, Any]:
|
||
"""Map flat from /sales/portal/table → domrf_kn_flats columns.
|
||
|
||
Real fields: flatId|id (numeric, may be null), odsId, elemId (uuid hash),
|
||
type, number, isStudio, totalArea, livingArea, rooms, status (free|booked|sold),
|
||
price, pricePerSquareMeter, numberFloors. Plus injected _objId and _floor.
|
||
"""
|
||
flat_id = _g(row, "flatId", "id")
|
||
if flat_id is None:
|
||
# Fallback: hash of elemId (stable, fits BIGINT). Without an id we cannot UPSERT.
|
||
elem = _g(row, "elemId")
|
||
if elem:
|
||
flat_id = abs(hash(elem)) % (2**63 - 1)
|
||
return {
|
||
"id": flat_id,
|
||
"ods_id": _g(row, "odsId"),
|
||
"flat_type": _g(row, "type", "flatType"),
|
||
"flat_number": _g(row, "number", "flatNumber"),
|
||
"is_studio": _to_bool(_g(row, "isStudio")),
|
||
"total_area": _g(row, "totalArea"),
|
||
"living_area": _g(row, "livingArea"),
|
||
"rooms": _g(row, "rooms"),
|
||
"status": _g(row, "status"),
|
||
"price_rub": _g(row, "price"),
|
||
"price_per_m2": _g(row, "pricePerSquareMeter"),
|
||
"floor": _g(row, "_floor", "floor"),
|
||
"num_floors": _g(row, "numberFloors"),
|
||
"obj_id": _g(row, "_objId", "objId"),
|
||
"city": _g(row, "city"),
|
||
"region_cd": region_cd,
|
||
"obj_name": _g(row, "objName") or _g(row.get("objInfo") or {}, "objCommercNm"),
|
||
}
|
||
|
||
|
||
# ── DB writes ────────────────────────────────────────────────────────────────
|
||
|
||
UPSERT_OBJECT_SQL = text(
|
||
"""
|
||
INSERT INTO domrf_kn_objects (
|
||
obj_id, hobj_id, comm_name, addr, short_addr, region_cd,
|
||
dev_id, dev_name, dev_inn,
|
||
floor_min, floor_max, flat_count, square_living, ready_dt,
|
||
problem_flag, site_status, green_house, escrow,
|
||
obj_class, wall_type, energy_eff,
|
||
latitude, longitude, obj_status, snapshot_date
|
||
) VALUES (
|
||
:obj_id, :hobj_id, :comm_name, :addr, :short_addr, :region_cd,
|
||
:dev_id, :dev_name, :dev_inn,
|
||
:floor_min, :floor_max, :flat_count, :square_living, :ready_dt,
|
||
:problem_flag, :site_status, :green_house, :escrow,
|
||
:obj_class, :wall_type, :energy_eff,
|
||
:latitude, :longitude, :obj_status, :snapshot_date
|
||
)
|
||
ON CONFLICT (obj_id, snapshot_date) DO UPDATE SET
|
||
comm_name = EXCLUDED.comm_name,
|
||
addr = EXCLUDED.addr,
|
||
dev_id = EXCLUDED.dev_id,
|
||
dev_name = EXCLUDED.dev_name,
|
||
flat_count = EXCLUDED.flat_count,
|
||
square_living = EXCLUDED.square_living,
|
||
ready_dt = EXCLUDED.ready_dt,
|
||
site_status = EXCLUDED.site_status,
|
||
escrow = EXCLUDED.escrow,
|
||
obj_class = EXCLUDED.obj_class,
|
||
latitude = EXCLUDED.latitude,
|
||
longitude = EXCLUDED.longitude,
|
||
obj_status = EXCLUDED.obj_status
|
||
"""
|
||
)
|
||
|
||
UPSERT_FLAT_SQL = text(
|
||
"""
|
||
INSERT INTO domrf_kn_flats (
|
||
id, ods_id, flat_type, flat_number, is_studio, total_area, living_area,
|
||
rooms, status, price_rub, price_per_m2, floor, num_floors, obj_id,
|
||
city, region_cd, obj_name, snapshot_date
|
||
) VALUES (
|
||
:id, :ods_id, :flat_type, :flat_number, :is_studio, :total_area, :living_area,
|
||
:rooms, :status, :price_rub, :price_per_m2, :floor, :num_floors, :obj_id,
|
||
:city, :region_cd, :obj_name, :snapshot_date
|
||
)
|
||
ON CONFLICT (id, snapshot_date) DO UPDATE SET
|
||
status = EXCLUDED.status,
|
||
price_rub = EXCLUDED.price_rub,
|
||
price_per_m2 = EXCLUDED.price_per_m2,
|
||
obj_id = EXCLUDED.obj_id,
|
||
region_cd = EXCLUDED.region_cd,
|
||
obj_name = EXCLUDED.obj_name
|
||
"""
|
||
)
|
||
|
||
INSERT_RAW_SQL = text(
|
||
"""
|
||
INSERT INTO domrf_raw_endpoints (
|
||
snapshot_date, section, endpoint, source_url, payload, payload_size
|
||
) VALUES (
|
||
:snapshot_date, :section, :endpoint, :source_url,
|
||
CAST(:payload AS JSONB), :payload_size
|
||
)
|
||
ON CONFLICT (snapshot_date, section, endpoint) DO UPDATE SET
|
||
payload = EXCLUDED.payload,
|
||
payload_size = EXCLUDED.payload_size,
|
||
source_url = EXCLUDED.source_url
|
||
"""
|
||
)
|
||
|
||
|
||
def _insert_raw(db: Session, snapshot_date: date, endpoint_label: str, payload: Any) -> None:
|
||
body = json.dumps(payload, ensure_ascii=False)
|
||
db.execute(
|
||
INSERT_RAW_SQL,
|
||
{
|
||
"snapshot_date": snapshot_date,
|
||
"section": RAW_SECTION,
|
||
"endpoint": endpoint_label[:255],
|
||
"source_url": f"https://xn--80az8a.xn--d1aqf.xn--p1ai{endpoint_label}",
|
||
"payload": body,
|
||
"payload_size": len(body.encode("utf-8")),
|
||
},
|
||
)
|
||
|
||
|
||
def upsert_objects(
|
||
db: Session, raw: list[dict[str, Any]], snapshot_date: date, region_cd: int
|
||
) -> int:
|
||
inserted = 0
|
||
for row in raw:
|
||
norm = _norm_object(row, region_cd=region_cd)
|
||
if not norm["obj_id"]:
|
||
continue
|
||
norm["snapshot_date"] = snapshot_date
|
||
try:
|
||
db.execute(UPSERT_OBJECT_SQL, norm)
|
||
inserted += 1
|
||
except Exception as e:
|
||
logger.warning("upsert object %s failed: %s", norm.get("obj_id"), e)
|
||
db.rollback()
|
||
return inserted
|
||
|
||
|
||
def upsert_flats(
|
||
db: Session, raw: list[dict[str, Any]], snapshot_date: date, region_cd: int
|
||
) -> int:
|
||
inserted = 0
|
||
skipped_no_id = 0
|
||
for row in raw:
|
||
norm = _norm_flat(row, region_cd)
|
||
if norm["id"] is None:
|
||
skipped_no_id += 1
|
||
continue
|
||
norm["snapshot_date"] = snapshot_date
|
||
try:
|
||
db.execute(UPSERT_FLAT_SQL, norm)
|
||
inserted += 1
|
||
except Exception as e:
|
||
logger.warning("upsert flat %s failed: %s", norm.get("id"), e)
|
||
db.rollback()
|
||
if skipped_no_id:
|
||
logger.info("skipped %d flats without numeric id", skipped_no_id)
|
||
return inserted
|
||
|
||
|
||
# ── extras (sale_graph, sales_agg, infrastructure, photos) ──────────────────
|
||
|
||
UPSERT_SALE_GRAPH_SQL = text(
|
||
"""
|
||
INSERT INTO domrf_kn_sale_graph (
|
||
obj_id, report_month, type, realised, contracted, area_sq, price_avg, snapshot_date
|
||
) VALUES (
|
||
:obj_id, :report_month, :type, :realised, :contracted, :area_sq, :price_avg, :snapshot_date
|
||
)
|
||
ON CONFLICT (obj_id, report_month, type, snapshot_date) DO UPDATE SET
|
||
realised = EXCLUDED.realised,
|
||
contracted = EXCLUDED.contracted,
|
||
area_sq = EXCLUDED.area_sq,
|
||
price_avg = EXCLUDED.price_avg
|
||
"""
|
||
)
|
||
|
||
UPSERT_SALES_AGG_SQL = text(
|
||
"""
|
||
INSERT INTO domrf_kn_sales_agg (
|
||
obj_id, type, name, total, realised, perc, snapshot_date
|
||
) VALUES (
|
||
:obj_id, :type, :name, :total, :realised, :perc, :snapshot_date
|
||
)
|
||
ON CONFLICT (obj_id, type, snapshot_date) DO UPDATE SET
|
||
name = EXCLUDED.name,
|
||
total = EXCLUDED.total,
|
||
realised = EXCLUDED.realised,
|
||
perc = EXCLUDED.perc
|
||
"""
|
||
)
|
||
|
||
UPSERT_INFRA_SQL = text(
|
||
"""
|
||
INSERT INTO domrf_kn_infrastructure (
|
||
obj_id, poi_name, poi_subtitle, poi_category, poi_address,
|
||
poi_lat, poi_lon, distance_m, snapshot_date
|
||
) VALUES (
|
||
:obj_id, :poi_name, :poi_subtitle, :poi_category, :poi_address,
|
||
:poi_lat, :poi_lon, :distance_m, :snapshot_date
|
||
)
|
||
ON CONFLICT (obj_id, poi_name, poi_lat, poi_lon, snapshot_date) DO UPDATE SET
|
||
poi_subtitle = EXCLUDED.poi_subtitle,
|
||
poi_category = EXCLUDED.poi_category,
|
||
poi_address = EXCLUDED.poi_address,
|
||
distance_m = EXCLUDED.distance_m
|
||
"""
|
||
)
|
||
|
||
UPSERT_PHOTO_SQL = text(
|
||
"""
|
||
INSERT INTO domrf_kn_photos (
|
||
obj_id, obj_file_id, ord_num, photo_url, photo_dttm, period_dt,
|
||
size_bytes, photo_name, ready_desc, build_type, hidden,
|
||
local_path, downloaded_at
|
||
) VALUES (
|
||
:obj_id, :obj_file_id, :ord_num, :photo_url, :photo_dttm, :period_dt,
|
||
:size_bytes, :photo_name, :ready_desc, :build_type, :hidden,
|
||
:local_path, :downloaded_at
|
||
)
|
||
ON CONFLICT (obj_id, obj_file_id) DO UPDATE SET
|
||
ord_num = EXCLUDED.ord_num,
|
||
photo_url = EXCLUDED.photo_url,
|
||
photo_dttm = EXCLUDED.photo_dttm,
|
||
period_dt = EXCLUDED.period_dt,
|
||
size_bytes = EXCLUDED.size_bytes,
|
||
photo_name = EXCLUDED.photo_name,
|
||
ready_desc = EXCLUDED.ready_desc,
|
||
hidden = EXCLUDED.hidden,
|
||
local_path = COALESCE(EXCLUDED.local_path, domrf_kn_photos.local_path),
|
||
downloaded_at = COALESCE(EXCLUDED.downloaded_at, domrf_kn_photos.downloaded_at)
|
||
"""
|
||
)
|
||
|
||
INSERT_FAILURE_SQL = text(
|
||
"""
|
||
INSERT INTO kn_scrape_failures (
|
||
run_id, obj_id, endpoint, full_url, http_status, error, body_preview
|
||
) VALUES (:run_id, :obj_id, :endpoint, :full_url, :http_status, :error, :body_preview)
|
||
"""
|
||
)
|
||
|
||
|
||
def _to_dt(s: Any) -> datetime | None:
|
||
"""Coerce 'DD-MM-YYYY HH:MM:SS' or 'YYYY-MM-DD ...' to datetime."""
|
||
if not s or not isinstance(s, str):
|
||
return None
|
||
for fmt in ("%d-%m-%Y %H:%M:%S", "%Y-%m-%d %H:%M:%S", "%d-%m-%Y", "%Y-%m-%d"):
|
||
try:
|
||
return datetime.strptime(s.strip(), fmt)
|
||
except ValueError:
|
||
pass
|
||
return None
|
||
|
||
|
||
def _to_month_date(s: Any) -> date | None:
|
||
"""Coerce '01.04.2025' (sale_graph reportMonthDt) to date."""
|
||
if not s or not isinstance(s, str):
|
||
return None
|
||
for fmt in ("%d.%m.%Y", "%Y-%m-%d"):
|
||
try:
|
||
return datetime.strptime(s.strip(), fmt).date()
|
||
except ValueError:
|
||
pass
|
||
return None
|
||
|
||
|
||
def _http_status_of(e: Exception) -> int | None:
|
||
"""Try to extract HTTP status from exception message like 'http 404: ...'."""
|
||
msg = str(e)
|
||
if msg.startswith("http "):
|
||
try:
|
||
return int(msg.split()[1].rstrip(":"))
|
||
except (IndexError, ValueError):
|
||
return None
|
||
return None
|
||
|
||
|
||
def _body_of(e: Exception) -> str:
|
||
"""Extract body preview from exception message."""
|
||
msg = str(e)
|
||
if ":" in msg:
|
||
return msg.split(":", 1)[1].strip()[:500]
|
||
return msg[:500]
|
||
|
||
|
||
def _classify_and_log(
|
||
db: Session,
|
||
run_id: int | None,
|
||
obj_id: int | None,
|
||
endpoint: str,
|
||
full_url: str,
|
||
e: Exception,
|
||
) -> None:
|
||
"""Log endpoint failure at appropriate severity and persist to kn_scrape_failures.
|
||
|
||
404 from sale_graph/sales_agg is *expected* for сданные ЖК (no active sales) —
|
||
log as DEBUG so it doesn't pollute worker stdout / Sentry. Other errors stay
|
||
WARNING. Failures table records everything for the admin journal.
|
||
"""
|
||
status = _http_status_of(e)
|
||
is_expected_404 = status == 404 and endpoint.startswith(("sale_graph", "sales_agg"))
|
||
if is_expected_404:
|
||
# У сданных ЖК нет активных продаж — 404 ожидаемое поведение API.
|
||
# Не пишем ни в logger.warning (Sentry/Docker), ни в kn_scrape_failures
|
||
# (admin UI). Только debug-уровень для редкого диагностирования.
|
||
logger.debug("%s obj=%s 404 (expected for сданные)", endpoint, obj_id)
|
||
return
|
||
logger.warning("%s obj=%s FAILED: %s\n URL: %s", endpoint, obj_id, e, full_url)
|
||
_log_failure(
|
||
db,
|
||
run_id,
|
||
obj_id,
|
||
endpoint,
|
||
full_url,
|
||
status,
|
||
str(e),
|
||
_body_of(e),
|
||
)
|
||
|
||
|
||
def _log_failure(
|
||
db: Session,
|
||
run_id: int | None,
|
||
obj_id: int | None,
|
||
endpoint: str,
|
||
full_url: str,
|
||
http_status: int | None,
|
||
error: str,
|
||
body_preview: str = "",
|
||
) -> None:
|
||
"""Record a per-request failure into kn_scrape_failures.
|
||
|
||
Caller commits later — this just stages the INSERT in the current transaction.
|
||
Uses a SAVEPOINT so a serialisation issue here doesn't poison the outer txn.
|
||
"""
|
||
if run_id is None:
|
||
return
|
||
try:
|
||
db.execute(
|
||
INSERT_FAILURE_SQL,
|
||
{
|
||
"run_id": run_id,
|
||
"obj_id": obj_id,
|
||
"endpoint": endpoint[:255] if endpoint else None,
|
||
"full_url": full_url[:1000] if full_url else None,
|
||
"http_status": http_status,
|
||
"error": error[:1000] if error else None,
|
||
"body_preview": body_preview[:500] if body_preview else None,
|
||
},
|
||
)
|
||
except Exception as e:
|
||
logger.warning("failed to log failure: %s", e)
|
||
db.rollback()
|
||
|
||
|
||
async def fetch_sale_graph(
|
||
sess: BrowserSession, obj_id: int, type_: str
|
||
) -> tuple[list[dict[str, Any]], str]:
|
||
"""Return (rows, full_url). On HTTP errors raises — caller logs."""
|
||
path = PATH_SALE_GRAPH.format(obj_id=obj_id)
|
||
payload = await sess.get_json(path, {"type": type_})
|
||
full_url = f"{BASE_URL}{path}?type={type_}"
|
||
rows = (payload.get("data") or {}).get("salesGraphDtos") or []
|
||
return rows, full_url
|
||
|
||
|
||
async def fetch_sales_agg(sess: BrowserSession, obj_id: int) -> tuple[dict[str, Any], str]:
|
||
path = PATH_SALES_AGG.format(obj_id=obj_id)
|
||
payload = await sess.get_json(path, {})
|
||
return (payload.get("data") or {}), f"{BASE_URL}{path}"
|
||
|
||
|
||
async def fetch_infrastructure(
|
||
sess: BrowserSession, obj_id: int
|
||
) -> tuple[list[dict[str, Any]], str]:
|
||
path = PATH_INFRA.format(obj_id=obj_id)
|
||
payload = await sess.get_json(path, {})
|
||
return (payload.get("data") or []), f"{BASE_URL}{path}"
|
||
|
||
|
||
async def fetch_photos(sess: BrowserSession, obj_id: int) -> tuple[list[dict[str, Any]], str]:
|
||
path = PATH_PHOTOS.format(obj_id=obj_id)
|
||
# /photo endpoint returns top-level array, not {data: ...}
|
||
payload = await sess.get_json(path, {})
|
||
full_url = f"{BASE_URL}{path}"
|
||
if isinstance(payload, list):
|
||
return payload, full_url
|
||
return (payload.get("data") or []), full_url
|
||
|
||
|
||
def upsert_sale_graph(
|
||
db: Session, obj_id: int, type_: str, rows: list[dict[str, Any]], snapshot_date: date
|
||
) -> int:
|
||
n = 0
|
||
for r in rows:
|
||
rm = _to_month_date(r.get("reportMonthDt"))
|
||
if rm is None:
|
||
continue
|
||
try:
|
||
db.execute(
|
||
UPSERT_SALE_GRAPH_SQL,
|
||
{
|
||
"obj_id": obj_id,
|
||
"report_month": rm,
|
||
"type": type_,
|
||
"realised": r.get("realised"),
|
||
"contracted": r.get("contracted"),
|
||
"area_sq": r.get("areaSq"),
|
||
"price_avg": r.get("priceAvg"),
|
||
"snapshot_date": snapshot_date,
|
||
},
|
||
)
|
||
n += 1
|
||
except Exception as e:
|
||
logger.warning("upsert sale_graph obj=%s type=%s month=%s: %s", obj_id, type_, rm, e)
|
||
db.rollback()
|
||
return n
|
||
|
||
|
||
def upsert_sales_agg(db: Session, obj_id: int, agg: dict[str, Any], snapshot_date: date) -> int:
|
||
"""agg is {apartmentsAggData: {...}, nonlivAggData: {...}, parkingAggData: {...}}."""
|
||
type_map = {
|
||
"apartmentsAggData": "apartments",
|
||
"nonlivAggData": "nonliv",
|
||
"parkingAggData": "parking",
|
||
}
|
||
n = 0
|
||
for key, type_ in type_map.items():
|
||
d = agg.get(key)
|
||
if not isinstance(d, dict):
|
||
continue
|
||
try:
|
||
db.execute(
|
||
UPSERT_SALES_AGG_SQL,
|
||
{
|
||
"obj_id": obj_id,
|
||
"type": type_,
|
||
"name": d.get("name"),
|
||
"total": d.get("total"),
|
||
"realised": d.get("realised"),
|
||
"perc": d.get("perc"),
|
||
"snapshot_date": snapshot_date,
|
||
},
|
||
)
|
||
n += 1
|
||
except Exception as e:
|
||
logger.warning("upsert sales_agg obj=%s type=%s: %s", obj_id, type_, e)
|
||
db.rollback()
|
||
return n
|
||
|
||
|
||
def upsert_infrastructure(
|
||
db: Session, obj_id: int, pois: list[dict[str, Any]], snapshot_date: date
|
||
) -> int:
|
||
n = 0
|
||
for poi in pois:
|
||
try:
|
||
db.execute(
|
||
UPSERT_INFRA_SQL,
|
||
{
|
||
"obj_id": obj_id,
|
||
"poi_name": (poi.get("name") or "").strip()[:500] or "—",
|
||
"poi_subtitle": poi.get("subtitle"),
|
||
"poi_category": poi.get("tabName"),
|
||
"poi_address": poi.get("address"),
|
||
"poi_lat": poi.get("latitude"),
|
||
"poi_lon": poi.get("longitude"),
|
||
"distance_m": poi.get("dist"),
|
||
"snapshot_date": snapshot_date,
|
||
},
|
||
)
|
||
n += 1
|
||
except Exception as e:
|
||
logger.warning("upsert infra obj=%s poi=%s: %s", obj_id, poi.get("name"), e)
|
||
db.rollback()
|
||
return n
|
||
|
||
|
||
def upsert_photos(
|
||
db: Session,
|
||
obj_id: int,
|
||
photos: list[dict[str, Any]],
|
||
local_paths: dict[str, str] | None = None,
|
||
) -> int:
|
||
"""Upsert photo metadata. local_paths maps file_id → relative path on disk."""
|
||
local_paths = local_paths or {}
|
||
now_dt = datetime.now()
|
||
n = 0
|
||
for p in photos:
|
||
file_id = p.get("objFileId")
|
||
if not file_id:
|
||
continue
|
||
local = local_paths.get(file_id)
|
||
try:
|
||
db.execute(
|
||
UPSERT_PHOTO_SQL,
|
||
{
|
||
"obj_id": obj_id,
|
||
"obj_file_id": file_id,
|
||
"ord_num": p.get("ordNum"),
|
||
"photo_url": p.get("objPhotoUrl"),
|
||
"photo_dttm": _to_dt(p.get("objPhotoDttm")),
|
||
"period_dt": _to_month_date(p.get("objPeriodDt")),
|
||
"size_bytes": p.get("objSize"),
|
||
"photo_name": p.get("objPhotoNm"),
|
||
"ready_desc": p.get("objReadyDesc"),
|
||
"build_type": p.get("objBuildTypeShortDesc"),
|
||
"hidden": _to_bool(p.get("objPhotoHiddenFlg")),
|
||
"local_path": local,
|
||
"downloaded_at": now_dt if local else None,
|
||
},
|
||
)
|
||
n += 1
|
||
except Exception as e:
|
||
logger.warning("upsert photo obj=%s file=%s: %s", obj_id, file_id, e)
|
||
db.rollback()
|
||
return n
|
||
|
||
|
||
async def download_photos(
|
||
sess: BrowserSession,
|
||
obj_id: int,
|
||
photos: list[dict[str, Any]],
|
||
base_dir: Path,
|
||
) -> dict[str, str]:
|
||
"""Download photo binaries to base_dir/{obj_id}/{file_id}.png (or .jpg).
|
||
Returns mapping {file_id: relative_path}. Skips if file already exists with
|
||
matching size_bytes.
|
||
"""
|
||
out: dict[str, str] = {}
|
||
obj_dir = base_dir / str(obj_id)
|
||
obj_dir.mkdir(parents=True, exist_ok=True)
|
||
for p in photos:
|
||
file_id = p.get("objFileId")
|
||
url = p.get("objPhotoUrl")
|
||
if not file_id or not url:
|
||
continue
|
||
# Extension hint from objPhotoNm or default png
|
||
name = (p.get("objPhotoNm") or "").lower()
|
||
ext = ".jpg" if name.endswith((".jpg", ".jpeg")) else ".png"
|
||
local = obj_dir / f"{file_id}{ext}"
|
||
# Skip if size matches
|
||
expected_size = p.get("objSize")
|
||
if local.exists() and expected_size and local.stat().st_size == expected_size:
|
||
out[file_id] = str(local)
|
||
continue
|
||
try:
|
||
data = await sess.download_binary(url)
|
||
local.write_bytes(data)
|
||
out[file_id] = str(local)
|
||
except Exception as e:
|
||
logger.warning("download photo obj=%s file=%s failed: %s", obj_id, file_id, e)
|
||
return out
|
||
|
||
|
||
# ── orchestrator ─────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _place_str(region_code: int) -> str:
|
||
"""Map region code → DOM.РФ `place` parameter. By default uses the regionCode itself
|
||
(e.g. 66=Свердл region). City-level codes (e.g. '0-4'=ЕКБ) can be passed as-is via
|
||
PLACE_OVERRIDE if more granularity is needed.
|
||
"""
|
||
return str(region_code)
|
||
|
||
|
||
async def fetch_objects_for_status(
|
||
sess: BrowserSession, place: str, status: int
|
||
) -> list[dict[str, Any]]:
|
||
"""Fetch ALL objects for a given (place, objStatus) — server returns up to limit=999999."""
|
||
payload = await sess.get_json(
|
||
PATH_OBJECTS,
|
||
{
|
||
"offset": 0,
|
||
"limit": 999999,
|
||
"sortField": "default",
|
||
"sortType": "desc",
|
||
"place": place,
|
||
"objStatus": status,
|
||
},
|
||
)
|
||
rows = _extract_list(payload)
|
||
total = _extract_total(payload)
|
||
logger.info("kn/object place=%s status=%d -> %d/%s rows", place, status, len(rows), total)
|
||
return rows
|
||
|
||
|
||
async def fetch_flats_for_object(sess: BrowserSession, obj_id: int) -> list[dict[str, Any]]:
|
||
"""Fetch flat-table for one object, return flat rows (entrance/floor flattened)."""
|
||
try:
|
||
payload = await sess.get_json(PATH_FLATS_TABLE, {"externalId": obj_id})
|
||
except Exception as e:
|
||
logger.warning("flats fetch obj=%s failed: %s", obj_id, e)
|
||
return []
|
||
# Body shape: {externalId, entrances: [{entranceNumber, floors:[{floorNumber, flats:[...]}]}]}
|
||
return _flatten_table(payload)
|
||
|
||
|
||
async def probe_endpoint(
|
||
url_or_path: str,
|
||
region_code: int = 66,
|
||
headed: bool = False,
|
||
load_state: str | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Quick one-shot probe through a real Playwright session for debugging.
|
||
|
||
Returns {status, content_type, body_preview, full_url}. Use to verify that
|
||
a URL works in our session before/after a failing scraper run.
|
||
"""
|
||
if url_or_path.startswith("http"):
|
||
# split into path+query for BrowserSession.get_json
|
||
from urllib.parse import parse_qsl, urlsplit
|
||
|
||
s = urlsplit(url_or_path)
|
||
path = s.path
|
||
params = dict(parse_qsl(s.query, keep_blank_values=True))
|
||
else:
|
||
path = url_or_path.split("?", 1)[0]
|
||
params = {}
|
||
if "?" in url_or_path:
|
||
from urllib.parse import parse_qsl
|
||
|
||
params = dict(parse_qsl(url_or_path.split("?", 1)[1], keep_blank_values=True))
|
||
|
||
async with BrowserSession(
|
||
region_code=region_code, headed=headed, load_state=load_state
|
||
) as sess:
|
||
try:
|
||
payload = await sess.get_json(path, params)
|
||
body = json.dumps(payload, ensure_ascii=False)[:2000]
|
||
qs = "&".join(f"{k}={v}" for k, v in params.items())
|
||
full_url = f"{BASE_URL}{path}" + (f"?{qs}" if qs else "")
|
||
return {
|
||
"status": 200,
|
||
"content_type": "application/json",
|
||
"body_preview": body,
|
||
"full_url": full_url,
|
||
}
|
||
except Exception as e:
|
||
return {
|
||
"status": _http_status_of(e),
|
||
"content_type": "error",
|
||
"body_preview": str(e)[:2000],
|
||
"full_url": f"{BASE_URL}{path}",
|
||
}
|
||
|
||
|
||
async def run_region_sweep(
|
||
region_code: int,
|
||
developers: list[str] | None = None,
|
||
headed: bool = False,
|
||
snapshot_date: date | None = None,
|
||
load_state: str | None = None,
|
||
save_state: str | None = None,
|
||
statuses: tuple[int, ...] = DEFAULT_STATUSES,
|
||
fetch_flats: bool = True,
|
||
extras: bool = True,
|
||
download_photos_binary: bool = False,
|
||
photos_dir: Path | str | None = None,
|
||
place_override: str | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Full sweep for one region: bootstrap → fetch objects (each objStatus) → fetch flats.
|
||
|
||
`developers` filters the OBJECT set locally by `developer.companyGroup` matching the
|
||
"<group>_0" id pattern (since DOM.РФ doesn't honour developer filter on the API).
|
||
"""
|
||
snapshot_date = snapshot_date or date.today()
|
||
db = SessionLocal()
|
||
run_id = db.execute(
|
||
text(
|
||
"""
|
||
INSERT INTO kn_scrape_runs (region_codes, developer_ids, snapshot_date, status)
|
||
VALUES (ARRAY[:rc]::int[], CAST(:devs AS text[]), :snap, 'running')
|
||
RETURNING run_id
|
||
"""
|
||
),
|
||
{
|
||
"rc": region_code,
|
||
"devs": developers if developers else None,
|
||
"snap": snapshot_date,
|
||
},
|
||
).scalar_one()
|
||
db.commit()
|
||
|
||
place = place_override or _place_str(region_code)
|
||
|
||
try:
|
||
async with BrowserSession(
|
||
region_code=region_code,
|
||
headed=headed,
|
||
load_state=load_state,
|
||
save_state=save_state,
|
||
) as sess:
|
||
# --- objects across statuses -------------------------------------
|
||
all_objects: list[dict[str, Any]] = []
|
||
for status in statuses:
|
||
rows = await fetch_objects_for_status(sess, place, status)
|
||
all_objects.extend(rows)
|
||
logger.info("place=%s total objects across %s = %d", place, statuses, len(all_objects))
|
||
|
||
# local developer filter
|
||
if developers:
|
||
wanted_groups = {int(d.split("_")[0]) for d in developers if "_" in d}
|
||
all_objects = [
|
||
o
|
||
for o in all_objects
|
||
if isinstance(o.get("developer"), dict)
|
||
and o["developer"].get("companyGroup") in wanted_groups
|
||
]
|
||
logger.info("after developers filter: %d objects", len(all_objects))
|
||
|
||
# --- flats per object --------------------------------------------
|
||
all_flats: list[dict[str, Any]] = []
|
||
if fetch_flats:
|
||
for i, obj in enumerate(all_objects):
|
||
obj_id = obj.get("objId") or obj.get("obj_id")
|
||
if not obj_id:
|
||
continue
|
||
flats = await fetch_flats_for_object(sess, obj_id)
|
||
all_flats.extend(flats)
|
||
if (i + 1) % 25 == 0:
|
||
logger.info(
|
||
"flats progress: %d/%d objs, %d flats so far",
|
||
i + 1,
|
||
len(all_objects),
|
||
len(all_flats),
|
||
)
|
||
logger.info("total flats: %d (across %d objs)", len(all_flats), len(all_objects))
|
||
|
||
# --- extras (sale_graph, sales_agg, infra, photos) per object ----
|
||
extras_counts = {
|
||
"sale_graph_rows": 0,
|
||
"sales_agg_rows": 0,
|
||
"infra_rows": 0,
|
||
"photos_rows": 0,
|
||
"photos_downloaded": 0,
|
||
}
|
||
if extras:
|
||
pdir = Path(photos_dir) if photos_dir else PHOTOS_DIR_DEFAULT
|
||
for i, obj in enumerate(all_objects):
|
||
obj_id = obj.get("objId") or obj.get("obj_id")
|
||
if not obj_id:
|
||
continue
|
||
|
||
# sale_graph (apartments + parking)
|
||
for type_ in SALE_GRAPH_TYPES:
|
||
try:
|
||
rows, full_url = await fetch_sale_graph(sess, obj_id, type_)
|
||
extras_counts["sale_graph_rows"] += upsert_sale_graph(
|
||
db, obj_id, type_, rows, snapshot_date
|
||
)
|
||
except Exception as e:
|
||
full_url = (
|
||
f"{BASE_URL}{PATH_SALE_GRAPH.format(obj_id=obj_id)}?type={type_}"
|
||
)
|
||
_classify_and_log(
|
||
db, run_id, obj_id, f"sale_graph_{type_}", full_url, e
|
||
)
|
||
|
||
# sales_agg
|
||
try:
|
||
agg, full_url = await fetch_sales_agg(sess, obj_id)
|
||
extras_counts["sales_agg_rows"] += upsert_sales_agg(
|
||
db, obj_id, agg, snapshot_date
|
||
)
|
||
except Exception as e:
|
||
full_url = f"{BASE_URL}{PATH_SALES_AGG.format(obj_id=obj_id)}"
|
||
_classify_and_log(db, run_id, obj_id, "sales_agg", full_url, e)
|
||
|
||
# infrastructure
|
||
try:
|
||
pois, full_url = await fetch_infrastructure(sess, obj_id)
|
||
extras_counts["infra_rows"] += upsert_infrastructure(
|
||
db, obj_id, pois, snapshot_date
|
||
)
|
||
except Exception as e:
|
||
full_url = f"{BASE_URL}{PATH_INFRA.format(obj_id=obj_id)}"
|
||
_classify_and_log(db, run_id, obj_id, "infrastructure", full_url, e)
|
||
|
||
# photos (metadata + optional binary)
|
||
try:
|
||
photos, full_url = await fetch_photos(sess, obj_id)
|
||
local_paths: dict[str, str] = {}
|
||
if download_photos_binary and photos:
|
||
local_paths = await download_photos(sess, obj_id, photos, pdir)
|
||
extras_counts["photos_downloaded"] += len(local_paths)
|
||
extras_counts["photos_rows"] += upsert_photos(
|
||
db, obj_id, photos, local_paths
|
||
)
|
||
except Exception as e:
|
||
full_url = f"{BASE_URL}{PATH_PHOTOS.format(obj_id=obj_id)}"
|
||
_classify_and_log(db, run_id, obj_id, "photos", full_url, e)
|
||
|
||
if (i + 1) % 10 == 0:
|
||
logger.info(
|
||
"extras progress: %d/%d objs, %s",
|
||
i + 1,
|
||
len(all_objects),
|
||
extras_counts,
|
||
)
|
||
# Periodic commit so failures are visible in admin UI even mid-run
|
||
db.commit()
|
||
logger.info("extras done: %s", extras_counts)
|
||
|
||
request_count = sess.request_count
|
||
|
||
# --- write all in one transaction ------------------------------------
|
||
obj_count = upsert_objects(db, all_objects, snapshot_date, region_code)
|
||
flat_count = upsert_flats(db, all_flats, snapshot_date, region_code) if fetch_flats else 0
|
||
# Save raw payloads (one row per status; flats are huge so we store the count, not bodies).
|
||
_insert_raw(db, snapshot_date, f"kn_object_place_{place}", all_objects)
|
||
|
||
db.execute(
|
||
text(
|
||
"""
|
||
UPDATE kn_scrape_runs
|
||
SET finished_at=NOW(), objects_count=:oc, flats_count=:fc,
|
||
requests_count=:rq, status='done'
|
||
WHERE run_id=:rid
|
||
"""
|
||
),
|
||
{"oc": obj_count, "fc": flat_count, "rq": request_count, "rid": run_id},
|
||
)
|
||
db.commit()
|
||
return {
|
||
"run_id": run_id,
|
||
"region_code": region_code,
|
||
"place": place,
|
||
"objects": obj_count,
|
||
"flats": flat_count,
|
||
"requests": request_count,
|
||
"snapshot_date": snapshot_date.isoformat(),
|
||
**extras_counts,
|
||
}
|
||
except Exception as e:
|
||
logger.exception("kn sweep failed for region=%d", region_code)
|
||
db.rollback()
|
||
db.execute(
|
||
text(
|
||
"""
|
||
UPDATE kn_scrape_runs
|
||
SET finished_at=NOW(), status='failed', error=:err
|
||
WHERE run_id=:rid
|
||
"""
|
||
),
|
||
{"err": str(e)[:2000], "rid": run_id},
|
||
)
|
||
db.commit()
|
||
raise
|
||
finally:
|
||
db.close()
|