feat: harden sync flow, proxy bridge, and data normalization

iaai_scraper/storage/db.py

@@ -4,7 +4,7 @@ from contextlib import contextmanager
 from datetime import datetime, timezone
 from typing import Iterator
 
-from sqlalchemy import create_engine, select
+from sqlalchemy import create_engine, or_, select
 from sqlalchemy.orm import Session, sessionmaker
 
 from ..core.config import Settings
@@ -76,6 +76,16 @@ class PersistenceService:
 	def start_sync_run(self, lane: str) -> int:
 		# Создаём запись о запуске синхронизации.
 		with self.session_scope() as session:
+			# Если предыдущий процесс умер, оставив run в `running`,
+			# помечаем его как failed перед новым запуском.
+			now = datetime.now(timezone.utc)
+			stale_runs = session.execute(select(SyncRun).where(SyncRun.status == "running")).scalars().all()
+			for stale in stale_runs:
+				stale.status = "failed"
+				stale.finished_at = now
+				if not stale.error_summary:
+					stale.error_summary = "Recovered stale running sync run before starting a new run"
+
 			run = SyncRun(status="running", lane=lane, ids_fetched=0, cars_upserted=0, cars_failed=0, images_upserted=0)
 			session.add(run)
 			session.flush()
@@ -95,6 +105,22 @@ class PersistenceService:
 			run.images_upserted = images_upserted
 			run.error_summary = error_summary
 
+	def get_existing_origin_urls(self, origin_urls: list[str]) -> set[str]:
+		# Возвращает уже существующие в БД origin_url для фильтрации only-new запусков.
+		if not origin_urls:
+			return set()
+		with self.session_scope() as session:
+			rows = session.execute(select(Car.origin_url).where(Car.origin_url.in_(origin_urls))).all()
+			return {str(row[0]) for row in rows if row and row[0]}
+
+	def get_existing_origin_ids(self, origin_ids: list[str]) -> set[str]:
+		# Возвращает уже существующие в БД origin_id для фильтрации только новых авто.
+		if not origin_ids:
+			return set()
+		with self.session_scope() as session:
+			rows = session.execute(select(Car.origin_id).where(Car.origin_id.in_(origin_ids))).all()
+			return {str(row[0]) for row in rows if row and row[0]}
+
 	@staticmethod
 	def _add_images(session: Session, car_id: int, images: list[dict[str, object]]) -> None:
 		for image_payload in images:
@@ -104,7 +130,7 @@ class PersistenceService:
 	def _car_payload(record: CarRecord) -> dict[str, object]:
 		payload = record.model_dump(mode="python")
 		result = {key: value for key, value in payload.items() if key in CAR_DB_FIELDS}
-		# Serialize raw_attributes dict to JSON string for Text column.
+	
 		if "raw_attributes" in result and isinstance(result["raw_attributes"], dict):
 			result["raw_attributes"] = json.dumps(result["raw_attributes"], ensure_ascii=False, default=str)
 		return result
@@ -116,8 +142,11 @@ class PersistenceService:
 		images = [image.model_dump(mode="python") for image in record.images]
 		content_hash = str(payload.get("content_hash") or "")
 		with self.session_scope() as session:
-			# поиск по origin_id
-			car = session.execute(select(Car).where(Car.origin_id == record.origin_id)).scalar_one_or_none()
+			# Сначала пытаемся найти по origin_id, а если ранее origin_id был неполный,
+			# подхватываем существующую запись по origin_url, чтобы не плодить дубли.
+			car = session.execute(
+				select(Car).where(or_(Car.origin_id == record.origin_id, Car.origin_url == record.origin_url))
+			).scalar_one_or_none()
 			action = "inserted"
 			if car is None:
 				car = Car(**payload)

iaai_scraper/storage/enums.py

@@ -5,4 +5,4 @@ STEERING_WHEEL_ENUM_VALUES = ("LEFT", "RIGHT", "left", "right", "NA")
 BODY_TYPE_ENUM_VALUES = ("COUPE", "SUV", "HATCHBACK", "MINIVAN", "SEDAN", "NA", "Station Wagon", "Pickup", "Truck", "Open", "RV", "Other", "STATION_WAGON", "PICKUP", "TRUCK", "OPEN", "OTHER")
 COUNTRY_ENUM_VALUES = ("JP", "KR", "US", "CA", "NA")
 ORIGIN_ENUM_VALUES = ("TAU", "CARSENSOR", "HANAMARU", "ENCAR", "KURUMA_TRADER", "carsensor", "encar", "kuruma_trader", "asnet", "kababa", "ACV", "COPART", "copart", "NA", "ASNET", "KABABA", "IAAI")
-SELLING_TYPE_ENUM_VALUES = ("STOCK", "AUCTION", "TENDER", "stock", "auction", "tender", "NA")
+SELLING_TYPE_ENUM_VALUES = ("STOCK", "AUCTION", "TENDER", "stock", "auction", "tender", "NA")