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

2026-04-08 18:19:02 +03:00
17 changed files with 683 additions and 24 deletions
--- a/.dockerignore
+++ b/.dockerignore
@@ -14,3 +14,4 @@ coverage.xml
 *.egg-info/
 dist/
 build/
+artifacts/
--- a/.env.example
+++ b/.env.example
@@ -36,6 +36,7 @@ IAAI_AFTER_PAGE_CHANGE_MAX_S=6.0

 # Scheduler (default once per hour)
 IAAI_SCHEDULER_INTERVAL_MINUTES=60
+IAAI_SYNC_ONLY_NEW=true

 # Retry / backoff
 IAAI_RETRY_DELAY_SECONDS=2.5
--- a/4
+++ b/4
@@ -5,6 +5,10 @@ ENV PYTHONDONTWRITEBYTECODE=1 \

 WORKDIR /app

+# Xvfb for headed Chromium in container (IAAI blocks headless)
+RUN apt-get update -qq && apt-get install -y --no-install-recommends xvfb \
+    && rm -rf /var/lib/apt/lists/*
+
 COPY requirements.txt ./
 RUN pip install --no-cache-dir -r requirements.txt

--- a/README.md
+++ b/README.md
@@ -6,6 +6,8 @@

 По умолчанию работает последовательно одна машина за раз, с паузами между запросами.

+JSON-результаты CLI по умолчанию сохраняются в `artifacts/json/`, чтобы не засорять корень проекта.
+
 ## Что делает

 1. Открывает страницу листинга `Vehiclelisting/Cars`, собирает ссылки на карточки.
@@ -77,6 +79,7 @@ SQLite подходит для локальной отладки, но не дл

 - запуск раз в **1 час**
 - лимит **30 машин за цикл**
+- обрабатываются только **новые авто** (по умолчанию `IAAI_SYNC_ONLY_NEW=true`)

 Это уже отражено в актуальных env-настройках.

@@ -91,6 +94,16 @@ IAAI использует anti-bot / fraud protection.
 - Chromium **не поддерживает SOCKS5 с аутентификацией напрямую**
 - поэтому для production желательно покупать прокси, который отдаёт именно HTTP/HTTPS доступ

+Если используется встроенный bridge `iaai_scraper/proxy_bridge.py` (HTTP/HTTPS → SOCKS5),
+в нём добавлены базовые меры стабильности:
+
+- корректное чтение request body через `rfile`
+- поддержка `Transfer-Encoding: chunked` для request body
+- базовое логирование запросов и ошибок
+- таймауты relay-соединений
+- ограничение числа рабочих потоков (`PROXY_BRIDGE_MAX_WORKERS`)
+- безопасный ответ `502 Bad Gateway` без утечки внутренних исключений
+
 Пример:

 ```env
@@ -117,10 +130,10 @@ IAAI_PROXY_PASSWORD=password
 python main.py init-db

 # собрать ссылки из листинга
-python main.py collect-listing --make Toyota --model Camry --output listing.json
+python main.py collect-listing --make Toyota --model Camry --output artifacts/json/listing.json

 # scrape одной карточки
-python main.py scrape-vehicle "https://www.iaai.com/VehicleDetail/41180634~US" --output result.json
+python main.py scrape-vehicle "https://www.iaai.com/VehicleDetail/41180634~US" --output artifacts/json/result.json

 # scrape + запись в БД
 python main.py sync-vehicle "https://www.iaai.com/VehicleDetail/41180634~US" --lane iaai
@@ -128,6 +141,9 @@ python main.py sync-vehicle "https://www.iaai.com/VehicleDetail/41180634~US" --l
 # массовая синхронизация листинга
 python main.py sync-listing --make Toyota --model Camry --lane iaai_cars --limit 30

+# при необходимости можно принудительно отключить фильтр only-new
+python main.py sync-listing --limit 30 --only-new false
+
 # daemon-режим (цикл каждые N минут)
 python main.py run-daemon --interval 60
 ```
--- a/entrypoint.sh
+++ b/entrypoint.sh
@@ -1,4 +1,31 @@
 #!/bin/bash
 set -e

+# Start HTTP→SOCKS5 proxy bridge if SOCKS5 upstream is configured
+if [ -n "$SOCKS5_PROXY_HOST" ]; then
+  echo "[entrypoint] Starting proxy bridge (HTTP :8899 → SOCKS5 $SOCKS5_PROXY_HOST:${SOCKS5_PROXY_PORT:-1002})..."
+  if [ -z "$IAAI_PROXY_SERVER" ]; then
+    export IAAI_PROXY_SERVER="http://127.0.0.1:8899"
+  elif [ "$IAAI_PROXY_SERVER" = "http://localhost:8899" ]; then
+    export IAAI_PROXY_SERVER="http://127.0.0.1:8899"
+  fi
+  echo "[entrypoint] Using browser proxy: $IAAI_PROXY_SERVER"
+  python -m iaai_scraper.proxy_bridge &
+  BRIDGE_PID=$!
+  sleep 1
+  if ! kill -0 $BRIDGE_PID 2>/dev/null; then
+    echo "[entrypoint] ERROR: iaai_scraper.proxy_bridge failed to start"
+    exit 1
+  fi
+  echo "[entrypoint] Proxy bridge started (PID $BRIDGE_PID)"
+fi
+
+# IAAI blocks headless Chromium on Linux; run headed via Xvfb virtual display
+export DISPLAY=:99
+export IAAI_HEADLESS=false
+Xvfb :99 -screen 0 1920x1080x24 -nolisten tcp &
+XVFB_PID=$!
+sleep 0.5
+echo "[entrypoint] Xvfb started (PID $XVFB_PID, DISPLAY=$DISPLAY)"
+
 exec "$@"
--- a/iaai_scraper/cli.py
+++ b/iaai_scraper/cli.py
@@ -13,39 +13,41 @@ def build_parser() -> argparse.ArgumentParser:
    parser.add_argument("--debug", action="store_true", help="Enable DEBUG logging")
    subparsers = parser.add_subparsers(dest="command", required=True)

+    default_output_dir = Path("artifacts/json")
+
    init_db_parser = subparsers.add_parser("init-db", help="Create local DB tables")
-    init_db_parser.add_argument("--output", default="iaai_db_init.json", help="Path to output JSON")
+    init_db_parser.add_argument("--output", default=str(default_output_dir / "iaai_db_init.json"), help="Path to output JSON")

    listing_parser = subparsers.add_parser("collect-listing", help="Collect vehicle URLs from Vehiclelisting/Cars")
    listing_parser.add_argument("--make", default=None, help="Optional make filter")
    listing_parser.add_argument("--model", default=None, help="Optional model filter")
-    listing_parser.add_argument("--output", default="iaai_listing_links.json", help="Path to output JSON")
+    listing_parser.add_argument("--output", default=str(default_output_dir / "iaai_listing_links.json"), help="Path to output JSON")

    open_parser = subparsers.add_parser(
        "open-vehicle",
        help="Open a vehicle page in gentle mode and save only DOM-based hints",
    )
    open_parser.add_argument("vehicle_url", help="IAAI vehicle detail URL")
-    open_parser.add_argument("--output", default="iaai_vehicle_opened.json", help="Path to output JSON")
+    open_parser.add_argument("--output", default=str(default_output_dir / "iaai_vehicle_opened.json"), help="Path to output JSON")

    scrape_parser = subparsers.add_parser(
        "scrape-vehicle",
        help="Open a vehicle page and capture a limited set of likely useful JSON responses",
    )
    scrape_parser.add_argument("vehicle_url", help="IAAI vehicle detail URL")
-    scrape_parser.add_argument("--output", default="iaai_vehicle_detail.json", help="Path to output JSON")
+    scrape_parser.add_argument("--output", default=str(default_output_dir / "iaai_vehicle_detail.json"), help="Path to output JSON")

    export_parser = subparsers.add_parser(
        "export-db-json",
        help="Scrape a vehicle page and save only the DB-ready car record JSON",
    )
    export_parser.add_argument("vehicle_url", help="IAAI vehicle detail URL")
-    export_parser.add_argument("--output", default="iaai_vehicle_db_record.json", help="Path to output JSON")
+    export_parser.add_argument("--output", default=str(default_output_dir / "iaai_vehicle_db_record.json"), help="Path to output JSON")

    sync_vehicle_parser = subparsers.add_parser("sync-vehicle", help="Scrape one vehicle and upsert it into the DB")
    sync_vehicle_parser.add_argument("vehicle_url", help="IAAI vehicle detail URL")
    sync_vehicle_parser.add_argument("--lane", default="iaai", help="Logical lane name for sync_runs")
-    sync_vehicle_parser.add_argument("--output", default="iaai_sync_vehicle.json", help="Path to output JSON")
+    sync_vehicle_parser.add_argument("--output", default=str(default_output_dir / "iaai_sync_vehicle.json"), help="Path to output JSON")

    sync_listing_parser = subparsers.add_parser(
        "sync-listing",
@@ -55,7 +57,13 @@ def build_parser() -> argparse.ArgumentParser:
    sync_listing_parser.add_argument("--model", default=None, help="Optional model filter")
    sync_listing_parser.add_argument("--lane", default="iaai_cars", help="Logical lane name for sync_runs")
    sync_listing_parser.add_argument("--limit", type=int, default=None, help="Limit number of vehicles to sync")
-    sync_listing_parser.add_argument("--output", default="iaai_sync_listing.json", help="Path to output JSON")
+    sync_listing_parser.add_argument(
+        "--only-new",
+        choices=["true", "false"],
+        default=None,
+        help="Process only new vehicles (default from IAAI_SYNC_ONLY_NEW)",
+    )
+    sync_listing_parser.add_argument("--output", default=str(default_output_dir / "iaai_sync_listing.json"), help="Path to output JSON")

    daemon_parser = subparsers.add_parser(
        "run-daemon",
@@ -108,7 +116,14 @@ def main() -> None:
        elif args.command == "sync-vehicle":
            data = scraper.sync_vehicle(args.vehicle_url, lane=args.lane)
        else:
-            data = scraper.sync_listing(make=args.make, model=args.model, lane=args.lane, limit=args.limit)
+            only_new = None if args.only_new is None else args.only_new == "true"
+            data = scraper.sync_listing(
+                make=args.make,
+                model=args.model,
+                lane=args.lane,
+                limit=args.limit,
+                only_new=only_new,
+            )

        save_to_json(data, Path(args.output))
        print(f"Saved result to {Path(args.output).resolve()}")
--- a/iaai_scraper/core/config.py
+++ b/iaai_scraper/core/config.py
@@ -126,6 +126,7 @@ class Settings:
    log_file: str | None = os.getenv("IAAI_LOG_FILE") or None
    enable_trace_id_logs: bool = os.getenv("IAAI_ENABLE_TRACE_ID_LOGS", "true").strip().lower() in {"1", "true", "yes", "on"}
    scheduler_interval_minutes: int = int(os.getenv("IAAI_SCHEDULER_INTERVAL_MINUTES", "60"))
+    sync_only_new: bool = os.getenv("IAAI_SYNC_ONLY_NEW", "true").strip().lower() in {"1", "true", "yes", "on"}
    fingerprint: FingerprintConfig = field(default_factory=FingerprintConfig)
    gentle: GentleModeConfig = field(default_factory=GentleModeConfig)
    pace: HumanPaceConfig = field(default_factory=HumanPaceConfig)
--- a/iaai_scraper/core/utils.py
+++ b/iaai_scraper/core/utils.py
@@ -7,7 +7,9 @@ from typing import Any, Iterable


 def save_to_json(data: Any, filename: str | Path) -> None:
-    Path(filename).write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
+    path = Path(filename)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")


 def short_sleep(a: float = 0.10, b: float = 0.35) -> None:
--- a/iaai_scraper/parsing/mapper.py
+++ b/iaai_scraper/parsing/mapper.py
@@ -62,9 +62,21 @@ class CarMapper:
 		parser_id = f"iaai:{origin_id}"
 		brand = self._as_str(first_non_empty([core.get("make"), vehicle_summary.get("make")])) or "UNKNOWN"
 		model = self._as_str(first_non_empty([core.get("model"), vehicle_summary.get("model")])) or "UNKNOWN"
-		year = self._to_int(first_non_empty([core.get("year"), vehicle_summary.get("year")]))
-		price = self._to_int(first_non_empty([pricing.get("buy_now"), pricing.get("current_bid"), vehicle_summary.get("buy_now"), vehicle_summary.get("current_bid")]))
-		mileage = self._to_int(first_non_empty([core.get("odometer"), vehicle_summary.get("odometer"), 0])) or 0
+		year = self._to_year(first_non_empty([core.get("year"), vehicle_summary.get("year")]))
+		price = self._first_parsed_int(
+			[
+				pricing.get("buy_now"),
+				pricing.get("current_bid"),
+				vehicle_summary.get("buy_now"),
+				vehicle_summary.get("current_bid"),
+				pricing.get("actual_cash_value"),
+			],
+			self._to_money_int,
+		)
+		mileage = self._first_parsed_int(
+			[core.get("odometer"), vehicle_summary.get("odometer"), 0],
+			self._to_int,
+		) or 0
 		color = self._normalize_color(first_non_empty([core.get("color"), vehicle_summary.get("color"), "other"]))
 		drive = self._normalize_drive(first_non_empty([core.get("drive"), vehicle_summary.get("drive")]))
 		gearbox = self._normalize_gearbox(first_non_empty([core.get("gearbox"), vehicle_summary.get("gearbox")]))
@@ -83,7 +95,19 @@ class CarMapper:
 		repair_history = self._boolish(first_non_empty([core.get("repair_history"), vehicle_summary.get("repair_history"), False]))
 		non_smoking = self._boolish(first_non_empty([core.get("non_smoking"), vehicle_summary.get("non_smoking"), True]))
 		evaluation = self._as_str(first_non_empty([core.get("grade"), core.get("evaluation"), vehicle_summary.get("evaluation")])) or None
-		currency = self._normalize_currency(first_non_empty([pricing.get("currency"), vehicle_summary.get("currency"), "USD"]))
+		currency = self._normalize_currency(
+			first_non_empty(
+				[
+					pricing.get("currency"),
+					vehicle_summary.get("currency"),
+					pricing.get("buy_now"),
+					pricing.get("current_bid"),
+					vehicle_summary.get("buy_now"),
+					vehicle_summary.get("current_bid"),
+					"USD",
+				]
+			)
+		)
 		slug = self._slugify(" ".join(filter(None, [str(year or ""), brand, model, origin_id])))
 		images_records = self._build_images(images.get("urls") or vehicle_summary.get("image_urls") or [])
 		origin = "IAAI" if "IAAI" in ORIGIN_ENUM_VALUES else "NA"
@@ -162,6 +186,76 @@ class CarMapper:
 		digits = re.sub(r"[^\d]", "", str(value))
 		return int(digits) if digits else None

+	@classmethod
+	def _to_money_int(cls, value: Any) -> int | None:
+		# Нормализация «грязной» стоимости: "$4,500", "4 500 USD", "4.500,00 €", "USD 4,500 - 5,200".
+		if value is None:
+			return None
+		if isinstance(value, bool):
+			return None
+		if isinstance(value, (int, float)):
+			return int(value)
+
+		text = str(value).strip()
+		if not text:
+			return None
+
+		lowered = text.lower()
+		if any(token in lowered for token in ["n/a", "na", "tbd", "unknown", "call", "contact"]):
+			return None
+
+		numbers = re.findall(r"\d[\d\s.,]*", text)
+		if not numbers:
+			return None
+
+		best: int | None = None
+		for number in numbers:
+			clean = number.replace(" ", "")
+			if "," in clean and "." in clean:
+				# Поддержка и 1,234.56, и 1.234,56.
+				if clean.rfind(",") > clean.rfind("."):
+					clean = clean.replace(".", "").replace(",", ".")
+				else:
+					clean = clean.replace(",", "")
+			elif "," in clean:
+				parts = clean.split(",")
+				# Десятичный формат 123,45 -> 123.45 иначе считаем разделителем тысяч.
+				if len(parts[-1]) in {1, 2} and len(parts) == 2:
+					clean = clean.replace(",", ".")
+				else:
+					clean = clean.replace(",", "")
+			elif "." in clean:
+				parts = clean.split(".")
+				if not (len(parts[-1]) in {1, 2} and len(parts) == 2):
+					clean = clean.replace(".", "")
+
+			try:
+				parsed = int(float(clean))
+			except ValueError:
+				continue
+
+			if parsed > 0 and (best is None or parsed > best):
+				best = parsed
+
+		return best
+
+	@staticmethod
+	def _first_parsed_int(values: list[Any], parser) -> int | None:
+		for value in values:
+			parsed = parser(value)
+			if parsed is not None:
+				return parsed
+		return None
+
+	@staticmethod
+	def _to_year(value: Any) -> int | None:
+		parsed = CarMapper._to_int(value)
+		if parsed is None:
+			return None
+		if 1900 <= parsed <= 2100:
+			return parsed
+		return None
+
 	def _to_engine_cc(self, value: Any) -> int | None:
 		# Поддерживаем и литры, и уже готовые cc.
 		text = str(value).lower().strip() if value is not None else ""
@@ -174,7 +268,24 @@ class CarMapper:
 		return int(text) if re.match(r"^\d+$", text) else None

 	def _normalize_currency(self, value: Any) -> str:
-		text = self._as_str(value).upper() or "USD"
+		text = self._as_str(value)
+		upper = text.upper() or "USD"
+		if any(token in text for token in ["€", "EUR"]):
+			return "EUR"
+		if any(token in text for token in ["¥", "JPY"]):
+			return "JPY"
+		if any(token in text for token in ["₩", "KRW"]):
+			return "KRW"
+		if any(token in text for token in ["£", "GBP"]):
+			return "GBP"
+		if any(token in text for token in ["₽", "RUB"]):
+			return "RUB"
+		if any(token in text for token in ["AED", "د.إ"]):
+			return "AED"
+		if any(token in text for token in ["CA$", "CAD"]):
+			return "CAD"
+
+		text = upper
 		if text in CURRENCY_ENUM_VALUES:
 			return text
 		return "USD" if "$" in str(value) else "USD"
--- a/iaai_scraper/proxy_bridge.py
+++ b/iaai_scraper/proxy_bridge.py
@@ -0,0 +1,317 @@
+from __future__ import annotations
+
+import logging
+import os
+import select
+import socket
+import socketserver
+import struct
+import threading
+from urllib.parse import urlsplit
+
+BUFFER_SIZE = 65536
+CRLF = b"\r\n"
+DEFAULT_LISTEN_HOST = os.getenv("PROXY_BRIDGE_HOST", "0.0.0.0")
+DEFAULT_LISTEN_PORT = int(os.getenv("PROXY_BRIDGE_PORT", "8899"))
+SOCKS5_HOST = os.getenv("SOCKS5_PROXY_HOST", "")
+SOCKS5_PORT = int(os.getenv("SOCKS5_PROXY_PORT", "1002"))
+SOCKS5_USER = os.getenv("SOCKS5_PROXY_USER", "")
+SOCKS5_PASS = os.getenv("SOCKS5_PROXY_PASS", "")
+RELAY_IDLE_TIMEOUT_SECONDS = int(os.getenv("PROXY_BRIDGE_RELAY_IDLE_TIMEOUT_SECONDS", "60"))
+MAX_WORKERS = int(os.getenv("PROXY_BRIDGE_MAX_WORKERS", "64"))
+
+logger = logging.getLogger("proxy_bridge")
+
+
+class ThreadingTCPServer(socketserver.ThreadingMixIn, socketserver.TCPServer):
+    allow_reuse_address = True
+    daemon_threads = True
+
+    def __init__(self, server_address, request_handler_class):
+        super().__init__(server_address, request_handler_class)
+        self._worker_semaphore = threading.BoundedSemaphore(MAX_WORKERS)
+
+    def process_request_thread(self, request, client_address):
+        with self._worker_semaphore:
+            super().process_request_thread(request, client_address)
+
+
+def _recv_exact(sock: socket.socket, size: int) -> bytes:
+    data = b""
+    while len(data) < size:
+        chunk = sock.recv(size - len(data))
+        if not chunk:
+            raise ConnectionError("Unexpected EOF from SOCKS5 server")
+        data += chunk
+    return data
+
+
+def _socks5_connect(host: str, port: int) -> socket.socket:
+    if not SOCKS5_HOST:
+        raise RuntimeError("SOCKS5_PROXY_HOST is not configured")
+
+    upstream = socket.create_connection((SOCKS5_HOST, SOCKS5_PORT), timeout=30)
+    upstream.settimeout(30)
+
+    methods = [0x00]
+    if SOCKS5_USER or SOCKS5_PASS:
+        methods = [0x02]
+    upstream.sendall(bytes([0x05, len(methods), *methods]))
+    version, method = _recv_exact(upstream, 2)
+    if version != 0x05 or method == 0xFF:
+        upstream.close()
+        raise ConnectionError("SOCKS5 authentication negotiation failed")
+
+    if method == 0x02:
+        username = SOCKS5_USER.encode("utf-8")
+        password = SOCKS5_PASS.encode("utf-8")
+        if len(username) > 255 or len(password) > 255:
+            upstream.close()
+            raise ValueError("SOCKS5 username/password too long")
+        upstream.sendall(bytes([0x01, len(username)]) + username + bytes([len(password)]) + password)
+        auth_version, auth_status = _recv_exact(upstream, 2)
+        if auth_version != 0x01 or auth_status != 0x00:
+            upstream.close()
+            raise ConnectionError("SOCKS5 username/password authentication failed")
+
+    try:
+        socket.inet_aton(host)
+        addr_type = 0x01
+        addr_payload = socket.inet_aton(host)
+    except OSError:
+        host_bytes = host.encode("idna")
+        if len(host_bytes) > 255:
+            upstream.close()
+            raise ValueError("Target host is too long for SOCKS5 domain format")
+        addr_type = 0x03
+        addr_payload = bytes([len(host_bytes)]) + host_bytes
+
+    request = bytes([0x05, 0x01, 0x00, addr_type]) + addr_payload + struct.pack("!H", port)
+    upstream.sendall(request)
+
+    response_head = _recv_exact(upstream, 4)
+    version, reply, _reserved, reply_addr_type = response_head
+    if version != 0x05 or reply != 0x00:
+        upstream.close()
+        raise ConnectionError(f"SOCKS5 connect failed with code {reply}")
+
+    if reply_addr_type == 0x01:
+        _recv_exact(upstream, 4)
+    elif reply_addr_type == 0x03:
+        domain_len = _recv_exact(upstream, 1)[0]
+        _recv_exact(upstream, domain_len)
+    elif reply_addr_type == 0x04:
+        _recv_exact(upstream, 16)
+    _recv_exact(upstream, 2)
+
+    upstream.settimeout(RELAY_IDLE_TIMEOUT_SECONDS)
+    return upstream
+
+
+def _relay_bidirectional(left: socket.socket, right: socket.socket) -> None:
+    sockets = [left, right]
+    left.settimeout(RELAY_IDLE_TIMEOUT_SECONDS)
+    right.settimeout(RELAY_IDLE_TIMEOUT_SECONDS)
+    try:
+        while True:
+            readable, _, exceptional = select.select(sockets, [], sockets, RELAY_IDLE_TIMEOUT_SECONDS)
+            if exceptional:
+                break
+            if not readable:
+                logger.debug("Relay idle timeout reached; closing sockets")
+                return
+            for current in readable:
+                other = right if current is left else left
+                data = current.recv(BUFFER_SIZE)
+                if not data:
+                    return
+                other.sendall(data)
+    finally:
+        for sock in sockets:
+            try:
+                sock.shutdown(socket.SHUT_RDWR)
+            except OSError:
+                pass
+            try:
+                sock.close()
+            except OSError:
+                pass
+
+
+class ProxyHandler(socketserver.StreamRequestHandler):
+    def handle(self) -> None:
+        try:
+            request_line = self.rfile.readline(BUFFER_SIZE).decode("iso-8859-1").strip()
+            if not request_line:
+                return
+
+            method, target, version = request_line.split()
+            headers = self._read_headers()
+            logger.info("%s %s", method, target)
+
+            if method.upper() == "CONNECT":
+                host, port = self._parse_connect_target(target)
+                logger.info("CONNECT %s:%s", host, port)
+                upstream = _socks5_connect(host, port)
+                self.wfile.write(f"{version} 200 Connection Established".encode("ascii") + CRLF + CRLF)
+                self.wfile.flush()
+                _relay_bidirectional(self.connection, upstream)
+                return
+
+            host, port, path = self._parse_forward_target(target, headers)
+            upstream = _socks5_connect(host, port)
+            self._send_forward_request(upstream, method, path, version, headers)
+            body = self._read_request_body(headers)
+            if body:
+                upstream.sendall(body)
+            _relay_bidirectional(self.connection, upstream)
+        except Exception as exc:
+            logger.exception("Proxy bridge request failed: %s", exc)
+            try:
+                self.wfile.write(
+                    b"HTTP/1.1 502 Bad Gateway" + CRLF
+                    + b"Connection: close" + CRLF
+                    + b"Content-Type: text/plain; charset=utf-8" + CRLF + CRLF
+                    + b"Bad Gateway"
+                )
+                self.wfile.flush()
+            except OSError:
+                pass
+
+    def _read_headers(self) -> list[tuple[str, str]]:
+        headers: list[tuple[str, str]] = []
+        while True:
+            line = self.rfile.readline(BUFFER_SIZE)
+            if line in {CRLF, b"\n", b""}:
+                break
+            decoded = line.decode("iso-8859-1")
+            if ":" not in decoded:
+                continue
+            name, value = decoded.split(":", 1)
+            headers.append((name.strip(), value.strip()))
+        return headers
+
+    @staticmethod
+    def _parse_connect_target(target: str) -> tuple[str, int]:
+        if target.startswith("["):
+            end = target.find("]")
+            if end == -1 or len(target) <= end + 2 or target[end + 1] != ":":
+                raise ValueError("Invalid CONNECT target")
+            host = target[1:end]
+            port_text = target[end + 2 :]
+            return host, int(port_text)
+
+        host, port_text = target.rsplit(":", 1)
+        return host, int(port_text)
+
+    @staticmethod
+    def _parse_forward_target(target: str, headers: list[tuple[str, str]]) -> tuple[str, int, str]:
+        if target.startswith("http://"):
+            parts = urlsplit(target)
+            port = parts.port or 80
+            path = parts.path or "/"
+            if parts.query:
+                path += f"?{parts.query}"
+            return parts.hostname or "", port, path
+
+        if target.startswith("https://"):
+            raise ValueError("HTTPS absolute-form request must use CONNECT")
+
+        host_header = next((value for name, value in headers if name.lower() == "host"), "")
+        if not host_header:
+            raise ValueError("Missing Host header")
+        if ":" in host_header:
+            host, port_text = host_header.rsplit(":", 1)
+            return host, int(port_text), target
+        return host_header, 80, target
+
+    def _send_forward_request(
+        self,
+        upstream: socket.socket,
+        method: str,
+        path: str,
+        version: str,
+        headers: list[tuple[str, str]],
+    ) -> None:
+        filtered_headers: list[tuple[str, str]] = []
+        hop_by_hop = {
+            "proxy-connection",
+            "proxy-authorization",
+            "connection",
+            "keep-alive",
+            "te",
+            "trailer",
+            "transfer-encoding",
+            "upgrade",
+        }
+        for name, value in headers:
+            if name.lower() in hop_by_hop:
+                continue
+            filtered_headers.append((name, value))
+
+        request_head = [f"{method} {path} {version}\r\n"]
+        request_head.extend(f"{name}: {value}\r\n" for name, value in filtered_headers)
+        request_head.append("\r\n")
+        upstream.sendall("".join(request_head).encode("iso-8859-1"))
+
+    def _read_request_body(self, headers: list[tuple[str, str]]) -> bytes:
+        transfer_encoding = next((value for name, value in headers if name.lower() == "transfer-encoding"), "")
+        if "chunked" in transfer_encoding.lower():
+            return self._read_chunked_request_body()
+
+        content_length = next((value for name, value in headers if name.lower() == "content-length"), None)
+        if not content_length:
+            return b""
+        return self.rfile.read(int(content_length))
+
+    def _read_chunked_request_body(self) -> bytes:
+        chunks: list[bytes] = []
+        while True:
+            size_line = self.rfile.readline(BUFFER_SIZE)
+            if not size_line:
+                raise ConnectionError("Unexpected EOF in chunked request")
+            size_text = size_line.strip().split(b";", 1)[0]
+            chunk_size = int(size_text, 16)
+            chunks.append(size_line)
+            if chunk_size == 0:
+                while True:
+                    trailer_line = self.rfile.readline(BUFFER_SIZE)
+                    if not trailer_line:
+                        raise ConnectionError("Unexpected EOF in chunked trailers")
+                    chunks.append(trailer_line)
+                    if trailer_line in {CRLF, b"\n"}:
+                        return b"".join(chunks)
+
+            chunk_data = self.rfile.read(chunk_size)
+            if len(chunk_data) != chunk_size:
+                raise ConnectionError("Unexpected EOF in chunk body")
+            chunks.append(chunk_data)
+            chunk_end = self.rfile.read(2)
+            if chunk_end != CRLF:
+                raise ConnectionError("Invalid chunk terminator")
+            chunks.append(chunk_end)
+
+
+def main() -> None:
+    if not SOCKS5_HOST:
+        raise SystemExit("SOCKS5_PROXY_HOST is required")
+
+    logging.basicConfig(
+        level=os.getenv("PROXY_BRIDGE_LOG_LEVEL", "INFO").upper(),
+        format="[%(asctime)s] [proxy_bridge] %(levelname)s: %(message)s",
+    )
+
+    with ThreadingTCPServer((DEFAULT_LISTEN_HOST, DEFAULT_LISTEN_PORT), ProxyHandler) as server:
+        logger.info(
+            "Listening on %s:%s -> socks5://%s:%s (max_workers=%s)",
+            DEFAULT_LISTEN_HOST,
+            DEFAULT_LISTEN_PORT,
+            SOCKS5_HOST,
+            SOCKS5_PORT,
+            MAX_WORKERS,
+        )
+        server.serve_forever()
+
+
+if __name__ == "__main__":
+    main()
--- a/iaai_scraper/scraper.py
+++ b/iaai_scraper/scraper.py
@@ -1,4 +1,5 @@
 import logging
+import re
 import signal
 import time
 import uuid
@@ -21,10 +22,18 @@ from .storage.listing import ListingCollector
 from .storage.schemas import CarRecord

 logger = logging.getLogger("iaai_scraper.scraper")
+VEHICLE_ID_RE = re.compile(r"/VehicleDetail/(\d+)(?:~[A-Z]{2})?", re.IGNORECASE)


 class IAAIScraper:

+    @staticmethod
+    def _extract_origin_id_from_url(vehicle_url: str) -> str | None:
+        match = VEHICLE_ID_RE.search(vehicle_url)
+        if not match:
+            return None
+        return match.group(1)
+
    def __init__(self, runtime_settings: Settings | None = None) -> None:
        # Базовые зависимости и сервисы скрапера.
        self.settings = runtime_settings or settings
@@ -260,7 +269,14 @@ class IAAIScraper:
                error_summary=error_summary,
            )

-    def sync_listing(self, make: str | None = None, model: str | None = None, lane: str = "iaai_cars", limit: int | None = None):
+    def sync_listing(
+        self,
+        make: str | None = None,
+        model: str | None = None,
+        lane: str = "iaai_cars",
+        limit: int | None = None,
+        only_new: bool | None = None,
+    ):
        """Листинг + sync всех найденных машин."""
        # Массовая синхронизация с общим run_id и сбором ошибок.
        trace_id = self._new_trace_id("sync-listing")
@@ -271,6 +287,7 @@ class IAAIScraper:
        cars_failed = 0
        images_upserted = 0
        total = 0
+        skipped_existing = 0
        failures: list[dict[str, str]] = []
        listing: dict = {}

@@ -280,6 +297,27 @@ class IAAIScraper:
            if limit is not None:
                vehicle_urls = vehicle_urls[:max(0, limit)]

+            effective_only_new = self.settings.sync_only_new if only_new is None else only_new
+            if effective_only_new:
+                existing_urls = self.persistence.get_existing_origin_urls(vehicle_urls)
+                url_to_origin_id = {
+                    url: self._extract_origin_id_from_url(url)
+                    for url in vehicle_urls
+                }
+                candidate_origin_ids = [origin_id for origin_id in url_to_origin_id.values() if origin_id]
+                existing_ids = self.persistence.get_existing_origin_ids(candidate_origin_ids)
+
+                known_urls = {
+                    url
+                    for url in vehicle_urls
+                    if (url in existing_urls) or (url_to_origin_id.get(url) in existing_ids)
+                }
+
+                skipped_existing = len(known_urls)
+                if skipped_existing:
+                    logger.info("Filtering already known vehicles: skipped %d", skipped_existing)
+                vehicle_urls = [url for url in vehicle_urls if url not in known_urls]
+
            total = len(vehicle_urls)
            logger.info("Starting sync: %d vehicles to process", total)

@@ -345,6 +383,7 @@ class IAAIScraper:
            "cars_upserted": cars_upserted,
            "cars_failed": cars_failed,
            "images_upserted": images_upserted,
+            "skipped_existing": skipped_existing,
            "elapsed_seconds": round(time.perf_counter() - started_at, 3),
            "failures": failures,
        }
--- a/iaai_scraper/storage/db.py
+++ b/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)
--- a/iaai_scraper/storage/enums.py
+++ b/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")
--- a/requirements.txt
+++ b/requirements.txt
@@ -2,5 +2,6 @@ playwright>=1.53.0
 python-dotenv>=1.0.1
 pydantic>=2.8.2
 SQLAlchemy>=2.0.32
+PySocks>=1.7.1
 pytest>=8.3.0
 pytest-cov>=5.0.0
--- a/tests/test_db.py
+++ b/tests/test_db.py
@@ -8,7 +8,7 @@ from sqlalchemy import select

 from iaai_scraper.core.config import Settings
 from iaai_scraper.storage.db import PersistenceService
-from iaai_scraper.storage.models import Car, Image
+from iaai_scraper.storage.models import Car, Image, SyncRun
 from iaai_scraper.storage.schemas import CarRecord, ImageRecord


@@ -116,6 +116,36 @@ class TestPersistenceServiceIntegration(unittest.TestCase):
            car = session.execute(select(Car).where(Car.origin_id == "1000")).scalar_one()
        self.assertEqual(car.origin_id, "1000")

+    def test_start_sync_run_marks_stale_running_runs_as_failed(self) -> None:
+        first_run_id = self.persistence.start_sync_run("lane-a")
+        second_run_id = self.persistence.start_sync_run("lane-b")
+
+        self.assertNotEqual(first_run_id, second_run_id)
+
+        with self.persistence.session_scope() as session:
+            first = session.get(SyncRun, first_run_id)
+            second = session.get(SyncRun, second_run_id)
+
+        self.assertEqual(first.status, "failed")
+        self.assertIsNotNone(first.finished_at)
+        self.assertEqual(second.status, "running")
+
+    def test_upsert_falls_back_to_origin_url_to_prevent_duplicates(self) -> None:
+        first = self._record("OLD-ID", content_hash="v1")
+        first.origin_url = "https://www.iaai.com/VehicleDetail/45089484~US"
+        self.persistence.upsert_car(first)
+
+        second = self._record("NEW-ID", content_hash="v2")
+        second.origin_url = "https://www.iaai.com/VehicleDetail/45089484~US"
+        result = self.persistence.upsert_car(second)
+
+        self.assertEqual(result["action"], "updated")
+        with self.persistence.session_scope() as session:
+            cars = session.execute(select(Car)).scalars().all()
+
+        self.assertEqual(len(cars), 1)
+        self.assertEqual(cars[0].origin_id, "NEW-ID")
+

 if __name__ == "__main__":
    unittest.main()
--- a/tests/test_mappers.py
+++ b/tests/test_mappers.py
@@ -22,7 +22,6 @@ class TestCarMapper(unittest.TestCase):
            },
        )

-        # Длина hex-представления SHA-256
        self.assertEqual(len(record.content_hash), 64)

    def test_content_hash_changes_when_image_set_changes(self) -> None:
@@ -104,6 +103,37 @@ class TestCarMapper(unittest.TestCase):
        self.assertEqual(record.drive, "FWD")
        self.assertEqual(record.gearbox, "AT")

+    def test_price_parsing_dirty_formats(self) -> None:
+        record = self.mapper.map_to_car_record(
+            vehicle_url="https://www.iaai.com/VehicleDetail/777~US",
+            vehicle_summary={"make": "Toyota", "model": "Corolla"},
+            payload_insights={
+                "vehicle_core": {},
+                "pricing": {"buy_now": "USD 4,500 - 5,200"},
+                "damage": {},
+                "auction": {},
+                "images": {},
+            },
+        )
+
+        self.assertEqual(record.price, 5200)
+
+    def test_currency_detection_from_symbol(self) -> None:
+        record = self.mapper.map_to_car_record(
+            vehicle_url="https://www.iaai.com/VehicleDetail/778~US",
+            vehicle_summary={"make": "Toyota", "model": "Corolla"},
+            payload_insights={
+                "vehicle_core": {},
+                "pricing": {"buy_now": "€4.500,00"},
+                "damage": {},
+                "auction": {},
+                "images": {},
+            },
+        )
+
+        self.assertEqual(record.currency, "EUR")
+        self.assertEqual(record.price, 4500)
+

 if __name__ == "__main__":
    unittest.main()
--- a/tests/test_scraper.py
+++ b/tests/test_scraper.py
@@ -50,6 +50,8 @@ class TestScraperSync(unittest.TestCase):
        scraper.persistence.start_sync_run = MagicMock(return_value=2)
        scraper.persistence.finish_sync_run = MagicMock()
        scraper.persistence.upsert_car = MagicMock(return_value={"action": "inserted", "images_upserted": 1})
+        scraper.persistence.get_existing_origin_urls = MagicMock(return_value=set())
+        scraper.persistence.get_existing_origin_ids = MagicMock(return_value=set())

        scraper.collect_listing = MagicMock(return_value={"vehicle_urls": ["https://www.iaai.com/VehicleDetail/222~US"]})
        page = MagicMock()
@@ -73,6 +75,8 @@ class TestScraperSync(unittest.TestCase):
        scraper.persistence.start_sync_run = MagicMock(return_value=3)
        scraper.persistence.finish_sync_run = MagicMock()
        scraper.persistence.upsert_car = MagicMock(return_value={"action": "inserted", "images_upserted": 1})
+        scraper.persistence.get_existing_origin_urls = MagicMock(return_value=set())
+        scraper.persistence.get_existing_origin_ids = MagicMock(return_value=set())
        scraper.collect_listing = MagicMock(return_value={
            "vehicle_urls": [
                "https://www.iaai.com/VehicleDetail/222~US",
@@ -93,6 +97,8 @@ class TestScraperSync(unittest.TestCase):
        scraper.persistence.start_sync_run = MagicMock(return_value=4)
        scraper.persistence.finish_sync_run = MagicMock()
        scraper.persistence.upsert_car = MagicMock(return_value={"action": "skipped", "images_upserted": 0})
+        scraper.persistence.get_existing_origin_urls = MagicMock(return_value=set())
+        scraper.persistence.get_existing_origin_ids = MagicMock(return_value=set())
        scraper.collect_listing = MagicMock(return_value={"vehicle_urls": ["https://www.iaai.com/VehicleDetail/444~US"]})
        scraper._scrape_on_page = MagicMock(return_value={"db_record": make_db_record("444")})
        scraper._get_page = MagicMock(return_value=MagicMock())
@@ -101,6 +107,35 @@ class TestScraperSync(unittest.TestCase):

        self.assertEqual(result["cars_upserted"], 0)

+    def test_sync_listing_only_new_filters_existing_by_url_and_origin_id(self) -> None:
+        scraper = self._make_scraper()
+
+        scraper.persistence.create_tables = MagicMock()
+        scraper.persistence.start_sync_run = MagicMock(return_value=5)
+        scraper.persistence.finish_sync_run = MagicMock()
+        scraper.persistence.upsert_car = MagicMock(return_value={"action": "inserted", "images_upserted": 0})
+        scraper.persistence.get_existing_origin_urls = MagicMock(return_value={"https://www.iaai.com/VehicleDetail/111~US"})
+        scraper.persistence.get_existing_origin_ids = MagicMock(return_value={"222"})
+
+        scraper.collect_listing = MagicMock(return_value={
+            "vehicle_urls": [
+                "https://www.iaai.com/VehicleDetail/111~US",  # exists by URL
+                "https://www.iaai.com/VehicleDetail/222~US",  # exists by ID
+                "https://www.iaai.com/VehicleDetail/333~US",  # new
+            ]
+        })
+        page = MagicMock()
+        scraper._get_page = MagicMock(return_value=page)
+        scraper._scrape_on_page = MagicMock(return_value={"db_record": make_db_record("333")})
+
+        result = scraper.sync_listing(only_new=True)
+
+        self.assertEqual(result["skipped_existing"], 2)
+        self.assertEqual(result["cars_upserted"], 1)
+        self.assertEqual(scraper._scrape_on_page.call_count, 1)
+        scraper.persistence.get_existing_origin_urls.assert_called_once()
+        scraper.persistence.get_existing_origin_ids.assert_called_once()
+
    def test_close_resets_browser_state(self) -> None:
        scraper = self._make_scraper()
        scraper.context = MagicMock()