Track sold cars

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

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,10 +104,12 @@ 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
 
+        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]
             )
@@ -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(

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,17 +318,34 @@ 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
+				if action == "inserted":
+					insert_images_by_car_id[car_id] = images
+					continue
+				updated_entries_needing_compare.append(entry)
+
+			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
@@ -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

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

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)
 

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={

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

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