feat: extract passive metadata via CDP performance logs and request interception

app/main.py

@@ -238,36 +238,85 @@ def _detect_block_challenge(html: str, status_code: Optional[int]) -> tuple[bool
     return blocked_detected, challenge_detected
 
 
+def _extract_passive_metadata(
+    driver: Driver, target_url: str
+) -> tuple[Optional[int], Optional[dict[str, str]], Optional[str]]:
+    reqs = getattr(driver, "requests", None)
+    if isinstance(reqs, (list, tuple)) and reqs:
+        for req in reversed(reqs):
+            resp = getattr(req, "response", None)
+            if resp:
+                status_code = getattr(resp, "status_code", None)
+                headers = getattr(resp, "headers", None)
+                req_url = getattr(req, "url", None)
+                if status_code is not None:
+                    hdr_dict = {str(k): str(v) for k, v in dict(headers).items()} if headers else None
+                    return int(status_code), hdr_dict, str(req_url) if req_url else None
+
+    get_log = getattr(driver, "get_log", None)
+    if callable(get_log):
+        try:
+            import json
+
+            logs = get_log("performance")
+            if isinstance(logs, list):
+                for entry in reversed(logs):
+                    raw_msg = entry.get("message", "{}") if isinstance(entry, dict) else "{}"
+                    msg_obj = json.loads(raw_msg) if isinstance(raw_msg, str) else raw_msg
+                    msg = msg_obj.get("message", {}) if isinstance(msg_obj, dict) else {}
+                    if msg.get("method") == "Network.responseReceived":
+                        params = msg.get("params", {})
+                        resp = params.get("response", {})
+                        res_type = params.get("type") or resp.get("type")
+                        if res_type in ("Document", "Other", None) or not res_type:
+                            status_code = resp.get("status")
+                            headers = resp.get("headers")
+                            url = resp.get("url")
+                            if status_code is not None:
+                                hdr_dict = {str(k): str(v) for k, v in headers.items()} if isinstance(headers, dict) else None
+                                return int(status_code), hdr_dict, str(url) if url else None
+        except Exception:
+            # Performance logs or CDP responses may be unsupported or malformed; continue fallback.
+            pass
+
+    return None, None, None
+
+
 def _fetch_metadata(
     driver: Driver, target_url: str
 ) -> tuple[Optional[int], Optional[dict[str, str]], str, Optional[str]]:
-    status_code: Optional[int] = None
-    headers: Optional[dict[str, str]] = None
     final_url = getattr(driver, "current_url", None) or target_url
 
+    # 1. Attempt passive CDP / network interception first (zero extra network calls)
+    try:
+        status_code, headers, passive_url = _extract_passive_metadata(driver, target_url)
+        if status_code is not None:
+            return status_code, headers, passive_url or str(final_url), None
+    except Exception:
+        # Passive metadata extraction is best-effort; fall back to active requests.get if supported.
+        pass
+
+    # 2. Fallback to active requests.get if driver provides it
     try:
         request_client = getattr(driver, "requests", None)
-        if request_client is None or not hasattr(request_client, "get"):
-            raise RuntimeError("driver.requests.get is unavailable")
-
-        try:
-            response = request_client.get(target_url, timeout=3)
-        except TypeError:
-            response = request_client.get(target_url)
-
-        status_code = getattr(response, "status_code", None)
-        response_headers = getattr(response, "headers", None)
-
-        if response_headers:
-            headers = {str(k): str(v) for k, v in dict(response_headers).items()}
-
-        response_url = getattr(response, "url", None)
-        if response_url:
-            final_url = str(response_url)
-
-        return status_code, headers, final_url, None
+        if request_client is not None and hasattr(request_client, "get"):
+            try:
+                response = request_client.get(target_url, timeout=3)
+            except TypeError:
+                response = request_client.get(target_url)
+
+            status_code = getattr(response, "status_code", None)
+            response_headers = getattr(response, "headers", None)
+            headers = {str(k): str(v) for k, v in dict(response_headers).items()} if response_headers else None
+            response_url = getattr(response, "url", None)
+            if response_url:
+                final_url = str(response_url)
+
+            return status_code, headers, str(final_url), None
     except Exception as exc:  # best-effort metadata only
-        return status_code, headers, str(final_url), str(exc)
+        return None, None, str(final_url), str(exc)
+
+    return None, None, str(final_url), None
 
 
 def _register_request_id(request_id: str) -> None:

tests/test_api_contract.py

@@ -142,6 +142,50 @@ def test_run_scrape_forwards_driver_kwargs(self):
             _CaptureDriver.last_init_kwargs["proxy"], "http://proxy.example:8080"
         )
 
+    def test_extract_passive_metadata_from_requests_list(self):
+        class _Req:
+            def __init__(self, status, headers, url):
+                self.response = type("Resp", (), {"status_code": status, "headers": headers})()
+                self.url = url
+
+        driver = type("D", (), {"requests": [_Req(200, {"content-type": "text/html"}, "https://example.com/final")]})()
+        status, headers, final_url = main._extract_passive_metadata(driver, "https://example.com")
+        self.assertEqual(status, 200)
+        self.assertEqual(headers, {"content-type": "text/html"})
+        self.assertEqual(final_url, "https://example.com/final")
+
+    def test_extract_passive_metadata_from_performance_logs(self):
+        import json
+
+        perf_log = [
+            {
+                "message": json.dumps(
+                    {
+                        "message": {
+                            "method": "Network.responseReceived",
+                            "params": {
+                                "type": "Document",
+                                "response": {
+                                    "status": 200,
+                                    "headers": {"content-type": "text/html"},
+                                    "url": "https://example.com/cdp-final",
+                                },
+                            },
+                        }
+                    }
+                )
+            }
+        ]
+        driver = type(
+            "D",
+            (),
+            {"get_log": lambda self, log_type: perf_log if log_type == "performance" else []},
+        )()
+        status, headers, final_url = main._extract_passive_metadata(driver, "https://example.com")
+        self.assertEqual(status, 200)
+        self.assertEqual(headers, {"content-type": "text/html"})
+        self.assertEqual(final_url, "https://example.com/cdp-final")
+
     def test_is_blocked_ip_allows_well_known_nat64_prefix(self):
         nat64_ip = ipaddress.ip_address("64:ff9b::3691:8e03")
         self.assertFalse(main._is_blocked_ip(nat64_ip))