add encar scraper

2026-04-16 18:01:05 +03:00
parent 5624c1973a
commit 4096d29a07
28 changed files with 3472 additions and 150 deletions
--- a/encar_scraper/init.py
+++ b/encar_scraper/init.py
@@ -0,0 +1 @@
+__all__: list[str] = []
--- a/encar_scraper/api/init.py
+++ b/encar_scraper/api/init.py
@@ -0,0 +1,3 @@
+from .app import create_app
+
+__all__ = ["create_app"]
--- a/encar_scraper/api/app.py
+++ b/encar_scraper/api/app.py
@@ -0,0 +1,40 @@
+# Создание FastAPI-приложения и настройка его жизненного цикла.
+
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI
+
+from ..core.config import Settings
+from ..storage.db import PersistenceService
+from .routes import cars, health, tasks
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    # Жизненный цикл.
+    # Таблицы создаются через Alembic-миграции (сервис migrate).
+    yield
+
+
+def create_app(settings: Settings | None = None) -> FastAPI:
+    _settings = settings or Settings()
+
+    app = FastAPI(
+        title="Encar Scraper API",
+        description="REST API для управления задачами скрапинга Encar и просмотра данных",
+        version="1.0.0",
+        lifespan=lifespan,
+    )
+
+    app.state.settings = _settings
+    app.state.persistence = PersistenceService(_settings)
+
+    app.include_router(health.router, tags=["health"])
+    app.include_router(cars.router, prefix="/api/v1", tags=["cars"])
+    app.include_router(tasks.router, prefix="/api/v1", tags=["tasks"])
+
+    return app
+
+
+# Экземпляр приложения для запуска через uvicorn.
+app = create_app()
--- a/encar_scraper/api/deps.py
+++ b/encar_scraper/api/deps.py
@@ -0,0 +1,9 @@
+# Вспомогательные зависимости для FastAPI-роутов.
+
+from fastapi import Request
+
+from ..storage.db import PersistenceService
+
+
+def get_persistence(request: Request) -> PersistenceService:
+    return request.app.state.persistence
--- a/encar_scraper/api/routes/init.py
+++ b/encar_scraper/api/routes/init.py
--- a/encar_scraper/api/routes/cars.py
+++ b/encar_scraper/api/routes/cars.py
@@ -0,0 +1,104 @@
+# Роуты для просмотра автомобилей и агрегированной статистики.
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+from sqlalchemy import func, select
+from sqlalchemy.orm import selectinload
+
+from ..deps import get_persistence
+from ...storage.db import PersistenceService
+from ...storage.models import Car, Image
+from ...storage.schemas import CarRead
+
+router = APIRouter()
+
+
+@router.get("/cars")
+def list_cars(
+    page: int = Query(1, ge=1),
+    per_page: int = Query(20, ge=1, le=100),
+    brand: str | None = None,
+    model: str | None = None,
+    year_min: int | None = None,
+    year_max: int | None = None,
+    is_sold: bool | None = None,
+    persistence: PersistenceService = Depends(get_persistence),
+):
+    # Список автомобилей с пагинацией и фильтрами
+    with persistence.session_scope() as session:
+        query = select(Car)
+
+        if brand:
+            query = query.where(Car.brand.ilike(f"%{brand}%"))
+        if model:
+            query = query.where(Car.model.ilike(f"%{model}%"))
+        if year_min is not None:
+            query = query.where(Car.year >= year_min)
+        if year_max is not None:
+            query = query.where(Car.year <= year_max)
+        if is_sold is not None:
+            query = query.where(Car.is_sold == is_sold)
+
+        count_query = select(func.count()).select_from(query.subquery())
+        total = session.execute(count_query).scalar() or 0
+
+        offset = (page - 1) * per_page
+        cars = session.execute(
+            query.options(selectinload(Car.images)).order_by(Car.last_seen_at.desc()).offset(offset).limit(per_page)
+        ).scalars().all()
+
+        return {
+            "total": total,
+            "page": page,
+            "per_page": per_page,
+            "pages": (total + per_page - 1) // per_page if per_page else 0,
+            "items": [CarRead.model_validate(car).model_dump(mode="json") for car in cars],
+        }
+
+
+@router.get("/cars/{car_id}")
+def get_car(
+    car_id: int,
+    persistence: PersistenceService = Depends(get_persistence),
+):
+    # Детальная информация об автомобиле с изображениями
+    with persistence.session_scope() as session:
+        car = session.get(Car, car_id)
+        if car is None:
+            raise HTTPException(status_code=404, detail="Car not found")
+        return CarRead.model_validate(car).model_dump(mode="json")
+
+
+@router.get("/cars/by-origin/{origin_id}")
+def get_car_by_origin(
+    origin_id: str,
+    persistence: PersistenceService = Depends(get_persistence),
+):
+    # Поиск автомобиля по origin_id
+    with persistence.session_scope() as session:
+        car = session.execute(
+            select(Car).where(Car.origin_id == origin_id)
+        ).scalars().first()
+        if car is None:
+            raise HTTPException(status_code=404, detail="Car not found")
+        return CarRead.model_validate(car).model_dump(mode="json")
+
+
+@router.get("/stats")
+def get_stats(persistence: PersistenceService = Depends(get_persistence)):
+    # Общая статистика по БД
+    with persistence.session_scope() as session:
+        total_cars = session.execute(select(func.count(Car.id))).scalar() or 0
+        total_images = session.execute(select(func.count(Image.id))).scalar() or 0
+        brands = session.execute(
+            select(Car.brand, func.count(Car.id))
+            .group_by(Car.brand)
+            .order_by(func.count(Car.id).desc())
+            .limit(20)
+        ).all()
+
+        return {
+            "total_cars": total_cars,
+            "total_images": total_images,
+            "top_brands": [{"brand": b, "count": c} for b, c in brands],
+        }
+
--- a/encar_scraper/api/routes/health.py
+++ b/encar_scraper/api/routes/health.py
@@ -0,0 +1,30 @@
+# Роут проверки доступности сервиса и соединения с БД.
+
+import logging
+
+from fastapi import APIRouter, Depends
+from sqlalchemy import text
+
+from ..deps import get_persistence
+from ...storage.db import PersistenceService
+
+router = APIRouter()
+logger = logging.getLogger("encar_scraper.api.health")
+
+
+@router.get("/health")
+def health_check(persistence: PersistenceService = Depends(get_persistence)):
+    # Проверка API и БД
+    db_ok = False
+    try:
+        with persistence.session_scope() as session:
+            session.execute(text("SELECT 1"))
+            db_ok = True
+    except Exception:
+        logger.warning("Health DB check failed", exc_info=True)
+
+    return {
+        "status": "ok" if db_ok else "degraded",
+        "service": "encar-scraper-api",
+        "database": "connected" if db_ok else "unavailable",
+    }
--- a/encar_scraper/api/routes/tasks.py
+++ b/encar_scraper/api/routes/tasks.py
@@ -0,0 +1,121 @@
+# Роуты запуска задач синхронизации Encar и просмотра истории sync-runs.
+
+from fastapi import APIRouter, Depends, Query
+from pydantic import BaseModel
+from sqlalchemy import select, func
+
+from ..deps import get_persistence
+from ...storage.db import PersistenceService
+from ...storage.models import SyncRun
+from ...worker.celery_app import celery_app
+from ...worker.tasks import encar_sync_listing_task, encar_sync_vehicle_task
+
+router = APIRouter()
+
+
+class SyncVehicleRequest(BaseModel):
+    vehicle_url: str
+    lane: str = "encar"
+
+
+class SyncListingRequest(BaseModel):
+    car_type: str = "all"  # all, domestic, import
+    manufacturer: str | None = None
+    year_from: int | None = None
+    year_to: int | None = None
+    price_max: int | None = None
+    lane: str = "encar"
+    limit: int | None = None
+
+
+@router.post("/tasks/sync-vehicle")
+def start_sync_vehicle(body: SyncVehicleRequest):
+    """Синхронизация одного авто Encar."""
+    result = encar_sync_vehicle_task.apply_async(
+        kwargs={"vehicle_url": body.vehicle_url, "lane": body.lane},
+        queue="encar",
+    )
+    return {
+        "task_id": result.id,
+        "status": "queued",
+        "vehicle_url": body.vehicle_url,
+    }
+
+
+@router.post("/tasks/sync-listing")
+def start_sync_listing(body: SyncListingRequest):
+    """Синхронизация листинга Encar через публичный API."""
+    result = encar_sync_listing_task.apply_async(
+        kwargs={
+            "car_type": body.car_type,
+            "manufacturer": body.manufacturer,
+            "year_from": body.year_from,
+            "year_to": body.year_to,
+            "price_max": body.price_max,
+            "lane": body.lane,
+            "limit": body.limit,
+        },
+        queue="encar",
+    )
+    return {
+        "task_id": result.id,
+        "status": "queued",
+        "car_type": body.car_type,
+        "manufacturer": body.manufacturer,
+    }
+
+
+@router.get("/tasks/{task_id}")
+def get_task_status(task_id: str):
+    """Статус задачи Celery."""
+    result = celery_app.AsyncResult(task_id)
+
+    payload: dict = {
+        "task_id": task_id,
+        "state": result.state,
+    }
+
+    if result.successful():
+        payload["result"] = result.result
+    elif result.failed():
+        payload["error"] = str(result.result)
+    elif result.info is not None:
+        payload["meta"] = result.info
+
+    return payload
+
+
+@router.get("/sync-runs")
+def list_sync_runs(
+    page: int = Query(1, ge=1),
+    per_page: int = Query(20, ge=1, le=100),
+    persistence: PersistenceService = Depends(get_persistence),
+):
+    """История запусков синхронизации."""
+    with persistence.session_scope() as session:
+        total = session.execute(select(func.count(SyncRun.id))).scalar() or 0
+        offset = (page - 1) * per_page
+        runs = session.execute(
+            select(SyncRun).order_by(SyncRun.started_at.desc()).offset(offset).limit(per_page)
+        ).scalars().all()
+
+        return {
+            "total": total,
+            "page": page,
+            "per_page": per_page,
+            "items": [
+                {
+                    "id": run.id,
+                    "started_at": run.started_at.isoformat() if run.started_at else None,
+                    "finished_at": run.finished_at.isoformat() if run.finished_at else None,
+                    "status": run.status,
+                    "lane": run.lane,
+                    "ids_fetched": run.ids_fetched,
+                    "cars_upserted": run.cars_upserted,
+                    "cars_failed": run.cars_failed,
+                    "images_upserted": run.images_upserted,
+                    "error_summary": run.error_summary,
+                }
+                for run in runs
+            ],
+        }
--- a/encar_scraper/cli.py
+++ b/encar_scraper/cli.py
@@ -0,0 +1,103 @@
+import argparse
+from pathlib import Path
+
+from .core.config import Settings
+from .core.utils import save_to_json
+from .encar import EncarScraper, EncarFilters, CAR_TYPE_ALL, CAR_TYPE_DOMESTIC, CAR_TYPE_IMPORT
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description="Encar scraper CLI")
+    subparsers = parser.add_subparsers(dest="command", required=True)
+
+    default_output_dir = Path("artifacts/json")
+
+    subparsers.add_parser("init-db", help="Create DB tables")
+
+    # collect-listing
+    collect_parser = subparsers.add_parser("collect-listing", help="Collect Encar listing")
+    collect_parser.add_argument("--limit", type=int, default=None, help="Max items to collect")
+    collect_parser.add_argument("--car-type", choices=["all", "domestic", "import"], default="all",
+                                help="all=все, domestic=корейские, import=импорт")
+    collect_parser.add_argument("--manufacturer", default=None, help="Filter by manufacturer (e.g. BMW, 현대)")
+    collect_parser.add_argument("--year-from", type=int, default=None, help="Min year")
+    collect_parser.add_argument("--year-to", type=int, default=None, help="Max year")
+    collect_parser.add_argument("--price-max", type=int, default=None, help="Max price in 만원 (10k KRW)")
+    collect_parser.add_argument("--output", default=str(default_output_dir / "encar_listing.json"))
+
+    # scrape-vehicle
+    scrape_parser = subparsers.add_parser("scrape-vehicle", help="Fetch single vehicle data")
+    scrape_parser.add_argument("vehicle_url")
+    scrape_parser.add_argument("--output", default=str(default_output_dir / "encar_vehicle_detail.json"))
+
+    # sync-vehicle
+    sync_parser = subparsers.add_parser("sync-vehicle", help="Sync single vehicle into DB")
+    sync_parser.add_argument("vehicle_url")
+    sync_parser.add_argument("--lane", default="encar")
+    sync_parser.add_argument("--output", default=str(default_output_dir / "encar_sync_vehicle.json"))
+
+    # sync-listing
+    sync_listing_parser = subparsers.add_parser("sync-listing", help="Collect listing and sync all to DB")
+    sync_listing_parser.add_argument("--limit", type=int, default=None, help="Max items to sync")
+    sync_listing_parser.add_argument("--car-type", choices=["all", "domestic", "import"], default="all")
+    sync_listing_parser.add_argument("--manufacturer", default=None, help="Filter by manufacturer")
+    sync_listing_parser.add_argument("--year-from", type=int, default=None)
+    sync_listing_parser.add_argument("--year-to", type=int, default=None)
+    sync_listing_parser.add_argument("--price-max", type=int, default=None)
+    sync_listing_parser.add_argument("--lane", default="encar")
+    sync_listing_parser.add_argument("--only-new", choices=["true", "false"], default="true")
+    sync_listing_parser.add_argument("--output", default=str(default_output_dir / "encar_sync_listing.json"))
+
+    return parser
+
+
+def _build_filters_from_args(args) -> EncarFilters:
+    """Строит фильтры из аргументов CLI."""
+    car_type_map = {"all": CAR_TYPE_ALL, "domestic": CAR_TYPE_DOMESTIC, "import": CAR_TYPE_IMPORT}
+    return EncarFilters(
+        car_type=car_type_map.get(getattr(args, "car_type", "all"), CAR_TYPE_ALL),
+        manufacturer=getattr(args, "manufacturer", None),
+        year_from=getattr(args, "year_from", None),
+        year_to=getattr(args, "year_to", None),
+        price_to=getattr(args, "price_max", None),
+    )
+
+
+def main() -> None:
+    import sys
+    import logging
+
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+        stream=sys.stderr,
+    )
+
+    parser = build_parser()
+    args = parser.parse_args()
+    scraper = EncarScraper()
+
+    if args.command == "init-db":
+        result = scraper.init_db()
+        print(f"DB initialized: {result}")
+        return
+    elif args.command == "collect-listing":
+        filters = _build_filters_from_args(args)
+        result = scraper.collect_listing(limit=args.limit, filters=filters)
+    elif args.command == "scrape-vehicle":
+        result = scraper.scrape_vehicle_detail(args.vehicle_url)
+    elif args.command == "sync-vehicle":
+        result = scraper.sync_vehicle(args.vehicle_url, lane=args.lane)
+    elif args.command == "sync-listing":
+        filters = _build_filters_from_args(args)
+        only_new = args.only_new == "true"
+        result = scraper.sync_listing(limit=args.limit, filters=filters, lane=args.lane, only_new=only_new)
+    else:
+        raise SystemExit(f"Unknown command: {args.command}")
+
+    save_to_json(result, Path(args.output))
+    print(f"Saved to {Path(args.output).resolve()}")
+
+
+if __name__ == "__main__":
+    main()
--- a/encar_scraper/core/init.py
+++ b/encar_scraper/core/init.py
@@ -0,0 +1 @@
+__all__: list[str] = []
--- a/encar_scraper/core/config.py
+++ b/encar_scraper/core/config.py
@@ -0,0 +1,106 @@
+import os
+from dataclasses import dataclass, field
+from pathlib import Path
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+
+TRUE_VALUES = {"1", "true", "yes", "on"}
+
+
+#  Хелперы для чтения env-переменных с приведением типов 
+
+def _env_str(name: str, default: str) -> str:
+    value = os.getenv(name)
+    return value if value is not None else default
+
+
+def _env_optional_str(name: str) -> str | None:
+    value = os.getenv(name)
+    if value is None:
+        return None
+    value = value.strip()
+    return value or None
+
+
+def _env_path_str(name: str) -> str | None:
+    value = _env_optional_str(name)
+    if value is None:
+        return None
+    return str(Path(value).expanduser())
+
+
+def _env_bool(name: str, default: bool) -> bool:
+    fallback = "true" if default else "false"
+    return _env_str(name, fallback).strip().lower() in TRUE_VALUES
+
+
+def _env_int(name: str, default: int) -> int:
+    return int(_env_str(name, str(default)).strip())
+
+
+def _env_float(name: str, default: float) -> float:
+    return float(_env_str(name, str(default)).strip())
+
+
+#  Конфиг браузерного отпечатка (User-Agent для HTTP запросов) 
+
+@dataclass(slots=True)
+class FingerprintConfig:
+    user_agent: str = (
+        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+        "AppleWebKit/537.36 (KHTML, like Gecko) "
+        "Chrome/135.0.0.0 Safari/537.36"
+    )
+
+
+# - Конфиг PostgreSQL (URL, пул соединений, pool_recycle) 
+
+@dataclass(slots=True)
+class DatabaseConfig:
+    url: str = _env_str("ENCAR_DATABASE_URL", "postgresql+psycopg2://encar:encar@localhost:5434/encar_db")
+    echo: bool = _env_bool("ENCAR_DATABASE_ECHO", False)
+    pool_size: int = _env_int("ENCAR_DATABASE_POOL_SIZE", 5)
+    max_overflow: int = _env_int("ENCAR_DATABASE_MAX_OVERFLOW", 10)
+    pool_recycle_seconds: int = _env_int("ENCAR_DATABASE_POOL_RECYCLE_SECONDS", 1800)
+    auto_create_tables: bool = _env_bool("ENCAR_DATABASE_AUTO_CREATE_TABLES", False)
+
+
+# --- Конфиг Redis (URL для Celery broker) ---
+
+@dataclass(slots=True)
+class RedisConfig:
+    url: str = _env_str("ENCAR_REDIS_URL", "redis://localhost:6380/1")
+
+
+# --- Конфиг Celery (лимиты задач, concurrency, beat-расписание) ---
+
+@dataclass(slots=True)
+class CeleryConfig:
+    broker_url: str = _env_str("CELERY_BROKER_URL", "")
+    result_backend: str = _env_str("CELERY_RESULT_BACKEND", "")
+    task_soft_time_limit: int = _env_int("CELERY_TASK_SOFT_TIME_LIMIT", 14400)
+    task_time_limit: int = _env_int("CELERY_TASK_TIME_LIMIT", 14520)
+    worker_concurrency: int = _env_int("CELERY_WORKER_CONCURRENCY", 4)
+    worker_max_tasks_per_child: int = _env_int("CELERY_WORKER_MAX_TASKS_PER_CHILD", 50)
+    broker_visibility_timeout: int = _env_int("CELERY_BROKER_VISIBILITY_TIMEOUT", 7200)
+    # Encar settings
+    encar_beat_interval_minutes: int = _env_int("ENCAR_BEAT_INTERVAL_MINUTES", 60)
+    encar_beat_limit: int | None = _env_int("ENCAR_BEAT_LIMIT", 0) or None
+
+
+# --- Главный объект настроек: собирает все блоки конфигурации ---
+
+@dataclass(slots=True)
+class Settings:
+    log_level: str = _env_str("ENCAR_LOG_LEVEL", "INFO")
+    log_file: str | None = _env_optional_str("ENCAR_LOG_FILE")
+    fingerprint: FingerprintConfig = field(default_factory=FingerprintConfig)
+    database: DatabaseConfig = field(default_factory=DatabaseConfig)
+    redis: RedisConfig = field(default_factory=RedisConfig)
+    celery: CeleryConfig = field(default_factory=CeleryConfig)
+
+# Глобальный синглтон — используется по умолчанию во всех модулях.
+settings = Settings()
--- a/encar_scraper/core/exceptions.py
+++ b/encar_scraper/core/exceptions.py
@@ -0,0 +1,2 @@
+class ScraperError(Exception):
+    """Базовое исключение скрапера."""
--- a/encar_scraper/core/logs.py
+++ b/encar_scraper/core/logs.py
@@ -0,0 +1,32 @@
+import logging
+import sys
+from contextvars import ContextVar
+
+# Храним trace_id текущего потока/корутины.
+TRACE_ID: ContextVar[str] = ContextVar("trace_id", default="-")
+
+
+class TraceIdFilter(logging.Filter):
+    # Добавляет trace_id в каждую запись лога для сквозной трассировки.
+    def filter(self, record: logging.LogRecord) -> bool:
+        record.trace_id = TRACE_ID.get()
+        return True
+
+
+def set_trace_id(trace_id: str) -> None:
+    TRACE_ID.set(trace_id)
+
+
+def setup_logging(level: str = "INFO", log_file: str | None = None) -> None:
+    handlers: list[logging.Handler] = [logging.StreamHandler(sys.stdout)]
+    if log_file:
+        handlers.append(logging.FileHandler(log_file, encoding="utf-8"))
+    trace_filter = TraceIdFilter()
+    for handler in handlers:
+        handler.addFilter(trace_filter)
+    logging.basicConfig(
+        level=getattr(logging, level.upper(), logging.INFO),
+        format="%(asctime)s | %(levelname)s | %(name)s | trace=%(trace_id)s | %(message)s",
+        handlers=handlers,
+        force=True,
+    )
--- a/encar_scraper/core/runtime_config.py
+++ b/encar_scraper/core/runtime_config.py
@@ -0,0 +1,97 @@
+"""Загрузчик runtime_config.json для управления фильтрами синхронизации."""
+from __future__ import annotations
+
+import json
+import logging
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+logger = logging.getLogger("encar_scraper.runtime_config")
+
+DEFAULT_CONFIG_PATH = Path(__file__).resolve().parent.parent.parent / "runtime_config.json"
+
+
+@dataclass
+class SyncConfig:
+    name: str | None = None
+    lane: str = "encar_cars"
+    only_new: bool = False
+    limit: int | None = None
+    probe_all_photos: bool = False
+
+
+@dataclass
+class FiltersConfig:
+    price_min: int | None = None
+    price_max: int | None = None
+    mileage_min: int | None = None
+    mileage_max: int | None = None
+    brands: list[str] = field(default_factory=list)
+    models: list[str] = field(default_factory=list)
+    years: list[int] = field(default_factory=list)
+    body_types: list[str] = field(default_factory=list)
+    colors: list[str] = field(default_factory=list)
+    drives: list[str] = field(default_factory=list)
+    gearboxes: list[str] = field(default_factory=list)
+    exclude_brands: list[str] = field(default_factory=list)
+    exclude_models: list[str] = field(default_factory=list)
+    exclude_years: list[int] = field(default_factory=list)
+    exclude_body_types: list[str] = field(default_factory=list)
+
+
+@dataclass
+class RuntimeConfig:
+    sync: SyncConfig = field(default_factory=SyncConfig)
+    filters: FiltersConfig = field(default_factory=FiltersConfig)
+
+
+def load_runtime_config(path: str | Path | None = None) -> RuntimeConfig:
+    """Загружает runtime_config.json. Если файл не найден — возвращает дефолт."""
+    config_path = Path(path) if path else DEFAULT_CONFIG_PATH
+    if not config_path.exists():
+        logger.info("runtime_config.json not found at %s, using defaults", config_path)
+        return RuntimeConfig()
+
+    try:
+        raw: dict[str, Any] = json.loads(config_path.read_text(encoding="utf-8"))
+    except Exception as exc:
+        logger.warning("Failed to parse runtime_config.json: %s, using defaults", exc)
+        return RuntimeConfig()
+
+    sync_raw = raw.get("sync") or {}
+    filters_raw = raw.get("filters") or {}
+    price_raw = filters_raw.get("price") or {}
+    mileage_raw = filters_raw.get("mileage") or {}
+
+    sync = SyncConfig(
+        name=sync_raw.get("name"),
+        lane=sync_raw.get("lane", "encar_cars"),
+        only_new=bool(sync_raw.get("only_new", False)),
+        limit=sync_raw.get("limit"),
+        probe_all_photos=bool(sync_raw.get("probe_all_photos", False)),
+    )
+
+    filters = FiltersConfig(
+        price_min=price_raw.get("min"),
+        price_max=price_raw.get("max"),
+        mileage_min=mileage_raw.get("min"),
+        mileage_max=mileage_raw.get("max"),
+        brands=[b.lower() for b in (filters_raw.get("brands") or [])],
+        models=[m.lower() for m in (filters_raw.get("models") or [])],
+        years=filters_raw.get("years") or [],
+        body_types=[b.lower() for b in (filters_raw.get("body_types") or [])],
+        colors=[c.lower() for c in (filters_raw.get("colors") or [])],
+        drives=[d.lower() for d in (filters_raw.get("drives") or [])],
+        gearboxes=[g.lower() for g in (filters_raw.get("gearboxes") or [])],
+        exclude_brands=[b.lower() for b in (filters_raw.get("exclude_brands") or [])],
+        exclude_models=[m.lower() for m in (filters_raw.get("exclude_models") or [])],
+        exclude_years=filters_raw.get("exclude_years") or [],
+        exclude_body_types=[b.lower() for b in (filters_raw.get("exclude_body_types") or [])],
+    )
+
+    logger.info(
+        "runtime_config loaded: lane=%s, limit=%s, brands=%d, exclude_brands=%d",
+        sync.lane, sync.limit, len(filters.brands), len(filters.exclude_brands),
+    )
+    return RuntimeConfig(sync=sync, filters=filters)
--- a/encar_scraper/core/utils.py
+++ b/encar_scraper/core/utils.py
@@ -0,0 +1,72 @@
+import json
+import re
+from pathlib import Path
+from typing import Any, Callable, Iterable
+
+
+def save_to_json(data: Any, filename: str | Path) -> None:
+    path = Path(filename)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
+
+
+def first_non_empty(values: Iterable[Any]) -> Any | None:
+    for value in values:
+        if value not in (None, "", [], {}, ()):
+            return value
+    return None
+
+
+# Регулярные выражения для VIN, lot и price.
+VIN_RE = re.compile(r"\b([A-HJ-NPR-Z0-9]{17})\b", re.IGNORECASE)
+LOT_RE = re.compile(r"\b(\d{7,10})\b")
+PRICE_RE = re.compile(r"\$\s?([\d,]+(?:\.\d{1,2})?)")
+
+
+def deep_find_key(obj, target_keys: set[str], max_depth: int = 64, _depth: int = 0) -> list:
+    # Рекурсивно ищет значения по набору ключей в произвольном JSON-дереве.
+    found = []
+    if _depth >= max_depth:
+        return found
+    if isinstance(obj, dict):
+        for key, value in obj.items():
+            if key.lower() in target_keys:
+                found.append(value)
+            found.extend(deep_find_key(value, target_keys, max_depth=max_depth, _depth=_depth + 1))
+    elif isinstance(obj, list):
+        for item in obj:
+            found.extend(deep_find_key(item, target_keys, max_depth=max_depth, _depth=_depth + 1))
+    return found
+
+
+def deep_find_all_keys(
+    payloads: list,
+    field_map: dict[str, set[str]],
+    max_depth: int = 64,
+) -> dict[str, list]:
+    """Извлекает все нужные поля за один проход по JSON."""
+    # Готовим обратную карту: нормализованный ключ -> имя поля.
+    reverse: dict[str, str] = {}
+    for field_name, keys in field_map.items():
+        for k in keys:
+            reverse[k.lower()] = field_name
+
+    result: dict[str, list] = {f: [] for f in field_map}
+
+    def _recurse(obj: Any, depth: int) -> None:
+        if depth >= max_depth:
+            return
+        if isinstance(obj, dict):
+            for k, v in obj.items():
+                field = reverse.get(k.lower())
+                if field is not None:
+                    result[field].append(v)
+                _recurse(v, depth + 1)
+        elif isinstance(obj, list):
+            for item in obj:
+                _recurse(item, depth + 1)
+
+    for payload in payloads:
+        _recurse(payload, 0)
+
+    return result
--- a/encar_scraper/encar.py
+++ b/encar_scraper/encar.py
--- a/encar_scraper/storage/init.py
+++ b/encar_scraper/storage/init.py
@@ -0,0 +1 @@
+__all__: list[str] = []
--- a/encar_scraper/storage/db.py
+++ b/encar_scraper/storage/db.py
@@ -0,0 +1,494 @@
+import logging
+from contextlib import contextmanager
+from datetime import datetime, timezone
+from typing import Any, Iterator
+
+from sqlalchemy import create_engine, delete, or_, select, text, update
+from sqlalchemy.dialects.postgresql import insert as pg_insert
+from sqlalchemy.orm import Session, sessionmaker
+
+from ..core.config import Settings
+from .models import Base, Car, Image, SyncRun
+from .schemas import CarRecord
+
+logger = logging.getLogger("encar_scraper.db")
+
+
+CAR_DB_FIELDS = {
+	col.key for col in Car.__table__.columns
+	if col.key not in ("id",)
+}
+
+_IN_CHUNK_SIZE = 5000
+
+
+class PersistenceService:
+
+	def __init__(self, settings: Settings) -> None:
+		self.settings = settings
+		engine_kwargs = {
+			"echo": settings.database.echo,
+			"future": True,
+		}
+		if "postgresql" in settings.database.url:
+			engine_kwargs["pool_size"] = settings.database.pool_size
+			engine_kwargs["max_overflow"] = settings.database.max_overflow
+			engine_kwargs["pool_pre_ping"] = True
+			engine_kwargs["pool_recycle"] = settings.database.pool_recycle_seconds
+		self.engine = create_engine(settings.database.url, **engine_kwargs)
+		self.session_factory = sessionmaker(bind=self.engine, expire_on_commit=False, future=True)
+
+	def create_tables(self) -> None:
+		# В тестах/локально на SQLite разрешаем create_all; для non-SQLite в проде — только через миграции.
+		is_sqlite = self.settings.database.url.startswith("sqlite")
+		if not is_sqlite and not self.settings.database.auto_create_tables:
+			return
+		try:
+			Base.metadata.create_all(self.engine)
+		except Exception:
+			logger.debug("create_tables skipped (schema already exists)")
+
+	@contextmanager
+	def session_scope(self) -> Iterator[Session]:
+		session = self.session_factory()
+		try:
+			yield session
+			session.commit()
+		except Exception:
+			session.rollback()
+			raise
+		finally:
+			session.close()
+
+	def start_sync_run(self, lane: str) -> int:
+		with self.session_scope() as session:
+			now = datetime.now(timezone.utc)
+			stale_runs = session.execute(select(SyncRun).where(SyncRun.status == "running")).scalars().all()
+			for stale in stale_runs:
+				stale.status = "failed"
+				stale.finished_at = now
+				if not stale.error_summary:
+					stale.error_summary = "Recovered stale running sync run before starting a new run"
+
+			run = SyncRun(status="running", lane=lane, ids_fetched=0, cars_upserted=0, cars_failed=0, images_upserted=0)
+			session.add(run)
+			session.flush()
+			return int(run.id)
+
+	def finish_sync_run(self, run_id: int, *, status: str, ids_fetched: int, cars_upserted: int, cars_failed: int, images_upserted: int, error_summary: str | None = None) -> None:
+		with self.session_scope() as session:
+			run = session.get(SyncRun, run_id)
+			if run is None:
+				return
+			run.finished_at = datetime.now(timezone.utc)
+			run.status = status
+			run.ids_fetched = ids_fetched
+			run.cars_upserted = cars_upserted
+			run.cars_failed = cars_failed
+			run.images_upserted = images_upserted
+			run.error_summary = error_summary
+
+	@staticmethod
+	def _add_images(session: Session, car_id: int, images: list[dict[str, object]]) -> None:
+		if not images:
+			return
+		session.add_all([
+			Image(
+				fullres_image=str(img["fullres_image"]),
+				preview_image=str(img["preview_image"]),
+				order_index=int(img.get("order_index", 0)),
+				car_id=car_id,
+			)
+			for img in images
+		])
+
+	@staticmethod
+	def _car_payload(record: CarRecord) -> dict[str, object]:
+		payload = record.model_dump(mode="python")
+		return {key: value for key, value in payload.items() if key in CAR_DB_FIELDS}
+
+	def _is_postgres(self) -> bool:
+		return self.engine.dialect.name == "postgresql"
+
+	def _load_existing_cars(
+		self,
+		session: Session,
+		origin_ids: list[str],
+		origin_urls: list[str],
+	) -> tuple[dict[str, Car], dict[str, Car]]:
+		existing_by_id: dict[str, Car] = {}
+		existing_by_url: dict[str, Car] = {}
+		if not origin_ids and not origin_urls:
+			return existing_by_id, existing_by_url
+
+		existing_cars: list[Car] = []
+		max_len = max(len(origin_ids), len(origin_urls), 1)
+		for i in range(0, max_len, _IN_CHUNK_SIZE):
+			id_chunk = origin_ids[i:i + _IN_CHUNK_SIZE]
+			url_chunk = origin_urls[i:i + _IN_CHUNK_SIZE]
+			conditions = []
+			if id_chunk:
+				conditions.append(Car.origin_id.in_(id_chunk))
+			if url_chunk:
+				conditions.append(Car.origin_url.in_(url_chunk))
+			if not conditions:
+				continue
+			rows = session.execute(
+				select(Car).where(or_(*conditions))
+			).scalars().all()
+			existing_cars.extend(rows)
+
+		for car in existing_cars:
+			if car.origin_id:
+				existing_by_id[car.origin_id] = car
+			if car.origin_url:
+				existing_by_url[car.origin_url] = car
+		return existing_by_id, existing_by_url
+
+	def _load_existing_image_urls(self, session: Session, car_ids: set[int]) -> dict[int, set[str]]:
+		existing_images_map: dict[int, set[str]] = {}
+		if not car_ids:
+			return existing_images_map
+
+		car_id_list = list(car_ids)
+		for i in range(0, len(car_id_list), _IN_CHUNK_SIZE):
+			chunk = car_id_list[i:i + _IN_CHUNK_SIZE]
+			img_rows = session.execute(
+				select(Image.car_id, Image.fullres_image).where(Image.car_id.in_(chunk))
+			).all()
+			for cid, furl in img_rows:
+				existing_images_map.setdefault(int(cid), set()).add(str(furl))
+		return existing_images_map
+
+	@staticmethod
+	def _postgres_upsert_set_map(insert_stmt) -> dict[str, object]:
+		return {
+			key: getattr(insert_stmt.excluded, key)
+			for key in CAR_DB_FIELDS
+		}
+
+	def _replace_images_for_car(
+		self,
+		session: Session,
+		car_id: int,
+		images: list[dict[str, object]],
+		origin_id: str,
+	) -> int:
+		nested = session.begin_nested()
+		try:
+			session.execute(delete(Image).where(Image.car_id == car_id))
+			self._add_images(session, car_id, images)
+			session.flush()
+			nested.commit()
+			return len(images)
+		except Exception:
+			nested.rollback()
+			logger.warning("Image replacement failed for car %s, keeping old images", origin_id, exc_info=True)
+			return 0
+
+	def _upsert_cars_batch_postgres(self, records: list[CarRecord]) -> dict[str, int]:
+		inserted = 0
+		updated = 0
+		images_total = 0
+
+		with self.session_scope() as session:
+			origin_ids = [r.origin_id for r in records if r.origin_id]
+			origin_urls = [r.origin_url for r in records if r.origin_url]
+			existing_by_id, existing_by_url = self._load_existing_cars(session, origin_ids, origin_urls)
+
+			entries: list[dict[str, object]] = []
+			upsert_payloads: list[dict[str, object]] = []
+			for record in records:
+				payload = self._car_payload(record)
+				images = [image.model_dump(mode="python") for image in record.images]
+				car_by_id = existing_by_id.get(record.origin_id)
+				car_by_url = existing_by_url.get(record.origin_url)
+				entry: dict[str, object] = {"record": record, "images": images, "car_id": None}
+
+				if car_by_url is not None and car_by_url.origin_id != record.origin_id and car_by_id is None:
+					for key, value in payload.items():
+						setattr(car_by_url, key, value)
+					car_by_url.last_seen_at = record.last_seen_at
+					entry["car_id"] = int(car_by_url.id)
+					updated += 1
+				else:
+					upsert_payloads.append(payload)
+					if car_by_id is not None:
+						updated += 1
+					else:
+						inserted += 1
+				entries.append(entry)
+
+			session.flush()
+
+			if upsert_payloads:
+				insert_stmt = pg_insert(Car).values(upsert_payloads)
+				upsert_stmt = insert_stmt.on_conflict_do_update(
+					index_elements=[Car.origin_id],
+					set_=self._postgres_upsert_set_map(insert_stmt),
+				).returning(Car.id, Car.origin_id)
+				rows = session.execute(upsert_stmt).all()
+				car_ids_by_origin_id = {str(origin_id): int(car_id) for car_id, origin_id in rows}
+				for entry in entries:
+					if entry["car_id"] is not None:
+						continue
+					record = entry["record"]
+					car_id = car_ids_by_origin_id.get(record.origin_id)
+					if car_id is None:
+						raise RuntimeError(f"PostgreSQL upsert did not return car_id for {record.origin_id}")
+					entry["car_id"] = car_id
+
+			car_ids = {int(entry["car_id"]) for entry in entries if entry["car_id"] is not None}
+			existing_images_map = self._load_existing_image_urls(session, car_ids)
+			images_by_car_id: dict[int, list[dict[str, object]]] = {}
+			replace_ids: list[int] = []
+
+			for entry in entries:
+				car_id = int(entry["car_id"])
+				images = entry["images"]
+				new_image_urls = {
+					str(img.get("fullres_image", ""))
+					for img in images
+					if img.get("fullres_image")
+				}
+				old_image_urls = existing_images_map.get(car_id, set())
+				if new_image_urls != old_image_urls:
+					replace_ids.append(car_id)
+					images_by_car_id[car_id] = images
+				else:
+					images_total += len(old_image_urls)
+
+			if replace_ids:
+				for i in range(0, len(replace_ids), _IN_CHUNK_SIZE):
+					chunk = replace_ids[i:i + _IN_CHUNK_SIZE]
+					session.execute(delete(Image).where(Image.car_id.in_(chunk)))
+				for car_id in replace_ids:
+					images = images_by_car_id[car_id]
+					self._add_images(session, car_id, images)
+					images_total += len(images)
+
+		return {"inserted": inserted, "updated": updated, "images_upserted": images_total}
+
+	def upsert_car(self, record: CarRecord):
+		# Вставка или обновление автомобиля по origin_id/origin_url.
+		payload = self._car_payload(record)
+		images = [image.model_dump(mode="python") for image in record.images]
+		with self.session_scope() as session:
+			if self._is_postgres():
+				car_by_url = session.execute(
+					select(Car).where(Car.origin_url == record.origin_url)
+				).scalar_one_or_none()
+				if car_by_url is not None and car_by_url.origin_id != record.origin_id:
+					for key, value in payload.items():
+						setattr(car_by_url, key, value)
+					car_by_url.last_seen_at = record.last_seen_at
+					session.flush()
+					car_id = int(car_by_url.id)
+					action = "updated"
+				else:
+					existed = session.execute(
+						select(Car.id).where(Car.origin_id == record.origin_id)
+					).scalar_one_or_none() is not None
+					insert_stmt = pg_insert(Car).values(**payload)
+					upsert_stmt = insert_stmt.on_conflict_do_update(
+						index_elements=[Car.origin_id],
+						set_=self._postgres_upsert_set_map(insert_stmt),
+					).returning(Car.id)
+					car_id = int(session.execute(upsert_stmt).scalar_one())
+					action = "updated" if existed or car_by_url is not None else "inserted"
+
+				images_upserted = self._replace_images_for_car(
+					session, car_id, images, record.origin_id,
+				)
+				return {"car_id": car_id, "images_upserted": images_upserted, "action": action}
+
+			car = session.execute(
+				select(Car).where(or_(Car.origin_id == record.origin_id, Car.origin_url == record.origin_url))
+			).scalar_one_or_none()
+			action = "inserted"
+			if car is None:
+				car = Car(**payload)
+				session.add(car)
+				session.flush()
+			else:
+				action = "updated"
+				for key, value in payload.items():
+					setattr(car, key, value)
+				car.last_seen_at = record.last_seen_at
+				session.flush()
+				images_upserted = self._replace_images_for_car(session, int(car.id), images, record.origin_id)
+				return {"car_id": int(car.id), "images_upserted": images_upserted, "action": action}
+			self._add_images(session, int(car.id), images)
+			session.flush()
+			return {"car_id": int(car.id), "images_upserted": len(images), "action": action}
+	def upsert_cars_batch(self, records: list[CarRecord]) -> dict[str, int]:
+		"""Пакетный upsert нескольких автомобилей в одной транзакции.
+
+		Оптимизации:
+		- Дедупликация записей по origin_id перед вставкой.
+		- Chunked IN-queries для больших списков (обход лимита PG параметров).
+		- Пропуск перезаписи изображений, если набор URL не изменился.
+		- Один DELETE по car_id IN (...) вместо удаления по одному.
+		- Fallback на по-одному upsert если batch commit упал.
+		 """
+		# ── Дедупликация записей внутри батча ──
+		seen_ids: dict[str, int] = {}
+		unique_records: list[CarRecord] = []
+		for idx, r in enumerate(records):
+			key = r.origin_id or r.origin_url
+			if key in seen_ids:
+				logger.debug("Dedup: skipping duplicate record %s (idx %d vs %d)", key, idx, seen_ids[key])
+				continue
+			seen_ids[key] = idx
+			unique_records.append(r)
+
+		if len(unique_records) < len(records):
+			logger.info("Deduped batch: %d → %d records", len(records), len(unique_records))
+		records = unique_records
+
+		try:
+			return self._upsert_cars_batch_inner(records)
+		except Exception as exc:
+			logger.warning("Batch upsert failed (%s), falling back to individual upserts", exc)
+			return self._upsert_cars_individually(records)
+
+	def _upsert_cars_batch_inner(self, records: list[CarRecord]) -> dict[str, int]:
+		"""Внутренняя реализация batched upsert (одна транзакция)."""
+		if self._is_postgres():
+			return self._upsert_cars_batch_postgres(records)
+
+		inserted = 0
+		updated = 0
+		images_total = 0
+
+		with self.session_scope() as session:
+			# Получаем существующие записи chunked-запросами.
+			origin_ids = [r.origin_id for r in records if r.origin_id]
+			origin_urls = [r.origin_url for r in records if r.origin_url]
+
+			existing_by_id, existing_by_url = self._load_existing_cars(session, origin_ids, origin_urls)
+
+			# Предзагружаем ВСЕ изображения для обновляемых машин одним запросом.
+			existing_car_ids = set()
+			for record in records:
+				car = existing_by_id.get(record.origin_id) or existing_by_url.get(record.origin_url)
+				if car is not None:
+					existing_car_ids.add(int(car.id))
+
+			# Строим маппинг car_id → set(image_urls) для сравнения.
+			existing_images_map = self._load_existing_image_urls(session, existing_car_ids)
+
+			new_cars: list[tuple[Car, list[dict]]] = []
+			update_cars_needing_images: list[tuple[Car, list[dict]]] = []
+
+			for record in records:
+				payload = self._car_payload(record)
+				images = [image.model_dump(mode="python") for image in record.images]
+
+				car = existing_by_id.get(record.origin_id) or existing_by_url.get(record.origin_url)
+				if car is None:
+					car = Car(**payload)
+					session.add(car)
+					inserted += 1
+					new_cars.append((car, images))
+				else:
+					for key, value in payload.items():
+						setattr(car, key, value)
+					car.last_seen_at = record.last_seen_at
+					updated += 1
+
+					# Проверяем, изменились ли изображения.
+					new_image_urls = {img.get("fullres_image", "") for img in images}
+					old_image_urls = existing_images_map.get(int(car.id), set())
+					if new_image_urls != old_image_urls:
+						update_cars_needing_images.append((car, images))
+					else:
+						images_total += len(old_image_urls)
+
+			# Один flush для всех вставок.
+			session.flush()
+
+			# Добавляем изображения для новых автомобилей.
+			for car, images in new_cars:
+				self._add_images(session, int(car.id), images)
+				images_total += len(images)
+
+			# Массово обновляем изображения только для машин с изменёнными картинками.
+			if update_cars_needing_images:
+				update_ids = [int(car.id) for car, _ in update_cars_needing_images]
+				for i in range(0, len(update_ids), _IN_CHUNK_SIZE):
+					chunk = update_ids[i:i + _IN_CHUNK_SIZE]
+					session.execute(delete(Image).where(Image.car_id.in_(chunk)))
+				for car, images in update_cars_needing_images:
+					self._add_images(session, int(car.id), images)
+					images_total += len(images)
+
+		return {"inserted": inserted, "updated": updated, "images_upserted": images_total}
+
+	def _upsert_cars_individually(self, records: list[CarRecord]) -> dict[str, int]:
+		# Fallback на поштучный upsert.
+		inserted = 0
+		updated = 0
+		images_total = 0
+		for record in records:
+			try:
+				result = self.upsert_car(record)
+				action = result.get("action", "inserted")
+				if action == "inserted":
+					inserted += 1
+				else:
+					updated += 1
+				images_total += int(result.get("images_upserted", 0))
+			except Exception as exc:
+				logger.error("Individual upsert failed for %s: %s", record.origin_id, exc)
+		return {"inserted": inserted, "updated": updated, "images_upserted": images_total}
+
+	def mark_sold_not_in_listing(self, active_origin_ids: set[str], lane: str = "encar") -> int:
+		"""Помечает авто как проданные, если их нет в активном листинге (по origin_id).
+
+		Для PostgreSQL использует временную таблицу + LEFT JOIN вместо NOT IN,
+		что кардинально быстрее при больших объёмах (200K+ IDs).
+		"""
+		if not active_origin_ids:
+			return 0
+		with self.session_scope() as session:
+			if self._is_postgres():
+				session.execute(text("CREATE TEMP TABLE IF NOT EXISTS _active_ids (origin_id TEXT NOT NULL) ON COMMIT DROP"))
+				session.execute(text("TRUNCATE _active_ids"))
+
+				id_list = list(active_origin_ids)
+				for i in range(0, len(id_list), _IN_CHUNK_SIZE):
+					chunk = id_list[i:i + _IN_CHUNK_SIZE]
+					values = ",".join(f"(:{f'u{j}'})" for j in range(len(chunk)))
+					params = {f"u{j}": oid for j, oid in enumerate(chunk)}
+					session.execute(text(f"INSERT INTO _active_ids (origin_id) VALUES {values}"), params)
+
+				session.execute(text("CREATE INDEX IF NOT EXISTS _ix_active_ids ON _active_ids (origin_id)"))
+
+				result = session.execute(text("""
+					UPDATE cars
+					SET is_sold = TRUE
+					FROM (
+						SELECT c.id
+						FROM cars c
+						LEFT JOIN _active_ids a ON c.origin_id = a.origin_id
+						WHERE a.origin_id IS NULL
+						AND c.is_sold = FALSE
+						AND c.origin_id LIKE 'encar:%%'
+					) sub
+					WHERE cars.id = sub.id
+				"""))
+				count = result.rowcount or 0
+			else:
+				stmt = (
+					update(Car)
+					.where(Car.origin_id.notin_(active_origin_ids))
+					.where(Car.is_sold == False)  # noqa: E712
+					.where(Car.origin_id.like("encar:%"))
+					.values(is_sold=True)
+				)
+				result = session.execute(stmt)
+				count = result.rowcount or 0
+			if count:
+				logger.info("Marked %d cars as sold (no longer in listing)", count)
+			return count
--- a/encar_scraper/storage/enums.py
+++ b/encar_scraper/storage/enums.py
@@ -0,0 +1,36 @@
+CURRENCY_ENUM_VALUES = ("JPY", "USD", "EUR", "RUB", "KRW", "AED", "GBP", "CAD")
+DRIVE_ENUM_VALUES = ("FWD", "RWD", "2WD", "4WD", "NA")
+GEARBOX_ENUM_VALUES = ("AT", "CVT", "MT", "EV", "NA")
+STEERING_WHEEL_ENUM_VALUES = ("LEFT", "RIGHT", "NA")
+BODY_TYPE_ENUM_VALUES = (
+	"COUPE",
+	"SUV",
+	"HATCHBACK",
+	"MINIVAN",
+	"SEDAN",
+	"STATION_WAGON",
+	"PICKUP",
+	"TRUCK",
+	"OPEN",
+	"RV",
+	"VAN",
+	"CONVERTIBLE",
+	"BUS",
+	"COMPACT",
+	"MID_SIZE",
+	"FULL_SIZE",
+	"SPORTS",
+	"CLASSIC",
+	"ELECTRIC",
+	"HYBRID",
+	"SPECIAL",
+	"WAGON",
+	"OTHER",
+	"NA",
+)
+COUNTRY_ENUM_VALUES = ("JP", "KR", "US", "CA", "NA")
+ORIGIN_ENUM_VALUES = (
+	"ENCAR",
+	"NA",
+)
+SELLING_TYPE_ENUM_VALUES = ("STOCK", "AUCTION", "TENDER", "NA")
--- a/encar_scraper/storage/models.py
+++ b/encar_scraper/storage/models.py
@@ -0,0 +1,82 @@
+from datetime import datetime
+
+from sqlalchemy import BigInteger, Boolean, DateTime, Enum, ForeignKey, Index, Integer, String, Text, func
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
+
+from .enums import (
+	BODY_TYPE_ENUM_VALUES,
+	COUNTRY_ENUM_VALUES,
+	CURRENCY_ENUM_VALUES,
+	DRIVE_ENUM_VALUES,
+	GEARBOX_ENUM_VALUES,
+	ORIGIN_ENUM_VALUES,
+	SELLING_TYPE_ENUM_VALUES,
+	STEERING_WHEEL_ENUM_VALUES,
+)
+
+
+class Base(DeclarativeBase):
+	pass
+
+
+class Car(Base):
+	__tablename__ = "cars"
+	__table_args__ = (
+		Index("ix_cars_brand_model", "brand", "model"),
+		Index("ix_cars_origin_id_not_sold", "origin_id", "is_sold"),
+	)
+	id: Mapped[int] = mapped_column(BigInteger().with_variant(Integer, "sqlite"), primary_key=True, autoincrement=True)
+	parser_id: Mapped[str] = mapped_column(String(255), nullable=False, unique=True)
+	brand: Mapped[str] = mapped_column(String(100), nullable=False, index=True)
+	model: Mapped[str] = mapped_column(String(500), nullable=False)
+	year: Mapped[int | None] = mapped_column(Integer, nullable=True, index=True)
+	price: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
+	currency: Mapped[str] = mapped_column(Enum(*CURRENCY_ENUM_VALUES, name="currencyenum", native_enum=False, 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=False, create_constraint=False), nullable=False, default="NA")
+	is_sold: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, index=True)
+	color: Mapped[str] = mapped_column(String(), nullable=False, default="other")
+	drive: Mapped[str | None] = mapped_column(Enum(*DRIVE_ENUM_VALUES, name="driveenum", native_enum=False, create_constraint=False), nullable=True)
+	gearbox: Mapped[str | None] = mapped_column(Enum(*GEARBOX_ENUM_VALUES, name="gearboxenum", native_enum=False, create_constraint=False), nullable=True)
+	steering_wheel: Mapped[str | None] = mapped_column(Enum(*STEERING_WHEEL_ENUM_VALUES, name="steeringwheelenum", native_enum=False, create_constraint=False), nullable=True)
+	body_type: Mapped[str] = mapped_column(Enum(*BODY_TYPE_ENUM_VALUES, name="bodytypeenum", native_enum=False, 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=False, 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=False, create_constraint=False), nullable=False, default="NA")
+	origin_url: Mapped[str] = mapped_column(String(), nullable=False, index=True)
+	origin_id: Mapped[str] = mapped_column(String(), nullable=False, unique=True, index=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(), index=True)
+	images: Mapped[list["Image"]] = relationship("Image", back_populates="car", cascade="all, delete-orphan")
+
+
+class Image(Base):
+	__tablename__ = "images"
+	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(BigInteger().with_variant(Integer, "sqlite"), ForeignKey("cars.id", ondelete="CASCADE"), nullable=False, index=True)
+	car: Mapped[Car] = relationship("Car", back_populates="images")
+
+
+class SyncRun(Base):
+	__tablename__ = "sync_runs"
+	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, index=True)
+	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)
--- a/encar_scraper/storage/schemas.py
+++ b/encar_scraper/storage/schemas.py
@@ -0,0 +1,87 @@
+from datetime import datetime, timezone
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class ImageRecord(BaseModel):
+	fullres_image: str
+	preview_image: str
+	order_index: int = 0
+
+
+class CarRecord(BaseModel):
+	parser_id: str
+	brand: str
+	model: str
+	year: int | None = None
+	price: int | None = None
+	currency: str = "USD"
+	mileage: int = 0
+	country: str = "US"
+	is_sold: bool = False
+	color: str = "other"
+	drive: str | None = None
+	gearbox: str | None = None
+	steering_wheel: str | None = None
+	body_type: str = "OTHER"
+	engine_volume: int | None = None
+	selling_type: str = "AUCTION"
+	one_owner: bool = False
+	new_car: bool = False
+	is_hidden: bool = False
+	origin: str = "NA"
+	origin_url: str
+	origin_id: str
+	is_damaged: bool = False
+	evaluation: str | None = None
+	non_smoking: bool = True
+	rental: bool = False
+	repair_history: bool = False
+	slug: str
+	last_seen_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+	images: list[ImageRecord] = Field(default_factory=list)
+
+
+class ImageRead(BaseModel):
+	model_config = ConfigDict(from_attributes=True)
+
+	id: int
+	fullres_image: str
+	preview_image: str
+	order_index: int = 0
+
+
+class CarRead(BaseModel):
+	model_config = ConfigDict(from_attributes=True)
+
+	id: int
+	parser_id: str
+	brand: str
+	model: str
+	year: int | None = None
+	price: int | None = None
+	currency: str = "USD"
+	mileage: int = 0
+	country: str = "US"
+	is_sold: bool = False
+	color: str = "other"
+	drive: str | None = None
+	gearbox: str | None = None
+	steering_wheel: str | None = None
+	body_type: str = "OTHER"
+	engine_volume: int | None = None
+	selling_type: str = "AUCTION"
+	one_owner: bool = False
+	new_car: bool = False
+	is_hidden: bool = False
+	origin: str = "NA"
+	origin_url: str = ""
+	origin_id: str = ""
+	is_damaged: bool = False
+	evaluation: str | None = None
+	non_smoking: bool = True
+	rental: bool = False
+	repair_history: bool = False
+	slug: str = ""
+	last_seen_at: datetime | None = None
+	images: list[ImageRead] = Field(default_factory=list)
+
--- a/encar_scraper/worker/init.py
+++ b/encar_scraper/worker/init.py
@@ -0,0 +1,3 @@
+from .celery_app import celery_app
+
+__all__ = ["celery_app"]
--- a/encar_scraper/worker/celery_app.py
+++ b/encar_scraper/worker/celery_app.py
@@ -0,0 +1,71 @@
+# Инициализация Celery-приложения и периодических задач для Encar.
+
+from celery import Celery
+
+from ..core.config import settings
+
+
+def _broker_url() -> str:
+    return settings.celery.broker_url or settings.redis.url
+
+
+def _result_backend() -> str:
+    return settings.celery.result_backend or settings.redis.url
+
+
+celery_app = Celery(
+    "encar_scraper",
+    broker=_broker_url(),
+    backend=_result_backend(),
+)
+
+celery_app.conf.update(
+    task_serializer="json",
+    accept_content=["json"],
+    result_serializer="json",
+    timezone="UTC",
+    enable_utc=True,
+    task_soft_time_limit=settings.celery.task_soft_time_limit,
+    task_time_limit=settings.celery.task_time_limit,
+    task_acks_late=True,
+    task_reject_on_worker_lost=True,
+    task_track_started=True,
+    worker_concurrency=settings.celery.worker_concurrency,
+    worker_max_tasks_per_child=settings.celery.worker_max_tasks_per_child,
+    worker_pool="prefork",
+    worker_prefetch_multiplier=1,
+    broker_connection_retry_on_startup=True,
+    broker_transport_options={
+        "visibility_timeout": settings.celery.broker_visibility_timeout,
+    },
+    result_expires=86400,
+    worker_hijack_root_logger=False,
+    beat_schedule={
+        "periodic-encar-sync": {
+            "task": "encar_scraper.worker.tasks.encar_sync_listing_task",
+            "schedule": settings.celery.encar_beat_interval_minutes * 60.0,
+            "args": (),
+            "kwargs": {
+                "car_type": "all",
+            },
+            "options": {"queue": "encar"},
+        },
+    },
+    task_routes={
+        "encar_scraper.worker.tasks.*": {"queue": "encar"},
+    },
+)
+
+celery_app.autodiscover_tasks(["encar_scraper.worker"])
+
+
+from celery.signals import setup_logging as celery_setup_logging
+
+
+@celery_setup_logging.connect
+def _configure_logging(loglevel, logfile, format, colorize, **kwargs):
+    """Перехватываем настройку логирования Celery, чтобы использовать свой формат."""
+    from ..core.logs import setup_logging
+    import logging
+    level_name = logging.getLevelName(loglevel) if isinstance(loglevel, int) else str(loglevel)
+    setup_logging(level=level_name, log_file=logfile)
--- a/encar_scraper/worker/tasks.py
+++ b/encar_scraper/worker/tasks.py
@@ -0,0 +1,244 @@
+# Celery tasks для синхронизации автомобилей Encar.
+
+import logging
+from threading import Event, Thread
+import uuid
+
+from celery import shared_task
+from redis import Redis
+
+from ..core.config import Settings
+from ..core.runtime_config import load_runtime_config
+from ..encar import EncarScraper, EncarFilters, expand_allowed_brands
+from ..storage.db import PersistenceService
+
+logger = logging.getLogger("encar_scraper.worker.tasks")
+
+ENCAR_SYNC_LOCK_KEY = "encar:locks:sync_listing"
+
+_persistence: PersistenceService | None = None
+_redis: Redis | None = None
+
+
+def _sync_listing_lock_ttl_seconds() -> int:
+    settings = Settings()
+    return max(settings.celery.task_time_limit + 120, 300)
+
+
+def _get_persistence() -> PersistenceService:
+    global _persistence
+    if _persistence is None:
+        _persistence = PersistenceService(Settings())
+    return _persistence
+
+
+def _get_redis() -> Redis:
+    global _redis
+    if _redis is None:
+        settings = Settings()
+        _redis = Redis.from_url(
+            settings.redis.url,
+            decode_responses=True,
+            socket_timeout=10,
+            socket_connect_timeout=5,
+        )
+    return _redis
+
+
+def _acquire_lock(redis_client: Redis, key: str, owner_token: str, ttl_seconds: int) -> bool:
+    try:
+        return bool(redis_client.set(key, owner_token, nx=True, ex=ttl_seconds))
+    except Exception as exc:
+        logger.warning("Failed to acquire lock %s", key, exc_info=True)
+        return False
+
+
+_REFRESH_LOCK_SCRIPT = """
+if redis.call('GET', KEYS[1]) == ARGV[1] then
+    return redis.call('EXPIRE', KEYS[1], tonumber(ARGV[2]))
+end
+return 0
+"""
+
+_RELEASE_LOCK_SCRIPT = """
+if redis.call('GET', KEYS[1]) == ARGV[1] then
+    return redis.call('DEL', KEYS[1])
+end
+return 0
+"""
+
+
+def _refresh_lock_if_owner(redis_client: Redis, key: str, owner_token: str, ttl_seconds: int) -> bool | None:
+    try:
+        script = redis_client.register_script(_REFRESH_LOCK_SCRIPT)
+        refreshed = script(keys=[key], args=[owner_token, int(ttl_seconds)])
+        return bool(refreshed)
+    except Exception as exc:
+        logger.warning("Failed to refresh lock %s", key, exc_info=True)
+        return None
+
+
+def _release_lock_if_owner(redis_client: Redis, key: str, owner_token: str) -> None:
+    try:
+        script = redis_client.register_script(_RELEASE_LOCK_SCRIPT)
+        script(keys=[key], args=[owner_token])
+    except Exception as exc:
+        logger.warning("Failed to release lock %s", key, exc_info=True)
+
+
+def _start_lock_heartbeat(
+    redis_client: Redis,
+    key: str,
+    owner_token: str,
+    ttl_seconds: int,
+) -> tuple[Event, Thread]:
+    stop_event = Event()
+    interval_seconds = max(5.0, min(30.0, ttl_seconds / 3))
+
+    def _heartbeat() -> None:
+        while not stop_event.wait(interval_seconds):
+            refreshed = _refresh_lock_if_owner(redis_client, key, owner_token, ttl_seconds)
+            if refreshed is False:
+                logger.warning("Lost sync_listing lock ownership for %s", owner_token)
+                return
+
+    thread = Thread(target=_heartbeat, name="encar-lock-heartbeat", daemon=True)
+    thread.start()
+    return stop_event, thread
+
+
+@shared_task(
+    name="encar_scraper.worker.tasks.encar_sync_listing_task",
+    bind=True,
+    max_retries=2,
+    default_retry_delay=60,
+    acks_late=True,
+)
+def encar_sync_listing_task(
+    self,
+    car_type: str = "all",
+    manufacturer: str | None = None,
+    year_from: int | None = None,
+    year_to: int | None = None,
+    price_max: int | None = None,
+    lane: str = "encar",
+    limit: int | None = None,
+):
+    """
+    Полная синхронизация листинга Encar.
+    Собирает ВСЕ авто, обновляет цены/данные, помечает проданные.
+    """
+    persistence = _get_persistence()
+    persistence.create_tables()
+    task_id = self.request.id or "unknown"
+    owner_token = f"{task_id}:{uuid.uuid4().hex}"
+    redis_client = _get_redis()
+
+    lock_ttl = _sync_listing_lock_ttl_seconds()
+    lock_acquired = _acquire_lock(redis_client, ENCAR_SYNC_LOCK_KEY, owner_token, lock_ttl)
+
+    if not lock_acquired:
+        logger.info("encar_sync_listing_task skipped: another sync is already running")
+        return {
+            "status": "skipped",
+            "reason": "sync_already_running",
+            "task_id": task_id,
+        }
+
+    heartbeat_stop: Event | None = None
+    heartbeat_thread: Thread | None = None
+
+    try:
+        heartbeat_stop, heartbeat_thread = _start_lock_heartbeat(
+            redis_client,
+            ENCAR_SYNC_LOCK_KEY,
+            owner_token,
+            lock_ttl,
+        )
+        self.update_state(state="STARTED", meta={"stage": "encar_sync_started", "task_id": task_id})
+
+        # --- Загрузка runtime_config ---
+        rc = load_runtime_config()
+        effective_limit = limit or rc.sync.limit
+        effective_lane = lane or rc.sync.lane or "encar"
+
+        car_type_map = {"all": "A", "domestic": "Y", "import": "N"}
+        filters = EncarFilters(
+            car_type=car_type_map.get(car_type, "A"),
+            manufacturer=manufacturer,
+            year_from=year_from,
+            year_to=year_to,
+            price_to=price_max,
+        )
+
+        # Набор разрешённых брендов (lowercased) из runtime_config
+        # Расширяем английские имена alias'ами (корейский + русский перевод)
+        allowed_brands = expand_allowed_brands(
+            set(rc.filters.brands) if rc.filters.brands else None
+        )
+        excluded_brands = expand_allowed_brands(
+            set(rc.filters.exclude_brands) if rc.filters.exclude_brands else None
+        )
+
+        scraper = EncarScraper()
+        result = scraper.sync_listing(
+            limit=effective_limit,
+            filters=filters,
+            lane=effective_lane,
+            redis_client=redis_client,
+            allowed_brands=allowed_brands,
+            excluded_brands=excluded_brands,
+            probe_all_photos=rc.sync.probe_all_photos,
+        )
+
+        summary = {
+            "task_id": task_id,
+            "source": "encar",
+            "total_available": result.get("total_available", 0),
+            "items_collected": result.get("items_collected", 0),
+            "cars_synced": result.get("cars_synced", 0),
+            "cars_failed": result.get("cars_failed", 0),
+            "cars_marked_sold": result.get("cars_marked_sold", 0),
+        }
+        logger.info(
+            "encar_sync_listing_task completed: %d synced, %d failed, %d marked sold",
+            summary["cars_synced"], summary["cars_failed"], summary["cars_marked_sold"],
+        )
+        return {"status": "success", **summary}
+
+    except Exception as exc:
+        logger.error("encar_sync_listing_task failed: %s", exc, exc_info=True)
+        raise self.retry(exc=exc)
+    finally:
+        if heartbeat_stop is not None:
+            heartbeat_stop.set()
+        if heartbeat_thread is not None:
+            heartbeat_thread.join(timeout=max(1.0, min(5.0, lock_ttl / 10)))
+        if lock_acquired:
+            _release_lock_if_owner(redis_client, ENCAR_SYNC_LOCK_KEY, owner_token)
+
+
+@shared_task(
+    name="encar_scraper.worker.tasks.encar_sync_vehicle_task",
+    bind=True,
+    max_retries=2,
+    default_retry_delay=30,
+    acks_late=True,
+)
+def encar_sync_vehicle_task(self, vehicle_url: str, lane: str = "encar"):
+    """Синхронизация одного авто Encar."""
+    persistence = _get_persistence()
+    persistence.create_tables()
+
+    try:
+        scraper = EncarScraper()
+        result = scraper.sync_vehicle(vehicle_url, lane=lane)
+        logger.info("encar_sync_vehicle_task completed: %s", vehicle_url)
+        return {
+            "status": "success",
+            "vehicle_url": vehicle_url,
+            "origin_id": result.get("origin_id"),
+        }
+    except Exception as exc:
+        logger.error("encar_sync_vehicle_task failed: %s — %s", vehicle_url, exc, exc_info=True)
+        raise self.retry(exc=exc)