2378 lines
98 KiB
Python
2378 lines
98 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 asyncio
|
||
import hashlib
|
||
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.documents import extract_documents, upsert_documents
|
||
|
||
# obj_checks import temporarily disabled — endpoint /checks returns 404 (run #19).
|
||
# Re-enable with _fetch_obj_checks_safe when endpoint is found (see TODO in Phase B/C).
|
||
# from app.services.scrapers.obj_checks import extract_obj_checks, upsert_obj_checks
|
||
from app.services.scrapers.stealth import BASE_URL, BrowserSession, WafBlockedError
|
||
|
||
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}"
|
||
# Five separate document endpoints discovered via Playwright on obj=65136 (2026-05-17).
|
||
# Former single PATH_DOCUMENTS (/documents) returned 404 — replaced by these 5 real paths.
|
||
PATH_DOC_RPD = "/сервисы/api/object/{obj_id}/document/rpd"
|
||
PATH_DOC_DEVELOPER_REPORT = "/сервисы/api/object/{obj_id}/developer/report"
|
||
PATH_DOC_PROJECT_DOCUMENTATION = "/сервисы/api/object/{obj_id}/project/documentation"
|
||
PATH_DOC_DOCUMENTATION_OTHER = "/сервисы/api/object/{obj_id}/documentation/other"
|
||
PATH_DOC_PERMITS = "/сервисы/api/object/{obj_id}/document/permits"
|
||
# NOTE: PATH_CHECKS URL not verified via devtools — likely pattern analogous to /infrastructure.
|
||
# If endpoint returns 404/error the failure is logged to kn_scrape_failures and scrape continues.
|
||
PATH_CHECKS = "/сервисы/api/object/{obj_id}/checks"
|
||
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_int(v: Any) -> int | None:
|
||
"""Coerce a DOM.РФ numeric value into int. None / non-numeric / empty → None.
|
||
Accepts int, float (truncated), and numeric strings ('12', '12.0')."""
|
||
if v is None or isinstance(v, bool):
|
||
return None
|
||
if isinstance(v, int):
|
||
return v
|
||
if isinstance(v, float):
|
||
try:
|
||
return int(v) # int(inf) → OverflowError, int(nan) → ValueError
|
||
except (ValueError, OverflowError):
|
||
return None
|
||
if isinstance(v, str):
|
||
s = v.strip()
|
||
if not s:
|
||
return None
|
||
try:
|
||
return int(float(s))
|
||
except (ValueError, OverflowError):
|
||
return None
|
||
return None
|
||
|
||
|
||
def _to_float(v: Any) -> float | None:
|
||
"""Coerce a DOM.РФ numeric value into float. None / bool / non-numeric / empty → None.
|
||
Accepts int, float, and numeric strings ('45.2'). Mirror of _to_int for area/price."""
|
||
if v is None or isinstance(v, bool):
|
||
return None
|
||
if isinstance(v, int | float):
|
||
f = float(v)
|
||
if f != f or f in (float("inf"), float("-inf")): # NaN / ±inf guard
|
||
return None
|
||
return f
|
||
if isinstance(v, str):
|
||
s = v.strip()
|
||
if not s:
|
||
return None
|
||
try:
|
||
f = float(s)
|
||
except (ValueError, OverflowError):
|
||
return None
|
||
if f != f or f in (float("inf"), float("-inf")):
|
||
return None
|
||
return f
|
||
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
|
||
|
||
|
||
_OBJ_CLASS_PATTERNS: list[tuple[str, str]] = [
|
||
# Ordered from most to least specific to avoid 'бизнес' matching inside longer phrases.
|
||
# Each tuple: (regex pattern, canonical class name)
|
||
(r"элит", "Элит"),
|
||
(r"бизнес", "Бизнес"),
|
||
(r"премиум", "Премиум"),
|
||
(r"комфорт", "Комфорт"),
|
||
(r"стандарт|эконом", "Стандарт"),
|
||
# "типового класса" / "типовой серии" / "типовая застройка" — matches DOM.РФ AI-descriptions
|
||
# that use the word "типов*" to describe standard/economy housing. The canonical value
|
||
# "Типовой" matches what the catalog SSR scraper (domrf_catalog.py) extracts from pages.
|
||
# Bug #572: this pattern was absent → 131+ objects remained NULL despite matching aiDescription.
|
||
# Intentionally broad (r"типов" not r"типов[ыоа]й?\b"): DOM.RF aiDescription uses varied
|
||
# inflections; tightening risks dropping coverage below the 70% acceptance threshold.
|
||
(r"типов", "Типовой"),
|
||
]
|
||
|
||
|
||
def _extract_obj_class_from_ai(ai_description: str | None) -> str | None:
|
||
"""Извлечь класс жилья из AI-описания DOM.РФ.
|
||
|
||
DOM.РФ не отдаёт `objClass` в /kn/object list — поле всегда NULL.
|
||
Однако `aiDescription` содержит текстовое упоминание класса ('комфорт-класса',
|
||
'«Комфорт»', 'класс «Бизнес»', 'Класс объекта — комфорт' и т.д.,
|
||
'типового класса', 'типовой серии').
|
||
|
||
Возвращает каноническое название или None если описание пустое / класс не найден.
|
||
"""
|
||
import re
|
||
|
||
if not ai_description:
|
||
return None
|
||
for pattern, class_name in _OBJ_CLASS_PATTERNS:
|
||
if re.search(pattern, ai_description, re.IGNORECASE):
|
||
return class_name
|
||
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 (confirmed via payload audit 2026-05-17, obj=65136):
|
||
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, aiDescription, freeFlatsInfo{priceMin, numberFlats},
|
||
residentialBuildings (section_count), rpdNum (project_declaration_num),
|
||
objPublDt (project_published_at), metro{name, time, line, color, colors, isWalk, id}.
|
||
|
||
Note: DOM.РФ /kn/object list endpoint never populates `objClass` in API responses
|
||
(field absent from payload). Class is extracted from `aiDescription` text as fallback.
|
||
|
||
22e fields NOT in kn-API list payload (remain NULL until catalog scraper 22d):
|
||
first_floor_type, elevators_passenger/cargo_count, parking_total_slots,
|
||
guest_parking_inside/outside_count, ceiling_height_m, finishing_variants_count,
|
||
has_free_planning, playground_kids/sport_count, has_bike_paths, trash_areas_count,
|
||
has_ramp, has_low_platforms, has_wheelchair_lift, flat_area_min/max,
|
||
price_max_rub, price_per_m2_min/max, parking_provision_pct, avg_flat_area_m2,
|
||
domrf_score_location/transport/infrastructure.
|
||
"""
|
||
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
|
||
|
||
# objClass is absent from the list endpoint — extract from aiDescription instead.
|
||
obj_class = _g(row, "objClass") or _extract_obj_class_from_ai(_g(row, "aiDescription"))
|
||
if obj_class is None:
|
||
obj_id = _g(row, "objId", "obj_id", "id")
|
||
logger.debug(
|
||
"obj_class not found for obj_id=%s (no objClass field, no aiDescription class match)",
|
||
obj_id,
|
||
)
|
||
|
||
# Metro: kn-API returns a single metro object (nearest station).
|
||
# Stored as metro_top3 JSON array (single element) for future multi-station extension.
|
||
metro_raw = row.get("metro")
|
||
metro_nearest_name: str | None = None
|
||
metro_nearest_walk_minutes: int | None = None
|
||
metro_top3: list[dict[str, Any]] | None = None
|
||
if isinstance(metro_raw, dict) and metro_raw.get("name"):
|
||
metro_nearest_name = metro_raw.get("name")
|
||
raw_time = metro_raw.get("time")
|
||
if raw_time is not None:
|
||
try:
|
||
metro_nearest_walk_minutes = round(float(raw_time))
|
||
except (TypeError, ValueError):
|
||
pass
|
||
metro_top3 = [
|
||
{
|
||
"name": metro_raw.get("name"),
|
||
"time": raw_time,
|
||
"line": metro_raw.get("line"),
|
||
"color": metro_raw.get("color"),
|
||
"isWalk": metro_raw.get("isWalk"),
|
||
}
|
||
]
|
||
|
||
# freeFlatsInfo.priceMin — price of cheapest available flat in the object
|
||
# freeFlatsInfo.numberFlats — кол-во квартир в текущей продаже (#950 Layer2).
|
||
free_info = row.get("freeFlatsInfo") if isinstance(row.get("freeFlatsInfo"), dict) else {}
|
||
price_min_rub = _g(free_info, "priceMin") if free_info else None
|
||
free_flats = _to_int(_g(free_info, "numberFlats")) if free_info 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": obj_class,
|
||
"wall_type": _g(row, "wallType"),
|
||
"energy_eff": _g(row, "energyEff"),
|
||
"latitude": _g(row, "latitude"),
|
||
"longitude": _g(row, "longitude"),
|
||
"obj_status": _g(row, "objStatus"),
|
||
# 22e: fields mappable from kn-API list payload
|
||
"section_count": _g(row, "residentialBuildings"),
|
||
"project_declaration_num": _g(row, "rpdNum"),
|
||
"project_published_at": _to_date(_g(row, "objPublDt")),
|
||
"price_min_rub": price_min_rub,
|
||
"free_flats": free_flats,
|
||
"dev_group_name": _g(dev, "groupName"),
|
||
# 22e + 22h: metro (nearest station from kn-API, single object)
|
||
"metro_nearest_name": metro_nearest_name,
|
||
"metro_nearest_walk_minutes": metro_nearest_walk_minutes,
|
||
"metro_top3": json.dumps(metro_top3, ensure_ascii=False) if metro_top3 else None,
|
||
# 22e: fields NOT in kn-API list payload — filled by catalog scraper (22d)
|
||
"first_floor_type": None,
|
||
"elevators_passenger_count": None,
|
||
"elevators_cargo_count": None,
|
||
"parking_total_slots": None,
|
||
"guest_parking_inside_count": None,
|
||
"guest_parking_outside_count": None,
|
||
"ceiling_height_m": None,
|
||
"finishing_variants_count": None,
|
||
"has_free_planning": None,
|
||
"avg_flat_area_m2": None,
|
||
"playground_kids_count": None,
|
||
"playground_sport_count": None,
|
||
"has_bike_paths": None,
|
||
"trash_areas_count": None,
|
||
"has_ramp": None,
|
||
"has_low_platforms": None,
|
||
"has_wheelchair_lift": None,
|
||
"flat_area_min": None,
|
||
"flat_area_max": None,
|
||
"price_max_rub": None,
|
||
"price_per_m2_min": None,
|
||
"price_per_m2_max": None,
|
||
"parking_provision_pct": None,
|
||
"domrf_score_location": None,
|
||
"domrf_score_transport": None,
|
||
"domrf_score_infrastructure": None,
|
||
}
|
||
|
||
|
||
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.
|
||
|
||
Note: /portal/table returns price=null for most objects (sold/booked flats and
|
||
objects that don't expose per-flat pricing via this endpoint). price_per_m2 is
|
||
derived from price_rub / total_area when the API returns price_rub but omits
|
||
pricePerSquareMeter — defensive fallback for partial API responses.
|
||
"""
|
||
flat_id = _g(row, "flatId", "id")
|
||
if flat_id is None:
|
||
# Fallback: ДЕТЕРМИНИРОВАННЫЙ sha256-digest от elemId (#1208). Раньше
|
||
# использовался hash(elem), который в CPython РАНДОМИЗИРОВАН per-process
|
||
# (PYTHONHASHSEED нигде в репо не зафиксирован) → при resume упавшего
|
||
# sweep (новый процесс воркера) та же квартира получала ДРУГОЙ id →
|
||
# ON CONFLICT (id, snapshot_date) не срабатывал → дубли строк одной
|
||
# квартиры в одном snapshot, инфляция агрегаций. sha256 стабилен между
|
||
# процессами/перезапусками. Берём 8 байт (BIGINT'у достаточно).
|
||
elem = _g(row, "elemId")
|
||
if elem:
|
||
digest = hashlib.sha256(str(elem).encode("utf-8")).digest()[:8]
|
||
flat_id = int.from_bytes(digest, "big") % (2**63 - 1)
|
||
|
||
price_rub = _g(row, "price")
|
||
price_per_m2 = _g(row, "pricePerSquareMeter")
|
||
total_area = _g(row, "totalArea")
|
||
|
||
# Derive price_per_m2 when API returns price_rub but omits pricePerSquareMeter.
|
||
# Covers cases where the table endpoint has the flat price but no pre-computed m² rate.
|
||
# Coerce оба операнда в float ДО сравнения/деления: API иногда отдаёт totalArea/price
|
||
# строкой ('45.2'), и `total_area > 0` на str роняло бы _norm_flat с TypeError
|
||
# → весь run падал бы в status='failed' (#1644).
|
||
price_rub_num = _to_float(price_rub)
|
||
total_area_num = _to_float(total_area)
|
||
if price_per_m2 is None and price_rub_num is not None and total_area_num and total_area_num > 0:
|
||
price_per_m2 = round(price_rub_num / total_area_num, 2)
|
||
logger.info(
|
||
"derive price_per_m2=%.2f for flat ods_id=%s obj_id=%s",
|
||
price_per_m2,
|
||
_g(row, "odsId"),
|
||
_g(row, "_objId", "objId"),
|
||
)
|
||
|
||
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": total_area,
|
||
"living_area": _g(row, "livingArea"),
|
||
"rooms": _g(row, "rooms"),
|
||
"status": _g(row, "status"),
|
||
"price_rub": price_rub,
|
||
"price_per_m2": price_per_m2,
|
||
"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"),
|
||
# 22b: новые поля квартиры
|
||
"section_no": _g(row, "_entrance"), # entranceNumber injected by _flatten_table
|
||
"finishing_type": None, # заполнит catalog scraper 22d
|
||
"ceiling_height_m": None, # заполнит catalog scraper 22d
|
||
"key_handover_dt": None, # заполнит catalog scraper 22d
|
||
"catalog_updated_at": None, # заполнит catalog scraper 22d
|
||
"catalog_url_hash": None, # заполнит catalog scraper 22d
|
||
}
|
||
|
||
|
||
# ── 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,
|
||
section_count, project_declaration_num, project_published_at,
|
||
price_min_rub, free_flats, dev_group_name,
|
||
metro_nearest_name, metro_nearest_walk_minutes, metro_top3,
|
||
first_floor_type, elevators_passenger_count, elevators_cargo_count,
|
||
parking_total_slots, guest_parking_inside_count, guest_parking_outside_count,
|
||
ceiling_height_m, finishing_variants_count, has_free_planning, avg_flat_area_m2,
|
||
playground_kids_count, playground_sport_count, has_bike_paths, trash_areas_count,
|
||
has_ramp, has_low_platforms, has_wheelchair_lift,
|
||
flat_area_min, flat_area_max, price_max_rub, price_per_m2_min, price_per_m2_max,
|
||
parking_provision_pct, domrf_score_location, domrf_score_transport,
|
||
domrf_score_infrastructure,
|
||
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,
|
||
:section_count, :project_declaration_num, :project_published_at,
|
||
:price_min_rub, :free_flats, :dev_group_name,
|
||
:metro_nearest_name, :metro_nearest_walk_minutes, CAST(:metro_top3 AS jsonb),
|
||
:first_floor_type, :elevators_passenger_count, :elevators_cargo_count,
|
||
:parking_total_slots, :guest_parking_inside_count, :guest_parking_outside_count,
|
||
:ceiling_height_m, :finishing_variants_count, :has_free_planning, :avg_flat_area_m2,
|
||
:playground_kids_count, :playground_sport_count, :has_bike_paths, :trash_areas_count,
|
||
:has_ramp, :has_low_platforms, :has_wheelchair_lift,
|
||
:flat_area_min, :flat_area_max, :price_max_rub, :price_per_m2_min, :price_per_m2_max,
|
||
:parking_provision_pct, :domrf_score_location, :domrf_score_transport,
|
||
:domrf_score_infrastructure,
|
||
: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 = COALESCE(EXCLUDED.obj_class, domrf_kn_objects.obj_class),
|
||
wall_type = COALESCE(EXCLUDED.wall_type, domrf_kn_objects.wall_type),
|
||
energy_eff = COALESCE(EXCLUDED.energy_eff, domrf_kn_objects.energy_eff),
|
||
latitude = EXCLUDED.latitude,
|
||
longitude = EXCLUDED.longitude,
|
||
obj_status = EXCLUDED.obj_status,
|
||
section_count = COALESCE(EXCLUDED.section_count, domrf_kn_objects.section_count),
|
||
project_declaration_num = COALESCE(
|
||
EXCLUDED.project_declaration_num, domrf_kn_objects.project_declaration_num
|
||
),
|
||
project_published_at = COALESCE(
|
||
EXCLUDED.project_published_at, domrf_kn_objects.project_published_at
|
||
),
|
||
price_min_rub = EXCLUDED.price_min_rub,
|
||
free_flats = EXCLUDED.free_flats,
|
||
dev_group_name = COALESCE(EXCLUDED.dev_group_name, domrf_kn_objects.dev_group_name),
|
||
metro_nearest_name = COALESCE(
|
||
EXCLUDED.metro_nearest_name, domrf_kn_objects.metro_nearest_name
|
||
),
|
||
metro_nearest_walk_minutes = COALESCE(
|
||
EXCLUDED.metro_nearest_walk_minutes, domrf_kn_objects.metro_nearest_walk_minutes
|
||
),
|
||
metro_top3 = COALESCE(EXCLUDED.metro_top3, domrf_kn_objects.metro_top3),
|
||
first_floor_type = COALESCE(
|
||
EXCLUDED.first_floor_type, domrf_kn_objects.first_floor_type
|
||
),
|
||
elevators_passenger_count = COALESCE(
|
||
EXCLUDED.elevators_passenger_count, domrf_kn_objects.elevators_passenger_count
|
||
),
|
||
elevators_cargo_count = COALESCE(
|
||
EXCLUDED.elevators_cargo_count, domrf_kn_objects.elevators_cargo_count
|
||
),
|
||
parking_total_slots = COALESCE(
|
||
EXCLUDED.parking_total_slots, domrf_kn_objects.parking_total_slots
|
||
),
|
||
guest_parking_inside_count = COALESCE(
|
||
EXCLUDED.guest_parking_inside_count, domrf_kn_objects.guest_parking_inside_count
|
||
),
|
||
guest_parking_outside_count = COALESCE(
|
||
EXCLUDED.guest_parking_outside_count, domrf_kn_objects.guest_parking_outside_count
|
||
),
|
||
ceiling_height_m = COALESCE(
|
||
EXCLUDED.ceiling_height_m, domrf_kn_objects.ceiling_height_m
|
||
),
|
||
finishing_variants_count = COALESCE(
|
||
EXCLUDED.finishing_variants_count, domrf_kn_objects.finishing_variants_count
|
||
),
|
||
has_free_planning = COALESCE(
|
||
EXCLUDED.has_free_planning, domrf_kn_objects.has_free_planning
|
||
),
|
||
avg_flat_area_m2 = COALESCE(
|
||
EXCLUDED.avg_flat_area_m2, domrf_kn_objects.avg_flat_area_m2
|
||
),
|
||
playground_kids_count = COALESCE(
|
||
EXCLUDED.playground_kids_count, domrf_kn_objects.playground_kids_count
|
||
),
|
||
playground_sport_count = COALESCE(
|
||
EXCLUDED.playground_sport_count, domrf_kn_objects.playground_sport_count
|
||
),
|
||
has_bike_paths = COALESCE(EXCLUDED.has_bike_paths, domrf_kn_objects.has_bike_paths),
|
||
trash_areas_count = COALESCE(
|
||
EXCLUDED.trash_areas_count, domrf_kn_objects.trash_areas_count
|
||
),
|
||
has_ramp = COALESCE(EXCLUDED.has_ramp, domrf_kn_objects.has_ramp),
|
||
has_low_platforms = COALESCE(
|
||
EXCLUDED.has_low_platforms, domrf_kn_objects.has_low_platforms
|
||
),
|
||
has_wheelchair_lift = COALESCE(
|
||
EXCLUDED.has_wheelchair_lift, domrf_kn_objects.has_wheelchair_lift
|
||
),
|
||
flat_area_min = COALESCE(EXCLUDED.flat_area_min, domrf_kn_objects.flat_area_min),
|
||
flat_area_max = COALESCE(EXCLUDED.flat_area_max, domrf_kn_objects.flat_area_max),
|
||
price_max_rub = COALESCE(EXCLUDED.price_max_rub, domrf_kn_objects.price_max_rub),
|
||
price_per_m2_min = COALESCE(
|
||
EXCLUDED.price_per_m2_min, domrf_kn_objects.price_per_m2_min
|
||
),
|
||
price_per_m2_max = COALESCE(
|
||
EXCLUDED.price_per_m2_max, domrf_kn_objects.price_per_m2_max
|
||
),
|
||
parking_provision_pct = COALESCE(
|
||
EXCLUDED.parking_provision_pct, domrf_kn_objects.parking_provision_pct
|
||
),
|
||
domrf_score_location = COALESCE(
|
||
EXCLUDED.domrf_score_location, domrf_kn_objects.domrf_score_location
|
||
),
|
||
domrf_score_transport = COALESCE(
|
||
EXCLUDED.domrf_score_transport, domrf_kn_objects.domrf_score_transport
|
||
),
|
||
domrf_score_infrastructure = COALESCE(
|
||
EXCLUDED.domrf_score_infrastructure, domrf_kn_objects.domrf_score_infrastructure
|
||
)
|
||
"""
|
||
)
|
||
|
||
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,
|
||
section_no, finishing_type, ceiling_height_m,
|
||
key_handover_dt, catalog_updated_at, catalog_url_hash,
|
||
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,
|
||
:section_no, :finishing_type, :ceiling_height_m,
|
||
:key_handover_dt, :catalog_updated_at, :catalog_url_hash,
|
||
:snapshot_date
|
||
)
|
||
ON CONFLICT (id, snapshot_date) DO UPDATE SET
|
||
status = EXCLUDED.status,
|
||
price_rub = EXCLUDED.price_rub,
|
||
-- COALESCE: keep a backfilled / sale_graph-derived price_per_m2 when the
|
||
-- scraper produces NULL (NULL-status historical flats have no price_rub).
|
||
-- Direct scraper value always wins when non-NULL.
|
||
price_per_m2 = COALESCE(EXCLUDED.price_per_m2, domrf_kn_flats.price_per_m2),
|
||
obj_id = EXCLUDED.obj_id,
|
||
region_cd = EXCLUDED.region_cd,
|
||
obj_name = EXCLUDED.obj_name,
|
||
flat_number = COALESCE(EXCLUDED.flat_number, domrf_kn_flats.flat_number),
|
||
living_area = COALESCE(EXCLUDED.living_area, domrf_kn_flats.living_area),
|
||
is_studio = COALESCE(EXCLUDED.is_studio, domrf_kn_flats.is_studio),
|
||
total_area = COALESCE(EXCLUDED.total_area, domrf_kn_flats.total_area),
|
||
rooms = COALESCE(EXCLUDED.rooms, domrf_kn_flats.rooms),
|
||
floor = COALESCE(EXCLUDED.floor, domrf_kn_flats.floor),
|
||
num_floors = COALESCE(EXCLUDED.num_floors, domrf_kn_flats.num_floors),
|
||
section_no = COALESCE(EXCLUDED.section_no, domrf_kn_flats.section_no),
|
||
finishing_type = COALESCE(EXCLUDED.finishing_type, domrf_kn_flats.finishing_type),
|
||
ceiling_height_m = COALESCE(
|
||
EXCLUDED.ceiling_height_m, domrf_kn_flats.ceiling_height_m
|
||
),
|
||
key_handover_dt = COALESCE(EXCLUDED.key_handover_dt, domrf_kn_flats.key_handover_dt),
|
||
catalog_updated_at = COALESCE(
|
||
EXCLUDED.catalog_updated_at, domrf_kn_flats.catalog_updated_at
|
||
),
|
||
catalog_url_hash = COALESCE(
|
||
EXCLUDED.catalog_url_hash, domrf_kn_flats.catalog_url_hash
|
||
)
|
||
"""
|
||
)
|
||
|
||
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 _ensure_snapshot(db: Session, snapshot_date: date) -> None:
|
||
"""Гарантирует наличие row в domrf_snapshots для данной даты. Все дочерние
|
||
таблицы (domrf_raw_endpoints, domrf_kn_objects, etc) имеют FK на эту
|
||
таблицу, поэтому row должен существовать ДО первого INSERT, иначе
|
||
ForeignKeyViolation."""
|
||
db.execute(
|
||
text(
|
||
"""
|
||
INSERT INTO domrf_snapshots (snapshot_date, fetched_at, notes)
|
||
VALUES (:snap, NOW(), 'kn-API sweep')
|
||
ON CONFLICT (snapshot_date) DO NOTHING
|
||
"""
|
||
),
|
||
{"snap": snapshot_date},
|
||
)
|
||
db.commit()
|
||
|
||
|
||
def _backfill_district_names(db: Session, snapshot_date: date, region_code: int) -> None:
|
||
"""Spatial-join свежезаписанных ЖК с ekb_districts_geom через PostGIS.
|
||
Best-effort: если ekb_districts_geom пустая (загрузчик не запущен) —
|
||
UPDATE затронет 0 строк, ничего не ломается.
|
||
|
||
Только для региона 66 (ЕКБ) — для других регионов полигонов нет.
|
||
"""
|
||
if region_code != 66:
|
||
return
|
||
try:
|
||
db.execute(
|
||
text(
|
||
"""
|
||
UPDATE domrf_kn_objects o
|
||
SET district_name = d.district_name
|
||
FROM ekb_districts_geom d
|
||
WHERE o.snapshot_date = :snap
|
||
AND o.region_cd = :rc
|
||
AND o.latitude IS NOT NULL
|
||
AND o.longitude IS NOT NULL
|
||
AND (o.district_name IS NULL OR o.district_name <> d.district_name)
|
||
AND ST_Contains(
|
||
d.geom,
|
||
ST_SetSRID(ST_MakePoint(o.longitude, o.latitude), 4326)
|
||
)
|
||
"""
|
||
),
|
||
{"snap": snapshot_date, "rc": region_code},
|
||
)
|
||
db.commit()
|
||
except Exception as e:
|
||
logger.warning("backfill district_name failed: %s", e)
|
||
try:
|
||
db.rollback()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
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")),
|
||
},
|
||
)
|
||
|
||
|
||
_UPSERT_DEVELOPER_SQL = text(
|
||
"""
|
||
INSERT INTO domrf_developers (developer_id, developer_name)
|
||
VALUES (:id, :name)
|
||
ON CONFLICT (developer_id) DO NOTHING
|
||
"""
|
||
)
|
||
|
||
|
||
def _ensure_developers(db: Session, raw: list[dict[str, Any]]) -> int:
|
||
"""Pre-upsert всех developer'ов, на которых ссылаются объекты — иначе
|
||
FK fk_kn_obj_developer (dev_id → domrf_developers) валит upsert_objects.
|
||
Возвращает кол-во новых записей."""
|
||
seen: dict[str, str] = {} # dev_id → name
|
||
for row in raw:
|
||
dev = row.get("developer") if isinstance(row.get("developer"), dict) else {}
|
||
cg = _g(dev, "companyGroup") if dev else None
|
||
if not cg:
|
||
continue
|
||
dev_id = f"{cg}_0"
|
||
if dev_id in seen:
|
||
continue
|
||
# developer_name NOT NULL — fallback на companyGroup как строку.
|
||
name = (
|
||
_g(dev, "groupName")
|
||
or _g(dev, "shortName")
|
||
or _g(dev, "fullName")
|
||
or f"Developer #{cg}"
|
||
)
|
||
seen[dev_id] = name
|
||
inserted = 0
|
||
for dev_id, name in seen.items():
|
||
try:
|
||
with db.begin_nested():
|
||
db.execute(_UPSERT_DEVELOPER_SQL, {"id": dev_id, "name": name})
|
||
inserted += 1
|
||
except Exception as e:
|
||
logger.warning("ensure developer %s failed: %s", dev_id, e)
|
||
db.commit()
|
||
return inserted
|
||
|
||
|
||
def upsert_objects(
|
||
db: Session, raw: list[dict[str, Any]], snapshot_date: date, region_cd: int
|
||
) -> int:
|
||
# Без этого FK fk_kn_obj_developer валит upsert на новых девелоперах.
|
||
_ensure_developers(db, raw)
|
||
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
|
||
# SAVEPOINT (db.begin_nested) — при FK/CHECK violation одного объекта
|
||
# rollback'ается ТОЛЬКО он, остальные uplinks этой транзакции живы.
|
||
# Старый db.rollback() сбрасывал всю транзакцию целиком (теряя сотни
|
||
# уже успешно закоммиченных объектов).
|
||
try:
|
||
with db.begin_nested():
|
||
db.execute(UPSERT_OBJECT_SQL, norm)
|
||
inserted += 1
|
||
except Exception as e:
|
||
logger.warning("upsert object %s failed: %s", norm.get("obj_id"), e)
|
||
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:
|
||
with db.begin_nested():
|
||
db.execute(UPSERT_FLAT_SQL, norm)
|
||
inserted += 1
|
||
except Exception as e:
|
||
logger.warning("upsert flat %s failed: %s", norm.get("id"), e)
|
||
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
|
||
)
|
||
-- Issue #297 22j: новый UNIQUE (obj_id, poi_category, poi_name, poi_address) — без
|
||
-- snapshot_date, чтобы каждый scrape run не накапливал ×N дубликаты POI.
|
||
ON CONFLICT (obj_id, poi_category, poi_name, poi_address) DO NOTHING
|
||
"""
|
||
)
|
||
|
||
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, thumb_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, :thumb_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),
|
||
thumb_path = COALESCE(EXCLUDED.thumb_path, domrf_kn_photos.thumb_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:
|
||
with db.begin_nested():
|
||
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)
|
||
|
||
|
||
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
|
||
|
||
|
||
async def _fetch_doc_endpoint(
|
||
sess: BrowserSession, path_template: str, obj_id: int
|
||
) -> tuple[list[dict[str, Any]], str]:
|
||
"""Generic helper: fetch one document endpoint, return (items, full_url)."""
|
||
path = path_template.format(obj_id=obj_id)
|
||
payload = await sess.get_json(path, {})
|
||
full_url = f"{BASE_URL}{path}"
|
||
if isinstance(payload, list):
|
||
return payload, full_url
|
||
data = payload.get("data")
|
||
if isinstance(data, list):
|
||
return data, full_url
|
||
return [], full_url
|
||
|
||
|
||
async def fetch_doc_rpd(sess: BrowserSession, obj_id: int) -> tuple[list[dict[str, Any]], str]:
|
||
"""214-ФЗ отчёт (РПД)."""
|
||
return await _fetch_doc_endpoint(sess, PATH_DOC_RPD, obj_id)
|
||
|
||
|
||
async def fetch_doc_developer_report(
|
||
sess: BrowserSession, obj_id: int
|
||
) -> tuple[list[dict[str, Any]], str]:
|
||
"""Отчёт застройщика."""
|
||
return await _fetch_doc_endpoint(sess, PATH_DOC_DEVELOPER_REPORT, obj_id)
|
||
|
||
|
||
async def fetch_doc_project_documentation(
|
||
sess: BrowserSession, obj_id: int
|
||
) -> tuple[list[dict[str, Any]], str]:
|
||
"""Проектная декларация."""
|
||
return await _fetch_doc_endpoint(sess, PATH_DOC_PROJECT_DOCUMENTATION, obj_id)
|
||
|
||
|
||
async def fetch_doc_documentation_other(
|
||
sess: BrowserSession, obj_id: int
|
||
) -> tuple[list[dict[str, Any]], str]:
|
||
"""Прочие документы."""
|
||
return await _fetch_doc_endpoint(sess, PATH_DOC_DOCUMENTATION_OTHER, obj_id)
|
||
|
||
|
||
async def fetch_doc_permits(sess: BrowserSession, obj_id: int) -> tuple[list[dict[str, Any]], str]:
|
||
"""Разрешения на строительство."""
|
||
return await _fetch_doc_endpoint(sess, PATH_DOC_PERMITS, obj_id)
|
||
|
||
|
||
async def fetch_obj_checks(sess: BrowserSession, obj_id: int) -> tuple[Any, str]:
|
||
"""Fetch 6 «Проверено на наш.дом.рф» checks for one object. Returns (payload, full_url).
|
||
|
||
Endpoint URL (/checks) не верифицирован через devtools — выведен по паттерну
|
||
/infrastructure, /documents. При HTTP-ошибке вызывающий код запишет failure в
|
||
kn_scrape_failures; scrape не прерывается.
|
||
"""
|
||
path = PATH_CHECKS.format(obj_id=obj_id)
|
||
payload = await sess.get_json(path, {})
|
||
return payload, f"{BASE_URL}{path}"
|
||
|
||
|
||
# ── _fetch_*_safe wrappers for asyncio.gather in Phase B/C ───────────────────
|
||
# Каждый wrapper возвращает (kind, full_url, result_or_exception).
|
||
# Exceptions НЕ raise — помещаются в возвращаемый tuple.
|
||
# BrowserSession._sem (size = per-instance concurrency, #1945 KN=2) bounds concurrency.
|
||
|
||
|
||
async def _fetch_flats_safe(
|
||
sess: BrowserSession, obj_id: int
|
||
) -> tuple[str, str, list[dict[str, Any]] | Exception]:
|
||
full_url = f"{BASE_URL}{PATH_FLATS_TABLE}?externalId={obj_id}"
|
||
try:
|
||
flats = await fetch_flats_for_object(sess, obj_id)
|
||
return ("flats", full_url, flats)
|
||
except Exception as e:
|
||
return ("flats", full_url, e)
|
||
|
||
|
||
async def _fetch_sale_graph_safe(
|
||
sess: BrowserSession, obj_id: int, type_: str
|
||
) -> tuple[str, str, tuple[list[dict[str, Any]], str] | Exception]:
|
||
full_url = f"{BASE_URL}{PATH_SALE_GRAPH.format(obj_id=obj_id)}?type={type_}"
|
||
try:
|
||
rows, url = await fetch_sale_graph(sess, obj_id, type_)
|
||
return (f"sale_graph_{type_}", url, (rows, url))
|
||
except Exception as e:
|
||
return (f"sale_graph_{type_}", full_url, e)
|
||
|
||
|
||
async def _fetch_sales_agg_safe(
|
||
sess: BrowserSession, obj_id: int
|
||
) -> tuple[str, str, tuple[dict[str, Any], str] | Exception]:
|
||
full_url = f"{BASE_URL}{PATH_SALES_AGG.format(obj_id=obj_id)}"
|
||
try:
|
||
agg, url = await fetch_sales_agg(sess, obj_id)
|
||
return ("sales_agg", url, (agg, url))
|
||
except Exception as e:
|
||
return ("sales_agg", full_url, e)
|
||
|
||
|
||
async def _fetch_infrastructure_safe(
|
||
sess: BrowserSession, obj_id: int
|
||
) -> tuple[str, str, tuple[list[dict[str, Any]], str] | Exception]:
|
||
full_url = f"{BASE_URL}{PATH_INFRA.format(obj_id=obj_id)}"
|
||
try:
|
||
pois, url = await fetch_infrastructure(sess, obj_id)
|
||
return ("infrastructure", url, (pois, url))
|
||
except Exception as e:
|
||
return ("infrastructure", full_url, e)
|
||
|
||
|
||
async def _fetch_photos_safe(
|
||
sess: BrowserSession, obj_id: int
|
||
) -> tuple[str, str, tuple[list[dict[str, Any]], str] | Exception]:
|
||
full_url = f"{BASE_URL}{PATH_PHOTOS.format(obj_id=obj_id)}"
|
||
try:
|
||
photos, url = await fetch_photos(sess, obj_id)
|
||
return ("photos", url, (photos, url))
|
||
except Exception as e:
|
||
return ("photos", full_url, e)
|
||
|
||
|
||
async def _fetch_doc_rpd_safe(
|
||
sess: BrowserSession, obj_id: int
|
||
) -> tuple[str, str, tuple[list[dict[str, Any]], str] | Exception]:
|
||
full_url = f"{BASE_URL}{PATH_DOC_RPD.format(obj_id=obj_id)}"
|
||
try:
|
||
items, url = await fetch_doc_rpd(sess, obj_id)
|
||
return ("doc_rpd", url, (items, url))
|
||
except Exception as e:
|
||
return ("doc_rpd", full_url, e)
|
||
|
||
|
||
async def _fetch_doc_developer_report_safe(
|
||
sess: BrowserSession, obj_id: int
|
||
) -> tuple[str, str, tuple[list[dict[str, Any]], str] | Exception]:
|
||
full_url = f"{BASE_URL}{PATH_DOC_DEVELOPER_REPORT.format(obj_id=obj_id)}"
|
||
try:
|
||
items, url = await fetch_doc_developer_report(sess, obj_id)
|
||
return ("doc_developer_report", url, (items, url))
|
||
except Exception as e:
|
||
return ("doc_developer_report", full_url, e)
|
||
|
||
|
||
async def _fetch_doc_project_documentation_safe(
|
||
sess: BrowserSession, obj_id: int
|
||
) -> tuple[str, str, tuple[list[dict[str, Any]], str] | Exception]:
|
||
full_url = f"{BASE_URL}{PATH_DOC_PROJECT_DOCUMENTATION.format(obj_id=obj_id)}"
|
||
try:
|
||
items, url = await fetch_doc_project_documentation(sess, obj_id)
|
||
return ("doc_project_documentation", url, (items, url))
|
||
except Exception as e:
|
||
return ("doc_project_documentation", full_url, e)
|
||
|
||
|
||
async def _fetch_doc_documentation_other_safe(
|
||
sess: BrowserSession, obj_id: int
|
||
) -> tuple[str, str, tuple[list[dict[str, Any]], str] | Exception]:
|
||
full_url = f"{BASE_URL}{PATH_DOC_DOCUMENTATION_OTHER.format(obj_id=obj_id)}"
|
||
try:
|
||
items, url = await fetch_doc_documentation_other(sess, obj_id)
|
||
return ("doc_documentation_other", url, (items, url))
|
||
except Exception as e:
|
||
return ("doc_documentation_other", full_url, e)
|
||
|
||
|
||
async def _fetch_doc_permits_safe(
|
||
sess: BrowserSession, obj_id: int
|
||
) -> tuple[str, str, tuple[list[dict[str, Any]], str] | Exception]:
|
||
full_url = f"{BASE_URL}{PATH_DOC_PERMITS.format(obj_id=obj_id)}"
|
||
try:
|
||
items, url = await fetch_doc_permits(sess, obj_id)
|
||
return ("doc_permits", url, (items, url))
|
||
except Exception as e:
|
||
return ("doc_permits", full_url, e)
|
||
|
||
|
||
async def _fetch_obj_checks_safe(
|
||
sess: BrowserSession, obj_id: int
|
||
) -> tuple[str, str, tuple[Any, str] | Exception]:
|
||
full_url = f"{BASE_URL}{PATH_CHECKS.format(obj_id=obj_id)}"
|
||
try:
|
||
payload, url = await fetch_obj_checks(sess, obj_id)
|
||
return ("obj_checks", url, (payload, url))
|
||
except Exception as e:
|
||
return ("obj_checks", full_url, e)
|
||
|
||
|
||
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:
|
||
with db.begin_nested():
|
||
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)
|
||
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:
|
||
with db.begin_nested():
|
||
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)
|
||
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:
|
||
with db.begin_nested():
|
||
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)
|
||
return n
|
||
|
||
|
||
def upsert_photos(
|
||
db: Session,
|
||
obj_id: int,
|
||
photos: list[dict[str, Any]],
|
||
local_paths: dict[str, str] | None = None,
|
||
thumb_paths: dict[str, str] | None = None,
|
||
) -> int:
|
||
"""Upsert photo metadata. local_paths/thumb_paths map file_id → path."""
|
||
local_paths = local_paths or {}
|
||
thumb_paths = thumb_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)
|
||
thumb = thumb_paths.get(file_id)
|
||
try:
|
||
with db.begin_nested():
|
||
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,
|
||
"thumb_path": thumb,
|
||
"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)
|
||
return n
|
||
|
||
|
||
async def download_photos(
|
||
sess: BrowserSession,
|
||
obj_id: int,
|
||
photos: list[dict[str, Any]],
|
||
base_dir: Path,
|
||
) -> tuple[dict[str, str], dict[str, str]]:
|
||
"""Download photo binaries to base_dir/{obj_id}/{file_id}.png (or .jpg) +
|
||
generate WebP thumbnails next to them.
|
||
|
||
Returns a pair of mappings: (local_paths, thumb_paths) — both keyed by
|
||
file_id, value = relative path on disk. Skips originals already present
|
||
with matching size_bytes; thumbs skipped if already exist.
|
||
"""
|
||
from app.services.photos.thumbs import make_thumbnail, thumb_path_for
|
||
|
||
locals_out: dict[str, str] = {}
|
||
thumbs_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:
|
||
locals_out[file_id] = str(local)
|
||
else:
|
||
try:
|
||
data = await sess.download_binary(url)
|
||
local.write_bytes(data)
|
||
locals_out[file_id] = str(local)
|
||
except Exception as e:
|
||
logger.warning("download photo obj=%s file=%s failed: %s", obj_id, file_id, e)
|
||
continue
|
||
# Always (re)check thumbnail. make_thumbnail is no-op if it already exists.
|
||
existing_thumb = thumb_path_for(local)
|
||
if existing_thumb.exists():
|
||
thumbs_out[file_id] = str(existing_thumb)
|
||
else:
|
||
t = make_thumbnail(local)
|
||
if t:
|
||
thumbs_out[file_id] = str(t)
|
||
return locals_out, thumbs_out
|
||
|
||
|
||
# ── result processing (shared by combined + isolated #1945 passes) ────────────
|
||
|
||
# #1945 isolated extras pass: после стольких подряд полностью-WAF объектов
|
||
# обрываем пас (extras заблокированы целиком — нет смысла жечь ~1500 recycle).
|
||
_EXTRAS_ABORT_AFTER_CONSEC_WAF = 5
|
||
|
||
# 5 doc-endpoint kinds, accumulated into a single per-object documents upsert.
|
||
_DOC_KINDS = frozenset(
|
||
(
|
||
"doc_rpd",
|
||
"doc_developer_report",
|
||
"doc_project_documentation",
|
||
"doc_documentation_other",
|
||
"doc_permits",
|
||
)
|
||
)
|
||
|
||
|
||
def _process_flats_result(
|
||
db: Session,
|
||
run_id: int,
|
||
obj_id: int,
|
||
full_url: str,
|
||
result: list[dict[str, Any]] | Exception,
|
||
snapshot_date: date,
|
||
region_code: int,
|
||
) -> int:
|
||
"""Upsert flats from a _fetch_flats_safe result. Returns flats inserted.
|
||
|
||
Errors → kn_scrape_failures (via _classify_and_log), 0 returned. Вынесено из
|
||
Phase-B/C loop чтобы isolated-flats-pass (#1945) и combined-loop делили один код.
|
||
"""
|
||
if isinstance(result, Exception):
|
||
_classify_and_log(db, run_id, obj_id, "flats", full_url, result)
|
||
return 0
|
||
if result:
|
||
return upsert_flats(db, result, snapshot_date, region_code)
|
||
return 0
|
||
|
||
|
||
async def _process_extras_result(
|
||
sess: BrowserSession,
|
||
db: Session,
|
||
run_id: int,
|
||
obj_id: int,
|
||
kind_tag: str,
|
||
full_url: str,
|
||
result: Any,
|
||
snapshot_date: date,
|
||
extras_counts: dict[str, int],
|
||
all_docs: list[dict[str, Any]],
|
||
*,
|
||
download_photos_binary: bool,
|
||
pdir: Path,
|
||
) -> None:
|
||
"""Handle one extras endpoint result (sale_graph/sales_agg/infra/photos/docs).
|
||
|
||
Mutates extras_counts / all_docs in place. Errors → kn_scrape_failures.
|
||
Doc-kinds накапливаются в all_docs (единый upsert делает вызывающий код).
|
||
Вынесено из loop чтобы combined- и isolated-extras-pass (#1945) делили код.
|
||
"""
|
||
if isinstance(result, Exception):
|
||
_classify_and_log(db, run_id, obj_id, kind_tag, full_url, result)
|
||
return
|
||
|
||
if kind_tag in ("sale_graph_apartments", "sale_graph_parking"):
|
||
sg_type = kind_tag.replace("sale_graph_", "")
|
||
rows_sg, _ = result
|
||
extras_counts["sale_graph_rows"] += upsert_sale_graph(
|
||
db, obj_id, sg_type, rows_sg, snapshot_date
|
||
)
|
||
elif kind_tag == "sales_agg":
|
||
agg_data, _ = result
|
||
extras_counts["sales_agg_rows"] += upsert_sales_agg(db, obj_id, agg_data, snapshot_date)
|
||
elif kind_tag == "infrastructure":
|
||
pois_data, _ = result
|
||
extras_counts["infra_rows"] += upsert_infrastructure(db, obj_id, pois_data, snapshot_date)
|
||
elif kind_tag == "photos":
|
||
photos_data, _ = result
|
||
local_paths: dict[str, str] = {}
|
||
thumb_paths: dict[str, str] = {}
|
||
if download_photos_binary and photos_data:
|
||
local_paths, thumb_paths = await download_photos(sess, obj_id, photos_data, pdir)
|
||
extras_counts["photos_downloaded"] += len(local_paths)
|
||
extras_counts["photos_rows"] += upsert_photos(
|
||
db, obj_id, photos_data, local_paths, thumb_paths
|
||
)
|
||
elif kind_tag in _DOC_KINDS:
|
||
doc_items, _ = result
|
||
all_docs.extend(extract_documents(doc_items or []))
|
||
|
||
|
||
def _extras_coros(sess: BrowserSession, obj_id: int) -> list[Any]:
|
||
"""The 10 extras-endpoint coroutines for one object (flats NOT included)."""
|
||
return [
|
||
_fetch_sale_graph_safe(sess, obj_id, "apartments"),
|
||
_fetch_sale_graph_safe(sess, obj_id, "parking"),
|
||
_fetch_sales_agg_safe(sess, obj_id),
|
||
_fetch_infrastructure_safe(sess, obj_id),
|
||
_fetch_photos_safe(sess, obj_id),
|
||
_fetch_doc_rpd_safe(sess, obj_id),
|
||
_fetch_doc_developer_report_safe(sess, obj_id),
|
||
_fetch_doc_project_documentation_safe(sess, obj_id),
|
||
_fetch_doc_documentation_other_safe(sess, obj_id),
|
||
_fetch_doc_permits_safe(sess, obj_id),
|
||
]
|
||
|
||
|
||
def _is_waf_poisoned(results: list[tuple[str, str, Any]]) -> bool:
|
||
"""Любой WafBlockedError среди результатов = сессия отравлена (cookies флагнуты).
|
||
|
||
После этого ВСЕ запросы на этой сессии (incl. следующего объекта) пойдут в 403,
|
||
поэтому вызывающий код должен пересоздать BrowserSession перед продолжением.
|
||
"""
|
||
return any(isinstance(r, WafBlockedError) for _kind, _url, r in results)
|
||
|
||
|
||
async def _run_extras_pass(
|
||
*,
|
||
db: Session,
|
||
run_id: int,
|
||
region_code: int,
|
||
all_objects: list[dict[str, Any]],
|
||
start_index: int,
|
||
snapshot_date: date,
|
||
extras_counts: dict[str, int],
|
||
pdir: Path,
|
||
download_photos_binary: bool,
|
||
load_state: str | None,
|
||
headed: bool,
|
||
browser_concurrency: int | None,
|
||
request_jitter_min_ms: int | None,
|
||
request_jitter_max_ms: int | None,
|
||
proxy_url: str | None,
|
||
) -> int:
|
||
"""#1945 isolated extras pass: extras в ОТДЕЛЬНОЙ BrowserSession от flats.
|
||
|
||
Ключ изоляции: на ЛЮБОЙ WafBlockedError (extras-403 травит cookies сессии)
|
||
закрываем сессию и открываем НОВУЮ (свежие cookies/контекст) перед следующим
|
||
объектом — яд не накапливается и НИКОГДА не касается flats (flats-сессия уже
|
||
закрыта). Best-effort: каждый сбой → kn_scrape_failures, run не падает.
|
||
|
||
Возвращает суммарный request_count всех extras-сессий (для метрики run'а).
|
||
"""
|
||
total = len(all_objects)
|
||
total_requests = 0
|
||
log_progress(
|
||
db, run_id, f"extras-пас (isolated): старт {total - start_index} объектов", stage="extras"
|
||
)
|
||
|
||
def _new_session() -> BrowserSession:
|
||
return BrowserSession(
|
||
region_code=region_code,
|
||
headed=headed,
|
||
load_state=load_state,
|
||
concurrency=browser_concurrency,
|
||
jitter_min_ms=request_jitter_min_ms,
|
||
jitter_max_ms=request_jitter_max_ms,
|
||
proxy_url=proxy_url,
|
||
)
|
||
|
||
# extras сейчас мертвы (403 с 2026-06-03). Recycle на КАЖДЫЙ объект (полный
|
||
# перезапуск браузера ~1500×) бессмыслен, если extras заблокированы целиком.
|
||
# Поэтому считаем подряд идущие полностью-WAF объекты и обрываем пас после
|
||
# порога — extras остаются best-effort opt-in, но не жгут час впустую.
|
||
consecutive_all_waf = 0
|
||
aborted = False
|
||
sess = _new_session()
|
||
await sess.__aenter__()
|
||
try:
|
||
await sess.warm_up()
|
||
for i in range(start_index, total):
|
||
obj = all_objects[i]
|
||
obj_id = obj.get("objId") or obj.get("obj_id")
|
||
if not obj_id:
|
||
continue
|
||
|
||
results = await asyncio.gather(*_extras_coros(sess, obj_id), return_exceptions=False)
|
||
all_docs: list[dict[str, Any]] = []
|
||
for kind_tag, full_url, result in results:
|
||
await _process_extras_result(
|
||
sess,
|
||
db,
|
||
run_id,
|
||
obj_id,
|
||
kind_tag,
|
||
full_url,
|
||
result,
|
||
snapshot_date,
|
||
extras_counts,
|
||
all_docs,
|
||
download_photos_binary=download_photos_binary,
|
||
pdir=pdir,
|
||
)
|
||
if all_docs:
|
||
ins, _skip = upsert_documents(db, obj_id, all_docs)
|
||
extras_counts["documents_rows"] += ins
|
||
|
||
poisoned = _is_waf_poisoned(results)
|
||
# Полностью-WAF объект = ни один extras не отдал JSON (все 403).
|
||
all_waf = all(isinstance(r, WafBlockedError) for _k, _u, r in results)
|
||
consecutive_all_waf = consecutive_all_waf + 1 if all_waf else 0
|
||
if consecutive_all_waf >= _EXTRAS_ABORT_AFTER_CONSEC_WAF:
|
||
aborted = True
|
||
log_progress(
|
||
db,
|
||
run_id,
|
||
f"extras-пас оборван: {consecutive_all_waf} объектов подряд"
|
||
" полностью WAF-403 (extras заблокированы целиком)",
|
||
level="warn",
|
||
stage="extras",
|
||
)
|
||
break
|
||
|
||
# RECYCLE: если сессия отравлена WAF-403 — пересоздаём перед next obj.
|
||
if poisoned:
|
||
total_requests += sess.request_count
|
||
await sess.__aexit__(None, None, None)
|
||
sess = _new_session()
|
||
await sess.__aenter__()
|
||
await sess.warm_up()
|
||
|
||
if (i + 1) % 10 == 0:
|
||
_checkpoint(db, run_id, i + 1)
|
||
finally:
|
||
total_requests += sess.request_count
|
||
await sess.__aexit__(None, None, None)
|
||
if aborted:
|
||
logger.info("extras isolated pass aborted early — extras WAF-blocked wholesale")
|
||
|
||
log_progress(
|
||
db,
|
||
run_id,
|
||
f"extras-пас done: sale_graph={extras_counts['sale_graph_rows']}"
|
||
f" agg={extras_counts['sales_agg_rows']} infra={extras_counts['infra_rows']}"
|
||
f" photos={extras_counts['photos_rows']} docs={extras_counts['documents_rows']}",
|
||
stage="extras",
|
||
)
|
||
return total_requests
|
||
|
||
|
||
# ── 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)
|
||
|
||
|
||
OBJECTS_PAGE_SIZE = 500
|
||
|
||
|
||
async def fetch_objects_for_status(
|
||
sess: BrowserSession, place: str, status: int
|
||
) -> list[dict[str, Any]]:
|
||
"""Fetch ALL objects for a given (place, objStatus), пагинируя по страницам.
|
||
|
||
Раньше делали единственный запрос с limit=999999 и доверяли допущению, что сервер
|
||
вернёт всё. Если WAF/прокси DOM.РФ обрезает гигантский limit до своего max, хвост
|
||
объектов молча терялся (не скрейпился), а run всё равно рапортовал status='done' (#1605).
|
||
Теперь идём страницами по OBJECTS_PAGE_SIZE и аккумулируем, пока не наберём total
|
||
(из payload) либо страница не вернётся короче запрошенной / пустой.
|
||
"""
|
||
rows: list[dict[str, Any]] = []
|
||
offset = 0
|
||
total: int | None = None
|
||
while True:
|
||
payload = await sess.get_json(
|
||
PATH_OBJECTS,
|
||
{
|
||
"offset": offset,
|
||
"limit": OBJECTS_PAGE_SIZE,
|
||
"sortField": "default",
|
||
"sortType": "desc",
|
||
"place": place,
|
||
"objStatus": status,
|
||
},
|
||
)
|
||
page = _extract_list(payload)
|
||
if total is None:
|
||
total = _extract_total(payload)
|
||
rows.extend(page)
|
||
# Стоп-условия: пустая страница, недобор до размера страницы (последняя),
|
||
# либо набрали заявленный total. total может быть None (сервер его не отдал) —
|
||
# тогда полагаемся на размер страницы как сигнал конца.
|
||
if not page or len(page) < OBJECTS_PAGE_SIZE:
|
||
break
|
||
if total is not None and len(rows) >= total:
|
||
break
|
||
offset += OBJECTS_PAGE_SIZE
|
||
|
||
logger.info(
|
||
"kn/object place=%s status=%d -> %d/%s rows (paginated, page=%d)",
|
||
place,
|
||
status,
|
||
len(rows),
|
||
total,
|
||
OBJECTS_PAGE_SIZE,
|
||
)
|
||
if total is not None and len(rows) < total:
|
||
logger.warning(
|
||
"kn/object place=%s status=%d: получено %d < total=%d — возможен недобор хвоста",
|
||
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).
|
||
|
||
On HTTP / WAF / non-JSON errors raises — caller (_fetch_flats_safe) ловит и кладёт
|
||
Exception в result-tuple, который result-loop отдаёт в _classify_and_log →
|
||
kn_scrape_failures. Раньше try/except здесь возвращал [] на ЛЮБУЮ ошибку, из-за чего
|
||
провал /portal/table (429, 5xx, WAF-challenge) не попадал в журнал отказов, а run
|
||
рапортовал ложную полноту по квартирам (#1643). Поведение теперь как у остальных
|
||
fetch_* endpoint'ов, которые исключения не глотают.
|
||
"""
|
||
payload = await sess.get_json(PATH_FLATS_TABLE, {"externalId": obj_id})
|
||
# 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}",
|
||
}
|
||
|
||
|
||
def log_progress(
|
||
db: Session,
|
||
run_id: int,
|
||
message: str,
|
||
*,
|
||
level: str = "info",
|
||
stage: str | None = None,
|
||
obj_id: int | None = None,
|
||
) -> None:
|
||
"""Append a progress event to kn_scrape_log so the admin UI can display
|
||
live status. Best-effort: SQL errors are swallowed and logged so a flaky
|
||
log table never aborts an actual sweep."""
|
||
try:
|
||
db.execute(
|
||
text(
|
||
"""
|
||
INSERT INTO kn_scrape_log (run_id, level, stage, obj_id, message)
|
||
VALUES (:rid, :lvl, :stg, :oid, :msg)
|
||
"""
|
||
),
|
||
{
|
||
"rid": run_id,
|
||
"lvl": level,
|
||
"stg": stage,
|
||
"oid": obj_id,
|
||
"msg": message[:500],
|
||
},
|
||
)
|
||
db.commit()
|
||
except Exception as e:
|
||
logger.warning("log_progress failed: %s", e)
|
||
try:
|
||
db.rollback()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _checkpoint(db: Session, run_id: int, progress_index: int) -> None:
|
||
"""Записать прогресс + heartbeat. Вызывается раз в N объектов и в конце."""
|
||
try:
|
||
db.execute(
|
||
text(
|
||
"""
|
||
UPDATE kn_scrape_runs
|
||
SET progress_obj_index = :idx,
|
||
heartbeat_at = NOW()
|
||
WHERE run_id = :rid
|
||
"""
|
||
),
|
||
{"idx": progress_index, "rid": run_id},
|
||
)
|
||
db.commit()
|
||
except Exception as e:
|
||
logger.warning("checkpoint failed run=%s: %s", run_id, e)
|
||
try:
|
||
db.rollback()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
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,
|
||
resume_from_run_id: int | None = None,
|
||
browser_concurrency: int | None = None,
|
||
request_jitter_min_ms: int | None = None,
|
||
request_jitter_max_ms: int | None = None,
|
||
proxy_url: str | None = None,
|
||
extras_isolated: bool = True,
|
||
) -> dict[str, Any]:
|
||
"""Full sweep for one region: bootstrap → fetch objects → per-object processing.
|
||
|
||
#1945 anti-ban: ``browser_concurrency`` / ``request_jitter_*_ms`` / ``proxy_url``
|
||
прокидываются в BrowserSession sweep'а, чтобы троттлить/проксировать ТОЛЬКО KN-
|
||
прогон (другие скраперы используют модульные дефолты stealth.py без изменений).
|
||
None у каждого = дефолт stealth.py (concurrency=8, jitter 600–1500, без прокси).
|
||
|
||
``extras_isolated`` (#1945 ГЛАВНЫЙ unblock): True (дефолт) → flats тянутся в
|
||
ЧИСТОЙ flats-only сессии (extras на ней НЕ дёргаются НИКОГДА → 403-яд extras не
|
||
травит cookies flats-сессии → flats_count восстанавливается), а extras идут
|
||
ОТДЕЛЬНЫМ best-effort проходом с recycle сессии на WAF-403. False → старое
|
||
поведение (flats+extras в одной сессии, poison-prone).
|
||
|
||
Resume-aware: если `resume_from_run_id` задан, читаем objects_snapshot и
|
||
progress_obj_index из исходного run и продолжаем с этого индекса. Новый
|
||
run_id создаётся (с resumed_from_run_id ссылкой), исходный остаётся
|
||
'zombie' для аудита. Heartbeat обновляется каждые 10 объектов — старт-хук
|
||
worker'а ловит застывшие row >5 мин.
|
||
"""
|
||
snapshot_date = snapshot_date or date.today()
|
||
db = SessionLocal()
|
||
# Все дочерние таблицы имеют FK на domrf_snapshots(snapshot_date) —
|
||
# гарантируем что row существует ДО любых insert'ов в Phase A/B/C/raw.
|
||
_ensure_snapshot(db, snapshot_date)
|
||
|
||
# ── Init: либо новый run, либо resume ──────────────────────────────────
|
||
all_objects: list[dict[str, Any]] = []
|
||
start_index = 0
|
||
place = place_override or _place_str(region_code)
|
||
params_dict: dict[str, Any] = {
|
||
"extras": extras,
|
||
"fetch_flats": fetch_flats,
|
||
"download_photos": download_photos_binary,
|
||
"place": place,
|
||
"statuses": list(statuses),
|
||
}
|
||
|
||
if resume_from_run_id:
|
||
prev = (
|
||
db.execute(
|
||
text(
|
||
"""
|
||
SELECT region_codes, developer_ids, snapshot_date, params,
|
||
objects_snapshot,
|
||
COALESCE(progress_obj_index, 0) AS idx
|
||
FROM kn_scrape_runs
|
||
WHERE run_id = :rid
|
||
"""
|
||
),
|
||
{"rid": resume_from_run_id},
|
||
)
|
||
.mappings()
|
||
.first()
|
||
)
|
||
if not prev or not prev["objects_snapshot"]:
|
||
db.close()
|
||
raise RuntimeError(f"resume_from_run_id={resume_from_run_id}: нет objects_snapshot")
|
||
all_objects = list(prev["objects_snapshot"])
|
||
start_index = int(prev["idx"] or 0)
|
||
# Восстанавливаем параметры старого run'а — прелюдия может отличаться
|
||
# от вызывающего кода (например beat schedule поменял extras).
|
||
if prev["params"]:
|
||
saved = prev["params"]
|
||
extras = saved.get("extras", extras)
|
||
fetch_flats = saved.get("fetch_flats", fetch_flats)
|
||
download_photos_binary = saved.get("download_photos", download_photos_binary)
|
||
snapshot_date = prev["snapshot_date"] or snapshot_date
|
||
|
||
run_id = db.execute(
|
||
text(
|
||
"""
|
||
INSERT INTO kn_scrape_runs (
|
||
region_codes, developer_ids, snapshot_date, status,
|
||
params, objects_snapshot, progress_obj_index,
|
||
total_obj_count, heartbeat_at, resumed_from_run_id
|
||
)
|
||
VALUES (
|
||
CAST(:rc AS int[]), CAST(:devs AS text[]), :snap, 'running',
|
||
CAST(:params AS jsonb), CAST(:objs AS jsonb), :idx,
|
||
:total, NOW(), :prev
|
||
)
|
||
RETURNING run_id
|
||
"""
|
||
),
|
||
{
|
||
"rc": list(prev["region_codes"]),
|
||
"devs": list(prev["developer_ids"]) if prev["developer_ids"] else None,
|
||
"snap": snapshot_date,
|
||
"params": json.dumps(prev["params"] or params_dict),
|
||
"objs": json.dumps(all_objects),
|
||
"idx": start_index,
|
||
"total": len(all_objects),
|
||
"prev": resume_from_run_id,
|
||
},
|
||
).scalar_one()
|
||
db.commit()
|
||
log_progress(
|
||
db,
|
||
run_id,
|
||
f"♻ Resume from index {start_index}/{len(all_objects)}"
|
||
f" (предыдущий run #{resume_from_run_id})",
|
||
stage="resume",
|
||
)
|
||
else:
|
||
run_id = db.execute(
|
||
text(
|
||
"""
|
||
INSERT INTO kn_scrape_runs (
|
||
region_codes, developer_ids, snapshot_date, status,
|
||
params, heartbeat_at
|
||
)
|
||
VALUES (
|
||
ARRAY[:rc]::int[], CAST(:devs AS text[]), :snap, 'running',
|
||
CAST(:params AS jsonb), NOW()
|
||
)
|
||
RETURNING run_id
|
||
"""
|
||
),
|
||
{
|
||
"rc": region_code,
|
||
"devs": developers if developers else None,
|
||
"snap": snapshot_date,
|
||
"params": json.dumps(params_dict),
|
||
},
|
||
).scalar_one()
|
||
db.commit()
|
||
log_progress(
|
||
db,
|
||
run_id,
|
||
f"Старт sweep региона {region_code} (place={place}, devs={developers or '*'},"
|
||
f" extras={extras}, photos={download_photos_binary})",
|
||
stage="start",
|
||
)
|
||
|
||
extras_counts = {
|
||
"sale_graph_rows": 0,
|
||
"sales_agg_rows": 0,
|
||
"infra_rows": 0,
|
||
"photos_rows": 0,
|
||
"photos_downloaded": 0,
|
||
"documents_rows": 0,
|
||
"checks_rows": 0,
|
||
}
|
||
total_flats = 0
|
||
request_count = 0
|
||
|
||
try:
|
||
async with BrowserSession(
|
||
region_code=region_code,
|
||
headed=headed,
|
||
load_state=load_state,
|
||
save_state=save_state,
|
||
concurrency=browser_concurrency,
|
||
jitter_min_ms=request_jitter_min_ms,
|
||
jitter_max_ms=request_jitter_max_ms,
|
||
proxy_url=proxy_url,
|
||
) as sess:
|
||
# Warm-up: visit /сервисы/каталог-новостроек/ to obtain WAF cookies
|
||
# (___dmpkit___, domain_sid). Required since 2026-05-24 — without these
|
||
# cookies all /сервисы/api/object/*/* requests return 403 + WAF HTML.
|
||
await sess.warm_up()
|
||
|
||
# ── Phase A — fetch objects (skip on resume) ────────────────────
|
||
if not resume_from_run_id:
|
||
for status in statuses:
|
||
rows = await fetch_objects_for_status(sess, place, status)
|
||
all_objects.extend(rows)
|
||
log_progress(
|
||
db,
|
||
run_id,
|
||
f"objStatus={status}: получено {len(rows)} объектов",
|
||
stage="fetch_objects",
|
||
)
|
||
# Heartbeat-tick: Phase A для 1516 ЖК идёт ~3 мин, autoclean
|
||
# пропустит это окно если хотя бы каждые ~1 мин обновлять
|
||
# heartbeat_at. progress_obj_index пока 0 — snapshot ещё
|
||
# не сохранён.
|
||
_checkpoint(db, run_id, 0)
|
||
logger.info(
|
||
"place=%s total objects across %s = %d",
|
||
place,
|
||
statuses,
|
||
len(all_objects),
|
||
)
|
||
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
|
||
]
|
||
log_progress(
|
||
db,
|
||
run_id,
|
||
f"После фильтра devs={developers}: {len(all_objects)} объектов",
|
||
stage="filter_devs",
|
||
)
|
||
|
||
# Phase A commit: upsert_objects + raw + objects_snapshot.
|
||
# Это критично для resume — при падении ниже worker сможет
|
||
# подхватить с правильным списком объектов и кадастровым snapshot'ом.
|
||
obj_count = upsert_objects(db, all_objects, snapshot_date, region_code)
|
||
_insert_raw(db, snapshot_date, f"kn_object_place_{place}", all_objects)
|
||
db.execute(
|
||
text(
|
||
"""
|
||
UPDATE kn_scrape_runs
|
||
SET objects_snapshot = CAST(:objs AS jsonb),
|
||
total_obj_count = :total,
|
||
objects_count = :total,
|
||
heartbeat_at = NOW()
|
||
WHERE run_id = :rid
|
||
"""
|
||
),
|
||
{
|
||
"objs": json.dumps(all_objects),
|
||
"total": len(all_objects),
|
||
"rid": run_id,
|
||
},
|
||
)
|
||
db.commit()
|
||
# Spatial-join: проставить district_name свежезаписанным ЖК
|
||
# через ekb_districts_geom (PostGIS-полигоны OSM). Backfill
|
||
# лёгкий — только новые/изменённые row на этом snapshot_date.
|
||
_backfill_district_names(db, snapshot_date, region_code)
|
||
log_progress(
|
||
db,
|
||
run_id,
|
||
f"Phase A done: {obj_count} объектов в БД, snapshot закреплён",
|
||
stage="phase_a",
|
||
)
|
||
|
||
# ── Phase B/C — per-object processing (resumable, parallel per-object) ─
|
||
# Все endpoint'ы одного obj_id запускаются параллельно через asyncio.gather.
|
||
# BrowserSession._sem (size = browser_concurrency, #1945 троттлит KN до 2)
|
||
# ограничивает запросы.
|
||
# DB upserts выполняются последовательно после gather — один db Session
|
||
# не thread-safe для параллельной записи.
|
||
#
|
||
# #1945 ИЗОЛЯЦИЯ: при extras_isolated=True эта (warm) сессия делает
|
||
# ТОЛЬКО flats — extras на ней НЕ дёргаются (их 403-яд не травит cookies
|
||
# flats-сессии → flats_count восстанавливается). extras уходят отдельным
|
||
# best-effort проходом ниже (после закрытия flats-сессии) с recycle на
|
||
# WAF-403. При extras_isolated=False — старое combined-поведение.
|
||
pdir = Path(photos_dir) if photos_dir else PHOTOS_DIR_DEFAULT
|
||
total = len(all_objects)
|
||
for i in range(start_index, total):
|
||
obj = all_objects[i]
|
||
obj_id = obj.get("objId") or obj.get("obj_id")
|
||
if not obj_id:
|
||
continue
|
||
|
||
# Собираем корутины для параллельного запуска
|
||
coros: list[Any] = []
|
||
|
||
if fetch_flats:
|
||
coros.append(_fetch_flats_safe(sess, obj_id))
|
||
|
||
# extras в combined-сессии ТОЛЬКО когда изоляция выключена —
|
||
# иначе они отравят flats (#1945). В isolated-режиме extras
|
||
# обрабатываются отдельным проходом ниже.
|
||
combined_extras = extras and not extras_isolated
|
||
if combined_extras:
|
||
coros.extend(_extras_coros(sess, obj_id))
|
||
# TODO: obj_checks endpoint not found at /api/object/{id}/checks (404).
|
||
# 6 чек-боксов "Проверено на наш.дом.рф" вероятно inline в kn/object payload.
|
||
# Re-enable после investigation структуры объекта (separate PR).
|
||
# coros.append(_fetch_obj_checks_safe(sess, obj_id))
|
||
|
||
if not coros:
|
||
continue
|
||
|
||
# Параллельный fetch всех endpoint'ов одного объекта
|
||
results = await asyncio.gather(*coros, return_exceptions=False)
|
||
|
||
# Sequential upsert — DB session не thread-safe
|
||
all_docs: list[dict[str, Any]] = []
|
||
for kind_tag, full_url, result in results:
|
||
if kind_tag == "flats":
|
||
total_flats += _process_flats_result(
|
||
db, run_id, obj_id, full_url, result, snapshot_date, region_code
|
||
)
|
||
else:
|
||
await _process_extras_result(
|
||
sess,
|
||
db,
|
||
run_id,
|
||
obj_id,
|
||
kind_tag,
|
||
full_url,
|
||
result,
|
||
snapshot_date,
|
||
extras_counts,
|
||
all_docs,
|
||
download_photos_binary=download_photos_binary,
|
||
pdir=pdir,
|
||
)
|
||
|
||
# Единый upsert всех документов объекта после обработки doc-endpoint'ов.
|
||
if all_docs:
|
||
ins, _skip = upsert_documents(db, obj_id, all_docs)
|
||
extras_counts["documents_rows"] += ins
|
||
|
||
# Checkpoint раз в 10 объектов: запись прогресса + heartbeat.
|
||
if (i + 1) % 10 == 0:
|
||
_checkpoint(db, run_id, i + 1)
|
||
log_progress(
|
||
db,
|
||
run_id,
|
||
f"Progress: {i + 1}/{total} | flats={total_flats}"
|
||
f" sale_graph={extras_counts['sale_graph_rows']}"
|
||
f" agg={extras_counts['sales_agg_rows']}"
|
||
f" infra={extras_counts['infra_rows']}"
|
||
f" photos={extras_counts['photos_rows']}"
|
||
f" downloaded={extras_counts['photos_downloaded']}"
|
||
f" docs={extras_counts['documents_rows']}"
|
||
f" checks={extras_counts['checks_rows']}",
|
||
stage="extras" if combined_extras else "fetch_flats",
|
||
)
|
||
|
||
_checkpoint(db, run_id, total)
|
||
request_count = sess.request_count
|
||
|
||
# ── Phase C (isolated) — extras best-effort пас в ОТДЕЛЬНОЙ сессии ──────
|
||
# flats-сессия закрыта. Теперь (только если extras=True И extras_isolated)
|
||
# тянем extras в disposable-сессиях: на каждый WAF-403 recycle всю
|
||
# BrowserSession (свежие cookies/контекст) перед следующим объектом, чтобы
|
||
# яд НЕ накапливался. extras сейчас мертвы (403 с 2026-06-03) — пас целиком
|
||
# best-effort: любые сбои уходят в kn_scrape_failures и НЕ роняют run.
|
||
if extras and extras_isolated:
|
||
try:
|
||
ex_req = await _run_extras_pass(
|
||
db=db,
|
||
run_id=run_id,
|
||
region_code=region_code,
|
||
all_objects=all_objects,
|
||
start_index=start_index,
|
||
snapshot_date=snapshot_date,
|
||
extras_counts=extras_counts,
|
||
pdir=pdir,
|
||
download_photos_binary=download_photos_binary,
|
||
load_state=load_state,
|
||
headed=headed,
|
||
browser_concurrency=browser_concurrency,
|
||
request_jitter_min_ms=request_jitter_min_ms,
|
||
request_jitter_max_ms=request_jitter_max_ms,
|
||
proxy_url=proxy_url,
|
||
)
|
||
request_count += ex_req
|
||
except Exception as e:
|
||
# best-effort: extras-пас НИКОГДА не должен ронять уже собранные flats.
|
||
logger.warning("extras isolated pass failed (best-effort): %s", e)
|
||
log_progress(
|
||
db,
|
||
run_id,
|
||
f"extras-пас (best-effort) упал: {type(e).__name__}: {str(e)[:200]}",
|
||
level="warn",
|
||
stage="extras",
|
||
)
|
||
|
||
# ── Phase D — finalize ─────────────────────────────────────────────
|
||
db.execute(
|
||
text(
|
||
"""
|
||
UPDATE kn_scrape_runs
|
||
SET finished_at = NOW(),
|
||
flats_count = :fc,
|
||
requests_count = :rq,
|
||
status = 'done',
|
||
heartbeat_at = NOW()
|
||
WHERE run_id = :rid
|
||
"""
|
||
),
|
||
{"fc": total_flats, "rq": request_count, "rid": run_id},
|
||
)
|
||
db.commit()
|
||
log_progress(
|
||
db,
|
||
run_id,
|
||
f"Phase D done: objects={total} flats={total_flats} requests={request_count}",
|
||
stage="phase_d",
|
||
)
|
||
|
||
# ── Phase E — derive is_ekb for this snapshot ──────────────────────
|
||
# Проставляем is_ekb=TRUE для объектов Екатеринбурга/Свердловской обл.
|
||
# по district_name (заполнен PostGIS join в Phase A).
|
||
# Только новые/изменившиеся строки: COALESCE(is_ekb, FALSE) = FALSE.
|
||
try:
|
||
result_e = db.execute(
|
||
text(
|
||
"""
|
||
UPDATE domrf_kn_objects
|
||
SET is_ekb = TRUE
|
||
WHERE snapshot_date = :snap
|
||
AND (
|
||
district_name ILIKE '%екатеринбург%'
|
||
OR district_name ILIKE '%свердловск%'
|
||
)
|
||
AND COALESCE(is_ekb, FALSE) = FALSE
|
||
"""
|
||
),
|
||
{"snap": snapshot_date},
|
||
)
|
||
db.commit()
|
||
ekb_updated = result_e.rowcount if result_e.rowcount >= 0 else -1
|
||
log_progress(
|
||
db,
|
||
run_id,
|
||
f"Phase E done: is_ekb derived, updated={ekb_updated} rows"
|
||
f" for snap {snapshot_date}",
|
||
stage="phase_e",
|
||
)
|
||
except Exception as e:
|
||
logger.warning("Phase E is_ekb derive failed: %s", e)
|
||
try:
|
||
db.rollback()
|
||
except Exception:
|
||
pass
|
||
log_progress(
|
||
db,
|
||
run_id,
|
||
f"Phase E is_ekb derive FAILED: {type(e).__name__}: {str(e)[:200]}",
|
||
level="warn",
|
||
stage="phase_e",
|
||
)
|
||
|
||
log_progress(
|
||
db,
|
||
run_id,
|
||
f"Готово objects={total} flats={total_flats} requests={request_count}"
|
||
f" docs={extras_counts['documents_rows']} checks={extras_counts['checks_rows']}",
|
||
stage="done",
|
||
)
|
||
return {
|
||
"run_id": run_id,
|
||
"region_code": region_code,
|
||
"place": place,
|
||
"objects": total,
|
||
"flats": total_flats,
|
||
"requests": request_count,
|
||
"snapshot_date": snapshot_date.isoformat(),
|
||
"resumed_from": resume_from_run_id,
|
||
**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()
|
||
log_progress(
|
||
db,
|
||
run_id,
|
||
f"Ошибка ❌ {type(e).__name__}: {str(e)[:300]}",
|
||
level="error",
|
||
stage="failed",
|
||
)
|
||
raise
|
||
finally:
|
||
db.close()
|