diff --git a/alembic/versions/005_add_car_seen_sold_timestamps.py b/alembic/versions/005_add_car_seen_sold_timestamps.py new file mode 100644 index 0000000..993e1d1 --- /dev/null +++ b/alembic/versions/005_add_car_seen_sold_timestamps.py @@ -0,0 +1,31 @@ +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "005_add_car_seen_sold_timestamps" +down_revision: Union[str, None] = "004_add_ingestion_tables" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column( + "MOBILEDE_cars", + sa.Column("first_seen_at", sa.DateTime(timezone=True), nullable=True), + ) + op.add_column( + "MOBILEDE_cars", + sa.Column("sold_at", sa.DateTime(timezone=True), nullable=True), + ) + op.execute('UPDATE "MOBILEDE_cars" SET first_seen_at = last_seen_at WHERE first_seen_at IS NULL') + op.alter_column("MOBILEDE_cars", "first_seen_at", nullable=False) + op.create_index("ix_MOBILEDE_cars_first_seen_at", "MOBILEDE_cars", ["first_seen_at"]) + op.create_index("ix_MOBILEDE_cars_sold_at", "MOBILEDE_cars", ["sold_at"]) + + +def downgrade() -> None: + op.drop_index("ix_MOBILEDE_cars_sold_at", table_name="MOBILEDE_cars") + op.drop_index("ix_MOBILEDE_cars_first_seen_at", table_name="MOBILEDE_cars") + op.drop_column("MOBILEDE_cars", "sold_at") + op.drop_column("MOBILEDE_cars", "first_seen_at") diff --git a/mobilede_scraper/mobile_de/scraper.py b/mobilede_scraper/mobile_de/scraper.py index 35c2146..733d607 100644 --- a/mobilede_scraper/mobile_de/scraper.py +++ b/mobilede_scraper/mobile_de/scraper.py @@ -4,6 +4,7 @@ import logging import os from collections.abc import Callable from dataclasses import asdict +from datetime import datetime, timezone from typing import Any import requests @@ -19,6 +20,7 @@ logger = logging.getLogger("mobile_de.scraper") MOBILEDE_ONLY_NEW_STOP_ON_EXISTING_STREAK = max(0, int(os.getenv("MOBILEDE_ONLY_NEW_STOP_ON_EXISTING_STREAK", "2"))) MOBILEDE_ONLY_NEW_MIN_NEW_RECORDS = max(0, int(os.getenv("MOBILEDE_ONLY_NEW_MIN_NEW_RECORDS", "1"))) +MOBILEDE_LIGHT_REFRESH_EXISTING = os.getenv("MOBILEDE_LIGHT_REFRESH_EXISTING", "false").strip().lower() in {"1", "true", "yes", "on"} MOBILEDE_SEARCH_STRATEGY_NOTE = ( "mobile.de search pages return about 20 listings per page and are limited to about 50 pages; " "for full coverage split into narrower segments and deduplicate by id." @@ -102,13 +104,15 @@ class MobileDeScraper: skipped_existing: int, existing_streak: int, new_records_kept: int, + existing_origin_ids: set[str] | None = None, ) -> tuple[list[CarRecord], int, int, int, bool]: if not only_new or not page_records: return page_records, skipped_existing, existing_streak, new_records_kept, False - existing_origin_ids = self.persistence.get_existing_origin_ids( - [record.origin_id for record in page_records if record.origin_id] - ) + if existing_origin_ids is None: + existing_origin_ids = self.persistence.get_existing_origin_ids( + [record.origin_id for record in page_records if record.origin_id] + ) head_cut_triggered = False should_cut_tail = self._should_cut_only_new_tail(sort_by, sort_order) @@ -289,10 +293,14 @@ class MobileDeScraper: mileage_max: str | None = None, sort_by: str | None = None, sort_order: str | None = None, + seen_at: datetime | None = None, progress_callback: Callable[[str, dict[str, Any]], None] | None = None, ) -> dict[str, Any]: self.persistence.create_tables() run_id = self.persistence.start_sync_run(lane) + run_seen_at = seen_at or datetime.now(timezone.utc) + if run_seen_at.tzinfo is None: + run_seen_at = run_seen_at.replace(tzinfo=timezone.utc) pages_payload: list[dict[str, Any]] = [] unique_ids: set[str] = set() listing_count = 0 @@ -354,6 +362,17 @@ class MobileDeScraper: page_records = [self.mapper.listing_to_car_record(listing) for listing in page.listings] page_records = self._dedupe_page_records(page_records, seen_record_keys) + for record in page_records: + record.is_sold = False + record.first_seen_at = run_seen_at + record.last_seen_at = run_seen_at + record.sold_at = None + record.skip_image_sync = False + existing_origin_ids: set[str] = set() + if page_records and (only_new or MOBILEDE_LIGHT_REFRESH_EXISTING): + existing_origin_ids = self.persistence.get_existing_origin_ids( + [record.origin_id for record in page_records if record.origin_id] + ) page_records, skipped_existing, existing_streak, new_records_kept, head_cut_triggered = ( self._apply_only_new_page_policy( page_records=page_records, @@ -363,8 +382,13 @@ class MobileDeScraper: skipped_existing=skipped_existing, existing_streak=existing_streak, new_records_kept=new_records_kept, + existing_origin_ids=existing_origin_ids, ) ) + if MOBILEDE_LIGHT_REFRESH_EXISTING and existing_origin_ids: + for record in page_records: + if record.origin_id and record.origin_id in existing_origin_ids: + record.skip_image_sync = True if progress_callback is not None: progress_callback( diff --git a/mobilede_scraper/storage/db.py b/mobilede_scraper/storage/db.py index 926b97a..dfcef83 100644 --- a/mobilede_scraper/storage/db.py +++ b/mobilede_scraper/storage/db.py @@ -20,6 +20,7 @@ CAR_DB_FIELDS = { col.key for col in Car.__table__.columns if col.key not in ("id",) } +CAR_UPDATE_FIELDS = CAR_DB_FIELDS - {"first_seen_at"} _IN_CHUNK_SIZE = 5000 CAR_TABLE_NAME = Car.__tablename__ @@ -170,6 +171,16 @@ class PersistenceService: payload = record.model_dump(mode="python") return {key: value for key, value in payload.items() if key in CAR_DB_FIELDS} + @staticmethod + def _apply_update_payload(car: Car, payload: dict[str, object]) -> None: + for key, value in payload.items(): + if key in CAR_UPDATE_FIELDS: + setattr(car, key, value) + + @staticmethod + def _skip_image_sync(record: CarRecord) -> bool: + return bool(getattr(record, "skip_image_sync", False)) + def _is_postgres(self) -> bool: return self.engine.dialect.name == "postgresql" @@ -227,7 +238,7 @@ class PersistenceService: def _postgres_upsert_set_map(insert_stmt) -> dict[str, object]: return { key: getattr(insert_stmt.excluded, key) - for key in CAR_DB_FIELDS + for key in CAR_UPDATE_FIELDS } def _replace_images_for_car( @@ -266,12 +277,16 @@ class PersistenceService: 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, "action": "inserted"} + entry: dict[str, object] = { + "record": record, + "images": images, + "car_id": None, + "action": "inserted", + "skip_image_sync": self._skip_image_sync(record), + } 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 + self._apply_update_payload(car_by_url, payload) entry["car_id"] = int(car_by_url.id) entry["action"] = "updated" updated += 1 @@ -303,37 +318,54 @@ class PersistenceService: 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] = [] + insert_images_by_car_id: dict[int, list[dict[str, object]]] = {} + updated_entries_needing_compare: list[dict[str, object]] = [] for entry in entries: car_id = int(entry["car_id"]) action = str(entry.get("action") or "updated") images = entry["images"] - if MOBILEDE_SKIP_IMAGES_FOR_UPDATED and action == "updated": + skip_image_sync = bool(entry.get("skip_image_sync")) + if action == "updated" and (MOBILEDE_SKIP_IMAGES_FOR_UPDATED or skip_image_sync): continue - 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 action == "inserted": + insert_images_by_car_id[car_id] = images + continue + updated_entries_needing_compare.append(entry) - 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) + for car_id, images in insert_images_by_car_id.items(): + self._add_images(session, car_id, images) + images_total += len(images) + + if updated_entries_needing_compare: + car_ids = {int(entry["car_id"]) for entry in updated_entries_needing_compare} + 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 updated_entries_needing_compare: + 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} @@ -347,9 +379,7 @@ class PersistenceService: 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 + self._apply_update_payload(car_by_url, payload) session.flush() car_id = int(car_by_url.id) action = "updated" @@ -380,9 +410,7 @@ class PersistenceService: session.flush() else: action = "updated" - for key, value in payload.items(): - setattr(car, key, value) - car.last_seen_at = record.last_seen_at + self._apply_update_payload(car, payload) 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} @@ -436,18 +464,8 @@ class PersistenceService: 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)) - - # Готовим map car_id -> 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]]] = [] + update_image_candidates: list[tuple[Car, list[dict]]] = [] for record in records: payload = self._car_payload(record) @@ -460,18 +478,11 @@ class PersistenceService: 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 + self._apply_update_payload(car, payload) 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) + if MOBILEDE_SKIP_IMAGES_FOR_UPDATED or self._skip_image_sync(record): + continue + update_image_candidates.append((car, images)) # Один flush. session.flush() @@ -481,6 +492,18 @@ class PersistenceService: self._add_images(session, int(car.id), images) images_total += len(images) + update_cars_needing_images: list[tuple[Car, list[dict]]] = [] + if update_image_candidates: + update_ids = [int(car.id) for car, _ in update_image_candidates] + existing_images_map = self._load_existing_image_urls(session, set(update_ids)) + for car, images in update_image_candidates: + 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) + # Обновляем только изменённые картинки. if update_cars_needing_images: update_ids = [int(car.id) for car, _ in update_cars_needing_images] @@ -521,7 +544,7 @@ class PersistenceService: .where(Car.origin_id.notin_(active_origin_ids)) .where(Car.is_sold == False) # noqa: E712 .where(_origin_prefix_filter(Car.origin_id)) - .values(is_sold=True) + .values(is_sold=True, sold_at=datetime.now(timezone.utc)) ) result = session.execute(stmt) count = result.rowcount or 0 @@ -598,7 +621,7 @@ class PersistenceService: # Массовая пометка sold. result = session.execute(text(""" UPDATE {car_table} - SET is_sold = TRUE + SET is_sold = TRUE, sold_at = NOW() FROM ( SELECT c.id FROM {car_table} c @@ -616,7 +639,7 @@ class PersistenceService: update(Car) .where(Car.is_sold == False) # noqa: E712 .where(_origin_prefix_filter(Car.origin_id)) - .values(is_sold=True) + .values(is_sold=True, sold_at=datetime.now(timezone.utc)) ) # Загружаем active URL. all_active = session.execute( @@ -638,7 +661,7 @@ class PersistenceService: for i in range(0, len(mark_ids), _IN_CHUNK_SIZE): chunk = mark_ids[i:i + _IN_CHUNK_SIZE] - session.execute(update(Car).where(Car.id.in_(chunk)).values(is_sold=True)) + session.execute(update(Car).where(Car.id.in_(chunk)).values(is_sold=True, sold_at=datetime.now(timezone.utc))) count = len(mark_ids) if count: @@ -699,7 +722,7 @@ class PersistenceService: .where(_origin_prefix_filter(Car.origin_id, prefixes)) .where(Car.is_sold == False) # noqa: E712 .where(Car.last_seen_at < since_ts) - .values(is_sold=True) + .values(is_sold=True, sold_at=since_ts) ) count = int(result.rowcount or 0) if count: @@ -762,3 +785,41 @@ class PersistenceService: ).order_by(Car.last_seen_at.asc()).offset(offset).limit(limit) ) return [str(row[0]) for row in result if row and row[0]] + + def get_active_cars_batch_for_sold_probe( + self, + prefix: str | tuple[str, ...] = MOBILEDE_ORIGIN_PREFIXES, + *, + limit: int = 200, + newest_first: bool = True, + ) -> list[tuple[int, str, datetime]]: + prefixes = (prefix,) if isinstance(prefix, str) else tuple(prefix) + order_by = Car.last_seen_at.desc() if newest_first else Car.last_seen_at.asc() + with self.session_scope() as session: + result = session.execute( + select(Car.id, Car.origin_url, Car.last_seen_at).where( + _origin_prefix_filter(Car.origin_id, prefixes), + Car.is_sold == False, # noqa: E712 + ).order_by(order_by).limit(limit) + ) + return [ + (int(row[0]), str(row[1]), row[2]) + for row in result + if row and row[0] and row[1] and row[2] + ] + + def mark_cars_sold_by_ids(self, car_ids: list[int], *, sold_at: datetime | None = None) -> int: + if not car_ids: + return 0 + ts = sold_at or datetime.now(timezone.utc) + with self.session_scope() as session: + result = session.execute( + update(Car) + .where(Car.id.in_(car_ids)) + .where(Car.is_sold == False) # noqa: E712 + .values(is_sold=True, sold_at=ts) + ) + count = int(result.rowcount or 0) + if count: + logger.info("Marked %d cars as sold by explicit id probe", count) + return count diff --git a/mobilede_scraper/storage/models.py b/mobilede_scraper/storage/models.py index 8ae2847..4728abc 100644 --- a/mobilede_scraper/storage/models.py +++ b/mobilede_scraper/storage/models.py @@ -54,7 +54,9 @@ class Car(Base): 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) + first_seen_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=func.now(), index=True) last_seen_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=func.now(), index=True) + sold_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True) images: Mapped[list["Image"]] = relationship("Image", back_populates="car", cascade="all, delete-orphan") diff --git a/mobilede_scraper/storage/schemas.py b/mobilede_scraper/storage/schemas.py index 7af7c43..6591f84 100644 --- a/mobilede_scraper/storage/schemas.py +++ b/mobilede_scraper/storage/schemas.py @@ -37,7 +37,10 @@ class CarRecord(BaseModel): rental: bool = False repair_history: bool = False slug: str + first_seen_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) last_seen_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) + sold_at: datetime | None = None + skip_image_sync: bool = False images: list[ImageRecord] = Field(default_factory=list) @@ -82,6 +85,8 @@ class CarRead(BaseModel): rental: bool = False repair_history: bool = False slug: str = "" + first_seen_at: datetime | None = None last_seen_at: datetime | None = None + sold_at: datetime | None = None images: list[ImageRead] = Field(default_factory=list) diff --git a/mobilede_scraper/worker/celery_app.py b/mobilede_scraper/worker/celery_app.py index 619b21b..3a2b0f3 100644 --- a/mobilede_scraper/worker/celery_app.py +++ b/mobilede_scraper/worker/celery_app.py @@ -11,6 +11,7 @@ from redis import Redis from ..core.config import settings from ..core.logs import setup_logging +from .constants import GLOBAL_DB_PROGRESS_TS_KEY, GLOBAL_PROGRESS_TS_KEY logger = logging.getLogger("mobilede_scraper.worker.celery_app") STARTUP_SYNC_DISPATCH_KEY = "mobilede:state:startup_sync_dispatched" @@ -23,6 +24,9 @@ def _env_bool(name: str, default: bool) -> bool: return raw in {"1", "true", "yes", "on"} +MOBILEDE_BEAT_SYNC_ENABLED = _env_bool("MOBILEDE_BEAT_SYNC_ENABLED", True) + + def _has_fresh_active_progress(redis_client: Redis, *, max_age_seconds: int = 180) -> bool: now = int(time.time()) try: @@ -43,6 +47,18 @@ def _has_fresh_active_progress(redis_client: Redis, *, max_age_seconds: int = 18 return False +def _has_recent_global_progress(redis_client: Redis, *, max_age_seconds: int = 300) -> bool: + now = int(time.time()) + try: + progress_ts = int(redis_client.get(GLOBAL_PROGRESS_TS_KEY) or 0) + db_progress_ts = int(redis_client.get(GLOBAL_DB_PROGRESS_TS_KEY) or 0) + except Exception: + logger.warning("Failed to inspect global startup progress keys", exc_info=True) + return False + freshest_ts = max(progress_ts, db_progress_ts) + return freshest_ts > 0 and now - freshest_ts <= max_age_seconds + + @celery_setup_logging.connect def _configure_logging(loglevel=None, **kwargs): # Перехватываем логирование Celery и пишем только в stderr (Docker logs). @@ -85,6 +101,25 @@ if _hard > _max_hard: ) _hard = _max_hard +beat_schedule = {} +if MOBILEDE_BEAT_SYNC_ENABLED: + beat_schedule = { + "periodic-mobilede-sync-search": { + "task": "mobilede.sync_runtime_segments", + "schedule": settings.celery.beat_sync_interval_minutes * 60.0, + "args": (), + "kwargs": { + "delay_seconds": float(os.getenv("MOBILEDE_REQUEST_DELAY_SECONDS", "0.7")), + "use_cursor": _env_bool("MOBILEDE_CURSOR_ENABLED", True), + "continuous": _env_bool("MOBILEDE_CONTINUOUS_SYNC_ENABLED", True), + }, + "options": { + "queue": MOBILEDE_SYNC_QUEUE, + "expires": settings.celery.beat_sync_interval_minutes * 60.0, + }, + } + } + celery_app.conf.update( task_serializer="json", accept_content=["json"], @@ -107,22 +142,7 @@ celery_app.conf.update( result_expires=86400, worker_redirect_stdouts=False, worker_hijack_root_logger=False, - beat_schedule={ - "periodic-mobilede-sync-search": { - "task": "mobilede.sync_runtime_segments", - "schedule": settings.celery.beat_sync_interval_minutes * 60.0, - "args": (), - "kwargs": { - "delay_seconds": float(os.getenv("MOBILEDE_REQUEST_DELAY_SECONDS", "0.7")), - "use_cursor": _env_bool("MOBILEDE_CURSOR_ENABLED", True), - "continuous": _env_bool("MOBILEDE_CONTINUOUS_SYNC_ENABLED", True), - }, - "options": { - "queue": MOBILEDE_SYNC_QUEUE, - "expires": settings.celery.beat_sync_interval_minutes * 60.0, - }, - } - }, + beat_schedule=beat_schedule, task_routes={ "mobilede.sync_runtime_segments": {"queue": MOBILEDE_SYNC_QUEUE}, "mobilede.sync_search": {"queue": MOBILEDE_SYNC_QUEUE}, @@ -153,17 +173,24 @@ def _on_worker_ready(**kwargs): ) has_fresh_progress = _has_fresh_active_progress(redis_client) + has_recent_global_progress = _has_recent_global_progress(redis_client) + has_live_progress = bool(has_fresh_progress or has_recent_global_progress) try: queue_len = int(redis_client.llen(MOBILEDE_SYNC_QUEUE) or 0) except Exception: queue_len = 0 - if queue_len > 0: + if queue_len > 0 and has_live_progress: logger.info("Worker ready: MOBILEDE_sync queue already has %d task(s); skip startup dispatch", queue_len) return + if queue_len > 0 and not has_live_progress: + logger.warning( + "Worker ready: MOBILEDE_sync queue has %d task(s), but no fresh progress is visible; forcing runtime sync dispatch", + queue_len, + ) should_dispatch = bool(redis_client.set(STARTUP_SYNC_DISPATCH_KEY, "1", nx=True, ex=600)) - if not should_dispatch and not has_fresh_progress: + if not should_dispatch and not has_live_progress: redis_client.delete(STARTUP_SYNC_DISPATCH_KEY) should_dispatch = bool(redis_client.set(STARTUP_SYNC_DISPATCH_KEY, "1", nx=True, ex=600)) if should_dispatch: @@ -182,7 +209,7 @@ def _on_worker_ready(**kwargs): logger.info("Worker ready immediate sync already dispatched recently; skipping duplicate enqueue") return - logger.info("Worker ready — dispatching initial mobile.de sync_search task") + logger.info("Worker ready - dispatching initial mobile.de sync_runtime_segments task") celery_app.send_task( "mobilede.sync_runtime_segments", kwargs={ diff --git a/tests/test_db.py b/tests/test_db.py index b38b6ac..b3f8cb9 100644 --- a/tests/test_db.py +++ b/tests/test_db.py @@ -2,6 +2,7 @@ import tempfile import unittest +from datetime import datetime, timedelta, timezone from pathlib import Path from sqlalchemy import select @@ -48,6 +49,14 @@ class TestPersistenceServiceIntegration(unittest.TestCase): ], ) + @staticmethod + def _as_utc(value: datetime | None) -> datetime | None: + if value is None: + return None + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + def test_insert_update_and_skip_flow(self) -> None: first = self._record("777", price=1000) inserted = self.persistence.upsert_car(first) @@ -122,6 +131,57 @@ class TestPersistenceServiceIntegration(unittest.TestCase): self.assertEqual(len(cars), 1) self.assertEqual(cars[0].origin_id, "NEW-ID") + def test_upsert_preserves_first_seen_and_reactivates_seen_car(self) -> None: + first_seen = datetime(2026, 5, 1, tzinfo=timezone.utc) + sold_at = datetime(2026, 5, 2, tzinfo=timezone.utc) + seen_at = datetime(2026, 5, 5, tzinfo=timezone.utc) + first = self._record("mobile.de:777", price=1000) + first.first_seen_at = first_seen + first.last_seen_at = first_seen + self.persistence.upsert_car(first) + + with self.persistence.session_scope() as session: + car = session.execute(select(Car).where(Car.origin_id == "mobile.de:777")).scalar_one() + car.is_sold = True + car.sold_at = sold_at + + updated = self._record("mobile.de:777", price=1500) + updated.first_seen_at = seen_at + updated.last_seen_at = seen_at + updated.is_sold = False + updated.sold_at = None + self.persistence.upsert_car(updated) + + with self.persistence.session_scope() as session: + car = session.execute(select(Car).where(Car.origin_id == "mobile.de:777")).scalar_one() + + self.assertEqual(self._as_utc(car.first_seen_at), first_seen) + self.assertEqual(self._as_utc(car.last_seen_at), seen_at) + self.assertFalse(car.is_sold) + self.assertIsNone(car.sold_at) + self.assertEqual(car.price, 1500) + + def test_mark_sold_not_seen_since_sets_sold_at_to_cycle_seen_at(self) -> None: + seen_at = datetime(2026, 5, 5, tzinfo=timezone.utc) + old = self._record("mobile.de:old") + old.first_seen_at = seen_at - timedelta(days=2) + old.last_seen_at = seen_at - timedelta(days=1) + current = self._record("mobile.de:current") + current.first_seen_at = seen_at + current.last_seen_at = seen_at + self.persistence.upsert_car(old) + self.persistence.upsert_car(current) + + marked = self.persistence.mark_sold_not_seen_since(seen_at) + + self.assertEqual(marked, 1) + with self.persistence.session_scope() as session: + cars = session.execute(select(Car).order_by(Car.origin_id.asc())).scalars().all() + + sold_map = {car.origin_id: (car.is_sold, self._as_utc(car.sold_at)) for car in cars} + self.assertEqual(sold_map["mobile.de:old"], (True, seen_at)) + self.assertEqual(sold_map["mobile.de:current"], (False, None)) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_mobilede_scraper.py b/tests/test_mobilede_scraper.py index c780110..eb17f70 100644 --- a/tests/test_mobilede_scraper.py +++ b/tests/test_mobilede_scraper.py @@ -1,6 +1,7 @@ from __future__ import annotations import unittest +from datetime import datetime, timezone from mobilede_scraper.mobile_de.models import MobileDeListing, MobileDeSearchPage from mobilede_scraper.mobile_de.scraper import MobileDeScraper @@ -9,6 +10,7 @@ from mobilede_scraper.mobile_de.scraper import MobileDeScraper class _FakePersistence: def __init__(self) -> None: self.upsert_calls: list[list[str]] = [] + self.upsert_records: list[object] = [] self.finish_payload: dict[str, object] | None = None def create_tables(self) -> None: @@ -22,6 +24,7 @@ class _FakePersistence: def upsert_cars_batch(self, records): self.upsert_calls.append([record.origin_id for record in records]) + self.upsert_records.extend(records) return { "inserted": len(records), "updated": 0, @@ -82,6 +85,31 @@ class TestMobileDeScraperStreamingSync(unittest.TestCase): self.assertIsNotNone(persistence.finish_payload) self.assertEqual(int(persistence.finish_payload["cars_upserted"]), 3) + def test_sync_search_applies_one_seen_at_to_all_records_in_run(self) -> None: + seen_at = datetime(2026, 5, 5, 12, 30, tzinfo=timezone.utc) + pages = [ + MobileDeSearchPage( + url="https://example.test/page-1", + page_number=1, + total_results=2, + listings=[ + MobileDeListing(id="1", url="https://example.test/1", title="Car 1"), + MobileDeListing(id="2", url="https://example.test/2", title="Car 2"), + ], + ), + ] + persistence = _FakePersistence() + scraper = MobileDeScraper(client=_FakeClient(pages), persistence=persistence) + + scraper.sync_search(max_pages=1, seen_at=seen_at) + + records = persistence.upsert_calls + self.assertEqual(records, [["mobile.de:1", "mobile.de:2"]]) + self.assertTrue(all(record.first_seen_at == seen_at for record in persistence.upsert_records)) + self.assertTrue(all(record.last_seen_at == seen_at for record in persistence.upsert_records)) + self.assertTrue(all(record.is_sold is False for record in persistence.upsert_records)) + self.assertTrue(all(record.sold_at is None for record in persistence.upsert_records)) + if __name__ == "__main__": unittest.main()