diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..95641c6 --- /dev/null +++ b/.env.example @@ -0,0 +1,21 @@ +POSTGRES_HOST=postgres +POSTGRES_PORT=5432 +POSTGRES_USER=postgres +POSTGRES_PASSWORD=postgres +POSTGRES_DB=iaai +IAAI_DATABASE_URL=postgresql+psycopg://postgres:postgres@postgres:5432/iaai +CELERY_BROKER_URL=redis://redis:6379/0 +IAAI_BASE_URL=https://www.iaai.com +IAAI_LISTING_START_URL=https://www.iaai.com/Search?url=B%2bU066dM8%2flZtRvnzTwIEi8Pib9%2fM2fLpCTQDIgQRm4%3d +IAAI_STORAGE_STATE_PATH=/app/iaai_storage_state.json +IAAI_HTTP_TIMEOUT=30 +IAAI_HTTP_RETRIES=2 +IAAI_HTTP_RETRY_BACKOFF_MS=700 +IAAI_FETCH_CONCURRENCY=24 +IAAI_DB_COMMIT_BATCH_SIZE=200 +IAAI_SESSION_KEEPALIVE_ENABLED=true +IAAI_SESSION_KEEPALIVE_INTERVAL_MINUTES=15 +IAAI_SCHEMA_BOOTSTRAP_ENABLED=true +SYNC_RUNTIME_CONFIG_FILE=/app/sync_runtime_config.json +SYNC_ADVISORY_LOCK_KEY=7642200 +SYNC_ERROR_SUMMARY_MAX_LEN=2000 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..0a8d523 --- /dev/null +++ b/.gitignore @@ -0,0 +1,219 @@ +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[codz] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +share/python-wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# PyInstaller +# Usually these files are written by a python script from a template +# before PyInstaller builds the exe, so as to inject date/other infos into it. +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py.cover +.hypothesis/ +.pytest_cache/ +cover/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py +db.sqlite3 +db.sqlite3-journal + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +.pybuilder/ +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# IPython +profile_default/ +ipython_config.py + +# pyenv +# For a library or package, you might want to ignore these files since the code is +# intended to run in multiple environments; otherwise, check them in: +# .python-version + +# pipenv +# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control. +# However, in case of collaboration, if having platform-specific dependencies or dependencies +# having no cross-platform support, pipenv may install dependencies that don't work, or not +# install all needed dependencies. +# Pipfile.lock + +# UV +# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control. +# This is especially recommended for binary packages to ensure reproducibility, and is more +# commonly ignored for libraries. +# uv.lock + +# poetry +# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control. +# This is especially recommended for binary packages to ensure reproducibility, and is more +# commonly ignored for libraries. +# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control +# poetry.lock +# poetry.toml + +# pdm +# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control. +# pdm recommends including project-wide configuration in pdm.toml, but excluding .pdm-python. +# https://pdm-project.org/en/latest/usage/project/#working-with-version-control +# pdm.lock +# pdm.toml +.pdm-python +.pdm-build/ + +# pixi +# Similar to Pipfile.lock, it is generally recommended to include pixi.lock in version control. +# pixi.lock +# Pixi creates a virtual environment in the .pixi directory, just like venv module creates one +# in the .venv directory. It is recommended not to include this directory in version control. +.pixi + +# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm +__pypackages__/ + +# Celery stuff +celerybeat-schedule +celerybeat.pid + +# Redis +*.rdb +*.aof +*.pid + +# RabbitMQ +mnesia/ +rabbitmq/ +rabbitmq-data/ + +# ActiveMQ +activemq-data/ + +# SageMath parsed files +*.sage.py + +# Environments +.env +.envrc +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# Spyder project settings +.spyderproject +.spyproject + +# Rope project settings +.ropeproject + +# mkdocs documentation +/site + +# mypy +.mypy_cache/ +.dmypy.json +dmypy.json + +# Pyre type checker +.pyre/ + +# pytype static type analyzer +.pytype/ + +# Cython debug symbols +cython_debug/ + +# PyCharm +# JetBrains specific template is maintained in a separate JetBrains.gitignore that can +# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore +# and can be added to the global gitignore or merged into this file. For a more nuclear +# option (not recommended) you can uncomment the following to ignore the entire idea folder. +# .idea/ + +# Abstra +# Abstra is an AI-powered process automation framework. +# Ignore directories containing user credentials, local state, and settings. +# Learn more at https://abstra.io/docs +.abstra/ + +# Visual Studio Code +# Visual Studio Code specific template is maintained in a separate VisualStudioCode.gitignore +# that can be found at https://github.com/github/gitignore/blob/main/Global/VisualStudioCode.gitignore +# and can be added to the global gitignore or merged into this file. However, if you prefer, +# you could uncomment the following to ignore the entire vscode folder +# .vscode/ + +# Ruff stuff: +.ruff_cache/ + +# PyPI configuration file +.pypirc + +# Marimo +marimo/_static/ +marimo/_lsp/ +__marimo__/ + +# Streamlit +.streamlit/secrets.toml +iaai_storage_state.json +html/catalog.html +html/page.html diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..cedf220 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,19 @@ +FROM python:3.13-slim + +ENV PYTHONDONTWRITEBYTECODE=1 +ENV PYTHONUNBUFFERED=1 + +WORKDIR /app + +RUN pip install --no-cache-dir --upgrade pip + +COPY pyproject.toml /app/ +COPY iaai_sync_service /app/iaai_sync_service +COPY alembic.ini /app/alembic.ini +COPY alembic /app/alembic +COPY sync_runtime_config.json /app/sync_runtime_config.json + +RUN pip install --no-cache-dir . \ + && playwright install --with-deps chromium + +CMD ["python", "-m", "iaai_sync_service.cli"] diff --git a/alembic.ini b/alembic.ini new file mode 100644 index 0000000..725a864 --- /dev/null +++ b/alembic.ini @@ -0,0 +1,38 @@ +[alembic] +script_location = alembic +prepend_sys_path = . +version_path_separator = os +sqlalchemy.url = postgresql+psycopg://postgres:postgres@localhost:5432/iaai + +[loggers] +keys = root,sqlalchemy,alembic + +[handlers] +keys = console + +[formatters] +keys = generic + +[logger_root] +level = WARN +handlers = console +qualname = + +[logger_sqlalchemy] +level = WARN +handlers = +qualname = sqlalchemy.engine + +[logger_alembic] +level = INFO +handlers = +qualname = alembic + +[handler_console] +class = StreamHandler +args = (sys.stderr,) +level = NOTSET +formatter = generic + +[formatter_generic] +format = %(levelname)-5.5s [%(name)s] %(message)s diff --git a/alembic/env.py b/alembic/env.py new file mode 100644 index 0000000..7891e6a --- /dev/null +++ b/alembic/env.py @@ -0,0 +1,85 @@ +from __future__ import annotations + +import os +from logging.config import fileConfig + +from alembic import context +from sqlalchemy import engine_from_config, pool, text + +from iaai_sync_service.models import Base + +config = context.config + +if config.config_file_name is not None: + fileConfig(config.config_file_name) + +target_metadata = Base.metadata +SYNC_VERSION_TABLE = "iaai_sync_alembic_version" +LEGACY_VERSION_TABLE = "alembic_version" +SYNC_REVISIONS = { + "0001_ensure_sync_schema", + "0002_extend_origin_enum_for_iaai", + "0003_create_iaai_source_tables", +} + + +def _normalize_database_url(url: str) -> str: + normalized = url.strip() + if normalized.startswith("jdbc:"): + normalized = normalized[len("jdbc:") :] + if normalized.startswith("postgres://"): + normalized = "postgresql+psycopg://" + normalized[len("postgres://") :] + elif normalized.startswith("postgresql://"): + normalized = "postgresql+psycopg://" + normalized[len("postgresql://") :] + return normalized + + +def _database_url() -> str: + raw_url = os.getenv("DATABASE_URL", config.get_main_option("sqlalchemy.url")) + return _normalize_database_url(raw_url) + + +def _select_version_table(connection) -> str: # noqa: ANN001 + inspector = connection.dialect + if inspector.has_table(connection, SYNC_VERSION_TABLE): + return SYNC_VERSION_TABLE + if inspector.has_table(connection, LEGACY_VERSION_TABLE): + version_num = connection.execute(text(f"SELECT version_num FROM {LEGACY_VERSION_TABLE} LIMIT 1")).scalar() + if version_num in SYNC_REVISIONS: + return LEGACY_VERSION_TABLE + return SYNC_VERSION_TABLE + + +def run_migrations_offline() -> None: + url = _database_url() + context.configure( + url=url, + target_metadata=target_metadata, + version_table=SYNC_VERSION_TABLE, + literal_binds=True, + dialect_opts={"paramstyle": "named"}, + ) + with context.begin_transaction(): + context.run_migrations() + + +def run_migrations_online() -> None: + configuration = config.get_section(config.config_ini_section) or {} + configuration["sqlalchemy.url"] = _database_url() + connectable = engine_from_config(configuration, prefix="sqlalchemy.", poolclass=pool.NullPool) + + with connectable.connect() as connection: + version_table = _select_version_table(connection) + context.configure( + connection=connection, + target_metadata=target_metadata, + version_table=version_table, + ) + with context.begin_transaction(): + context.run_migrations() + + +if context.is_offline_mode(): + run_migrations_offline() +else: + run_migrations_online() diff --git a/alembic/script.py.mako b/alembic/script.py.mako new file mode 100644 index 0000000..6b99c2a --- /dev/null +++ b/alembic/script.py.mako @@ -0,0 +1,27 @@ +"""${message} + +Revision ID: ${up_revision} +Revises: ${down_revision | comma,n} +Create Date: ${create_date} + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +${imports if imports else ""} + + +# revision identifiers, used by Alembic. +revision: str = ${repr(up_revision)} +down_revision: Union[str, Sequence[str], None] = ${repr(down_revision)} +branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)} +depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)} + + +def upgrade() -> None: + ${upgrades if upgrades else "pass"} + + +def downgrade() -> None: + ${downgrades if downgrades else "pass"} diff --git a/alembic/versions/0001_ensure_sync_schema.py b/alembic/versions/0001_ensure_sync_schema.py new file mode 100644 index 0000000..bd16239 --- /dev/null +++ b/alembic/versions/0001_ensure_sync_schema.py @@ -0,0 +1,165 @@ +"""bootstrap standalone iaai schema + +Revision ID: 0001_ensure_sync_schema +Revises: +Create Date: 2026-04-08 00:00:00 +""" + +from typing import Sequence, Union + +from alembic import op + + +# revision identifiers, used by Alembic. +revision: str = "0001_ensure_sync_schema" +down_revision: Union[str, Sequence[str], None] = None +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def _create_enum_if_missing(type_name: str, values: tuple[str, ...]) -> None: + values_sql = ", ".join("'" + value.replace("'", "''") + "'" for value in values) + op.execute( + f""" + DO $$ + BEGIN + IF NOT EXISTS ( + SELECT 1 + FROM pg_type t + JOIN pg_namespace n ON n.oid = t.typnamespace + WHERE t.typname = '{type_name}' + AND n.nspname = 'public' + ) THEN + CREATE TYPE public.{type_name} AS ENUM ({values_sql}); + END IF; + END + $$; + """ + ) + + +def upgrade() -> None: + _create_enum_if_missing("currencyenum", ("JPY", "USD", "EUR", "RUB", "KRW", "AED", "GBP", "CAD")) + _create_enum_if_missing("driveenum", ("FWD", "RWD", "TWO_WD", "FOUR_WD", "2WD", "4WD", "NA")) + _create_enum_if_missing("gearboxenum", ("AT", "CVT", "MT", "EV", "NA")) + _create_enum_if_missing("steeringwheelenum", ("LEFT", "RIGHT", "left", "right", "NA")) + _create_enum_if_missing( + "bodytypeenum", + ( + "COUPE", + "SUV", + "HATCHBACK", + "MINIVAN", + "SEDAN", + "NA", + "Station Wagon", + "Pickup", + "Truck", + "Open", + "RV", + "Other", + "STATION_WAGON", + "PICKUP", + "TRUCK", + "OPEN", + "OTHER", + ), + ) + _create_enum_if_missing("countryenum", ("JP", "KR", "US", "CA", "NA")) + _create_enum_if_missing( + "originenum", + ( + "TAU", + "CARSENSOR", + "HANAMARU", + "ENCAR", + "KURUMA_TRADER", + "carsensor", + "encar", + "kuruma_trader", + "asnet", + "kababa", + "ACV", + "COPART", + "copart", + "NA", + "ASNET", + "KABABA", + ), + ) + _create_enum_if_missing("sellingtypeenum", ("STOCK", "AUCTION", "TENDER", "stock", "auction", "tender", "NA")) + + op.execute( + """ + CREATE TABLE IF NOT EXISTS public.cars ( + id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, + parser_id VARCHAR(50) NOT NULL UNIQUE, + brand VARCHAR(50) NOT NULL, + model VARCHAR(50) NOT NULL, + year INTEGER NULL, + price BIGINT NULL, + currency public.currencyenum NOT NULL DEFAULT 'USD', + mileage INTEGER NOT NULL DEFAULT 0, + country public.countryenum NOT NULL DEFAULT 'NA', + is_sold BOOLEAN NOT NULL DEFAULT FALSE, + color VARCHAR NOT NULL DEFAULT 'other', + drive public.driveenum NULL, + gearbox public.gearboxenum NULL, + steering_wheel public.steeringwheelenum NULL, + body_type public.bodytypeenum NOT NULL DEFAULT 'OTHER', + engine_volume INTEGER NULL, + selling_type public.sellingtypeenum NOT NULL DEFAULT 'NA', + one_owner BOOLEAN NOT NULL DEFAULT FALSE, + new_car BOOLEAN NOT NULL DEFAULT FALSE, + is_hidden BOOLEAN NOT NULL DEFAULT FALSE, + origin public.originenum NOT NULL DEFAULT 'NA', + origin_url VARCHAR NOT NULL, + origin_id VARCHAR NOT NULL UNIQUE, + is_damaged BOOLEAN NOT NULL DEFAULT FALSE, + evaluation VARCHAR NULL, + non_smoking BOOLEAN NOT NULL DEFAULT TRUE, + rental BOOLEAN NOT NULL DEFAULT FALSE, + repair_history BOOLEAN NOT NULL DEFAULT FALSE, + slug VARCHAR NOT NULL, + last_seen_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + ) + """ + ) + + op.execute( + """ + CREATE TABLE IF NOT EXISTS public.images ( + id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, + fullres_image VARCHAR NOT NULL, + preview_image VARCHAR NOT NULL, + order_index INTEGER NOT NULL, + car_id INTEGER NOT NULL REFERENCES public.cars(id) ON DELETE CASCADE + ) + """ + ) + + op.execute( + """ + CREATE TABLE IF NOT EXISTS public.sync_runs ( + id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, + started_at TIMESTAMPTZ NOT NULL, + finished_at TIMESTAMPTZ NULL, + status TEXT NOT NULL, + lane TEXT NOT NULL, + ids_fetched INTEGER NOT NULL DEFAULT 0, + cars_upserted INTEGER NOT NULL DEFAULT 0, + cars_failed INTEGER NOT NULL DEFAULT 0, + images_upserted INTEGER NOT NULL DEFAULT 0, + error_summary TEXT NULL + ) + """ + ) + + op.execute("CREATE INDEX IF NOT EXISTS ix_images_car_id ON public.images (car_id)") + op.execute("CREATE INDEX IF NOT EXISTS ix_sync_runs_started_at ON public.sync_runs (started_at)") + op.execute("CREATE INDEX IF NOT EXISTS ix_sync_runs_status ON public.sync_runs (status)") + + +def downgrade() -> None: + # Compatibility-only migration for shared production databases. + pass diff --git a/alembic/versions/0002_extend_origin_enum_for_iaai.py b/alembic/versions/0002_extend_origin_enum_for_iaai.py new file mode 100644 index 0000000..11e9501 --- /dev/null +++ b/alembic/versions/0002_extend_origin_enum_for_iaai.py @@ -0,0 +1,86 @@ +"""extend origin enum for iaai + +Revision ID: 0002_extend_origin_enum_for_iaai +Revises: 0001_ensure_sync_schema +Create Date: 2026-04-08 00:00:00 +""" + +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = "0002_extend_origin_enum_for_iaai" +down_revision: Union[str, Sequence[str], None] = "0001_ensure_sync_schema" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def _enum_type_exists(type_name: str) -> bool: + bind = op.get_bind() + if bind.dialect.name != "postgresql": + return False + exists = bind.execute( + sa.text( + """ + SELECT 1 + FROM pg_type t + JOIN pg_namespace n ON n.oid = t.typnamespace + WHERE n.nspname = 'public' + AND t.typname = :type_name + AND t.typtype = 'e' + LIMIT 1 + """ + ), + {"type_name": type_name}, + ).scalar_one_or_none() + return bool(exists) + + +def _enum_type_for_column(table_name: str, column_name: str) -> str | None: + bind = op.get_bind() + if bind.dialect.name != "postgresql": + return None + return bind.execute( + sa.text( + """ + SELECT t.typname + FROM pg_attribute a + JOIN pg_class c ON c.oid = a.attrelid + JOIN pg_namespace n ON n.oid = c.relnamespace + JOIN pg_type t ON t.oid = a.atttypid + WHERE n.nspname = 'public' + AND c.relname = :table_name + AND a.attname = :column_name + AND a.attnum > 0 + AND NOT a.attisdropped + AND t.typtype = 'e' + LIMIT 1 + """ + ), + {"table_name": table_name, "column_name": column_name}, + ).scalar_one_or_none() + + +def _add_enum_value_if_missing(type_name: str | None, value: str) -> None: + if not type_name: + return + if not _enum_type_exists(type_name): + return + bind = op.get_bind() + if bind.dialect.name != "postgresql": + return + escaped = value.replace("'", "''") + op.execute(f"ALTER TYPE public.{type_name} ADD VALUE IF NOT EXISTS '{escaped}'") + + +def upgrade() -> None: + origin_type = _enum_type_for_column("cars", "origin") or "originenum" + _add_enum_value_if_missing(origin_type, "IAAI") + + +def downgrade() -> None: + # Enum value removal is intentionally skipped for compatibility. + pass diff --git a/alembic/versions/0003_create_iaai_source_tables.py b/alembic/versions/0003_create_iaai_source_tables.py new file mode 100644 index 0000000..25bca25 --- /dev/null +++ b/alembic/versions/0003_create_iaai_source_tables.py @@ -0,0 +1,100 @@ +"""create iaai-specific storage tables + +Revision ID: 0003_create_iaai_source_tables +Revises: 0002_extend_origin_enum_for_iaai +Create Date: 2026-04-09 00:00:00 +""" + +from typing import Sequence, Union + +from alembic import op + + +# revision identifiers, used by Alembic. +revision: str = "0003_create_iaai_source_tables" +down_revision: Union[str, Sequence[str], None] = "0002_extend_origin_enum_for_iaai" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.execute( + """ + CREATE TABLE IF NOT EXISTS public.iaai_cars ( + id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, + parser_id VARCHAR(50) NOT NULL UNIQUE, + brand VARCHAR(50) NOT NULL, + model VARCHAR(50) NOT NULL, + year INTEGER NULL, + price BIGINT NULL, + currency public.currencyenum NOT NULL DEFAULT 'USD', + mileage INTEGER NOT NULL DEFAULT 0, + country public.countryenum NOT NULL DEFAULT 'NA', + is_sold BOOLEAN NOT NULL DEFAULT FALSE, + color VARCHAR NOT NULL DEFAULT 'other', + drive public.driveenum NULL, + gearbox public.gearboxenum NULL, + steering_wheel public.steeringwheelenum NULL, + body_type public.bodytypeenum NOT NULL DEFAULT 'OTHER', + engine_volume INTEGER NULL, + selling_type public.sellingtypeenum NOT NULL DEFAULT 'NA', + one_owner BOOLEAN NOT NULL DEFAULT FALSE, + new_car BOOLEAN NOT NULL DEFAULT FALSE, + is_hidden BOOLEAN NOT NULL DEFAULT FALSE, + origin public.originenum NOT NULL DEFAULT 'NA', + origin_url VARCHAR NOT NULL, + origin_id VARCHAR NOT NULL UNIQUE, + is_damaged BOOLEAN NOT NULL DEFAULT FALSE, + evaluation VARCHAR NULL, + non_smoking BOOLEAN NOT NULL DEFAULT TRUE, + rental BOOLEAN NOT NULL DEFAULT FALSE, + repair_history BOOLEAN NOT NULL DEFAULT FALSE, + slug VARCHAR NOT NULL, + last_seen_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + ) + """ + ) + + op.execute( + """ + CREATE TABLE IF NOT EXISTS public.iaai_images ( + id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, + fullres_image VARCHAR NOT NULL, + preview_image VARCHAR NOT NULL, + order_index INTEGER NOT NULL, + car_id INTEGER NOT NULL REFERENCES public.iaai_cars(id) ON DELETE CASCADE + ) + """ + ) + + op.execute( + """ + CREATE TABLE IF NOT EXISTS public.iaai_sync_runs ( + id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, + started_at TIMESTAMPTZ NOT NULL, + finished_at TIMESTAMPTZ NULL, + status TEXT NOT NULL, + lane TEXT NOT NULL, + ids_fetched INTEGER NOT NULL DEFAULT 0, + cars_upserted INTEGER NOT NULL DEFAULT 0, + cars_failed INTEGER NOT NULL DEFAULT 0, + images_upserted INTEGER NOT NULL DEFAULT 0, + error_summary TEXT NULL + ) + """ + ) + + op.execute("CREATE INDEX IF NOT EXISTS ix_iaai_images_car_id ON public.iaai_images (car_id)") + op.execute("CREATE INDEX IF NOT EXISTS ix_iaai_images_car_order ON public.iaai_images (car_id, order_index)") + op.execute("CREATE INDEX IF NOT EXISTS ix_iaai_cars_last_seen_id ON public.iaai_cars (last_seen_at DESC, id DESC)") + op.execute( + "CREATE INDEX IF NOT EXISTS ix_iaai_cars_active_feed ON public.iaai_cars " + "(last_seen_at DESC, id DESC) WHERE is_sold = FALSE AND is_hidden = FALSE" + ) + op.execute("CREATE INDEX IF NOT EXISTS ix_iaai_sync_runs_started_at ON public.iaai_sync_runs (started_at)") + op.execute("CREATE INDEX IF NOT EXISTS ix_iaai_sync_runs_status ON public.iaai_sync_runs (status)") + + +def downgrade() -> None: + # Compatibility-only migration for shared production databases. + pass diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..0311d92 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,105 @@ +x-app-env: &app-env + POSTGRES_HOST: ${POSTGRES_HOST:-postgres} + POSTGRES_PORT: ${POSTGRES_PORT:-5432} + POSTGRES_USER: ${POSTGRES_USER:-postgres} + POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:-postgres} + POSTGRES_DB: ${POSTGRES_DB:-iaai} + DATABASE_URL: ${IAAI_DATABASE_URL:-postgresql+psycopg://${POSTGRES_USER:-postgres}:${POSTGRES_PASSWORD:-postgres}@${POSTGRES_HOST:-postgres}:${POSTGRES_PORT:-5432}/${POSTGRES_DB:-iaai}} + CELERY_BROKER_URL: redis://redis:6379/0 + CELERY_RESULT_BACKEND: ${CELERY_RESULT_BACKEND:-redis://redis:6379/1} + IAAI_BASE_URL: ${IAAI_BASE_URL:-https://www.iaai.com} + IAAI_LISTING_START_URL: ${IAAI_LISTING_START_URL:-} + IAAI_SESSION_COOKIES: ${IAAI_SESSION_COOKIES:-} + IAAI_STORAGE_STATE_PATH: /app/iaai_storage_state.json + IAAI_LOGIN: ${IAAI_LOGIN:-} + IAAI_PASSWORD: ${IAAI_PASSWORD:-} + IAAI_HTTP_TIMEOUT: ${IAAI_HTTP_TIMEOUT:-30} + IAAI_HTTP_RETRIES: ${IAAI_HTTP_RETRIES:-2} + IAAI_HTTP_RETRY_BACKOFF_MS: ${IAAI_HTTP_RETRY_BACKOFF_MS:-700} + IAAI_FETCH_CONCURRENCY: ${IAAI_FETCH_CONCURRENCY:-8} + IAAI_DB_COMMIT_BATCH_SIZE: ${IAAI_DB_COMMIT_BATCH_SIZE:-20} + IAAI_SESSION_KEEPALIVE_ENABLED: ${IAAI_SESSION_KEEPALIVE_ENABLED:-true} + IAAI_SESSION_KEEPALIVE_INTERVAL_MINUTES: ${IAAI_SESSION_KEEPALIVE_INTERVAL_MINUTES:-15} + IAAI_SCHEMA_BOOTSTRAP_ENABLED: ${IAAI_SCHEMA_BOOTSTRAP_ENABLED:-true} + SYNC_RUNTIME_CONFIG_FILE: /app/sync_runtime_config.json + +services: + postgres: + image: postgres:latest + environment: + POSTGRES_DB: ${POSTGRES_DB:-iaai} + POSTGRES_USER: ${POSTGRES_USER:-postgres} + POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:-postgres} + PGDATA: /var/lib/postgresql/data/pgdata + restart: unless-stopped + ports: + - "127.0.0.1:5432:5432" + volumes: + - postgres_data:/var/lib/postgresql/data + healthcheck: + test: ["CMD-SHELL", "pg_isready -U ${POSTGRES_USER:-postgres} -d ${POSTGRES_DB:-iaai}"] + interval: 10s + timeout: 5s + retries: 5 + + redis: + image: redis:7-alpine + restart: unless-stopped + ports: + - "6379" + healthcheck: + test: ["CMD", "redis-cli", "ping"] + interval: 10s + timeout: 5s + retries: 5 + + migrate: + build: . + environment: *app-env + command: ["alembic", "upgrade", "head"] + volumes: + - ./sync_runtime_config.json:/app/sync_runtime_config.json:ro + - ./iaai_storage_state.json:/app/iaai_storage_state.json + depends_on: + postgres: + condition: service_healthy + + worker: + build: . + environment: *app-env + command: + [ + "celery", + "-A", + "iaai_sync_service.celery_app:app", + "worker", + "--queues=iaai_sync", + "--concurrency=1", + "--loglevel=INFO" + ] + volumes: + - ./sync_runtime_config.json:/app/sync_runtime_config.json:ro + - ./iaai_storage_state.json:/app/iaai_storage_state.json + depends_on: + migrate: + condition: service_completed_successfully + redis: + condition: service_healthy + restart: unless-stopped + + beat: + build: . + environment: *app-env + command: ["celery", "-A", "iaai_sync_service.celery_app:app", "beat", "--loglevel=INFO"] + volumes: + - ./sync_runtime_config.json:/app/sync_runtime_config.json:ro + - ./iaai_storage_state.json:/app/iaai_storage_state.json + depends_on: + migrate: + condition: service_completed_successfully + redis: + condition: service_healthy + restart: unless-stopped + +volumes: + postgres_data: diff --git a/iaai_sync_service/__init__.py b/iaai_sync_service/__init__.py new file mode 100644 index 0000000..5430203 --- /dev/null +++ b/iaai_sync_service/__init__.py @@ -0,0 +1 @@ +"""IAAI Cars sync service package.""" diff --git a/iaai_sync_service/__main__.py b/iaai_sync_service/__main__.py new file mode 100644 index 0000000..5036958 --- /dev/null +++ b/iaai_sync_service/__main__.py @@ -0,0 +1,4 @@ +from iaai_sync_service.cli import main + +if __name__ == "__main__": + main() diff --git a/iaai_sync_service/celery_app.py b/iaai_sync_service/celery_app.py new file mode 100644 index 0000000..01d17cc --- /dev/null +++ b/iaai_sync_service/celery_app.py @@ -0,0 +1,38 @@ +from __future__ import annotations + +from celery import Celery +from celery.schedules import crontab + +from iaai_sync_service.settings import get_settings + +settings = get_settings() + +app = Celery("iaai_sync_service") +app.conf.broker_url = settings.celery_broker_url +app.conf.result_backend = settings.celery_result_backend +app.conf.task_default_queue = "iaai_sync" +app.conf.timezone = "UTC" +app.conf.enable_utc = True +app.conf.broker_connection_retry = True +app.conf.broker_connection_retry_on_startup = True +app.conf.broker_connection_max_retries = None + +def build_beat_schedule() -> dict[str, dict]: + schedule: dict[str, dict] = { + "sync-iaai-cars-every-hour": { + "task": "iaai.sync_cars_feed", + "schedule": crontab(minute=0), + "options": {"queue": "iaai_sync"}, + } + } + if settings.session_keepalive_enabled: + schedule["keep-iaai-session-alive"] = { + "task": "iaai.keep_session_alive", + "schedule": crontab(minute=f"*/{settings.session_keepalive_interval_minutes}"), + "options": {"queue": "iaai_sync"}, + } + return schedule + + +app.conf.beat_schedule = build_beat_schedule() +app.conf.imports = ("iaai_sync_service.tasks",) diff --git a/iaai_sync_service/cli.py b/iaai_sync_service/cli.py new file mode 100644 index 0000000..5103159 --- /dev/null +++ b/iaai_sync_service/cli.py @@ -0,0 +1,14 @@ +from __future__ import annotations + +import json + +from iaai_sync_service.tasks import execute_sync_job + + +def main() -> None: + result = execute_sync_job() + print(json.dumps(result, ensure_ascii=False, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/iaai_sync_service/client.py b/iaai_sync_service/client.py new file mode 100644 index 0000000..8480a2d --- /dev/null +++ b/iaai_sync_service/client.py @@ -0,0 +1,950 @@ +from __future__ import annotations + +import html +import json +import logging +import math +import re +import threading +import time +from collections.abc import Iterator +from dataclasses import dataclass +from pathlib import Path +from typing import Any +from urllib.parse import quote, urljoin + +import requests +from requests.adapters import HTTPAdapter + +from iaai_sync_service.runtime_config import RuntimeFilters +from iaai_sync_service.settings import Settings + +logger = logging.getLogger(__name__) + +TRANSIENT_HTTP_CODES = {408, 425, 429, 500, 502, 503, 504} +CHALLENGE_MARKERS = ( + "_incapsula_resource", + "incapsula", + "incident id", + "request unsuccessful", + "access denied", +) +COOKIE_ACCEPT_SELECTORS = ( + "button:has-text('Accept All')", + "button:has-text('Accept all')", + "button:has-text('I Agree')", + "button:has-text('Agree')", + "button:has-text('Only necessary')", + "button:has-text('Только необходимые')", + "button:has-text('Принять все')", + "[id*='accept']", + "[class*='accept']", +) +LISTING_MARKER = 'id="GBPSearchQuery"' +DETAIL_MARKER = 'id="ProductDetailsVM"' +RESIZER_URL = "https://vis.iaai.com/resizer" +DEFAULT_USER_AGENT = ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/124.0.0.0 Safari/537.36" +) +MAX_CHALLENGE_REFRESH_ATTEMPTS = 3 +PLAYWRIGHT_REFRESH_POLLS = 8 + + +@dataclass(frozen=True) +class ListingVehicle: + inventory_id: str + tenant: str | None + auction_id: str | None + auction_date: str | None + inventory_status: str | None + currency: str | None + timed_auction_closed: bool + timed_auction_indicator: bool + prebid_indicator: bool + buynow_indicator: bool + + +@dataclass(frozen=True) +class ListingPage: + vehicles: list[ListingVehicle] + result_count: int + page_size: int + current_page: int + gbp_search_query: dict[str, Any] + + +class HybridSessionAuth: + def __init__(self, settings: Settings) -> None: + self._settings = settings + self._thread_local = threading.local() + self._lock = threading.Lock() + self._refresh_lock = threading.Lock() + self._bootstrap_cookies_loaded = False + self._anonymous_bootstrap_attempted = False + self._refresh_generation = 0 + self._latest_refresh_cookies: list[dict[str, Any]] = [] + + def request( + self, + method: str, + url: str, + *, + timeout: int, + retries: int, + retry_backoff_ms: int, + headers: dict[str, str] | None = None, + data: Any | None = None, + json_body: Any | None = None, + expected_marker: str | None = None, + ) -> requests.Response: + session = self._get_session() + self._ensure_anonymous_session_bootstrap(session=session) + self._sync_session_with_latest_refresh(session) + last_error: Exception | None = None + refresh_attempts = 0 + attempt = 0 + while attempt <= retries: + try: + response = session.request( + method=method, + url=url, + headers=headers, + data=data, + json=json_body, + timeout=timeout, + ) + except requests.RequestException as exc: + last_error = exc + if attempt >= retries: + break + self._sleep_backoff(retry_backoff_ms, attempt) + attempt += 1 + continue + + if response.status_code in TRANSIENT_HTTP_CODES and attempt < retries: + response.close() + self._sleep_backoff(retry_backoff_ms, attempt) + attempt += 1 + continue + + if is_challenge_response( + status_code=response.status_code, + body_text=response.text, + expected_marker=expected_marker, + ): + response.close() + if refresh_attempts >= MAX_CHALLENGE_REFRESH_ATTEMPTS: + raise RuntimeError( + "IAAI challenge persisted after " + f"{refresh_attempts} Playwright refresh attempts for url={url}" + ) + refresh_attempts += 1 + logger.warning( + "Challenge detected for url=%s status=%s marker=%s refresh_attempt=%s/%s", + url, + response.status_code, + expected_marker, + refresh_attempts, + MAX_CHALLENGE_REFRESH_ATTEMPTS, + ) + self._refresh_session_via_playwright(expected_marker=LISTING_MARKER, session=session) + if refresh_attempts > 1: + self._sleep_backoff(retry_backoff_ms, refresh_attempts - 1) + continue + + return response + + if last_error is not None: + raise RuntimeError(f"Request failed url={url}: {last_error}") from last_error + raise RuntimeError(f"Request failed url={url} after retries") + + def persist_storage_state(self) -> None: + session = self._get_session() + with self._lock: + self._save_storage_state(session) + + def _get_session(self) -> requests.Session: + session = getattr(self._thread_local, "session", None) + if session is None: + session = requests.Session() + adapter = HTTPAdapter(pool_connections=20, pool_maxsize=20) + session.mount("http://", adapter) + session.mount("https://", adapter) + session.headers.update( + { + "user-agent": DEFAULT_USER_AGENT, + "accept-language": "en-US,en;q=0.9", + "cache-control": "no-cache", + "pragma": "no-cache", + } + ) + with self._lock: + self._bootstrap_session_cookies(session) + self._thread_local.session = session + self._thread_local.session_generation = 0 + self._sync_session_with_latest_refresh(session) + return session + + def _sync_session_with_latest_refresh(self, session: requests.Session) -> None: + with self._lock: + latest_generation = self._refresh_generation + session_generation = getattr(self._thread_local, "session_generation", 0) + if latest_generation <= session_generation or not self._latest_refresh_cookies: + return + cookies = list(self._latest_refresh_cookies) + self._apply_cookies_to_session(session, cookies) + self._thread_local.session_generation = latest_generation + + def _ensure_anonymous_session_bootstrap(self, *, session: requests.Session) -> None: + if self._anonymous_bootstrap_attempted: + return + if self._session_has_iaai_cookies(session): + self._anonymous_bootstrap_attempted = True + return + with self._lock: + if self._anonymous_bootstrap_attempted: + return + self._anonymous_bootstrap_attempted = True + logger.info("No IAAI cookies preloaded. Attempting anonymous session bootstrap via Playwright.") + try: + self._refresh_session_via_playwright(expected_marker=LISTING_MARKER, session=session) + except Exception as exc: # noqa: BLE001 + logger.warning( + "Anonymous session bootstrap via Playwright failed; continuing with direct HTTP flow: %s", + exc, + ) + + @staticmethod + def _session_has_iaai_cookies(session: requests.Session) -> bool: + cookies = getattr(session, "cookies", None) + if cookies is None: + return False + for item in cookies: + domain = "" + if isinstance(item, dict): + domain = parse_text(item.get("domain")) or "" + else: + domain = str(getattr(item, "domain", "") or "") + if not domain: + return True + if "iaai.com" in domain.lower(): + return True + return False + + def _bootstrap_session_cookies(self, session: requests.Session) -> None: + if self._bootstrap_cookies_loaded: + return + self._load_storage_state_cookies(session) + self._load_env_cookies(session) + self._bootstrap_cookies_loaded = True + + def _load_storage_state_cookies(self, session: requests.Session) -> None: + path = self._settings.iaai_storage_state_path + if not path.exists(): + return + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except Exception as exc: # noqa: BLE001 + logger.warning("Failed to read storage state file '%s': %s", path, exc) + return + cookies = payload.get("cookies") if isinstance(payload, dict) else None + if not isinstance(cookies, list): + return + applied = 0 + for item in cookies: + if not isinstance(item, dict): + continue + name = parse_text(item.get("name")) + value = parse_text(item.get("value")) + if not name or value is None: + continue + domain = parse_text(item.get("domain")) or ".iaai.com" + cookie_path = parse_text(item.get("path")) or "/" + expires_raw = item.get("expires") + expires = parse_int(expires_raw) + session.cookies.set(name, value, domain=domain, path=cookie_path, expires=expires) + applied += 1 + if applied: + logger.info("Loaded %s cookies from storage state", applied) + + def _load_env_cookies(self, session: requests.Session) -> None: + raw = self._settings.iaai_session_cookies + if not raw: + return + + try: + payload = json.loads(raw) + except Exception: # noqa: BLE001 + payload = None + + applied = 0 + if isinstance(payload, list): + for item in payload: + if not isinstance(item, dict): + continue + name = parse_text(item.get("name")) + value = parse_text(item.get("value")) + if not name or value is None: + continue + domain = parse_text(item.get("domain")) or ".iaai.com" + cookie_path = parse_text(item.get("path")) or "/" + expires = parse_int(item.get("expires")) + session.cookies.set(name, value, domain=domain, path=cookie_path, expires=expires) + applied += 1 + elif isinstance(payload, dict): + for name, value in payload.items(): + if not isinstance(name, str) or not isinstance(value, str): + continue + session.cookies.set(name.strip(), value.strip(), domain=".iaai.com", path="/") + applied += 1 + else: + for part in raw.split(";"): + if "=" not in part: + continue + name, value = part.split("=", 1) + name = name.strip() + value = value.strip() + if not name: + continue + session.cookies.set(name, value, domain=".iaai.com", path="/") + applied += 1 + + if applied: + logger.info("Loaded %s cookies from IAAI_SESSION_COOKIES", applied) + + def _refresh_session_via_playwright( + self, + *, + expected_marker: str | None = None, + session: requests.Session | None = None, + ) -> None: + target_session = session or self._get_session() + with self._lock: + baseline_generation = self._refresh_generation + + with self._refresh_lock: + with self._lock: + if self._refresh_generation > baseline_generation and self._latest_refresh_cookies: + self._apply_cookies_to_session(target_session, self._latest_refresh_cookies) + self._thread_local.session_generation = self._refresh_generation + return + + logger.warning("IAAI session challenge detected. Refreshing session via Playwright.") + cookies = self._fetch_cookies_via_playwright(expected_marker=expected_marker) + self._apply_cookies_to_session(target_session, cookies) + + with self._lock: + self._refresh_generation += 1 + self._latest_refresh_cookies = list(cookies) + self._anonymous_bootstrap_attempted = True + refreshed_generation = self._refresh_generation + self._thread_local.session_generation = refreshed_generation + self._save_storage_state(target_session) + + def _fetch_cookies_via_playwright(self, *, expected_marker: str | None = None) -> list[dict[str, Any]]: + try: + from playwright.sync_api import TimeoutError as PlaywrightTimeoutError + from playwright.sync_api import sync_playwright + except Exception as exc: # noqa: BLE001 + raise RuntimeError( + "Playwright is required for challenge fallback. Install dependency and browsers." + ) from exc + + with sync_playwright() as playwright: + browser = playwright.chromium.launch(headless=True) + try: + context = browser.new_context( + locale="en-US", + viewport={"width": 1366, "height": 768}, + user_agent=DEFAULT_USER_AGENT, + ) + page = context.new_page() + home_target = self._settings.iaai_base_url + "/" + target = urljoin(self._settings.iaai_base_url + "/", "Vehiclelisting/Cars") + timeout_ms = max(30_000, self._settings.http_timeout * 1000) + page.goto(home_target, wait_until="domcontentloaded", timeout=timeout_ms) + self._accept_cookie_banner(page) + page.goto(target, wait_until="domcontentloaded", timeout=timeout_ms) + self._accept_cookie_banner(page) + self._try_login(page) + self._wait_until_non_challenge( + page=page, + target=target, + timeout_ms=timeout_ms, + expected_marker=expected_marker or LISTING_MARKER, + ) + state = context.storage_state() + except PlaywrightTimeoutError as exc: + raise RuntimeError(f"Playwright refresh timed out: {exc}") from exc + finally: + browser.close() + + cookies = state.get("cookies") if isinstance(state, dict) else None + if not isinstance(cookies, list) or not cookies: + raise RuntimeError("Playwright refresh did not return cookies") + return [cookie for cookie in cookies if isinstance(cookie, dict)] + + @staticmethod + def _apply_cookies_to_session(session: requests.Session, cookies: list[dict[str, Any]]) -> None: + session.cookies.clear() + for cookie in cookies: + name = parse_text(cookie.get("name")) + value = parse_text(cookie.get("value")) + if not name or value is None: + continue + domain = parse_text(cookie.get("domain")) or ".iaai.com" + cookie_path = parse_text(cookie.get("path")) or "/" + expires = parse_int(cookie.get("expires")) + session.cookies.set(name, value, domain=domain, path=cookie_path, expires=expires) + + @staticmethod + def _wait_until_non_challenge( + *, + page: Any, + target: str, + timeout_ms: int, + expected_marker: str | None, + ) -> None: + # Give Incapsula redirect/challenge flow time to settle and verify page really opened. + poll_ms = max(1000, min(5000, timeout_ms // PLAYWRIGHT_REFRESH_POLLS)) + navigation_error_count = 0 + for poll_index in range(PLAYWRIGHT_REFRESH_POLLS): + try: + page.wait_for_load_state("domcontentloaded", timeout=poll_ms) + except Exception: # noqa: BLE001 + # The challenge page frequently redirects; continue polling. + pass + page.wait_for_timeout(poll_ms) + + body: str | None = None + for _ in range(3): + try: + body = page.content() + break + except Exception as exc: # noqa: BLE001 + message = str(exc).lower() + if "page.content" not in message or "navigating and changing the content" not in message: + raise + navigation_error_count += 1 + page.wait_for_timeout(max(200, poll_ms // 4)) + continue + + if body is not None and not is_challenge_response( + status_code=200, + body_text=body, + expected_marker=expected_marker, + ): + return + try: + page.goto(target, wait_until="domcontentloaded", timeout=timeout_ms) + except Exception as exc: # noqa: BLE001 + logger.debug( + "Playwright challenge retry navigation failed poll=%s/%s: %s", + poll_index + 1, + PLAYWRIGHT_REFRESH_POLLS, + exc, + ) + + raise RuntimeError( + "Playwright refresh completed but challenge page is still active " + f"(navigation_content_errors={navigation_error_count})" + ) + + def _try_login(self, page: Any) -> None: + if not self._settings.iaai_login or not self._settings.iaai_password: + return + + try: + login_selectors = ( + "input[type='email']", + "input[name*='email' i]", + "input[id*='email' i]", + "input[name*='username' i]", + ) + password_selector = "input[type='password']" + submit_selector = "button[type='submit'],input[type='submit']" + + email_locator = None + for selector in login_selectors: + locator = page.locator(selector) + if locator.count() > 0: + email_locator = locator.first + break + if email_locator is None: + return + + email_locator.fill(self._settings.iaai_login) + password_locator = page.locator(password_selector) + if password_locator.count() == 0: + return + password_locator.first.fill(self._settings.iaai_password) + submit_locator = page.locator(submit_selector) + if submit_locator.count() > 0: + submit_locator.first.click() + page.wait_for_load_state("domcontentloaded", timeout=30_000) + except Exception as exc: # noqa: BLE001 + logger.warning("Playwright login step failed: %s", exc) + + def _accept_cookie_banner(self, page: Any) -> None: + for selector in COOKIE_ACCEPT_SELECTORS: + try: + locator = page.locator(selector).first + if locator.count() == 0: + continue + if not locator.is_visible(timeout=500): + continue + locator.click(timeout=2_000) + page.wait_for_timeout(250) + return + except Exception: # noqa: BLE001 + continue + + def _save_storage_state(self, session: requests.Session) -> None: + path = self._settings.iaai_storage_state_path + cookies: list[dict[str, Any]] = [] + for cookie in session.cookies: + cookie_payload: dict[str, Any] = { + "name": cookie.name, + "value": cookie.value, + "domain": cookie.domain or ".iaai.com", + "path": cookie.path or "/", + "httpOnly": False, + "secure": bool(cookie.secure), + "sameSite": "Lax", + } + if cookie.expires is not None: + cookie_payload["expires"] = int(cookie.expires) + cookies.append(cookie_payload) + + payload = {"cookies": cookies, "origins": []} + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") + + @staticmethod + def _sleep_backoff(retry_backoff_ms: int, attempt: int) -> None: + if retry_backoff_ms <= 0: + return + delay = retry_backoff_ms * (2**attempt) / 1000 + time.sleep(delay) + + +class IAAIClient: + def __init__(self, settings: Settings) -> None: + self._settings = settings + self._auth = HybridSessionAuth(settings) + + def persist_session_state(self) -> None: + self._auth.persist_storage_state() + + def keep_session_alive(self) -> dict[str, Any]: + started = time.perf_counter() + scope = resolve_listing_scope_paths( + listing_start_url=self._settings.iaai_listing_start_url, + brands=set(), + )[0] + page_html = self._fetch_listing_first_page(scope) + marker_present = LISTING_MARKER in page_html + result_count = parse_int(parse_hidden_input_value(page_html, "ResultCount")) + self.persist_session_state() + return { + "scope": scope, + "marker_present": marker_present, + "result_count": result_count, + "elapsed_ms": int((time.perf_counter() - started) * 1000), + } + + def iter_listing_vehicles(self, filters: RuntimeFilters) -> Iterator[ListingVehicle]: + seen_inventory_ids: set[str] = set() + if self._settings.iaai_listing_start_url.strip() and filters.brands: + logger.info( + "Using explicit listing start URL; runtime brand scopes in listing step are ignored." + ) + scope_paths = resolve_listing_scope_paths( + listing_start_url=self._settings.iaai_listing_start_url, + brands=filters.brands, + ) + listing_started_at = time.perf_counter() + + for scope_index, scope_path in enumerate(scope_paths, start=1): + logger.info( + "IAAI listing scope start scope=%s/%s path=%s", + scope_index, + len(scope_paths), + scope_path, + ) + first_page_html = self._fetch_listing_first_page(scope_path) + first_page = parse_listing_page(first_page_html) + for vehicle in first_page.vehicles: + if vehicle.inventory_id in seen_inventory_ids: + continue + seen_inventory_ids.add(vehicle.inventory_id) + yield vehicle + + page_size = max(1, first_page.page_size) + total_pages = max(1, math.ceil(max(first_page.result_count, len(first_page.vehicles)) / page_size)) + gbp_search_query = first_page.gbp_search_query + logger.info( + "IAAI listing scope loaded scope=%s/%s path=%s result_count=%s page_size=%s total_pages=%s unique_ids=%s elapsed=%.1fs", + scope_index, + len(scope_paths), + scope_path, + first_page.result_count, + page_size, + total_pages, + len(seen_inventory_ids), + time.perf_counter() - listing_started_at, + ) + + for page_number in range(2, total_pages + 1): + page_html = self._fetch_listing_page(scope_path, gbp_search_query, page_number, page_size) + parsed_page = parse_listing_page(page_html) + gbp_search_query = parsed_page.gbp_search_query + for vehicle in parsed_page.vehicles: + if vehicle.inventory_id in seen_inventory_ids: + continue + seen_inventory_ids.add(vehicle.inventory_id) + yield vehicle + if page_number % 10 == 0 or page_number == total_pages: + logger.info( + "IAAI listing page progress scope=%s/%s path=%s page=%s/%s unique_ids=%s elapsed=%.1fs", + scope_index, + len(scope_paths), + scope_path, + page_number, + total_pages, + len(seen_inventory_ids), + time.perf_counter() - listing_started_at, + ) + + def fetch_vehicle_detail_payload(self, inventory_id: str) -> dict[str, Any]: + escaped_id = quote(inventory_id, safe="~") + url = urljoin(self._settings.iaai_base_url + "/", f"VehicleDetail/{escaped_id}") + response = self._auth.request( + "GET", + url, + timeout=self._settings.http_timeout, + retries=self._settings.http_retries, + retry_backoff_ms=self._settings.http_retry_backoff_ms, + headers={"accept": "text/html,application/xhtml+xml"}, + expected_marker=DETAIL_MARKER, + ) + with response: + if response.status_code >= 400: + raise RuntimeError(f"Vehicle detail request failed id={inventory_id} status={response.status_code}") + return parse_product_details_vm(response.text) + + def _fetch_listing_first_page(self, scope_path: str) -> str: + if scope_path.lower().startswith(("http://", "https://")): + url = scope_path + else: + url = urljoin(self._settings.iaai_base_url + "/", scope_path.lstrip("/")) + response = self._auth.request( + "GET", + url, + timeout=self._settings.http_timeout, + retries=self._settings.http_retries, + retry_backoff_ms=self._settings.http_retry_backoff_ms, + headers={"accept": "text/html,application/xhtml+xml"}, + expected_marker=LISTING_MARKER, + ) + with response: + if response.status_code >= 400: + raise RuntimeError(f"Listing request failed path={scope_path} status={response.status_code}") + return response.text + + def _fetch_listing_page( + self, + scope_path: str, + gbp_search_query: dict[str, Any], + page_number: int, + page_size: int, + ) -> str: + query_payload = dict(gbp_search_query) + query_payload["CurrentPage"] = page_number + query_payload["PageSize"] = page_size + + search_url = urljoin(self._settings.iaai_base_url + "/", "Search") + common_headers = { + "accept": "text/html,application/xhtml+xml,*/*", + "x-requested-with": "XMLHttpRequest", + } + + attempts: list[tuple[dict[str, str], Any, Any]] = [ + ({**common_headers, "content-type": "application/json"}, None, query_payload), + ({**common_headers, "content-type": "application/json"}, None, {"GBPSearchQuery": query_payload}), + ({**common_headers}, {"GBPSearchQuery": json.dumps(query_payload, separators=(",", ":"))}, None), + ( + {**common_headers, "content-type": "application/json"}, + json.dumps({"GBPSearchQuery": json.dumps(query_payload, separators=(",", ":"))}), + None, + ), + ] + + last_error: Exception | None = None + for headers, data, json_body in attempts: + try: + response = self._auth.request( + "POST", + search_url, + timeout=self._settings.http_timeout, + retries=self._settings.http_retries, + retry_backoff_ms=self._settings.http_retry_backoff_ms, + headers=headers, + data=data, + json_body=json_body, + expected_marker=LISTING_MARKER, + ) + with response: + if response.status_code >= 400: + raise RuntimeError( + f"Listing page request failed status={response.status_code} page={page_number}" + ) + body = response.text + if LISTING_MARKER not in body: + raise RuntimeError("Listing page response does not include GBPSearchQuery") + return body + except Exception as exc: # noqa: BLE001 + last_error = exc + continue + + if last_error is not None: + raise RuntimeError(f"Failed to load listing page={page_number} for {scope_path}: {last_error}") from last_error + raise RuntimeError(f"Failed to load listing page={page_number} for {scope_path}") + + +def build_brand_scope_paths(brands: set[str]) -> list[str]: + if not brands: + return ["/Vehiclelisting/Cars"] + + paths: list[str] = [] + for brand in sorted(brands): + raw = brand.strip() + if not raw: + continue + slug_hyphen = quote(raw.replace(" ", "-"), safe="-") + slug_raw = quote(raw, safe="") + for slug in (slug_hyphen, slug_raw): + path = f"/Vehiclelisting/Cars/{slug}" + if path not in paths: + paths.append(path) + return paths or ["/Vehiclelisting/Cars"] + + +def resolve_listing_scope_paths(*, listing_start_url: str, brands: set[str]) -> list[str]: + explicit_scope = listing_start_url.strip() + if explicit_scope: + return [explicit_scope] + return build_brand_scope_paths(brands) + + +def parse_listing_page(html_text: str) -> ListingPage: + gbp_raw = parse_hidden_input_value(html_text, "GBPSearchQuery") + vehicle_raw = parse_hidden_input_value(html_text, "VehicleDetails") + result_count_raw = parse_hidden_input_value(html_text, "ResultCount") + page_size_raw = parse_hidden_input_value(html_text, "PageSize") + current_page_raw = parse_hidden_input_value(html_text, "CurrentPage") + + if not gbp_raw: + raise RuntimeError("Listing page missing GBPSearchQuery") + if vehicle_raw is None: + raise RuntimeError("Listing page missing VehicleDetails") + + gbp_payload = json.loads(gbp_raw) + if not isinstance(gbp_payload, dict): + raise RuntimeError("GBPSearchQuery payload is not object") + + vehicle_payload = json.loads(vehicle_raw) + if not isinstance(vehicle_payload, list): + raise RuntimeError("VehicleDetails payload is not array") + + vehicles: list[ListingVehicle] = [] + for item in vehicle_payload: + if not isinstance(item, dict): + continue + inventory_id = parse_text(item.get("Id")) + if not inventory_id: + continue + vehicles.append( + ListingVehicle( + inventory_id=inventory_id, + tenant=parse_text(item.get("Tenant")), + auction_id=parse_text(item.get("ActnLnId")), + auction_date=parse_text(item.get("AuctionDate")) or parse_text(item.get("ActnDtTm")), + inventory_status=parse_text(item.get("InventoryStatus")), + currency=parse_text(item.get("Currency")), + timed_auction_closed=parse_bool(item.get("TimedAuctionClosedIndicator")), + timed_auction_indicator=parse_bool(item.get("TimedAuctionIndicator")), + prebid_indicator=parse_bool(item.get("PreBidIndicator")), + buynow_indicator=parse_bool(item.get("BuyNowIndicator")), + ) + ) + + return ListingPage( + vehicles=vehicles, + result_count=parse_int(result_count_raw) or len(vehicles), + page_size=parse_int(page_size_raw) or max(1, len(vehicles)), + current_page=parse_int(current_page_raw) or 1, + gbp_search_query=gbp_payload, + ) + + +def parse_product_details_vm(html_text: str) -> dict[str, Any]: + match = re.search( + r"]*id=\"ProductDetailsVM\"[^>]*>\s*(\{.*?\})\s*", + html_text, + flags=re.DOTALL, + ) + if match is None: + raise RuntimeError("ProductDetailsVM script not found") + + payload = json.loads(match.group(1)) + if not isinstance(payload, dict): + raise RuntimeError("ProductDetailsVM root is not object") + return payload + + +def parse_hidden_input_value(html_text: str, input_id: str) -> str | None: + escaped_id = re.escape(input_id) + patterns = ( + rf"]*\bid=\"{escaped_id}\"[^>]*\bvalue=\"([^\"]*)\"", + rf"]*\bid='{escaped_id}'[^>]*\bvalue='([^']*)'", + ) + for pattern in patterns: + match = re.search(pattern, html_text, flags=re.IGNORECASE) + if match is not None: + return html.unescape(match.group(1)) + return None + + +def build_resizer_images_from_keys(image_keys: list[dict[str, Any]]) -> list[dict[str, str | int]]: + seen_fullres: set[str] = set() + images: list[dict[str, str | int]] = [] + + for index, item in enumerate(image_keys): + if not isinstance(item, dict): + continue + key = parse_text(item.get("k")) + if key is None: + continue + + width = parse_int(item.get("w")) or 1600 + height = parse_int(item.get("h")) or 1200 + if width <= 0: + width = 1600 + if height <= 0: + height = 1200 + + order_index = parse_int(item.get("i")) + if order_index is None: + order_index = parse_int(item.get("in")) + if order_index is None: + order_index = index + + preview_width = min(640, width) + preview_height = max(1, int(round(height * (preview_width / width)))) + + escaped_key = quote(key, safe="~") + fullres = f"{RESIZER_URL}?imageKeys={escaped_key}&width={width}&height={height}" + preview = f"{RESIZER_URL}?imageKeys={escaped_key}&width={preview_width}&height={preview_height}" + + if fullres in seen_fullres: + continue + seen_fullres.add(fullres) + images.append( + { + "order_index": order_index, + "fullres_image": fullres, + "preview_image": preview, + } + ) + + images.sort(key=lambda row: (parse_int(row.get("order_index")) or 0, str(row.get("fullres_image")))) + return images + + +def is_challenge_response(*, status_code: int, body_text: str, expected_marker: str | None = None) -> bool: + if status_code in {401, 403}: + return True + + if _expected_marker_present(body_text=body_text, expected_marker=expected_marker): + # If expected listing/detail marker is present, this is a valid page even if + # Incapsula script references are embedded in the HTML. + return False + + lowered = (body_text or "").lower() + if any(marker in lowered for marker in CHALLENGE_MARKERS): + return True + + if expected_marker and not _expected_marker_present(body_text=body_text, expected_marker=expected_marker): + # Expected hidden marker/script missing from HTML often means anti-bot interstitial. + if " bool: + if not expected_marker: + return False + if expected_marker in body_text: + return True + + # Accept quote variants for marker fragments like id="GBPSearchQuery" / id='GBPSearchQuery'. + if '"' in expected_marker: + single_quoted = expected_marker.replace('"', "'") + if single_quoted in body_text: + return True + if "'" in expected_marker: + double_quoted = expected_marker.replace("'", '"') + if double_quoted in body_text: + return True + + marker_match = re.search(r"id=['\"]([^'\"]+)['\"]", expected_marker) + if marker_match is None: + return False + marker_id = re.escape(marker_match.group(1)) + return bool( + re.search( + rf"id\s*=\s*['\"]{marker_id}['\"]", + body_text, + flags=re.IGNORECASE, + ) + ) + + +def parse_text(value: Any) -> str | None: + if isinstance(value, str): + text = value.strip() + return text if text else None + return None + + +def parse_bool(value: Any) -> bool: + if isinstance(value, bool): + return value + if isinstance(value, str): + normalized = value.strip().lower() + return normalized in {"true", "1", "yes", "on"} + if isinstance(value, (int, float)) and not isinstance(value, bool): + return value != 0 + return False + + +def parse_int(value: Any) -> int | None: + if value is None or isinstance(value, bool): + return None + if isinstance(value, int): + return value + if isinstance(value, float): + return int(round(value)) + if isinstance(value, str): + text = value.strip() + if not text: + return None + normalized = text.replace(",", "").replace(" ", "") + try: + return int(round(float(normalized))) + except ValueError: + return None + return None diff --git a/iaai_sync_service/database.py b/iaai_sync_service/database.py new file mode 100644 index 0000000..85b8f8d --- /dev/null +++ b/iaai_sync_service/database.py @@ -0,0 +1,15 @@ +from __future__ import annotations + +from sqlalchemy import create_engine +from sqlalchemy.orm import Session, sessionmaker + +from iaai_sync_service.settings import get_settings + +_settings = get_settings() + +engine = create_engine(_settings.database_url, pool_pre_ping=True, future=True) +SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, expire_on_commit=False) + + +def get_session() -> Session: + return SessionLocal() diff --git a/iaai_sync_service/models.py b/iaai_sync_service/models.py new file mode 100644 index 0000000..4da0e54 --- /dev/null +++ b/iaai_sync_service/models.py @@ -0,0 +1,221 @@ +from __future__ import annotations + +from datetime import datetime + +from sqlalchemy import ( + BigInteger, + Boolean, + DateTime, + Enum, + ForeignKey, + Integer, + String, + Text, + func, +) +from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship + + +class Base(DeclarativeBase): + pass + + +CAR_TABLE_NAME = "iaai_cars" +IMAGE_TABLE_NAME = "iaai_images" +SYNC_RUN_TABLE_NAME = "iaai_sync_runs" + + +CURRENCY_ENUM_VALUES = ("JPY", "USD", "EUR", "RUB", "KRW", "AED", "GBP", "CAD") +DRIVE_ENUM_VALUES = ("FWD", "RWD", "TWO_WD", "FOUR_WD", "2WD", "4WD", "NA") +GEARBOX_ENUM_VALUES = ("AT", "CVT", "MT", "EV", "NA") +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", + "IAAI", + "copart", + "iaai", + "NA", + "ASNET", + "KABABA", +) +SELLING_TYPE_ENUM_VALUES = ("STOCK", "AUCTION", "TENDER", "stock", "auction", "tender", "NA") + + +class Car(Base): + __tablename__ = CAR_TABLE_NAME + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + parser_id: Mapped[str] = mapped_column(String(50), nullable=False, unique=True) + brand: Mapped[str] = mapped_column(String(50), nullable=False) + model: Mapped[str] = mapped_column(String(50), nullable=False) + year: Mapped[int | None] = mapped_column(Integer, nullable=True) + price: Mapped[int | None] = mapped_column(BigInteger, nullable=True) + currency: Mapped[str] = mapped_column( + Enum( + *CURRENCY_ENUM_VALUES, + name="currencyenum", + native_enum=True, + create_constraint=False, + ), + nullable=False, + default="USD", + ) + mileage: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + country: Mapped[str] = mapped_column( + Enum( + *COUNTRY_ENUM_VALUES, + name="countryenum", + native_enum=True, + create_constraint=False, + ), + nullable=False, + default="NA", + ) + is_sold: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + color: Mapped[str] = mapped_column(String(), nullable=False, default="other") + drive: Mapped[str | None] = mapped_column( + Enum( + *DRIVE_ENUM_VALUES, + name="driveenum", + native_enum=True, + create_constraint=False, + ), + nullable=True, + ) + gearbox: Mapped[str | None] = mapped_column( + Enum( + *GEARBOX_ENUM_VALUES, + name="gearboxenum", + native_enum=True, + create_constraint=False, + ), + nullable=True, + ) + steering_wheel: Mapped[str | None] = mapped_column( + Enum( + *STEERING_WHEEL_ENUM_VALUES, + name="steeringwheelenum", + native_enum=True, + create_constraint=False, + ), + nullable=True, + ) + body_type: Mapped[str] = mapped_column( + Enum( + *BODY_TYPE_ENUM_VALUES, + name="bodytypeenum", + native_enum=True, + create_constraint=False, + ), + nullable=False, + default="OTHER", + ) + engine_volume: Mapped[int | None] = mapped_column(Integer, nullable=True) + selling_type: Mapped[str] = mapped_column( + Enum( + *SELLING_TYPE_ENUM_VALUES, + name="sellingtypeenum", + native_enum=True, + create_constraint=False, + ), + nullable=False, + default="NA", + ) + one_owner: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + new_car: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + is_hidden: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + origin: Mapped[str] = mapped_column( + Enum( + *ORIGIN_ENUM_VALUES, + name="originenum", + native_enum=True, + create_constraint=False, + ), + nullable=False, + default="NA", + ) + origin_url: Mapped[str] = mapped_column(String(), nullable=False) + origin_id: Mapped[str] = mapped_column(String(), nullable=False, unique=True) + is_damaged: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + evaluation: Mapped[str | None] = mapped_column(String(), nullable=True) + non_smoking: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True) + rental: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + repair_history: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + slug: Mapped[str] = mapped_column(String(), nullable=False) + last_seen_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=func.now()) + + images: Mapped[list["Image"]] = relationship( + "Image", + back_populates="car", + cascade="all, delete-orphan", + ) + + +class Image(Base): + __tablename__ = IMAGE_TABLE_NAME + + id: Mapped[int] = mapped_column( + BigInteger().with_variant(Integer, "sqlite"), + primary_key=True, + autoincrement=True, + ) + fullres_image: Mapped[str] = mapped_column(String(), nullable=False) + preview_image: Mapped[str] = mapped_column(String(), nullable=False) + order_index: Mapped[int] = mapped_column(Integer, nullable=False) + car_id: Mapped[int] = mapped_column( + Integer, + ForeignKey(f"{CAR_TABLE_NAME}.id", ondelete="CASCADE"), + nullable=False, + ) + + car: Mapped[Car] = relationship("Car", back_populates="images") + + +class SyncRun(Base): + __tablename__ = SYNC_RUN_TABLE_NAME + + id: Mapped[int] = mapped_column( + BigInteger().with_variant(Integer, "sqlite"), + primary_key=True, + autoincrement=True, + ) + started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=func.now()) + finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + status: Mapped[str] = mapped_column(Text, nullable=False) + lane: Mapped[str] = mapped_column(Text, nullable=False) + ids_fetched: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + cars_upserted: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + cars_failed: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + images_upserted: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + error_summary: Mapped[str | None] = mapped_column(Text, nullable=True) diff --git a/iaai_sync_service/runtime_config.py b/iaai_sync_service/runtime_config.py new file mode 100644 index 0000000..2d5bb69 --- /dev/null +++ b/iaai_sync_service/runtime_config.py @@ -0,0 +1,103 @@ +from __future__ import annotations + +import json +import logging +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + + +@dataclass(frozen=True) +class RuntimeFilters: + brands: set[str] = field(default_factory=set) + models: set[str] = field(default_factory=set) + years: set[int] = field(default_factory=set) + body_types: set[str] = field(default_factory=set) + + def is_enabled(self) -> bool: + return bool(self.brands or self.models or self.years or self.body_types) + + +@dataclass(frozen=True) +class RuntimeConfig: + condition_check_enabled: bool | None = None + filters: RuntimeFilters = field(default_factory=RuntimeFilters) + + +def load_runtime_config(path: Path, logger: logging.Logger | None = None) -> RuntimeConfig: + if not path.exists(): + return RuntimeConfig() + + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except Exception as exc: # noqa: BLE001 + if logger is not None: + logger.warning("Failed to parse runtime config file '%s': %s", path, exc) + return RuntimeConfig() + + if not isinstance(payload, dict): + if logger is not None: + logger.warning("Runtime config file '%s' root must be object.", path) + return RuntimeConfig() + + sync_block = payload.get("sync") + filters_block = payload.get("filters") + if not isinstance(sync_block, dict): + sync_block = {} + if not isinstance(filters_block, dict): + filters_block = {} + + return RuntimeConfig( + condition_check_enabled=_to_optional_bool(sync_block.get("condition_check_enabled")), + filters=RuntimeFilters( + brands=_to_ci_set(filters_block.get("brands")), + models=_to_ci_set(filters_block.get("models")), + years=_to_int_set(filters_block.get("years"), min_value=1900), + body_types=_to_ci_set(filters_block.get("body_types")), + ) + ) + + +def _to_optional_bool(value: Any) -> bool | None: + if value is None: + return None + if isinstance(value, bool): + return value + if isinstance(value, str): + normalized = value.strip().lower() + if normalized in {"true", "1", "yes", "on"}: + return True + if normalized in {"false", "0", "no", "off"}: + return False + return None + + +def _to_ci_set(value: Any) -> set[str]: + if not isinstance(value, list): + return set() + result: set[str] = set() + for item in value: + if not isinstance(item, str): + continue + text = item.strip().casefold() + if text: + result.add(text) + return result + + +def _to_int_set(value: Any, *, min_value: int) -> set[int]: + if not isinstance(value, list): + return set() + result: set[int] = set() + for item in value: + if isinstance(item, bool): + continue + if isinstance(item, int): + if item >= min_value: + result.add(item) + continue + if isinstance(item, str) and item.strip().isdigit(): + parsed = int(item.strip()) + if parsed >= min_value: + result.add(parsed) + return result diff --git a/iaai_sync_service/schema_bootstrap.py b/iaai_sync_service/schema_bootstrap.py new file mode 100644 index 0000000..53fe2fd --- /dev/null +++ b/iaai_sync_service/schema_bootstrap.py @@ -0,0 +1,179 @@ +from __future__ import annotations + +from sqlalchemy import text +from sqlalchemy.orm import Session + +from iaai_sync_service.models import ( + BODY_TYPE_ENUM_VALUES, + CAR_TABLE_NAME, + COUNTRY_ENUM_VALUES, + CURRENCY_ENUM_VALUES, + DRIVE_ENUM_VALUES, + GEARBOX_ENUM_VALUES, + IMAGE_TABLE_NAME, + ORIGIN_ENUM_VALUES, + SELLING_TYPE_ENUM_VALUES, + SYNC_RUN_TABLE_NAME, + STEERING_WHEEL_ENUM_VALUES, +) + +SCHEMA_BOOTSTRAP_LOCK_KEY = 901_700_001 + + +def _create_enum_if_missing(session: Session, type_name: str, values: tuple[str, ...]) -> None: + values_sql = ", ".join("'" + value.replace("'", "''") + "'" for value in values) + session.execute( + text( + f""" + DO $$ + BEGIN + IF NOT EXISTS ( + SELECT 1 + FROM pg_type t + JOIN pg_namespace n ON n.oid = t.typnamespace + WHERE n.nspname = 'public' + AND t.typname = '{type_name}' + ) THEN + CREATE TYPE public.{type_name} AS ENUM ({values_sql}); + END IF; + END + $$; + """ + ) + ) + + +def _add_enum_value_if_missing(session: Session, type_name: str, value: str) -> None: + escaped = value.replace("'", "''") + session.execute(text(f"ALTER TYPE public.{type_name} ADD VALUE IF NOT EXISTS '{escaped}'")) + + +def _acquire_schema_bootstrap_lock(session: Session) -> None: + session.execute(text("SELECT pg_advisory_xact_lock(:key)"), {"key": SCHEMA_BOOTSTRAP_LOCK_KEY}) + + +def ensure_schema(session: Session) -> None: + bind = session.get_bind() + if bind is None or bind.dialect.name != "postgresql": + return + + _acquire_schema_bootstrap_lock(session) + + _create_enum_if_missing(session, "currencyenum", CURRENCY_ENUM_VALUES) + _create_enum_if_missing(session, "driveenum", DRIVE_ENUM_VALUES) + _create_enum_if_missing(session, "gearboxenum", GEARBOX_ENUM_VALUES) + _create_enum_if_missing(session, "steeringwheelenum", STEERING_WHEEL_ENUM_VALUES) + _create_enum_if_missing(session, "bodytypeenum", BODY_TYPE_ENUM_VALUES) + _create_enum_if_missing(session, "countryenum", COUNTRY_ENUM_VALUES) + _create_enum_if_missing(session, "originenum", ORIGIN_ENUM_VALUES) + _create_enum_if_missing(session, "sellingtypeenum", SELLING_TYPE_ENUM_VALUES) + + for value in CURRENCY_ENUM_VALUES: + _add_enum_value_if_missing(session, "currencyenum", value) + for value in DRIVE_ENUM_VALUES: + _add_enum_value_if_missing(session, "driveenum", value) + for value in GEARBOX_ENUM_VALUES: + _add_enum_value_if_missing(session, "gearboxenum", value) + for value in STEERING_WHEEL_ENUM_VALUES: + _add_enum_value_if_missing(session, "steeringwheelenum", value) + for value in BODY_TYPE_ENUM_VALUES: + _add_enum_value_if_missing(session, "bodytypeenum", value) + for value in COUNTRY_ENUM_VALUES: + _add_enum_value_if_missing(session, "countryenum", value) + for value in ORIGIN_ENUM_VALUES: + _add_enum_value_if_missing(session, "originenum", value) + for value in SELLING_TYPE_ENUM_VALUES: + _add_enum_value_if_missing(session, "sellingtypeenum", value) + + session.execute( + text( + f""" + CREATE TABLE IF NOT EXISTS public.{CAR_TABLE_NAME} ( + id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, + parser_id VARCHAR(50) NOT NULL UNIQUE, + brand VARCHAR(50) NOT NULL, + model VARCHAR(50) NOT NULL, + year INTEGER NULL, + price BIGINT NULL, + currency public.currencyenum NOT NULL DEFAULT 'USD', + mileage INTEGER NOT NULL DEFAULT 0, + country public.countryenum NOT NULL DEFAULT 'NA', + is_sold BOOLEAN NOT NULL DEFAULT FALSE, + color VARCHAR NOT NULL DEFAULT 'other', + drive public.driveenum NULL, + gearbox public.gearboxenum NULL, + steering_wheel public.steeringwheelenum NULL, + body_type public.bodytypeenum NOT NULL DEFAULT 'OTHER', + engine_volume INTEGER NULL, + selling_type public.sellingtypeenum NOT NULL DEFAULT 'NA', + one_owner BOOLEAN NOT NULL DEFAULT FALSE, + new_car BOOLEAN NOT NULL DEFAULT FALSE, + is_hidden BOOLEAN NOT NULL DEFAULT FALSE, + origin public.originenum NOT NULL DEFAULT 'NA', + origin_url VARCHAR NOT NULL, + origin_id VARCHAR NOT NULL UNIQUE, + is_damaged BOOLEAN NOT NULL DEFAULT FALSE, + evaluation VARCHAR NULL, + non_smoking BOOLEAN NOT NULL DEFAULT TRUE, + rental BOOLEAN NOT NULL DEFAULT FALSE, + repair_history BOOLEAN NOT NULL DEFAULT FALSE, + slug VARCHAR NOT NULL, + last_seen_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + ) + """ + ) + ) + session.execute( + text( + f""" + CREATE TABLE IF NOT EXISTS public.{IMAGE_TABLE_NAME} ( + id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, + fullres_image VARCHAR NOT NULL, + preview_image VARCHAR NOT NULL, + order_index INTEGER NOT NULL, + car_id INTEGER NOT NULL REFERENCES public.{CAR_TABLE_NAME}(id) ON DELETE CASCADE + ) + """ + ) + ) + session.execute( + text( + f""" + CREATE TABLE IF NOT EXISTS public.{SYNC_RUN_TABLE_NAME} ( + id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, + started_at TIMESTAMPTZ NOT NULL, + finished_at TIMESTAMPTZ NULL, + status TEXT NOT NULL, + lane TEXT NOT NULL, + ids_fetched INTEGER NOT NULL DEFAULT 0, + cars_upserted INTEGER NOT NULL DEFAULT 0, + cars_failed INTEGER NOT NULL DEFAULT 0, + images_upserted INTEGER NOT NULL DEFAULT 0, + error_summary TEXT NULL + ) + """ + ) + ) + + session.execute(text(f"CREATE INDEX IF NOT EXISTS ix_iaai_images_car_id ON public.{IMAGE_TABLE_NAME} (car_id)")) + session.execute( + text(f"CREATE INDEX IF NOT EXISTS ix_iaai_images_car_order ON public.{IMAGE_TABLE_NAME} (car_id, order_index)") + ) + session.execute( + text( + f"CREATE INDEX IF NOT EXISTS ix_iaai_cars_last_seen_id ON public.{CAR_TABLE_NAME} " + "(last_seen_at DESC, id DESC)" + ) + ) + session.execute( + text( + f"CREATE INDEX IF NOT EXISTS ix_iaai_cars_active_feed ON public.{CAR_TABLE_NAME} " + "(last_seen_at DESC, id DESC) WHERE is_sold = FALSE AND is_hidden = FALSE" + ) + ) + session.execute( + text(f"CREATE INDEX IF NOT EXISTS ix_iaai_sync_runs_started_at ON public.{SYNC_RUN_TABLE_NAME} (started_at)") + ) + session.execute( + text(f"CREATE INDEX IF NOT EXISTS ix_iaai_sync_runs_status ON public.{SYNC_RUN_TABLE_NAME} (status)") + ) diff --git a/iaai_sync_service/settings.py b/iaai_sync_service/settings.py new file mode 100644 index 0000000..5dd430f --- /dev/null +++ b/iaai_sync_service/settings.py @@ -0,0 +1,123 @@ +from __future__ import annotations + +import os +from dataclasses import dataclass +from functools import lru_cache +from pathlib import Path + + +def _read_env(name: str, default: str | None = None, *, required: bool = False) -> str: + value = os.getenv(name, default) + if required and (value is None or not value.strip()): + raise RuntimeError(f"Missing required environment variable: {name}") + if value is None: + return "" + return value.strip() + + +def _read_int(name: str, default: int) -> int: + raw = _read_env(name, str(default)) + try: + return int(raw) + except ValueError as exc: + raise RuntimeError(f"Environment variable {name} must be integer, got: {raw}") from exc + + +def _read_bool(name: str, default: bool) -> bool: + raw = _read_env(name, "true" if default else "false") + normalized = raw.strip().lower() + if normalized in {"1", "true", "yes", "on"}: + return True + if normalized in {"0", "false", "no", "off"}: + return False + raise RuntimeError(f"Environment variable {name} must be boolean, got: {raw}") + + +def _normalize_database_url(raw_url: str) -> str: + normalized = raw_url.strip() + if normalized.startswith("jdbc:"): + normalized = normalized[len("jdbc:") :] + if normalized.startswith("postgres://"): + normalized = "postgresql+psycopg://" + normalized[len("postgres://") :] + elif normalized.startswith("postgresql://"): + normalized = "postgresql+psycopg://" + normalized[len("postgresql://") :] + return normalized + + +@dataclass(frozen=True) +class Settings: + database_url: str + celery_broker_url: str + celery_result_backend: str + iaai_base_url: str + iaai_listing_start_url: str + sync_runtime_config_file: Path + iaai_session_cookies: str + iaai_storage_state_path: Path + iaai_login: str + iaai_password: str + http_timeout: int + http_retries: int + http_retry_backoff_ms: int + fetch_concurrency: int + db_commit_batch_size: int + session_keepalive_enabled: bool + session_keepalive_interval_minutes: int + schema_bootstrap_enabled: bool + advisory_lock_key: int + error_summary_max_len: int + + @classmethod + def from_env(cls) -> "Settings": + settings = cls( + database_url=_normalize_database_url(_read_env("DATABASE_URL", required=True)), + celery_broker_url=_read_env("CELERY_BROKER_URL", required=True), + celery_result_backend=_read_env("CELERY_RESULT_BACKEND", "redis://redis:6379/1"), + iaai_base_url=_read_env("IAAI_BASE_URL", "https://www.iaai.com").rstrip("/"), + iaai_listing_start_url=_read_env("IAAI_LISTING_START_URL", ""), + sync_runtime_config_file=Path(_read_env("SYNC_RUNTIME_CONFIG_FILE", "sync_runtime_config.json")), + iaai_session_cookies=_read_env("IAAI_SESSION_COOKIES", ""), + iaai_storage_state_path=Path(_read_env("IAAI_STORAGE_STATE_PATH", "iaai_storage_state.json")), + iaai_login=_read_env("IAAI_LOGIN", ""), + iaai_password=_read_env("IAAI_PASSWORD", ""), + http_timeout=_read_int("IAAI_HTTP_TIMEOUT", 30), + http_retries=_read_int("IAAI_HTTP_RETRIES", 2), + http_retry_backoff_ms=_read_int("IAAI_HTTP_RETRY_BACKOFF_MS", 700), + fetch_concurrency=_read_int("IAAI_FETCH_CONCURRENCY", 8), + db_commit_batch_size=_read_int("IAAI_DB_COMMIT_BATCH_SIZE", 20), + session_keepalive_enabled=_read_bool("IAAI_SESSION_KEEPALIVE_ENABLED", True), + session_keepalive_interval_minutes=_read_int("IAAI_SESSION_KEEPALIVE_INTERVAL_MINUTES", 15), + schema_bootstrap_enabled=_read_bool("IAAI_SCHEMA_BOOTSTRAP_ENABLED", False), + advisory_lock_key=_read_int("SYNC_ADVISORY_LOCK_KEY", 7642200), + error_summary_max_len=_read_int("SYNC_ERROR_SUMMARY_MAX_LEN", 2000), + ) + + if settings.http_timeout <= 0: + raise RuntimeError("IAAI_HTTP_TIMEOUT must be greater than 0") + if settings.http_retries < 0: + raise RuntimeError("IAAI_HTTP_RETRIES cannot be negative") + if settings.http_retry_backoff_ms < 0: + raise RuntimeError("IAAI_HTTP_RETRY_BACKOFF_MS cannot be negative") + if settings.fetch_concurrency <= 0: + raise RuntimeError("IAAI_FETCH_CONCURRENCY must be greater than 0") + if settings.db_commit_batch_size <= 0: + raise RuntimeError("IAAI_DB_COMMIT_BATCH_SIZE must be greater than 0") + if settings.session_keepalive_interval_minutes <= 0: + raise RuntimeError("IAAI_SESSION_KEEPALIVE_INTERVAL_MINUTES must be greater than 0") + if settings.session_keepalive_interval_minutes > 59: + raise RuntimeError("IAAI_SESSION_KEEPALIVE_INTERVAL_MINUTES must be <= 59") + if settings.error_summary_max_len <= 0: + raise RuntimeError("SYNC_ERROR_SUMMARY_MAX_LEN must be greater than 0") + if settings.iaai_base_url.lower().startswith("http") is False: + raise RuntimeError("IAAI_BASE_URL must be absolute URL") + if settings.iaai_listing_start_url and not ( + settings.iaai_listing_start_url.lower().startswith("http") + or settings.iaai_listing_start_url.startswith("/") + ): + raise RuntimeError("IAAI_LISTING_START_URL must be absolute URL or path starting with '/'") + return settings + + +@lru_cache(maxsize=1) +def get_settings() -> Settings: + return Settings.from_env() diff --git a/iaai_sync_service/sync_engine.py b/iaai_sync_service/sync_engine.py new file mode 100644 index 0000000..be469b7 --- /dev/null +++ b/iaai_sync_service/sync_engine.py @@ -0,0 +1,1276 @@ +from __future__ import annotations + +import concurrent.futures +import logging +import re +import secrets +import string +import time +from dataclasses import dataclass +from datetime import UTC, datetime +from typing import Any + +from sqlalchemy import delete, insert, select, text, update +from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.orm import Session + +from iaai_sync_service.client import IAAIClient, ListingVehicle, build_resizer_images_from_keys +from iaai_sync_service.models import Car, Image +from iaai_sync_service.runtime_config import RuntimeConfig, RuntimeFilters +from iaai_sync_service.settings import Settings + +logger = logging.getLogger(__name__) + +ORIGIN_PREFIX = "iaai:" +ORIGIN_URL_BASE = "https://www.iaai.com/VehicleDetail" +DAMAGE_NEUTRAL_VALUES = { + "NORMAL WEAR & TEAR", + "NORMAL WEAR", + "NORMALWEAR&TEAR", + "NONE", + "NO DAMAGE", + "NO VISIBLE DAMAGE", + "MINOR DENT/SCRATCHES", +} +INACTIVE_STATUS_VALUES = { + "SOLD", + "SO", + "CLOSED", + "CN", + "DELIVERED", + "WITHDRAWN", + "WDR", + "COMPLETE", + "COMPLETED", +} +PARSER_ID_PREFIX = "car-" +PARSER_ID_CHARS = string.ascii_letters + string.digits +PARSER_ID_RANDOM_LEN = 22 +PARSER_ID_PATTERN = re.compile(r"^car-[A-Za-z0-9]{22}$") +DB_IN_CLAUSE_CHUNK_SIZE = 2000 + + +@dataclass +class SyncStats: + ids_fetched: int = 0 + cars_upserted: int = 0 + cars_failed: int = 0 + images_upserted: int = 0 + + +@dataclass(frozen=True) +class PreparedCar: + inventory_id: str + brand: str + model: str + year: int | None + price: int | None + currency: str + mileage: int + country: str + color: str + drive: str | None + gearbox: str | None + steering_wheel: str + body_type: str + engine_volume: int | None + selling_type: str + origin: str + origin_url: str + origin_id: str + is_damaged: bool + evaluation: str | None + rental: bool + images: list[dict[str, str | int]] + + +class RowSkipError(RuntimeError): + pass + + +def utc_now() -> datetime: + return datetime.now(UTC) + + +def run_sync_once( + *, + session: Session, + client: IAAIClient, + settings: Settings, + runtime_config: RuntimeConfig | None = None, +) -> tuple[SyncStats, list[str]]: + runtime = runtime_config or RuntimeConfig() + stats = SyncStats() + errors: list[str] = [] + seen_at = utc_now() + + rows_seen = 0 + rows_skipped_condition = 0 + rows_selected = 0 + + candidates: dict[str, ListingVehicle] = {} + logger.info( + "IAAI listing collection started condition_check_enabled=%s filters_enabled=%s", + runtime.condition_check_enabled is True, + runtime.filters.is_enabled(), + ) + for vehicle in client.iter_listing_vehicles(runtime.filters): + rows_seen += 1 + if runtime.condition_check_enabled and not passes_condition_check(vehicle): + rows_skipped_condition += 1 + continue + candidates[vehicle.inventory_id] = vehicle + + rows_selected = len(candidates) + logger.info( + "IAAI listing summary rows_seen=%s rows_filtered_condition=%s condition_check_enabled=%s rows_selected=%s", + rows_seen, + rows_skipped_condition, + runtime.condition_check_enabled is True, + rows_selected, + ) + + if not candidates: + return stats, errors + + currency_labels = get_enum_labels(session, "currencyenum") + country_labels = get_enum_labels(session, "countryenum") + drive_labels = get_enum_labels(session, "driveenum") + gearbox_labels = get_enum_labels(session, "gearboxenum") + body_type_labels = get_enum_labels(session, "bodytypeenum") + body_type_default = resolve_body_type_default(session) + steering_left = resolve_steering_left(session) + selling_type_auction = resolve_selling_type_auction(session) + origin_code = resolve_origin_code(session) + + origin_ids = {f"{ORIGIN_PREFIX}{inventory_id}" for inventory_id in candidates} + origin_urls = {f"{ORIGIN_URL_BASE}/{inventory_id}" for inventory_id in candidates} + existing_cars_by_id, existing_cars_by_url = preload_existing_cars( + session=session, + origin_ids=origin_ids, + origin_urls=origin_urls, + ) + existing_image_signatures = preload_image_signatures( + session=session, + car_ids=list({car.id for car in existing_cars_by_id.values()}), + ) + + commit_batch_size = max(1, settings.db_commit_batch_size) + started_at = time.perf_counter() + prepared_rows: list[PreparedCar] = [] + + with concurrent.futures.ThreadPoolExecutor(max_workers=settings.fetch_concurrency) as executor: + future_to_vehicle = { + executor.submit(client.fetch_vehicle_detail_payload, inventory_id): vehicle + for inventory_id, vehicle in candidates.items() + } + + for index, future in enumerate(concurrent.futures.as_completed(future_to_vehicle), start=1): + vehicle = future_to_vehicle[future] + try: + payload = future.result() + prepared = prepare_car( + listing_vehicle=vehicle, + detail_payload=payload, + currency_labels=currency_labels, + country_labels=country_labels, + drive_labels=drive_labels, + gearbox_labels=gearbox_labels, + body_type_labels=body_type_labels, + body_type_default=body_type_default, + steering_left=steering_left, + selling_type_auction=selling_type_auction, + origin_code=origin_code, + ) + except Exception as exc: # noqa: BLE001 + stats.cars_failed += 1 + errors.append(f"inventory_id={vehicle.inventory_id}: detail parse failed: {exc}") + logger.exception("Detail parse failed inventory_id=%s: %s", vehicle.inventory_id, exc) + continue + + if not passes_runtime_filters_prepared(prepared, runtime.filters): + continue + + stats.ids_fetched += 1 + prepared_rows.append(prepared) + + if index % 100 == 0 or index == rows_selected: + elapsed = max(0.001, time.perf_counter() - started_at) + logger.info( + "IAAI sync progress %s/%s ids_fetched=%s cars_upserted=%s cars_failed=%s images_upserted=%s queued_for_db=%s throughput=%.2f details/s", + index, + rows_selected, + stats.ids_fetched, + stats.cars_upserted, + stats.cars_failed, + stats.images_upserted, + len(prepared_rows), + index / elapsed, + ) + + if prepared_rows: + logger.info( + "IAAI DB apply started rows=%s mode=%s commit_batch_size=%s", + len(prepared_rows), + "postgres_bulk" if is_postgresql_session(session) else "rowwise", + commit_batch_size, + ) + if is_postgresql_session(session): + apply_prepared_rows_postgres( + session=session, + prepared_rows=prepared_rows, + seen_at=seen_at, + commit_batch_size=commit_batch_size, + existing_cars_by_id=existing_cars_by_id, + existing_cars_by_url=existing_cars_by_url, + existing_image_signatures=existing_image_signatures, + stats=stats, + errors=errors, + ) + else: + apply_prepared_rows_rowwise( + session=session, + prepared_rows=prepared_rows, + seen_at=seen_at, + commit_batch_size=commit_batch_size, + existing_cars_by_id=existing_cars_by_id, + existing_cars_by_url=existing_cars_by_url, + existing_image_signatures=existing_image_signatures, + stats=stats, + errors=errors, + ) + + client.persist_session_state() + + if stats.cars_failed > 0: + logger.warning("Sold reconcile skipped due to failures cars_failed=%s", stats.cars_failed) + return stats, errors + + if runtime.filters.is_enabled(): + logger.info("Sold reconcile skipped because runtime filters are enabled") + return stats, errors + + if runtime.condition_check_enabled is True: + logger.info("Sold reconcile skipped because condition check is enabled") + return stats, errors + + if stats.ids_fetched == 0: + logger.warning("Sold reconcile skipped because no rows passed filters") + return stats, errors + + reconcile_sold_cars(session=session, seen_at=seen_at) + session.commit() + return stats, errors + + +def is_postgresql_session(session: Session) -> bool: + bind = session.get_bind() + return bind is not None and bind.dialect.name == "postgresql" + + +def apply_prepared_rows_rowwise( + *, + session: Session, + prepared_rows: list[PreparedCar], + seen_at: datetime, + commit_batch_size: int, + existing_cars_by_id: dict[str, Car], + existing_cars_by_url: dict[str, Car], + existing_image_signatures: dict[int, list[tuple[int, str, str]]], + stats: SyncStats, + errors: list[str], + allow_lookup: bool = False, +) -> None: + pending_db_changes = 0 + total_rows = len(prepared_rows) + started_at = time.perf_counter() + + for index, prepared in enumerate(prepared_rows, start=1): + try: + with session.begin_nested(): + car = upsert_car( + session=session, + row=prepared, + seen_at=seen_at, + existing_entity=existing_cars_by_id.get(prepared.origin_id) + or existing_cars_by_url.get(prepared.origin_url), + allow_lookup=allow_lookup, + ) + existing_cars_by_id[prepared.origin_id] = car + existing_cars_by_url[prepared.origin_url] = car + + existing_signature = existing_image_signatures.get(car.id) + images_written, normalized_images = replace_images_if_changed_cached( + session=session, + car_id=car.id, + images=prepared.images, + existing_normalized=existing_signature, + ) + existing_image_signatures[car.id] = normalized_images + car.is_hidden = not bool(normalized_images) + + stats.cars_upserted += 1 + stats.images_upserted += images_written + pending_db_changes += 1 + if pending_db_changes >= commit_batch_size: + session.commit() + pending_db_changes = 0 + except Exception as exc: # noqa: BLE001 + stats.cars_failed += 1 + errors.append(f"inventory_id={prepared.inventory_id}: upsert failed: {exc}") + logger.exception("Upsert failed inventory_id=%s: %s", prepared.inventory_id, exc) + + if index % 100 == 0 or index == total_rows: + elapsed = max(0.001, time.perf_counter() - started_at) + logger.info( + "IAAI DB progress %s/%s cars_upserted=%s cars_failed=%s images_upserted=%s throughput=%.2f rows/s", + index, + total_rows, + stats.cars_upserted, + stats.cars_failed, + stats.images_upserted, + index / elapsed, + ) + + if pending_db_changes > 0: + session.commit() + + +def apply_prepared_rows_postgres( + *, + session: Session, + prepared_rows: list[PreparedCar], + seen_at: datetime, + commit_batch_size: int, + existing_cars_by_id: dict[str, Car], + existing_cars_by_url: dict[str, Car], + existing_image_signatures: dict[int, list[tuple[int, str, str]]], + stats: SyncStats, + errors: list[str], +) -> None: + total_rows = len(prepared_rows) + started_at = time.perf_counter() + + for batch_start in range(0, total_rows, commit_batch_size): + batch = prepared_rows[batch_start : batch_start + commit_batch_size] + try: + _apply_prepared_batch_postgres( + session=session, + prepared_rows=batch, + seen_at=seen_at, + existing_cars_by_id=existing_cars_by_id, + existing_cars_by_url=existing_cars_by_url, + existing_image_signatures=existing_image_signatures, + stats=stats, + ) + except Exception as exc: # noqa: BLE001 + session.rollback() + logger.exception( + "PostgreSQL bulk apply failed batch_start=%s batch_size=%s. Falling back to row-wise mode: %s", + batch_start + 1, + len(batch), + exc, + ) + apply_prepared_rows_rowwise( + session=session, + prepared_rows=batch, + seen_at=seen_at, + commit_batch_size=max(1, min(commit_batch_size, 50)), + existing_cars_by_id=existing_cars_by_id, + existing_cars_by_url=existing_cars_by_url, + existing_image_signatures=existing_image_signatures, + stats=stats, + errors=errors, + allow_lookup=True, + ) + + processed = batch_start + len(batch) + if processed % 100 == 0 or processed == total_rows: + elapsed = max(0.001, time.perf_counter() - started_at) + logger.info( + "IAAI DB progress %s/%s cars_upserted=%s cars_failed=%s images_upserted=%s throughput=%.2f rows/s", + processed, + total_rows, + stats.cars_upserted, + stats.cars_failed, + stats.images_upserted, + processed / elapsed, + ) + + +def _apply_prepared_batch_postgres( + *, + session: Session, + prepared_rows: list[PreparedCar], + seen_at: datetime, + existing_cars_by_id: dict[str, Car], + existing_cars_by_url: dict[str, Car], + existing_image_signatures: dict[int, list[tuple[int, str, str]]], + stats: SyncStats, +) -> None: + if not prepared_rows: + return + + reserved_slugs: set[str] = set() + upsert_values: list[dict[str, Any]] = [] + normalized_images_by_origin: dict[str, list[tuple[int, str, str]]] = {} + + for row in prepared_rows: + existing = existing_cars_by_id.get(row.origin_id) or existing_cars_by_url.get(row.origin_url) + parser_id = parse_text(existing.parser_id) if existing is not None else None + if not is_valid_parser_id(parser_id): + parser_id = generate_parser_id(session) + + slug_needs_refresh = ( + existing is None + or not parse_text(existing.slug) + or existing.brand != row.brand + or existing.model != row.model + or existing.year != row.year + ) + if slug_needs_refresh: + slug = generate_unique_slug( + session=session, + brand=row.brand, + model=row.model, + year=row.year, + parser_id=parser_id, + ) + slug = reserve_slug_in_batch( + session=session, + base_slug=slug, + reserved_slugs=reserved_slugs, + ) + else: + slug = str(existing.slug) + reserved_slugs.add(slug) + + normalized_images = normalize_images(row.images) + normalized_images_by_origin[row.origin_id] = normalized_images + + upsert_values.append( + { + "parser_id": parser_id, + "brand": row.brand, + "model": row.model, + "year": row.year, + "price": row.price, + "currency": row.currency, + "mileage": row.mileage, + "country": row.country, + "is_sold": False, + "color": row.color, + "drive": row.drive, + "gearbox": row.gearbox, + "steering_wheel": row.steering_wheel, + "body_type": row.body_type, + "engine_volume": row.engine_volume, + "selling_type": row.selling_type, + "one_owner": False, + "new_car": False, + "is_hidden": not bool(normalized_images), + "origin": row.origin, + "origin_url": row.origin_url, + "origin_id": row.origin_id, + "is_damaged": row.is_damaged, + "evaluation": row.evaluation, + "non_smoking": True, + "rental": row.rental, + "repair_history": False, + "slug": slug, + "last_seen_at": seen_at, + } + ) + + statement = pg_insert(Car).values(upsert_values) + excluded = statement.excluded + update_values = { + "parser_id": excluded.parser_id, + "brand": excluded.brand, + "model": excluded.model, + "year": excluded.year, + "price": excluded.price, + "currency": excluded.currency, + "mileage": excluded.mileage, + "country": excluded.country, + "is_sold": excluded.is_sold, + "color": excluded.color, + "drive": excluded.drive, + "gearbox": excluded.gearbox, + "steering_wheel": excluded.steering_wheel, + "body_type": excluded.body_type, + "engine_volume": excluded.engine_volume, + "selling_type": excluded.selling_type, + "one_owner": excluded.one_owner, + "new_car": excluded.new_car, + "is_hidden": excluded.is_hidden, + "origin": excluded.origin, + "origin_url": excluded.origin_url, + "is_damaged": excluded.is_damaged, + "evaluation": excluded.evaluation, + "non_smoking": excluded.non_smoking, + "rental": excluded.rental, + "repair_history": excluded.repair_history, + "slug": excluded.slug, + "last_seen_at": excluded.last_seen_at, + } + result_rows = session.execute( + statement.on_conflict_do_update( + index_elements=[Car.origin_id], + set_=update_values, + ).returning(Car.id, Car.origin_id) + ).all() + + car_id_by_origin = {origin_id: car_id for car_id, origin_id in result_rows} + changed_car_ids: list[int] = [] + image_rows: list[dict[str, Any]] = [] + local_images_upserted = 0 + signature_updates: dict[int, list[tuple[int, str, str]]] = {} + + for payload in upsert_values: + origin_id = str(payload["origin_id"]) + car_id = car_id_by_origin.get(origin_id) + if car_id is None: + raise RuntimeError(f"Bulk upsert did not return car id for origin_id={origin_id}") + + normalized_new = normalized_images_by_origin[origin_id] + existing_normalized = existing_image_signatures.get(car_id) + if normalized_new == existing_normalized: + continue + + changed_car_ids.append(car_id) + signature_updates[car_id] = normalized_new + local_images_upserted += len(normalized_new) + for order_index, fullres_image, preview_image in normalized_new: + image_rows.append( + { + "car_id": car_id, + "order_index": order_index, + "fullres_image": fullres_image, + "preview_image": preview_image, + } + ) + + if changed_car_ids: + session.execute(delete(Image).where(Image.car_id.in_(changed_car_ids))) + if image_rows: + session.execute(insert(Image), image_rows) + + session.commit() + existing_image_signatures.update(signature_updates) + stats.cars_upserted += len(prepared_rows) + stats.images_upserted += local_images_upserted + + +def reserve_slug_in_batch( + *, + session: Session, + base_slug: str, + reserved_slugs: set[str], +) -> str: + if base_slug not in reserved_slugs: + reserved_slugs.add(base_slug) + return base_slug + + counter = 1 + while True: + candidate = f"{base_slug}-{counter}" + counter += 1 + if candidate in reserved_slugs: + continue + exists = session.execute(select(Car.id).where(Car.slug == candidate)).scalar_one_or_none() + if exists is None: + reserved_slugs.add(candidate) + return candidate + + +def passes_condition_check(row: ListingVehicle) -> bool: + if row.timed_auction_closed: + return False + status = (row.inventory_status or "").strip().upper() + if status in INACTIVE_STATUS_VALUES: + return False + return True + + +def passes_runtime_filters_prepared(row: PreparedCar, filters: RuntimeFilters) -> bool: + if not filters.is_enabled(): + return True + + brand_key = row.brand.casefold() + model_key = row.model.casefold() + body_key = row.body_type.casefold() + + if filters.brands and brand_key not in filters.brands: + return False + if filters.models and model_key not in filters.models: + return False + if filters.years and (row.year is None or row.year not in filters.years): + return False + if filters.body_types and body_key not in filters.body_types: + return False + return True + + +def prepare_car( + *, + listing_vehicle: ListingVehicle, + detail_payload: dict[str, Any], + currency_labels: set[str], + country_labels: set[str], + drive_labels: set[str], + gearbox_labels: set[str], + body_type_labels: set[str], + body_type_default: str, + steering_left: str, + selling_type_auction: str, + origin_code: str, +) -> PreparedCar: + inventory_view = detail_payload.get("inventoryView") + if not isinstance(inventory_view, dict): + raise RowSkipError("detail payload missing inventoryView") + + attributes = inventory_view.get("attributes") + if not isinstance(attributes, dict): + raise RowSkipError("detail payload missing inventoryView.attributes") + + inventory_id = parse_text(attributes.get("Id")) or listing_vehicle.inventory_id + if not inventory_id: + raise RowSkipError("missing inventory id") + + brand = limit_text(parse_text(attributes.get("Make")) or "unknown", 50) + model = build_model_name( + parse_text(attributes.get("Model")), + parse_text(attributes.get("Series")), + ) + if not model: + model = "unknown" + model = limit_text(model, 50) + + year = parse_year(attributes.get("Year")) + + auction_info = detail_payload.get("auctionInformation") + auction_info = auction_info if isinstance(auction_info, dict) else {} + bidding_info = auction_info.get("biddingInformation") + bidding_info = bidding_info if isinstance(bidding_info, dict) else {} + prebid_info = auction_info.get("prebidInformation") + prebid_info = prebid_info if isinstance(prebid_info, dict) else {} + + high_bid = first_positive_int( + prebid_info.get("decimalHighBidAmount"), + prebid_info.get("highBidAmount"), + bidding_info.get("highBidAmount"), + ) + buy_now = first_positive_int( + bidding_info.get("buyNowAmount"), + prebid_info.get("buyNowPrice"), + bidding_info.get("buyNowPrice"), + ) + price = high_bid if high_bid is not None else buy_now + + currency = map_currency( + parse_text(attributes.get("Currency")) or listing_vehicle.currency, + currency_labels=currency_labels, + ) + country = map_country( + inventory_id=inventory_id, + country_labels=country_labels, + ) + + mileage = parse_non_negative_int(attributes.get("ODOValue")) or 0 + color = normalize_color( + parse_text(attributes.get("ExteriorColor")) + or parse_text(attributes.get("ColorDesc")) + or parse_text(attributes.get("colorDesc")) + ) + drive = map_drive(parse_text(attributes.get("DriveLineTypeDesc")), drive_labels=drive_labels) + gearbox = map_gearbox(parse_text(attributes.get("Transmission")), gearbox_labels=gearbox_labels) + body_type = ( + map_body_type( + parse_text(attributes.get("BodyStyleName")) or parse_text(attributes.get("VehicleClass")), + body_type_labels=body_type_labels, + ) + or body_type_default + ) + engine_volume = parse_engine_volume( + parse_text(attributes.get("EngineSize")) or parse_text(attributes.get("EngineInformation")) + ) + + evaluation = parse_text(attributes.get("VehicleGrade")) + is_damaged = derive_is_damaged( + primary_damage=parse_text(attributes.get("PrimaryDamageDesc")), + secondary_damage=parse_text(attributes.get("SecondaryDamageDesc")), + ) + + image_dimensions = inventory_view.get("imageDimensions") + image_dimensions = image_dimensions if isinstance(image_dimensions, dict) else {} + keys_container = image_dimensions.get("keys") + keys_container = keys_container if isinstance(keys_container, dict) else {} + image_keys = keys_container.get("$values") + image_keys = image_keys if isinstance(image_keys, list) else [] + images = build_resizer_images_from_keys(image_keys) + + origin_id = f"{ORIGIN_PREFIX}{inventory_id}" + origin_url = f"{ORIGIN_URL_BASE}/{inventory_id}" + + return PreparedCar( + inventory_id=inventory_id, + brand=brand, + model=model, + year=year, + price=price, + currency=currency, + mileage=mileage, + country=country, + color=color, + drive=drive, + gearbox=gearbox, + steering_wheel=steering_left, + body_type=body_type, + engine_volume=engine_volume, + selling_type=selling_type_auction, + origin=origin_code, + origin_url=origin_url, + origin_id=origin_id, + is_damaged=is_damaged, + evaluation=evaluation, + rental=False, + images=images, + ) + + +def preload_existing_cars( + *, + session: Session, + origin_ids: set[str], + origin_urls: set[str], +) -> tuple[dict[str, Car], dict[str, Car]]: + by_origin_id: dict[str, Car] = {} + by_origin_url: dict[str, Car] = {} + + if origin_ids: + ids = sorted(origin_ids) + for chunk in chunked(ids, DB_IN_CLAUSE_CHUNK_SIZE): + rows = session.execute(select(Car).where(Car.origin_id.in_(chunk))).scalars().all() + for row in rows: + by_origin_id[row.origin_id] = row + by_origin_url[row.origin_url] = row + + if origin_urls: + urls_to_lookup = sorted(url for url in origin_urls if url not in by_origin_url) + for chunk in chunked(urls_to_lookup, DB_IN_CLAUSE_CHUNK_SIZE): + rows = session.execute(select(Car).where(Car.origin_url.in_(chunk))).scalars().all() + for row in rows: + by_origin_url[row.origin_url] = row + by_origin_id.setdefault(row.origin_id, row) + + return by_origin_id, by_origin_url + + +def preload_image_signatures(*, session: Session, car_ids: list[int]) -> dict[int, list[tuple[int, str, str]]]: + if not car_ids: + return {} + signatures: dict[int, list[tuple[int, str, str]]] = {} + for chunk in chunked(car_ids, DB_IN_CLAUSE_CHUNK_SIZE): + rows = session.execute( + select(Image) + .where(Image.car_id.in_(chunk)) + .order_by(Image.car_id, Image.order_index, Image.fullres_image, Image.preview_image) + ).scalars() + for row in rows: + signatures.setdefault(row.car_id, []).append( + ( + row.order_index, + row.fullres_image.strip(), + row.preview_image.strip(), + ) + ) + return signatures + + +def chunked(values: list[Any], size: int) -> list[list[Any]]: + if size <= 0: + return [values] + return [values[i : i + size] for i in range(0, len(values), size)] + + +def upsert_car( + *, + session: Session, + row: PreparedCar, + seen_at: datetime, + existing_entity: Car | None = None, + allow_lookup: bool = True, +) -> Car: + entity = existing_entity + if entity is None and allow_lookup: + entity = session.execute(select(Car).where(Car.origin_id == row.origin_id)).scalar_one_or_none() + if entity is None and allow_lookup: + entity = session.execute(select(Car).where(Car.origin_url == row.origin_url)).scalar_one_or_none() + + is_new_entity = False + if entity is None: + entity = Car( + parser_id=generate_parser_id(session), + brand=row.brand, + model=row.model, + origin_id=row.origin_id, + origin_url=row.origin_url, + ) + is_new_entity = True + elif not is_valid_parser_id(entity.parser_id): + entity.parser_id = generate_parser_id(session) + + slug_needs_refresh = ( + is_new_entity + or not parse_text(entity.slug) + or entity.brand != row.brand + or entity.model != row.model + or entity.year != row.year + ) + if slug_needs_refresh: + slug = generate_unique_slug( + session=session, + brand=row.brand, + model=row.model, + year=row.year, + parser_id=entity.parser_id, + ) + else: + slug = entity.slug + + entity.brand = row.brand + entity.model = row.model + entity.year = row.year + entity.price = row.price + entity.currency = row.currency + entity.mileage = row.mileage + entity.country = row.country + entity.is_sold = False + entity.color = row.color + entity.drive = row.drive + entity.gearbox = row.gearbox + entity.steering_wheel = row.steering_wheel + entity.body_type = row.body_type + entity.engine_volume = row.engine_volume + entity.selling_type = row.selling_type + entity.one_owner = False + entity.new_car = False + entity.origin = row.origin + entity.origin_url = row.origin_url + entity.origin_id = row.origin_id + entity.is_damaged = row.is_damaged + entity.evaluation = row.evaluation + entity.non_smoking = True + entity.rental = row.rental + entity.repair_history = False + entity.slug = slug + entity.last_seen_at = seen_at + + if is_new_entity: + session.add(entity) + + session.flush() + return entity + + +def reconcile_sold_cars(*, session: Session, seen_at: datetime) -> None: + session.execute( + update(Car) + .where( + Car.origin_id.like(f"{ORIGIN_PREFIX}%"), + Car.last_seen_at < seen_at, + Car.is_sold.is_(False), + ) + .values(is_sold=True) + ) + + +def replace_images_if_changed( + *, + session: Session, + car_id: int, + images: list[dict[str, str | int]], +) -> int: + written, _ = replace_images_if_changed_cached( + session=session, + car_id=car_id, + images=images, + existing_normalized=None, + ) + return written + + +def replace_images_if_changed_cached( + *, + session: Session, + car_id: int, + images: list[dict[str, str | int]], + existing_normalized: list[tuple[int, str, str]] | None, +) -> tuple[int, list[tuple[int, str, str]]]: + normalized_new = normalize_images(images) + + if existing_normalized is None: + existing_rows = session.execute(select(Image).where(Image.car_id == car_id)).scalars().all() + existing_normalized = sorted( + ((row.order_index, row.fullres_image.strip(), row.preview_image.strip()) for row in existing_rows), + key=lambda item: (item[0], item[1], item[2]), + ) + + if normalized_new == existing_normalized: + return 0, existing_normalized + + replace_images(session=session, car_id=car_id, images=images) + return len(normalized_new), normalized_new + + +def replace_images( + *, + session: Session, + car_id: int, + images: list[dict[str, str | int]], +) -> None: + session.execute(delete(Image).where(Image.car_id == car_id)) + for image in images: + fullres_image = parse_text(image.get("fullres_image")) + if fullres_image is None: + continue + preview_image = parse_text(image.get("preview_image")) or fullres_image + order_index = parse_non_negative_int(image.get("order_index")) or 0 + session.add( + Image( + car_id=car_id, + fullres_image=fullres_image, + preview_image=preview_image, + order_index=order_index, + ) + ) + + +def normalize_images(images: list[dict[str, str | int]]) -> list[tuple[int, str, str]]: + normalized: list[tuple[int, str, str]] = [] + for image in images: + fullres_image = parse_text(image.get("fullres_image")) + if fullres_image is None: + continue + preview_image = parse_text(image.get("preview_image")) or fullres_image + order_index = parse_non_negative_int(image.get("order_index")) or 0 + normalized.append((order_index, fullres_image.strip(), preview_image.strip())) + normalized.sort(key=lambda item: (item[0], item[1], item[2])) + return normalized + + +def build_model_name(model: str | None, series: str | None) -> str: + values = [model, series] + unique_parts: list[str] = [] + seen: set[str] = set() + for value in values: + if not value: + continue + normalized = " ".join(value.split()) + if not normalized: + continue + key = normalized.casefold() + if key in seen: + continue + seen.add(key) + unique_parts.append(normalized) + return " ".join(unique_parts).strip() + + +def parse_year(value: Any) -> int | None: + parsed = parse_non_negative_int(value) + if parsed is None or parsed <= 0: + return None + if parsed < 1900 or parsed > 2100: + return None + return parsed + + +def parse_non_negative_int(value: Any) -> int | None: + if value is None or isinstance(value, bool): + return None + if isinstance(value, int): + return value if value >= 0 else None + if isinstance(value, float): + if value < 0: + return None + return int(round(value)) + if isinstance(value, str): + raw = value.strip() + if not raw: + return None + compact = raw.replace(",", "").replace(" ", "") + compact = compact.replace("$", "") + match = re.search(r"-?\d+(?:\.\d+)?", compact) + if match is None: + return None + try: + parsed = float(match.group(0)) + except ValueError: + return None + if parsed < 0: + return None + return int(round(parsed)) + return None + + +def first_positive_int(*values: Any) -> int | None: + for value in values: + parsed = parse_non_negative_int(value) + if parsed is not None and parsed > 0: + return parsed + return None + + +def parse_text(value: Any) -> str | None: + if isinstance(value, str): + text = value.strip() + return text if text else None + return None + + +def normalize_color(value: str | None) -> str: + if not value: + return "other" + normalized = value.strip().lower() + return normalized or "other" + + +def map_currency(value: str | None, *, currency_labels: set[str]) -> str: + if value is None: + return select_enum_label(currency_labels, ("USD", "NA"), fallback="NA") or "NA" + normalized = value.strip().upper() + if normalized == "CAD": + return select_enum_label(currency_labels, ("CAD", "USD", "NA"), fallback="NA") or "NA" + if normalized == "USD": + return select_enum_label(currency_labels, ("USD", "NA"), fallback="NA") or "NA" + return select_enum_label(currency_labels, ("NA", "USD", "CAD"), fallback="NA") or "NA" + + +def map_country(*, inventory_id: str, country_labels: set[str]) -> str: + upper_id = inventory_id.strip().upper() + if upper_id.endswith("~US"): + return select_enum_label(country_labels, ("US", "NA"), fallback="NA") or "NA" + if upper_id.endswith("~CA"): + return select_enum_label(country_labels, ("CA", "US", "NA"), fallback="NA") or "NA" + return select_enum_label(country_labels, ("NA", "US", "CA"), fallback="NA") or "NA" + + +def map_drive(value: str | None, *, drive_labels: set[str]) -> str | None: + if not value: + return None + normalized = value.strip().lower() + candidates: tuple[str, ...] | None = None + if "front" in normalized or normalized == "fwd": + candidates = ("FWD",) + elif "all wheel" in normalized or "4x4" in normalized or normalized == "awd": + candidates = ("4WD", "FOUR_WD") + elif "rear" in normalized or normalized == "rwd": + candidates = ("RWD",) + elif "2wd" in normalized or "two wheel" in normalized: + candidates = ("2WD", "TWO_WD") + elif "unknown" in normalized or normalized in {"na", "n/a"}: + candidates = ("NA",) + if candidates is None: + return None + return select_enum_label(drive_labels, candidates, fallback="NA") + + +def map_gearbox(value: str | None, *, gearbox_labels: set[str]) -> str | None: + if not value: + return None + normalized = value.strip().lower() + candidates: tuple[str, ...] | None = None + if "cvt" in normalized: + candidates = ("CVT",) + elif "manual" in normalized or normalized == "mt": + candidates = ("MT",) + elif "electric" in normalized or normalized == "ev": + candidates = ("EV",) + elif "auto" in normalized or normalized == "at": + candidates = ("AT",) + elif "unknown" in normalized or normalized in {"na", "n/a"}: + candidates = ("NA",) + if candidates is None: + return None + return select_enum_label(gearbox_labels, candidates, fallback="NA") + + +def map_body_type(value: str | None, *, body_type_labels: set[str]) -> str | None: + if not value: + return None + normalized = value.strip().lower() + candidates: tuple[str, ...] | None = None + if "sedan" in normalized: + candidates = ("SEDAN",) + elif "sport utility" in normalized or normalized == "suv": + candidates = ("SUV",) + elif "hatch" in normalized: + candidates = ("HATCHBACK",) + elif "wagon" in normalized or normalized == "station": + candidates = ("STATION_WAGON", "Station Wagon") + elif "coupe" in normalized: + candidates = ("COUPE",) + elif "pickup" in normalized or "crew" in normalized and "cab" in normalized: + candidates = ("PICKUP", "Pickup") + elif "convertible" in normalized or "roadster" in normalized or "cabrio" in normalized: + candidates = ("OPEN", "Open") + elif "van" in normalized: + candidates = ("MINIVAN",) + elif "truck" in normalized or "chassis" in normalized: + candidates = ("TRUCK", "Truck") + elif "rv" in normalized or "motorized" in normalized: + candidates = ("RV",) + elif normalized in {"other", "unknown"}: + candidates = ("OTHER", "Other") + if candidates is None: + return None + return select_enum_label(body_type_labels, candidates, fallback="OTHER") + + +def parse_engine_volume(value: str | None) -> int | None: + if not value: + return None + match = re.search(r"(\d+(?:\.\d+)?)\s*[lL]\b", value) + if not match: + return None + try: + liters = float(match.group(1)) + except ValueError: + return None + cc = int(round(liters * 1000)) + if cc <= 0 or cc > 10000: + return None + return cc + + +def derive_is_damaged(*, primary_damage: str | None, secondary_damage: str | None) -> bool: + for value in (primary_damage, secondary_damage): + if not value: + continue + normalized = value.replace(" ", "").strip().upper() + if not normalized: + continue + if normalized in {item.replace(" ", "") for item in DAMAGE_NEUTRAL_VALUES}: + continue + return True + return False + + +def resolve_origin_code(session: Session) -> str: + labels = get_enum_labels(session, "originenum") + if labels: + return select_enum_label(labels, ("IAAI", "iaai", "NA"), fallback="NA") or "NA" + return "IAAI" + + +def resolve_selling_type_auction(session: Session) -> str: + labels = get_enum_labels(session, "sellingtypeenum") + if labels: + return select_enum_label(labels, ("AUCTION", "auction", "NA"), fallback="NA") or "NA" + return "AUCTION" + + +def resolve_steering_left(session: Session) -> str: + labels = get_enum_labels(session, "steeringwheelenum") + if labels: + return select_enum_label(labels, ("LEFT", "left", "NA"), fallback="NA") or "NA" + return "LEFT" + + +def resolve_body_type_default(session: Session) -> str: + labels = get_enum_labels(session, "bodytypeenum") + if labels: + return select_enum_label(labels, ("OTHER", "Other", "NA"), fallback="NA") or "NA" + return "OTHER" + + +def get_enum_labels(session: Session, type_name: str) -> set[str]: + bind = session.get_bind() + if bind is None or bind.dialect.name != "postgresql": + return set() + rows = session.execute( + text( + """ + SELECT e.enumlabel + FROM pg_type t + JOIN pg_namespace n ON n.oid = t.typnamespace + JOIN pg_enum e ON e.enumtypid = t.oid + WHERE n.nspname = 'public' + AND t.typname = :type_name + """ + ), + {"type_name": type_name}, + ).scalars() + return {value for value in rows} + + +def select_enum_label( + labels: set[str] | None, + candidates: tuple[str, ...], + *, + fallback: str | None = None, +) -> str | None: + if not labels: + return candidates[0] if candidates else fallback + for candidate in candidates: + if candidate in labels: + return candidate + if fallback and fallback in labels: + return fallback + return sorted(labels)[0] if labels else fallback + + +def limit_text(value: str, max_length: int) -> str: + if len(value) <= max_length: + return value + return value[:max_length].rstrip() + + +def is_valid_parser_id(value: str | None) -> bool: + if value is None: + return False + return bool(PARSER_ID_PATTERN.fullmatch(value)) + + +def generate_parser_id(session: Session) -> str: + while True: + candidate = PARSER_ID_PREFIX + "".join(secrets.choice(PARSER_ID_CHARS) for _ in range(PARSER_ID_RANDOM_LEN)) + exists = session.execute(select(Car.id).where(Car.parser_id == candidate)).scalar_one_or_none() + if exists is None: + return candidate + + +def slugify_text(value: str) -> str: + lowered = value.lower() + slug = re.sub(r"[^a-z0-9]+", "-", lowered).strip("-") + return slug or "car" + + +def generate_unique_slug( + *, + session: Session, + brand: str, + model: str, + year: int | None, + parser_id: str, +) -> str: + if year is None: + base_slug = slugify_text(f"{brand}-{model}") + else: + base_slug = slugify_text(f"{brand}-{model}-{year}") + + slug = base_slug + counter = 1 + while True: + existing = session.execute( + select(Car.id).where( + Car.slug == slug, + Car.parser_id != parser_id, + ) + ).scalar_one_or_none() + if existing is None: + return slug + slug = f"{base_slug}-{counter}" + counter += 1 diff --git a/iaai_sync_service/tasks.py b/iaai_sync_service/tasks.py new file mode 100644 index 0000000..94f3d8c --- /dev/null +++ b/iaai_sync_service/tasks.py @@ -0,0 +1,213 @@ +from __future__ import annotations + +from datetime import UTC, datetime +from typing import Any + +from celery.utils.log import get_task_logger +from sqlalchemy import text +from sqlalchemy.exc import ProgrammingError + +from iaai_sync_service.celery_app import app +from iaai_sync_service.client import IAAIClient +from iaai_sync_service.database import get_session +from iaai_sync_service.models import SYNC_RUN_TABLE_NAME, SyncRun +from iaai_sync_service.runtime_config import load_runtime_config +from iaai_sync_service.schema_bootstrap import ensure_schema +from iaai_sync_service.settings import get_settings +from iaai_sync_service.sync_engine import run_sync_once + +logger = get_task_logger(__name__) + +SYNC_LANE = "iaai_cars" + + +def utc_now() -> datetime: + return datetime.now(UTC) + + +def trim_error_summary(errors: list[str], max_len: int) -> str | None: + if not errors: + return None + joined = " | ".join(errors) + if len(joined) <= max_len: + return joined + return joined[: max_len - 3] + "..." + + +def _is_missing_sync_runs_error(exc: ProgrammingError) -> bool: + message = str(exc).lower() + return "does not exist" in message and SYNC_RUN_TABLE_NAME in message + + +def _create_sync_run(session, lane: str) -> SyncRun: # noqa: ANN001 + run = SyncRun( + status="running", + lane=lane, + started_at=utc_now(), + ids_fetched=0, + cars_upserted=0, + cars_failed=0, + images_upserted=0, + ) + session.add(run) + session.commit() + return run + + +@app.task(name="iaai.sync_cars_feed", queue="iaai_sync", bind=True) +def sync_cars_feed(self: Any) -> dict[str, Any]: + return execute_sync_job() + + +@app.task(name="iaai.keep_session_alive", queue="iaai_sync", bind=True) +def keep_session_alive(self: Any) -> dict[str, Any]: + return execute_keepalive_job() + + +def execute_sync_job() -> dict[str, Any]: + settings = get_settings() + session = get_session() + run_id: int | None = None + lock_acquired = False + + try: + logger.info("IAAI sync job started base_url=%s", settings.iaai_base_url) + if settings.schema_bootstrap_enabled: + ensure_schema(session) + session.commit() + logger.info("Schema bootstrap is enabled and completed.") + else: + logger.info("Schema bootstrap is disabled.") + + lock_acquired = bool( + session.execute( + text("SELECT pg_try_advisory_lock(:key)"), + {"key": settings.advisory_lock_key}, + ).scalar_one() + ) + if not lock_acquired: + logger.info("IAAI sync skipped: advisory lock is already held.") + return {"status": "skipped_locked"} + + try: + run = _create_sync_run(session=session, lane=SYNC_LANE) + except ProgrammingError as exc: + session.rollback() + if not _is_missing_sync_runs_error(exc): + raise + logger.warning( + "Table '%s' is missing. Running one-time schema bootstrap and retrying.", + SYNC_RUN_TABLE_NAME, + ) + ensure_schema(session) + session.commit() + run = _create_sync_run(session=session, lane=SYNC_LANE) + run_id = run.id + logger.info("IAAI sync run created run_id=%s", run_id) + + client = IAAIClient(settings=settings) + runtime_config = load_runtime_config(settings.sync_runtime_config_file, logger=logger) + logger.info("Loaded runtime config from '%s'", settings.sync_runtime_config_file) + + stats, errors = run_sync_once( + session=session, + client=client, + settings=settings, + runtime_config=runtime_config, + ) + + run = session.get(SyncRun, run_id) + if run is None: + raise RuntimeError(f"sync_runs row disappeared for id={run_id}") + run.finished_at = utc_now() + run.ids_fetched = stats.ids_fetched + run.cars_upserted = stats.cars_upserted + run.cars_failed = stats.cars_failed + run.images_upserted = stats.images_upserted + run.status = "partial" if stats.cars_failed > 0 else "success" + run.error_summary = trim_error_summary(errors, settings.error_summary_max_len) + session.commit() + + logger.info( + "IAAI sync completed status=%s ids=%s cars_upserted=%s cars_failed=%s images_upserted=%s", + run.status, + run.ids_fetched, + run.cars_upserted, + run.cars_failed, + run.images_upserted, + ) + return { + "status": run.status, + "ids_fetched": run.ids_fetched, + "cars_upserted": run.cars_upserted, + "cars_failed": run.cars_failed, + "images_upserted": run.images_upserted, + "run_id": run.id, + } + except Exception as exc: # noqa: BLE001 + session.rollback() + if run_id is not None: + failed = session.get(SyncRun, run_id) + if failed is not None: + failed.finished_at = utc_now() + failed.status = "failed" + failed.error_summary = trim_error_summary([str(exc)], settings.error_summary_max_len) + session.commit() + logger.exception("IAAI sync failed: %s", exc) + raise + finally: + if lock_acquired: + try: + session.execute( + text("SELECT pg_advisory_unlock(:key)"), + {"key": settings.advisory_lock_key}, + ) + session.commit() + except Exception: # noqa: BLE001 + session.rollback() + logger.exception("Failed to release advisory lock.") + session.close() + + +def execute_keepalive_job() -> dict[str, Any]: + settings = get_settings() + session = get_session() + lock_acquired = False + + try: + lock_acquired = bool( + session.execute( + text("SELECT pg_try_advisory_lock(:key)"), + {"key": settings.advisory_lock_key}, + ).scalar_one() + ) + if not lock_acquired: + logger.info("IAAI keepalive skipped: advisory lock is already held.") + return {"status": "skipped_locked"} + + logger.info("IAAI keepalive job started base_url=%s", settings.iaai_base_url) + client = IAAIClient(settings=settings) + result = client.keep_session_alive() + logger.info( + "IAAI keepalive completed scope=%s marker_present=%s result_count=%s elapsed_ms=%s", + result.get("scope"), + result.get("marker_present"), + result.get("result_count"), + result.get("elapsed_ms"), + ) + return {"status": "success", **result} + except Exception as exc: # noqa: BLE001 + logger.exception("IAAI keepalive failed: %s", exc) + raise + finally: + if lock_acquired: + try: + session.execute( + text("SELECT pg_advisory_unlock(:key)"), + {"key": settings.advisory_lock_key}, + ) + session.commit() + except Exception: # noqa: BLE001 + session.rollback() + logger.exception("Failed to release advisory lock for keepalive.") + session.close() diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..f11d6f9 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,26 @@ +[build-system] +requires = ["setuptools>=69", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "iaai" +version = "0.1.0" +description = "IAAI Cars sync service" +readme = "README.md" +requires-python = ">=3.13" +dependencies = [ + "alembic>=1.17.1", + "celery[redis]>=5.5.3", + "playwright>=1.58.0", + "psycopg[binary]>=3.2.12", + "requests>=2.32.5", + "sqlalchemy>=2.0.43", +] + +[project.optional-dependencies] +dev = [ + "pytest>=8.4.2", +] + +[tool.setuptools.packages.find] +include = ["iaai_sync_service*"] diff --git a/sync_runtime_config.json b/sync_runtime_config.json new file mode 100644 index 0000000..c70458e --- /dev/null +++ b/sync_runtime_config.json @@ -0,0 +1,11 @@ +{ + "sync": { + "condition_check_enabled": false + }, + "filters": { + "brands": [], + "models": [], + "years": [], + "body_types": [] + } +} diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..e80cb47 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,13 @@ +from __future__ import annotations + +import os +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +os.environ.setdefault("DATABASE_URL", "sqlite+pysqlite:///:memory:") +os.environ.setdefault("CELERY_BROKER_URL", "redis://localhost:6379/0") +os.environ.setdefault("IAAI_BASE_URL", "https://www.iaai.com") diff --git a/tests/test_client.py b/tests/test_client.py new file mode 100644 index 0000000..376e494 --- /dev/null +++ b/tests/test_client.py @@ -0,0 +1,231 @@ +from __future__ import annotations + +import json +from pathlib import Path + +from iaai_sync_service.client import ( + LISTING_MARKER, + HybridSessionAuth, + build_resizer_images_from_keys, + is_challenge_response, + parse_listing_page, + parse_product_details_vm, + resolve_listing_scope_paths, +) +from iaai_sync_service.settings import Settings + + +def _catalog_html() -> str: + return (Path(__file__).resolve().parents[1] / "html" / "catalog.html").read_text(encoding="utf-8") + + +def _page_html() -> str: + return (Path(__file__).resolve().parents[1] / "html" / "page.html").read_text(encoding="utf-8") + + +def test_parse_listing_page_extracts_hidden_payloads() -> None: + parsed = parse_listing_page(_catalog_html()) + + assert parsed.result_count >= len(parsed.vehicles) + assert parsed.page_size == 100 + assert parsed.current_page == 1 + assert parsed.vehicles + first = parsed.vehicles[0] + assert first.inventory_id.endswith("~US") + assert first.inventory_status == "RS" + + +def test_parse_product_details_vm_extracts_json_script() -> None: + payload = parse_product_details_vm(_page_html()) + inventory_view = payload["inventoryView"] + attrs = inventory_view["attributes"] + + assert attrs["Id"] == "45078011~US" + assert attrs["Make"] == "ACURA" + assert attrs["Model"] == "RSX" + + +def test_build_resizer_images_from_keys_deduplicates_and_orders() -> None: + keys = [ + {"k": "abc~1", "w": 2000, "h": 1500, "i": 2}, + {"k": "abc~1", "w": 2000, "h": 1500, "i": 2}, + {"k": "abc~2", "w": 1800, "h": 1200, "i": 1}, + ] + images = build_resizer_images_from_keys(keys) + + assert len(images) == 2 + assert images[0]["order_index"] == 1 + assert "imageKeys=abc~2" in str(images[0]["fullres_image"]) + + +def test_is_challenge_response_detects_missing_marker() -> None: + html = "

Blocked

" + assert is_challenge_response(status_code=200, body_text=html, expected_marker='id="GBPSearchQuery"') is True + + +def test_is_challenge_response_detects_incapsula_marker() -> None: + html = '' + assert is_challenge_response(status_code=200, body_text=html, expected_marker=None) is True + + +def test_is_challenge_response_accepts_expected_marker_even_with_incapsula_script() -> None: + html = f'' + assert is_challenge_response(status_code=200, body_text=html, expected_marker=LISTING_MARKER) is False + + +def test_catalog_hidden_vehicle_json_is_valid() -> None: + parsed = parse_listing_page(_catalog_html()) + # smoke-check we can serialize parsed query back to JSON + dumped = json.dumps(parsed.gbp_search_query) + assert "CurrentPage" in dumped + + +def test_resolve_listing_scope_paths_prefers_explicit_url() -> None: + paths = resolve_listing_scope_paths( + listing_start_url="https://www.iaai.com/Search?url=abc", + brands={"acura", "bmw"}, + ) + assert paths == ["https://www.iaai.com/Search?url=abc"] + + +class _FakeResponse: + def __init__(self, status_code: int, text: str) -> None: + self.status_code = status_code + self.text = text + + def close(self) -> None: + return + + +class _FakeSession: + def __init__(self, responses: list[_FakeResponse], *, cookies: list[object] | None = None) -> None: + self._responses = responses + self.calls = 0 + self.cookies = cookies or [] + + def request(self, *args, **kwargs): # noqa: ANN002, ANN003, ANN202 + self.calls += 1 + return self._responses.pop(0) + + +def _settings() -> Settings: + return Settings( + database_url="sqlite+pysqlite:///:memory:", + celery_broker_url="redis://localhost:6379/0", + celery_result_backend="redis://localhost:6379/1", + iaai_base_url="https://www.iaai.com", + iaai_listing_start_url="", + sync_runtime_config_file=Path("sync_runtime_config.json"), + iaai_session_cookies="", + iaai_storage_state_path=Path("iaai_storage_state.json"), + iaai_login="", + iaai_password="", + http_timeout=10, + http_retries=0, + http_retry_backoff_ms=1, + fetch_concurrency=1, + db_commit_batch_size=10, + session_keepalive_enabled=True, + session_keepalive_interval_minutes=15, + schema_bootstrap_enabled=False, + advisory_lock_key=1, + error_summary_max_len=2000, + ) + + +def test_hybrid_auth_retries_after_challenge(monkeypatch) -> None: # noqa: ANN001 + auth = HybridSessionAuth(_settings()) + cookie = type("Cookie", (), {"domain": ".iaai.com"})() + fake_session = _FakeSession( + [ + _FakeResponse(403, "forbidden"), + _FakeResponse(200, f"{LISTING_MARKER}"), + ], + cookies=[cookie], + ) + auth._thread_local.session = fake_session # type: ignore[attr-defined] + auth._bootstrap_cookies_loaded = True # type: ignore[attr-defined] + + refreshed = {"count": 0} + + def _fake_refresh(*, expected_marker=None, session=None) -> None: # noqa: ANN001 + refreshed["count"] += 1 + + monkeypatch.setattr(auth, "_refresh_session_via_playwright", _fake_refresh) + + response = auth.request( + "GET", + "https://www.iaai.com/Vehiclelisting/Cars", + timeout=5, + retries=0, + retry_backoff_ms=1, + expected_marker=LISTING_MARKER, + ) + assert response.status_code == 200 + assert refreshed["count"] == 1 + assert fake_session.calls == 2 + + +def test_hybrid_auth_bootstraps_anonymous_session_when_no_cookies(monkeypatch) -> None: # noqa: ANN001 + auth = HybridSessionAuth(_settings()) + fake_session = _FakeSession( + [ + _FakeResponse(200, f"{LISTING_MARKER}"), + ], + cookies=[], + ) + auth._thread_local.session = fake_session # type: ignore[attr-defined] + auth._bootstrap_cookies_loaded = True # type: ignore[attr-defined] + + refreshed = {"count": 0} + + def _fake_refresh(*, expected_marker=None, session=None) -> None: # noqa: ANN001 + refreshed["count"] += 1 + + monkeypatch.setattr(auth, "_refresh_session_via_playwright", _fake_refresh) + + response = auth.request( + "GET", + "https://www.iaai.com/Vehiclelisting/Cars", + timeout=5, + retries=0, + retry_backoff_ms=1, + expected_marker=LISTING_MARKER, + ) + assert response.status_code == 200 + assert refreshed["count"] == 1 + assert fake_session.calls == 1 + + +def test_wait_until_non_challenge_tolerates_transient_navigation_errors() -> None: + class _FakePage: + def __init__(self) -> None: + self.content_calls = 0 + self.goto_calls = 0 + + def wait_for_timeout(self, *_args, **_kwargs) -> None: # noqa: ANN002, ANN003 + return + + def wait_for_load_state(self, *_args, **_kwargs) -> None: # noqa: ANN002, ANN003 + return + + def content(self) -> str: + self.content_calls += 1 + if self.content_calls == 1: + raise RuntimeError( + "Page.content: Unable to retrieve content because the page is navigating and changing the content." + ) + return f"{LISTING_MARKER}" + + def goto(self, *_args, **_kwargs) -> None: # noqa: ANN002, ANN003 + self.goto_calls += 1 + + page = _FakePage() + HybridSessionAuth._wait_until_non_challenge( + page=page, + target="https://www.iaai.com/Vehiclelisting/Cars", + timeout_ms=5_000, + expected_marker=LISTING_MARKER, + ) + assert page.content_calls >= 2 + assert page.goto_calls == 0 diff --git a/tests/test_migrations.py b/tests/test_migrations.py new file mode 100644 index 0000000..ab4910d --- /dev/null +++ b/tests/test_migrations.py @@ -0,0 +1,36 @@ +from __future__ import annotations + +import importlib.util +from pathlib import Path + + +def _load_module(path: Path): # noqa: ANN202 + spec = importlib.util.spec_from_file_location(path.stem, path) + assert spec is not None + assert spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def test_revision_chain_is_correct() -> None: + base = Path(__file__).resolve().parents[1] / "alembic" / "versions" + rev1 = _load_module(base / "0001_ensure_sync_schema.py") + rev2 = _load_module(base / "0002_extend_origin_enum_for_iaai.py") + rev3 = _load_module(base / "0003_create_iaai_source_tables.py") + + assert rev1.revision == "0001_ensure_sync_schema" + assert rev1.down_revision is None + assert rev2.revision == "0002_extend_origin_enum_for_iaai" + assert rev2.down_revision == "0001_ensure_sync_schema" + assert rev3.revision == "0003_create_iaai_source_tables" + assert rev3.down_revision == "0002_extend_origin_enum_for_iaai" + + +def test_enum_migration_contains_required_values() -> None: + migration = ( + Path(__file__).resolve().parents[1] / "alembic" / "versions" / "0002_extend_origin_enum_for_iaai.py" + ).read_text(encoding="utf-8") + assert "originenum" in migration + assert "IAAI" in migration + assert "ADD VALUE IF NOT EXISTS" in migration diff --git a/tests/test_runtime_config.py b/tests/test_runtime_config.py new file mode 100644 index 0000000..9ed5856 --- /dev/null +++ b/tests/test_runtime_config.py @@ -0,0 +1,40 @@ +from __future__ import annotations + +import json + +from iaai_sync_service.runtime_config import RuntimeConfig, load_runtime_config + + +def test_load_runtime_config_reads_filters(tmp_path) -> None: # noqa: ANN001 + config_path = tmp_path / "sync_runtime_config.json" + config_path.write_text( + json.dumps( + { + "sync": { + "condition_check_enabled": "true", + }, + "filters": { + "brands": ["Acura", "BMW"], + "models": ["RSX", "3 SERIES"], + "years": [2021, "2022"], + "body_types": ["SUV"], + }, + } + ), + encoding="utf-8", + ) + + runtime = load_runtime_config(config_path) + assert runtime.condition_check_enabled is True + assert runtime.filters.brands == {"acura", "bmw"} + assert runtime.filters.models == {"rsx", "3 series"} + assert runtime.filters.years == {2021, 2022} + assert runtime.filters.body_types == {"suv"} + + +def test_load_runtime_config_returns_defaults_on_invalid_json(tmp_path) -> None: # noqa: ANN001 + config_path = tmp_path / "sync_runtime_config.json" + config_path.write_text("{invalid", encoding="utf-8") + + runtime = load_runtime_config(config_path) + assert runtime == RuntimeConfig() diff --git a/tests/test_settings.py b/tests/test_settings.py new file mode 100644 index 0000000..32eb684 --- /dev/null +++ b/tests/test_settings.py @@ -0,0 +1,66 @@ +from __future__ import annotations + +import pytest + +from iaai_sync_service.settings import Settings + + +def test_settings_from_env_reads_required_values(monkeypatch) -> None: # noqa: ANN001 + monkeypatch.setenv("DATABASE_URL", "postgres://user:pass@localhost:5432/db") + monkeypatch.setenv("CELERY_BROKER_URL", "redis://localhost:6379/0") + monkeypatch.setenv("IAAI_BASE_URL", "https://www.iaai.com") + monkeypatch.setenv("IAAI_LISTING_START_URL", "https://www.iaai.com/Search?url=abc") + monkeypatch.setenv("IAAI_FETCH_CONCURRENCY", "4") + monkeypatch.setenv("IAAI_SESSION_KEEPALIVE_ENABLED", "false") + monkeypatch.setenv("IAAI_SESSION_KEEPALIVE_INTERVAL_MINUTES", "20") + monkeypatch.setenv("IAAI_SCHEMA_BOOTSTRAP_ENABLED", "true") + + settings = Settings.from_env() + + assert settings.database_url == "postgresql+psycopg://user:pass@localhost:5432/db" + assert settings.celery_broker_url == "redis://localhost:6379/0" + assert settings.iaai_base_url == "https://www.iaai.com" + assert settings.iaai_listing_start_url == "https://www.iaai.com/Search?url=abc" + assert settings.fetch_concurrency == 4 + assert settings.session_keepalive_enabled is False + assert settings.session_keepalive_interval_minutes == 20 + assert settings.schema_bootstrap_enabled is True + + +def test_settings_rejects_non_positive_concurrency(monkeypatch) -> None: # noqa: ANN001 + monkeypatch.setenv("DATABASE_URL", "postgresql://user:pass@localhost:5432/db") + monkeypatch.setenv("CELERY_BROKER_URL", "redis://localhost:6379/0") + monkeypatch.setenv("IAAI_BASE_URL", "https://www.iaai.com") + monkeypatch.setenv("IAAI_FETCH_CONCURRENCY", "0") + + with pytest.raises(RuntimeError, match="IAAI_FETCH_CONCURRENCY"): + Settings.from_env() + + +def test_settings_bootstrap_flag_defaults_to_false(monkeypatch) -> None: # noqa: ANN001 + monkeypatch.setenv("DATABASE_URL", "postgresql://user:pass@localhost:5432/db") + monkeypatch.setenv("CELERY_BROKER_URL", "redis://localhost:6379/0") + monkeypatch.setenv("IAAI_BASE_URL", "https://www.iaai.com") + + settings = Settings.from_env() + assert settings.schema_bootstrap_enabled is False + + +def test_settings_rejects_invalid_listing_start_url(monkeypatch) -> None: # noqa: ANN001 + monkeypatch.setenv("DATABASE_URL", "postgresql://user:pass@localhost:5432/db") + monkeypatch.setenv("CELERY_BROKER_URL", "redis://localhost:6379/0") + monkeypatch.setenv("IAAI_BASE_URL", "https://www.iaai.com") + monkeypatch.setenv("IAAI_LISTING_START_URL", "Search?url=abc") + + with pytest.raises(RuntimeError, match="IAAI_LISTING_START_URL"): + Settings.from_env() + + +def test_settings_rejects_invalid_keepalive_interval(monkeypatch) -> None: # noqa: ANN001 + monkeypatch.setenv("DATABASE_URL", "postgresql://user:pass@localhost:5432/db") + monkeypatch.setenv("CELERY_BROKER_URL", "redis://localhost:6379/0") + monkeypatch.setenv("IAAI_BASE_URL", "https://www.iaai.com") + monkeypatch.setenv("IAAI_SESSION_KEEPALIVE_INTERVAL_MINUTES", "0") + + with pytest.raises(RuntimeError, match="IAAI_SESSION_KEEPALIVE_INTERVAL_MINUTES"): + Settings.from_env() diff --git a/tests/test_sync_engine.py b/tests/test_sync_engine.py new file mode 100644 index 0000000..1d18a5b --- /dev/null +++ b/tests/test_sync_engine.py @@ -0,0 +1,288 @@ +from __future__ import annotations + +from datetime import UTC, datetime +from pathlib import Path + +from sqlalchemy import create_engine +from sqlalchemy.orm import Session + +from iaai_sync_service.client import ListingVehicle +from iaai_sync_service.models import Base, Car, Image +from iaai_sync_service.runtime_config import RuntimeConfig, RuntimeFilters +from iaai_sync_service.settings import Settings +from iaai_sync_service.sync_engine import ( + prepare_car, + replace_images_if_changed, + run_sync_once, +) + + +def _session() -> Session: + engine = create_engine("sqlite+pysqlite:///:memory:", future=True) + Base.metadata.create_all(engine) + return Session(bind=engine) + + +def _settings() -> Settings: + return Settings( + database_url="sqlite+pysqlite:///:memory:", + celery_broker_url="redis://localhost:6379/0", + celery_result_backend="redis://localhost:6379/1", + iaai_base_url="https://www.iaai.com", + iaai_listing_start_url="", + sync_runtime_config_file=Path("sync_runtime_config.json"), + iaai_session_cookies="", + iaai_storage_state_path=Path("iaai_storage_state.json"), + iaai_login="", + iaai_password="", + http_timeout=10, + http_retries=1, + http_retry_backoff_ms=1, + fetch_concurrency=2, + db_commit_batch_size=20, + session_keepalive_enabled=True, + session_keepalive_interval_minutes=15, + schema_bootstrap_enabled=False, + advisory_lock_key=1, + error_summary_max_len=2000, + ) + + +def _listing_vehicle(inventory_id: str, *, status: str = "RS", closed: bool = False) -> ListingVehicle: + return ListingVehicle( + inventory_id=inventory_id, + tenant="US", + auction_id="1_1", + auction_date="2026-04-08T08:30:00+00:00", + inventory_status=status, + currency="USD", + timed_auction_closed=closed, + timed_auction_indicator=False, + prebid_indicator=True, + buynow_indicator=False, + ) + + +def _detail_payload( + inventory_id: str, + *, + make: str = "ACURA", + model: str = "RSX", + series: str = "BASE", + year: int = 2002, + bid: int = 500, +) -> dict: + return { + "inventoryView": { + "attributes": { + "Id": inventory_id, + "Year": str(year), + "Make": make, + "Model": model, + "Series": series, + "Currency": "USD", + "ODOValue": "219187", + "ExteriorColor": "GRAY", + "DriveLineTypeDesc": "FWD", + "Transmission": "Automatic", + "BodyStyleName": "COUPE", + "EngineSize": "2.0L I-4", + "VehicleGrade": "50", + "PrimaryDamageDesc": "NORMAL WEAR & TEAR", + "SecondaryDamageDesc": " ", + }, + "imageDimensions": { + "keys": { + "$values": [ + {"k": f"{inventory_id}~I1", "w": 1600, "h": 1200, "i": 0}, + {"k": f"{inventory_id}~I2", "w": 1600, "h": 1200, "i": 1}, + ] + } + }, + }, + "auctionInformation": { + "biddingInformation": { + "buyNowAmount": 0, + "buyNowPrice": "$0", + }, + "prebidInformation": { + "decimalHighBidAmount": str(bid), + "highBidAmount": f"${bid}", + "buyNowPrice": "$0", + }, + }, + } + + +class FakeClient: + def __init__(self, listing: list[ListingVehicle], detail_payloads: dict[str, dict]) -> None: + self._listing = listing + self._detail_payloads = detail_payloads + self.detail_calls = 0 + self.persist_calls = 0 + + def iter_listing_vehicles(self, filters: RuntimeFilters): # noqa: ANN201, ARG002 + return iter(self._listing) + + def fetch_vehicle_detail_payload(self, inventory_id: str): # noqa: ANN201 + self.detail_calls += 1 + return self._detail_payloads[inventory_id] + + def persist_session_state(self) -> None: + self.persist_calls += 1 + + +def test_prepare_car_maps_fields() -> None: + row = prepare_car( + listing_vehicle=_listing_vehicle("45078011~US"), + detail_payload=_detail_payload("45078011~US", bid=600), + currency_labels={"USD", "CAD", "NA"}, + country_labels={"US", "CA", "NA"}, + drive_labels={"FWD", "RWD", "4WD", "NA"}, + gearbox_labels={"AT", "MT", "NA"}, + body_type_labels={"COUPE", "OTHER"}, + body_type_default="OTHER", + steering_left="LEFT", + selling_type_auction="AUCTION", + origin_code="IAAI", + ) + + assert row.origin_id == "iaai:45078011~US" + assert row.origin_url == "https://www.iaai.com/VehicleDetail/45078011~US" + assert row.brand == "ACURA" + assert row.model == "RSX BASE" + assert row.year == 2002 + assert row.price == 600 + assert row.currency == "USD" + assert row.country == "US" + assert row.drive == "FWD" + assert row.gearbox == "AT" + assert row.body_type == "COUPE" + assert row.engine_volume == 2000 + assert row.is_damaged is False + assert len(row.images) == 2 + + +def test_replace_images_if_changed_returns_zero_for_same_set() -> None: + session = _session() + car = Car( + parser_id="car-aaaaaaaaaaaaaaaaaaaaaa", + brand="ACURA", + model="RSX", + origin_id="iaai:45078011~US", + origin_url="https://www.iaai.com/VehicleDetail/45078011~US", + slug="acura-rsx-2002", + is_sold=False, + last_seen_at=datetime(2026, 4, 1, 12, 0, tzinfo=UTC), + ) + session.add(car) + session.flush() + session.add( + Image( + car_id=car.id, + fullres_image="https://vis.iaai.com/resizer?imageKeys=a&width=1600&height=1200", + preview_image="https://vis.iaai.com/resizer?imageKeys=a&width=640&height=480", + order_index=0, + ) + ) + session.commit() + + written = replace_images_if_changed( + session=session, + car_id=car.id, + images=[ + { + "order_index": 0, + "fullres_image": "https://vis.iaai.com/resizer?imageKeys=a&width=1600&height=1200", + "preview_image": "https://vis.iaai.com/resizer?imageKeys=a&width=640&height=480", + } + ], + ) + + assert written == 0 + + +def test_run_sync_once_upserts_and_marks_missing_as_sold() -> None: + session = _session() + + listing_first = [_listing_vehicle("45078011~US"), _listing_vehicle("45268167~US")] + detail_first = { + "45078011~US": _detail_payload("45078011~US", make="ACURA", model="RSX", bid=500), + "45268167~US": _detail_payload("45268167~US", make="BMW", model="X5", bid=900), + } + client_first = FakeClient(listing_first, detail_first) + + stats1, errors1 = run_sync_once(session=session, client=client_first, settings=_settings()) + assert errors1 == [] + assert stats1.ids_fetched == 2 + assert stats1.cars_upserted == 2 + assert client_first.persist_calls == 1 + + listing_second = [_listing_vehicle("45078011~US")] + detail_second = { + "45078011~US": _detail_payload("45078011~US", make="ACURA", model="RSX", bid=650), + } + client_second = FakeClient(listing_second, detail_second) + + stats2, errors2 = run_sync_once(session=session, client=client_second, settings=_settings()) + assert errors2 == [] + assert stats2.ids_fetched == 1 + assert stats2.cars_upserted == 1 + + active = session.query(Car).filter_by(origin_id="iaai:45078011~US").one() + sold = session.query(Car).filter_by(origin_id="iaai:45268167~US").one() + assert active.is_sold is False + assert active.price == 650 + assert sold.is_sold is True + + +def test_run_sync_once_skips_reconcile_when_runtime_filters_enabled() -> None: + session = _session() + session.add( + Car( + parser_id="car-zzzzzzzzzzzzzzzzzzzzzz", + brand="BMW", + model="X5", + origin_id="iaai:99999999~US", + origin_url="https://www.iaai.com/VehicleDetail/99999999~US", + slug="bmw-x5-2020", + is_sold=False, + last_seen_at=datetime(2026, 4, 1, 12, 0, tzinfo=UTC), + ) + ) + session.commit() + + listing = [_listing_vehicle("45078011~US")] + details = {"45078011~US": _detail_payload("45078011~US", make="ACURA", model="RSX", bid=500)} + runtime = RuntimeConfig(filters=RuntimeFilters(brands={"bmw"})) + + stats, errors = run_sync_once( + session=session, + client=FakeClient(listing, details), + settings=_settings(), + runtime_config=runtime, + ) + + assert errors == [] + assert stats.ids_fetched == 0 + + row = session.query(Car).filter_by(origin_id="iaai:99999999~US").one() + assert row.is_sold is False + + +def test_run_sync_once_condition_check_skips_closed_rows() -> None: + session = _session() + listing = [_listing_vehicle("45078011~US", status="SOLD", closed=True)] + details = {"45078011~US": _detail_payload("45078011~US", bid=500)} + + runtime = RuntimeConfig(condition_check_enabled=True) + stats, errors = run_sync_once( + session=session, + client=FakeClient(listing, details), + settings=_settings(), + runtime_config=runtime, + ) + + assert errors == [] + assert stats.ids_fetched == 0 + assert session.query(Car).count() == 0 diff --git a/tests/test_tasks.py b/tests/test_tasks.py new file mode 100644 index 0000000..aca1895 --- /dev/null +++ b/tests/test_tasks.py @@ -0,0 +1,167 @@ +from __future__ import annotations + +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from iaai_sync_service.models import SyncRun +from iaai_sync_service.sync_engine import SyncStats +from iaai_sync_service.tasks import execute_keepalive_job, execute_sync_job + + +class _ScalarResult: + def __init__(self, value): # noqa: ANN001 + self._value = value + + def scalar_one(self): # noqa: ANN201 + return self._value + + +class FakeSession: + def __init__(self, *, lock_acquired: bool) -> None: + self.lock_acquired = lock_acquired + self.runs: dict[int, SyncRun] = {} + self.closed = False + + def execute(self, stmt, params=None): # noqa: ANN001, ANN201 + sql = str(stmt) + if "pg_try_advisory_lock" in sql: + return _ScalarResult(self.lock_acquired) + if "pg_advisory_unlock" in sql: + return _ScalarResult(True) + raise AssertionError(f"Unexpected SQL execute call: {sql}") + + def add(self, obj) -> None: # noqa: ANN001 + if isinstance(obj, SyncRun): + if obj.id is None: + obj.id = len(self.runs) + 1 + self.runs[obj.id] = obj + + def commit(self) -> None: + return + + def rollback(self) -> None: + return + + def get(self, model, run_id): # noqa: ANN001, ANN201 + assert model is SyncRun + return self.runs.get(run_id) + + def close(self) -> None: + self.closed = True + + +def _settings() -> SimpleNamespace: + return SimpleNamespace( + iaai_base_url="https://www.iaai.com", + sync_runtime_config_file=Path("sync_runtime_config.json"), + schema_bootstrap_enabled=True, + advisory_lock_key=123, + error_summary_max_len=2000, + ) + + +def test_execute_sync_job_skips_when_lock_not_available(monkeypatch) -> None: # noqa: ANN001 + session = FakeSession(lock_acquired=False) + monkeypatch.setattr("iaai_sync_service.tasks.get_settings", _settings) + monkeypatch.setattr("iaai_sync_service.tasks.get_session", lambda: session) + monkeypatch.setattr("iaai_sync_service.tasks.ensure_schema", lambda _session: None) + + result = execute_sync_job() + + assert result == {"status": "skipped_locked"} + assert session.closed is True + + +def test_execute_sync_job_success_writes_sync_run(monkeypatch) -> None: # noqa: ANN001 + session = FakeSession(lock_acquired=True) + monkeypatch.setattr("iaai_sync_service.tasks.get_settings", _settings) + monkeypatch.setattr("iaai_sync_service.tasks.get_session", lambda: session) + monkeypatch.setattr("iaai_sync_service.tasks.ensure_schema", lambda _session: None) + monkeypatch.setattr( + "iaai_sync_service.tasks.run_sync_once", + lambda **kwargs: (SyncStats(ids_fetched=5, cars_upserted=4, cars_failed=0, images_upserted=12), []), + ) + + result = execute_sync_job() + + assert result["status"] == "success" + assert result["ids_fetched"] == 5 + run = session.runs[result["run_id"]] + assert run.status == "success" + assert run.cars_upserted == 4 + + +def test_execute_sync_job_marks_failed_when_sync_raises(monkeypatch) -> None: # noqa: ANN001 + session = FakeSession(lock_acquired=True) + monkeypatch.setattr("iaai_sync_service.tasks.get_settings", _settings) + monkeypatch.setattr("iaai_sync_service.tasks.get_session", lambda: session) + monkeypatch.setattr("iaai_sync_service.tasks.ensure_schema", lambda _session: None) + + def _raise(**kwargs): # noqa: ANN003, ANN202 + raise RuntimeError("boom") + + monkeypatch.setattr("iaai_sync_service.tasks.run_sync_once", _raise) + + with pytest.raises(RuntimeError, match="boom"): + execute_sync_job() + + failed = session.runs[1] + assert failed.status == "failed" + assert failed.error_summary is not None + + +def test_execute_sync_job_skips_schema_bootstrap_when_disabled(monkeypatch) -> None: # noqa: ANN001 + session = FakeSession(lock_acquired=False) + settings = _settings() + settings.schema_bootstrap_enabled = False + + monkeypatch.setattr("iaai_sync_service.tasks.get_settings", lambda: settings) + monkeypatch.setattr("iaai_sync_service.tasks.get_session", lambda: session) + + def _fail_if_called(_session): # noqa: ANN001, ANN202 + raise AssertionError("ensure_schema should not be called") + + monkeypatch.setattr("iaai_sync_service.tasks.ensure_schema", _fail_if_called) + + result = execute_sync_job() + assert result == {"status": "skipped_locked"} + + +def test_execute_keepalive_job_skips_when_lock_not_available(monkeypatch) -> None: # noqa: ANN001 + session = FakeSession(lock_acquired=False) + monkeypatch.setattr("iaai_sync_service.tasks.get_settings", _settings) + monkeypatch.setattr("iaai_sync_service.tasks.get_session", lambda: session) + + result = execute_keepalive_job() + + assert result == {"status": "skipped_locked"} + assert session.closed is True + + +def test_execute_keepalive_job_success(monkeypatch) -> None: # noqa: ANN001 + session = FakeSession(lock_acquired=True) + monkeypatch.setattr("iaai_sync_service.tasks.get_settings", _settings) + monkeypatch.setattr("iaai_sync_service.tasks.get_session", lambda: session) + + class _FakeClient: + def __init__(self, settings): # noqa: ANN001 + self._settings = settings + + def keep_session_alive(self): # noqa: ANN201 + return { + "scope": "https://www.iaai.com/Search?url=abc", + "marker_present": True, + "result_count": 1234, + "elapsed_ms": 456, + } + + monkeypatch.setattr("iaai_sync_service.tasks.IAAIClient", _FakeClient) + + result = execute_keepalive_job() + + assert result["status"] == "success" + assert result["marker_present"] is True + assert result["result_count"] == 1234 + assert session.closed is True