Apply advanced bot bypass to ScreenerScraper

feature_pipeline.py

@@ -1,7 +1,11 @@
+from __future__ import annotations
 
 import re
 import threading
 from datetime import datetime
+from dataclasses import dataclass
+from typing import Optional, Tuple
+from urllib.parse import quote
 
 import numpy as np
 import pandas as pd
@@ -10,42 +14,92 @@ from bs4 import BeautifulSoup
 from requests.adapters import HTTPAdapter
 from urllib3.util.retry import Retry
 
+def _browser_headers() -> dict:
+    return {
+        "User-Agent": (
+            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+            "AppleWebKit/537.36 (KHTML, like Gecko) "
+            "Chrome/125.0.0.0 Safari/537.36"
+        ),
+        "Accept": (
+            "text/html,application/xhtml+xml,application/xml;"
+            "q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8"
+        ),
+        "Accept-Language": "en-US,en;q=0.9",
+        "Accept-Encoding": "gzip, deflate, br",
+        "Connection": "keep-alive",
+        "Upgrade-Insecure-Requests": "1",
+        "Cache-Control": "no-cache",
+        "Pragma": "no-cache",
+        "Referer": "https://www.screener.in/",
+    }
 
 class _ThreadLocalSessionFactory:
-    def __init__(self, headers=None, timeout=(3.5, 10.0), pool_maxsize=32):
+    def __init__(
+        self,
+        headers: Optional[dict] = None,
+        timeout: Tuple[float, float] = (5.0, 15.0),
+        pool_maxsize: int = 32,
+        prime_homepage: bool = True,
+    ):
         self._local = threading.local()
-        self.headers = headers or {"User-Agent": "Mozilla/5.0"}
+        self.headers = headers or _browser_headers()
         self.timeout = timeout
         self.pool_maxsize = pool_maxsize
+        self.prime_homepage = prime_homepage
+
+    def _build_session(self) -> requests.Session:
+        session = requests.Session()
+        session.headers.update(self.headers)
+
+        retry = Retry(
+            total=3,
+            connect=3,
+            read=3,
+            backoff_factor=0.35,
+            status_forcelist=(429, 500, 502, 503, 504),
+            allowed_methods=frozenset(["GET"]),
+            raise_on_status=False,
+        )
+
+        adapter = HTTPAdapter(
+            max_retries=retry,
+            pool_connections=self.pool_maxsize,
+            pool_maxsize=self.pool_maxsize,
+        )
+
+        session.mount("https://", adapter)
+        session.mount("http://", adapter)
+        return session
 
     def get(self) -> requests.Session:
         if not hasattr(self._local, "session"):
-            session = requests.Session()
-            session.headers.update(self.headers)
-            retry = Retry(
-                total=2,
-                connect=2,
-                read=2,
-                backoff_factor=0.15,
-                status_forcelist=(429, 500, 502, 503, 504),
-                allowed_methods=frozenset(["GET"]),
-                raise_on_status=False,
-            )
-            adapter = HTTPAdapter(
-                max_retries=retry,
-                pool_connections=self.pool_maxsize,
-                pool_maxsize=self.pool_maxsize,
-            )
-            session.mount("https://", adapter)
-            session.mount("http://", adapter)
-            self._local.session = session
-        return self._local.session
+            self._local.session = self._build_session()
+
+            # Warm up cookies / anti-bot checks by visiting the homepage first.
+            if self.prime_homepage:
+                try:
+                    self._local.session.get(
+                        "https://www.screener.in/",
+                        timeout=self.timeout,
+                    )
+                except requests.RequestException:
+                    pass
 
+        return self._local.session
 
 class ScreenerScraper:
-    def __init__(self):
-        self._session_factory = _ThreadLocalSessionFactory()
-        self.timeout = self._session_factory.timeout
+    def __init__(
+        self,
+        timeout: Tuple[float, float] = (5.0, 15.0),
+        pool_maxsize: int = 32,
+    ):
+        self._session_factory = _ThreadLocalSessionFactory(
+            timeout=timeout,
+            pool_maxsize=pool_maxsize,
+            prime_homepage=True,
+        )
+        self.timeout = timeout
 
     def _session(self) -> requests.Session:
         return self._session_factory.get()
@@ -57,31 +111,60 @@ class ScreenerScraper:
         except Exception:
             return BeautifulSoup(html, "html.parser")
 
-    def _fetch_html(self, ticker: str, consolidated: bool = True) -> str:
-        t = ticker.upper().strip()
-        urls = [
-            f"https://www.screener.in/company/{t}/consolidated/",
-            f"https://www.screener.in/company/{t}/",
-        ] if consolidated else [
-            f"https://www.screener.in/company/{t}/",
-            f"https://www.screener.in/company/{t}/consolidated/",
-        ]
-
-        last_status = None
-        last_text = ""
+    @staticmethod
+    def _normalize_ticker(ticker: str) -> str:
+        t = ticker.strip().upper()
+        return quote(t, safe="-._~")
+
+    def _fetch_url(self, url: str) -> str:
         session = self._session()
+        response = session.get(url, timeout=self.timeout)
+
+        if response.status_code == 200:
+            return response.text
+
+        preview = ""
+        try:
+            preview = response.text[:300].replace("\n", " ").replace("\r", " ")
+        except Exception:
+            preview = "<no response text>"
+
+        raise RuntimeError(
+            f"Request failed.\n"
+            f"URL: {url}\n"
+            f"Status: {response.status_code}\n"
+            f"Reason: {response.reason}\n"
+            f"Response preview: {preview}"
+        )
+
+    def _fetch_html(self, ticker: str, consolidated: bool = True) -> str:
+        t = self._normalize_ticker(ticker)
+
+        if consolidated:
+            urls = [
+                f"https://www.screener.in/company/{t}/consolidated/",
+                f"https://www.screener.in/company/{t}/",
+            ]
+        else:
+            urls = [
+                f"https://www.screener.in/company/{t}/",
+                f"https://www.screener.in/company/{t}/consolidated/",
+            ]
+
+        last_error: Optional[Exception] = None
 
         for url in urls:
             try:
-                response = session.get(url, timeout=self.timeout)
-                last_status = response.status_code
-                last_text = response.text
-                if response.status_code == 200:
-                    return response.text
-            except requests.RequestException:
+                html = self._fetch_url(url)
+                return html
+            except Exception as e:
+                last_error = e
                 continue
 
-        raise RuntimeError(f"Failed to fetch data for {t}. Last status: {last_status}")
+        raise RuntimeError(
+            f"Failed to fetch data for '{ticker}'. "
+            f"Last error: {last_error}"
+        ) from last_error
 
     @staticmethod
     def _clean_text(value: str) -> str: