IAAI scraper: Playwright + SQLAlchemy, парсинг авто с аукциона, Docker-ready

tests/conftest.py

@@ -0,0 +1,7 @@
+from __future__ import annotations
+
+import logging
+
+
+def pytest_configure() -> None:
+    logging.disable(logging.CRITICAL)

tests/test_db.py

@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from sqlalchemy import select
+
+from iaai_scraper.core.config import Settings
+from iaai_scraper.storage.db import PersistenceService
+from iaai_scraper.storage.models import Car, Image
+from iaai_scraper.storage.schemas import CarRecord, ImageRecord
+
+
+class TestPersistenceServiceIntegration(unittest.TestCase):
+    def setUp(self) -> None:
+        self.tmp_dir = tempfile.TemporaryDirectory()
+        db_path = Path(self.tmp_dir.name) / "test.sqlite"
+
+        self.settings = Settings()
+        self.settings.database.url = f"sqlite:///{db_path.as_posix()}"
+        self.settings.database.echo = False
+
+        self.persistence = PersistenceService(self.settings)
+        self.persistence.create_tables()
+
+    def tearDown(self) -> None:
+        self.persistence.engine.dispose()
+        self.tmp_dir.cleanup()
+
+    @staticmethod
+    def _record(origin_id: str, *, price: int = 1000, content_hash: str = "hash1") -> CarRecord:
+        return CarRecord(
+            parser_id=f"iaai:{origin_id}",
+            brand="Toyota",
+            model="Camry",
+            year=2014,
+            price=price,
+            origin_url=f"https://www.iaai.com/VehicleDetail/{origin_id}~US",
+            origin_id=origin_id,
+            slug=f"toyota-camry-{origin_id}",
+            content_hash=content_hash,
+            images=[
+                ImageRecord(
+                    fullres_image="https://vis.iaai.com/resizer?imageKeys=1&width=845&height=633",
+                    preview_image="https://vis.iaai.com/resizer?imageKeys=1&width=400&height=300",
+                    order_index=0,
+                )
+            ],
+            raw_attributes={"foo": "bar"},
+        )
+
+    def test_insert_update_and_skip_flow(self) -> None:
+        first = self._record("777", price=1000, content_hash="same")
+        inserted = self.persistence.upsert_car(first)
+        self.assertEqual(inserted["action"], "inserted")
+        self.assertEqual(inserted["images_upserted"], 1)
+
+        same = self._record("777", price=1000, content_hash="same")
+        updated_same = self.persistence.upsert_car(same)
+        self.assertEqual(updated_same["action"], "updated")
+        self.assertEqual(updated_same["images_upserted"], 1)
+
+        changed = self._record("777", price=1500, content_hash="changed")
+        updated = self.persistence.upsert_car(changed)
+        self.assertEqual(updated["action"], "updated")
+        self.assertEqual(updated["images_upserted"], 1)
+
+        with self.persistence.session_scope() as session:
+            cars = session.execute(select(Car)).scalars().all()
+            images = session.execute(select(Image)).scalars().all()
+
+        self.assertEqual(len(cars), 1)
+        self.assertEqual(cars[0].price, 1500)
+        self.assertEqual(len(images), 1)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_listing.py

@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+import unittest
+
+from iaai_scraper.browser.pace import HumanPacer
+from iaai_scraper.core.config import Settings
+from iaai_scraper.storage.listing import ListingCollector
+
+
+class _FakePage:
+    def __init__(self, counts: dict[str, int]) -> None:
+        self._counts = counts
+
+    class _Locator:
+        def __init__(self, count_value: int) -> None:
+            self._count_value = count_value
+
+        def count(self) -> int:
+            return self._count_value
+
+    def locator(self, selector: str) -> "_FakePage._Locator":
+        return _FakePage._Locator(self._counts.get(selector, 0))
+
+
+class TestListingUnit(unittest.TestCase):
+    def test_has_next_page_true_for_known_selector(self) -> None:
+        page = _FakePage({"a[aria-label*='Next']": 1})
+        self.assertTrue(ListingCollector._has_next_page(page))
+
+    def test_has_next_page_false_when_no_selectors(self) -> None:
+        page = _FakePage({})
+        self.assertFalse(ListingCollector._has_next_page(page))
+
+    def test_constructor_with_settings_and_pacer(self) -> None:
+        settings = Settings()
+        pacer = HumanPacer(settings)
+        collector = ListingCollector(settings, pacer)
+        self.assertIsNotNone(collector)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_mappers.py

@@ -0,0 +1,62 @@
+from __future__ import annotations
+
+import unittest
+
+from iaai_scraper.parsing.mapper import CarMapper
+
+
+class TestCarMapper(unittest.TestCase):
+    def setUp(self) -> None:
+        self.mapper = CarMapper()
+
+    def test_content_hash_is_sha256(self) -> None:
+        record = self.mapper.map_to_car_record(
+            vehicle_url="https://www.iaai.com/VehicleDetail/45089484~US",
+            vehicle_summary={"make": "Toyota", "model": "Camry", "year": "2014"},
+            payload_insights={
+                "vehicle_core": {"odometer": "120,000", "body_type": "sedan"},
+                "pricing": {"buy_now": "$4,500"},
+                "damage": {"primary": "normal wear"},
+                "auction": {},
+                "images": {"urls": []},
+            },
+        )
+
+        # Длина hex-представления SHA-256
+        self.assertEqual(len(record.content_hash), 64)
+
+    def test_deduplicates_images_by_image_key(self) -> None:
+        urls = [
+            "https://vis.iaai.com/resizer?imageKeys=1&width=200&height=150",
+            "https://vis.iaai.com/resizer?imageKeys=1&width=845&height=633",
+            "https://vis.iaai.com/resizer?imageKeys=2&width=400&height=300",
+        ]
+
+        record = self.mapper.map_to_car_record(
+            vehicle_url="https://www.iaai.com/VehicleDetail/123~US",
+            vehicle_summary={"make": "Honda", "model": "Civic", "image_urls": urls},
+            payload_insights={"vehicle_core": {}, "pricing": {}, "damage": {}, "auction": {}, "images": {}},
+        )
+
+        self.assertEqual(len(record.images), 2)
+        self.assertIn("width=845", record.images[0].fullres_image)
+        self.assertIn("height=633", record.images[0].fullres_image)
+
+    def test_no_damage_marker_is_not_damaged(self) -> None:
+        record = self.mapper.map_to_car_record(
+            vehicle_url="https://www.iaai.com/VehicleDetail/123~US",
+            vehicle_summary={"make": "Ford", "model": "Focus"},
+            payload_insights={
+                "vehicle_core": {},
+                "pricing": {},
+                "damage": {"primary": "normal wear"},
+                "auction": {},
+                "images": {},
+            },
+        )
+
+        self.assertFalse(record.is_damaged)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_parser.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+import unittest
+
+from iaai_scraper.parsing.parser import VehicleParser
+
+
+class TestVehicleParserUnit(unittest.TestCase):
+    def setUp(self) -> None:
+        self.parser = VehicleParser()
+
+    def test_parse_dom_key_value_pairs_extracts_known_fields(self) -> None:
+        dom_text = """
+        Stock #:
+        45089484
+        VIN (Status):
+        1HGCM82633A123456 (OK)
+        Primary Damage:
+        Front End
+        """
+        result = self.parser._parse_dom_key_value_pairs(dom_text)
+        self.assertEqual(result.get("lot_number"), "45089484")
+        self.assertEqual(result.get("vin"), "1HGCM82633A123456 (OK)")
+        self.assertEqual(result.get("primary_damage"), "Front End")
+
+    def test_parse_title_for_year_make_model(self) -> None:
+        parsed = self.parser._parse_title_for_year_make_model("2014 TOYOTA CAMRY for sale", "")
+        self.assertEqual(parsed["year"], "2014")
+        self.assertEqual(parsed["make"], "TOYOTA")
+        self.assertEqual(parsed["model"], "CAMRY")
+
+    def test_extract_image_urls_filters_other_vehicle(self) -> None:
+        vehicle_url = "https://www.iaai.com/VehicleDetail/45089484~US"
+        payloads = [
+            {
+                "imageUrls": [
+                    "https://vis.iaai.com/resizer?imageKeys=45089484~SID1&width=845&height=633",
+                    "https://vis.iaai.com/resizer?imageKeys=99999999~SID2&width=845&height=633",
+                ]
+            }
+        ]
+        urls = self.parser._extract_image_urls(payloads, "", vehicle_url)
+        self.assertEqual(len(urls), 1)
+        self.assertIn("45089484", urls[0])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_scraper.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+import unittest
+from unittest.mock import MagicMock
+
+from iaai_scraper.core.config import Settings
+from iaai_scraper.scraper import IAAIScraper
+from iaai_scraper.storage.schemas import CarRecord
+
+
+def make_db_record(origin_id: str) -> dict[str, object]:
+    return CarRecord(
+        parser_id=f"iaai:{origin_id}",
+        brand="Toyota",
+        model="Camry",
+        origin_url=f"https://www.iaai.com/VehicleDetail/{origin_id}~US",
+        origin_id=origin_id,
+        slug=f"toyota-camry-{origin_id}",
+    ).model_dump(mode="json")
+
+
+class TestScraperSync(unittest.TestCase):
+    def _make_scraper(self) -> IAAIScraper:
+        s = Settings()
+        s.log_level = "CRITICAL"
+        return IAAIScraper(s)
+
+    def test_sync_vehicle_uses_db_record_without_remapping(self) -> None:
+        scraper = self._make_scraper()
+
+        scraper.persistence.create_tables = MagicMock()
+        scraper.persistence.start_sync_run = MagicMock(return_value=1)
+        scraper.persistence.finish_sync_run = MagicMock()
+        scraper.persistence.upsert_car = MagicMock(return_value={"action": "inserted", "images_upserted": 0})
+
+        scraper.scrape_vehicle_detail = MagicMock(return_value={"db_record": make_db_record("111")})
+        scraper.car_mapper.map_to_car_record = MagicMock(side_effect=AssertionError("should not be called"))
+
+        result = scraper.sync_vehicle("https://www.iaai.com/VehicleDetail/111~US")
+
+        self.assertEqual(result["status"], "success")
+        self.assertEqual(scraper.persistence.upsert_car.call_count, 1)
+
+    def test_sync_listing_uses_db_record_without_remapping(self) -> None:
+        scraper = self._make_scraper()
+
+        scraper.persistence.create_tables = MagicMock()
+        scraper.persistence.start_sync_run = MagicMock(return_value=2)
+        scraper.persistence.finish_sync_run = MagicMock()
+        scraper.persistence.upsert_car = MagicMock(return_value={"action": "inserted", "images_upserted": 1})
+
+        scraper.collect_listing = MagicMock(return_value={"vehicle_urls": ["https://www.iaai.com/VehicleDetail/222~US"]})
+        scraper._scrape_on_page = MagicMock(return_value={"db_record": make_db_record("222")})
+        scraper._get_page = MagicMock(return_value=MagicMock())
+        scraper.car_mapper.map_to_car_record = MagicMock(side_effect=AssertionError("should not be called"))
+
+        result = scraper.sync_listing()
+
+        self.assertEqual(result["cars_upserted"], 1)
+        self.assertEqual(result["cars_failed"], 0)
+        self.assertEqual(scraper.persistence.upsert_car.call_count, 1)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_utils.py

@@ -0,0 +1,31 @@
+from __future__ import annotations
+
+import unittest
+
+from iaai_scraper.core.utils import deep_find_key
+
+
+class TestDeepFindKey(unittest.TestCase):
+    def test_finds_key_in_nested_structure(self) -> None:
+        payload = {
+            "root": {
+                "target": "a",
+                "nested": [{"target": "b"}, {"x": 1}],
+            }
+        }
+
+        found = deep_find_key(payload, {"target"})
+        self.assertEqual(found, ["a", "b"])
+
+    def test_respects_max_depth(self) -> None:
+        payload = {"l1": {"l2": {"l3": {"target": "value"}}}}
+
+        found_too_shallow = deep_find_key(payload, {"target"}, max_depth=2)
+        found_enough_depth = deep_find_key(payload, {"target"}, max_depth=8)
+
+        self.assertEqual(found_too_shallow, [])
+        self.assertEqual(found_enough_depth, ["value"])
+
+
+if __name__ == "__main__":
+    unittest.main()