#!/usr/bin/env python3
"""Residential-IP page fetch for the Melaya backlink presence check.

The Melaya cloud runs from a datacenter IP that Cloudflare "managed" Turnstile
forces into an interactive challenge no headless engine clears. The user's
runner is on a RESIDENTIAL IP, where the same real-browser fetch passes in
seconds. So the server hands us ONE public https url; we fetch it with the
local Scrapling stack and print the result as a single JSON line on stdout:

    {"status": int, "engine": str|null, "challenged": bool,
     "finalUrl": str, "html": str}   # html capped

Input (stdin, JSON): {"url": "...", "budget_seconds": 40}

Self-contained on purpose (mirrors localDbProbe.py): it imports only Scrapling
from the runner venv, never the shared tool bundle. HTTPS-only; the server
already validated the url, this is defence in depth. Never fetches anything but
the url it is given — no internal hosts, no secrets, no catalog data.
"""
import asyncio
import json
import sys

_HTML_CAP = 600_000
_CHALLENGE_MARKERS = (
    "just a moment", "cf_chl", "cf-challenge", "checking your browser",
    "enable javascript and cookies", "attention required",
)


def _html_of(page) -> str:
    for attr in ("html_content", "body", "content"):
        try:
            v = getattr(page, attr)
            if callable(v):
                v = v()
            if v:
                return v if isinstance(v, str) else str(v)
        except Exception:
            pass
    try:
        return str(page)
    except Exception:
        return ""


def _challenged(h: str) -> bool:
    h = (h or "").lower()
    return any(m in h for m in _CHALLENGE_MARKERS)


async def _fetch(url: str, budget: float) -> dict:
    ms = int(max(6.0, min(75.0, budget)) * 1000)
    # ── Tier 1: TLS-impersonating async GET (no browser) ──
    try:
        from scrapling.fetchers import AsyncFetcher
        p = await AsyncFetcher.get(url, timeout=min(ms, 15000), impersonate="chrome",
                                   stealthy_headers=True, follow_redirects=True)
        st = int(getattr(p, "status", 0) or 0)
        h = _html_of(p)
        final = str(getattr(p, "url", "") or url)
        if 200 <= st < 300 and h and not _challenged(h):
            return {"status": st, "engine": "fast", "challenged": False,
                    "finalUrl": final, "html": h[:_HTML_CAP]}
    except Exception:
        pass

    # ── Tier 2/3: stealth browser (real Chrome), plain then solve_cloudflare ──
    try:
        from scrapling.fetchers import StealthyFetcher
    except Exception:
        return {"status": 0, "engine": None, "challenged": False, "finalUrl": url, "html": ""}
    fn = getattr(StealthyFetcher, "async_fetch", None)
    if fn is None:
        return {"status": 0, "engine": None, "challenged": False, "finalUrl": url, "html": ""}

    last = {"status": 0, "engine": "stealth", "challenged": True, "finalUrl": url, "html": ""}
    for solve in (False, True):
        base = {"headless": True, "timeout": 60000 if solve else min(ms, 25000)}
        if solve:
            base.update(solve_cloudflare=True, block_webrtc=True, disable_resources=False)
        page = None
        engine = ("real_chrome_cf" if solve else "real_chrome")
        for use_real in (True, False):
            kw = dict(base)
            if use_real:
                kw["real_chrome"] = True
            else:
                engine = ("chromium_cf" if solve else "chromium")
            try:
                page = await fn(url, **kw)
                break
            except Exception:
                page = None
                continue
        if page is None:
            continue
        h = _html_of(page)
        st = int(getattr(page, "status", 0) or 0)
        final = str(getattr(page, "url", "") or url)
        last = {"status": st, "engine": engine, "challenged": _challenged(h),
                "finalUrl": final, "html": (h or "")[:_HTML_CAP]}
        if h and not _challenged(h):
            return last
        if not solve and not _challenged(h):
            break  # plain stealth returned something non-challenge-y but empty
    return last


async def _main() -> None:
    try:
        data = json.loads(sys.stdin.read() or "{}")
    except Exception:
        data = {}
    url = str(data.get("url") or "").strip()
    try:
        budget = float(data.get("budget_seconds") or 40)
    except Exception:
        budget = 40.0
    from urllib.parse import urlsplit
    if not url or urlsplit(url).scheme != "https":
        print(json.dumps({"status": 0, "engine": None, "challenged": False,
                          "finalUrl": url, "html": "", "error": "https_only"}))
        return
    try:
        out = await _fetch(url, budget)
    except Exception as e:
        out = {"status": 0, "engine": None, "challenged": False,
               "finalUrl": url, "html": "", "error": str(e)[:200]}
    print(json.dumps(out))


if sys.platform.startswith("win"):
    try:
        asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
    except Exception:
        pass
asyncio.run(_main())