1277 lines
42 KiB
Python
1277 lines
42 KiB
Python
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
|