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/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/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)