From 6821404d798793e8673edbe5d19838293b0e2a81 Mon Sep 17 00:00:00 2001 From: qananasikq Date: Wed, 8 Apr 2026 18:08:16 +0300 Subject: [PATCH] feat: harden sync flow, proxy bridge, and data normalization --- .dockerignore | 1 + .env.example | 1 + Dockerfile | 4 + README.md | 20 ++- entrypoint.sh | 27 +++ iaai_scraper/cli.py | 31 +++- iaai_scraper/core/config.py | 1 + iaai_scraper/core/utils.py | 4 +- iaai_scraper/parsing/mapper.py | 121 ++++++++++++- iaai_scraper/proxy_bridge.py | 317 +++++++++++++++++++++++++++++++++ iaai_scraper/scraper.py | 41 ++++- iaai_scraper/storage/db.py | 37 +++- iaai_scraper/storage/enums.py | 2 +- requirements.txt | 1 + tests/test_db.py | 32 +++- tests/test_mappers.py | 32 +++- tests/test_scraper.py | 35 ++++ 17 files changed, 683 insertions(+), 24 deletions(-) create mode 100644 iaai_scraper/proxy_bridge.py diff --git a/.dockerignore b/.dockerignore index 3351588..8a2cab3 100644 --- a/.dockerignore +++ b/.dockerignore @@ -14,3 +14,4 @@ coverage.xml *.egg-info/ dist/ build/ +artifacts/ diff --git a/.env.example b/.env.example index c0b32b1..8e87c4a 100644 --- 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 diff --git a/Dockerfile b/Dockerfile index 903b325..822174a 100644 --- a/Dockerfile +++ b/Dockerfile @@ -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 diff --git a/README.md b/README.md index fec8638..1d418ad 100644 --- 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 ``` diff --git a/entrypoint.sh b/entrypoint.sh index a645310..9423a62 100644 --- 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 "$@" diff --git a/iaai_scraper/cli.py b/iaai_scraper/cli.py index ae2029b..168840e 100644 --- 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()}") diff --git a/iaai_scraper/core/config.py b/iaai_scraper/core/config.py index 1d1dd81..5b3fe7c 100644 --- 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) diff --git a/iaai_scraper/core/utils.py b/iaai_scraper/core/utils.py index b02e385..30450a0 100644 --- 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: diff --git a/iaai_scraper/parsing/mapper.py b/iaai_scraper/parsing/mapper.py index 85d772c..0e16984 100644 --- 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" diff --git a/iaai_scraper/proxy_bridge.py b/iaai_scraper/proxy_bridge.py new file mode 100644 index 0000000..e2b0b99 --- /dev/null +++ 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() diff --git a/iaai_scraper/scraper.py b/iaai_scraper/scraper.py index 5e7007c..38fc552 100644 --- 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, } diff --git a/iaai_scraper/storage/db.py b/iaai_scraper/storage/db.py index b2c6c4f..861f708 100644 --- 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) diff --git a/iaai_scraper/storage/enums.py b/iaai_scraper/storage/enums.py index 88bda45..f8e34a3 100644 --- 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") \ No newline at end of file +SELLING_TYPE_ENUM_VALUES = ("STOCK", "AUCTION", "TENDER", "stock", "auction", "tender", "NA") diff --git a/requirements.txt b/requirements.txt index 064ce5e..847a1b3 100644 --- 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 diff --git a/tests/test_db.py b/tests/test_db.py index f4934bd..62ef84d 100644 --- 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() diff --git a/tests/test_mappers.py b/tests/test_mappers.py index dec9943..93b9fd7 100644 --- 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() diff --git a/tests/test_scraper.py b/tests/test_scraper.py index b8b9eb2..6228b92 100644 --- 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()