diff --git a/backend/app/services/scrapers/domrf_kn.py b/backend/app/services/scrapers/domrf_kn.py index 3cd5c3fe..15cc92b6 100644 --- a/backend/app/services/scrapers/domrf_kn.py +++ b/backend/app/services/scrapers/domrf_kn.py @@ -320,6 +320,24 @@ INSERT_RAW_SQL = text( ) +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 _insert_raw(db: Session, snapshot_date: date, endpoint_label: str, payload: Any) -> None: body = json.dumps(payload, ensure_ascii=False) db.execute( @@ -335,21 +353,69 @@ def _insert_raw(db: Session, snapshot_date: date, endpoint_label: str, payload: ) +_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: - db.execute(UPSERT_OBJECT_SQL, norm) + 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) - db.rollback() return inserted @@ -365,11 +431,11 @@ def upsert_flats( continue norm["snapshot_date"] = snapshot_date try: - db.execute(UPSERT_FLAT_SQL, norm) + 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) - db.rollback() if skipped_no_id: logger.info("skipped %d flats without numeric id", skipped_no_id) return inserted @@ -555,21 +621,21 @@ def _log_failure( 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, - }, - ) + 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) - db.rollback() async def fetch_sale_graph( @@ -616,23 +682,23 @@ def upsert_sale_graph( 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, - }, - ) + 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) - db.rollback() return n @@ -649,22 +715,22 @@ def upsert_sales_agg(db: Session, obj_id: int, agg: dict[str, Any], snapshot_dat 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, - }, - ) + 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) - db.rollback() return n @@ -674,24 +740,24 @@ def upsert_infrastructure( 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, - }, - ) + 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) - db.rollback() return n @@ -714,29 +780,29 @@ def upsert_photos( local = local_paths.get(file_id) thumb = thumb_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, - "thumb_path": thumb, - "downloaded_at": now_dt if local else None, - }, - ) + 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) - db.rollback() return n @@ -968,6 +1034,9 @@ async def run_region_sweep( """ 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]] = []