Prepare mobile de parser preview

iaai_scraper/browser/network.py

@@ -0,0 +1,130 @@
+import json
+import logging
+from dataclasses import dataclass, field
+from typing import Any
+from urllib.parse import urlparse
+
+from playwright.sync_api import Page, Request, Response
+
+from ..core.config import Settings
+
+logger = logging.getLogger("iaai_scraper.network")
+
+
+@dataclass
+class NetworkCapture:
+    # Перехватчик сетевых запросов.
+
+    settings: Settings
+    requests: list[dict[str, Any]] = field(default_factory=list)
+    json_responses: list[dict[str, Any]] = field(default_factory=list)
+    _seen_req: set[str] = field(default_factory=set)
+    _seen_resp: set[str] = field(default_factory=set)
+    _origin: str | None = None
+    _page: Any = field(default=None)
+
+    def attach(self, page: Page, origin_url: str | None = None) -> None:
+        # Снимаем старые подписки перед повторным attach.
+        try:
+            page.remove_listener("request", self._on_request)
+            page.remove_listener("response", self._on_response)
+        except Exception:
+            pass
+        # Подписка на сетевые события
+        try:
+            self._origin = urlparse(origin_url or page.url).netloc.lower() or None
+        except Exception:
+            self._origin = None
+        self._page = page
+        page.on("request", self._on_request)
+        page.on("response", self._on_response)
+
+    def _is_same_origin(self, url: str) -> bool:
+        # Фильтр по домену
+        if not self.settings.capture.capture_same_origin_only or not self._origin:
+            return True
+        netloc = urlparse(url).netloc.lower()
+        return netloc == self._origin or netloc.endswith(".iaai.com")
+
+    def _on_request(self, request: Request) -> None:
+        # Берём только xhr/fetch
+        if request.resource_type not in {"xhr", "fetch"}:
+            return
+        if not self._is_same_origin(request.url):
+            return
+        if len(self.requests) >= self.settings.capture.max_requests:
+            return
+        key = f"{request.method}:{request.url}:{request.post_data or ''}"
+        # Убираем дубли
+        if key in self._seen_req:
+            return
+        self._seen_req.add(key)
+        self.requests.append({
+            "url": request.url, "method": request.method,
+            "resource_type": request.resource_type, "post_data": request.post_data,
+        })
+
+    def _on_response(self, response: Response) -> None:
+        # Сохраняем JSON
+        request = response.request
+        if request.resource_type not in {"xhr", "fetch"}:
+            return
+        if not self._is_same_origin(response.url):
+            return
+        if len(self.json_responses) >= self.settings.capture.max_json_responses:
+            return
+        if not self._is_json(response):
+            return
+        key = f"{request.method}:{response.url}:{response.status}"
+        if key in self._seen_resp:
+            return
+        self._seen_resp.add(key)
+        try:
+            payload = response.json()
+        except Exception:
+            try:
+                payload = json.loads(response.text())
+            except Exception:
+                return
+        self.json_responses.append({
+            "url": response.url, "status": response.status,
+            "request_method": request.method, "post_data": request.post_data,
+            "resource_type": request.resource_type, "payload": payload,
+            "category": self._categorize(response.url),
+        })
+
+    @staticmethod
+    def _is_json(response: Response) -> bool:
+        ct = (response.headers.get("content-type") or "").lower()
+        if "application/json" in ct or "+json" in ct:
+            return True
+        url = response.url.lower()
+        return any(m in url for m in ["/api/", "/graphql", "vehicledetail", "vehicle", "auction", "bid", "images", "media"])
+
+    @staticmethod
+    def _categorize(url: str) -> str:
+        # Простая категория URL
+        low = url.lower()
+        mapping = {
+            "images": ["image", "media", "photos", "gallery"],
+            "bids": ["bid", "offer", "buy-now", "buynow"],
+            "auction": ["auction", "sale", "lane", "branch"],
+            "vehicle": ["vehicle", "detail", "vin", "damage", "runanddrive"],
+            "documents": ["title", "document", "report"],
+        }
+        for cat, markers in mapping.items():
+            if any(m in low for m in markers):
+                return cat
+        return "other"
+
+    def export(self) -> dict[str, Any]:
+        responses = sorted(self.json_responses, key=lambda i: (i["category"] == "other", i["url"]))
+        return {
+            "requests": self.requests,
+            "json_responses": responses,
+            "capture_limits": {
+                "same_origin_only": self.settings.capture.capture_same_origin_only,
+                "max_requests": self.settings.capture.max_requests,
+                "max_json_responses": self.settings.capture.max_json_responses,
+            },
+        }