"""TikTok page signal detection. Single entry point: PageSignals.from_page(...) """ from __future__ import annotations import re from typing import ClassVar from pydantic import BaseModel class PageSignals(BaseModel): """Normalised browser observations the LLM can act on.""" block_signal: str = ( "none" # none | login_redirect | captcha | soft_block | empty_response ) page_load_quality: str = "full" # full | partial | empty redirect_detected: bool = False slow_response: bool = False response_time_ms: int = 0 items_load_ratio: float | None = None zero_metrics: bool = False # ------------------------------------------------------------------ constants _SLOW_MS: ClassVar[int] = 15_000 _EMPTY_THRESHOLD: ClassVar[int] = 3_000 _LOGIN_URL_TOKENS: ClassVar[tuple] = ("/login", "loginpage", "login?") _CAPTCHA_TOKENS: ClassVar[tuple] = ( "captcha", "verifyaccount", "verify-account", "robot-check", ) # ------------------------------------------------------------------ factory @classmethod def from_page( cls, *, html: str, requested_url: str, final_url: str, response_time_ms: int, items_loaded: int = 0, count_text: str | None = None, metrics: dict | None = None, ) -> PageSignals: return cls( block_signal=cls._block_signal(html, final_url), page_load_quality=cls._load_quality(items_loaded), redirect_detected=cls._is_redirect(requested_url, final_url), slow_response=response_time_ms > cls._SLOW_MS, response_time_ms=response_time_ms, items_load_ratio=cls._load_ratio(items_loaded, count_text), zero_metrics=cls._zero_metrics(metrics or {}), ) # ------------------------------------------------------------------ private helpers @classmethod def _block_signal(cls, html: str, final_url: str) -> str: lower_url = final_url.lower() if any(t in lower_url for t in cls._LOGIN_URL_TOKENS): return "login_redirect" lower_html = html.lower() if any(t in lower_html for t in cls._CAPTCHA_TOKENS): return "captcha" if len(html) < cls._EMPTY_THRESHOLD: return "empty_response" if "tiktok.com" not in lower_url: return "soft_block" return "none" @classmethod def _load_quality(cls, items_loaded: int, threshold: int = 5) -> str: if items_loaded == 0: return "empty" if items_loaded < threshold: return "partial" return "full" @classmethod def _is_redirect(cls, requested: str, final: str) -> bool: def norm(u: str) -> str: return u.rstrip("/").split("?")[0].lower() return norm(requested) != norm(final) @classmethod def _load_ratio(cls, loaded: int, count_text: str | None) -> float | None: total = cls._parse_count(count_text) if not total: return None return round(loaded / total, 4) @staticmethod def _parse_count(text: str | None) -> int | None: if not text: return None m = re.search(r"(\d[\d,.]*)\s*([KkMmBb]?)", text) if not m: return None num = float(m.group(1).replace(",", "")) suffix = m.group(2).upper() return int( num * {"K": 1_000, "M": 1_000_000, "B": 1_000_000_000}.get(suffix, 1) ) @staticmethod def _zero_metrics(metrics: dict) -> bool: return all(not metrics.get(k) for k in ("views", "likes", "comments"))