"""One isolated Scrapy single-page Spider per frozen URL; private captures only.
Requires Scrapy==2.19.0. No full crawl, browser or JavaScript.
"""
import argparse
import asyncio
from collections import Counter
from datetime import datetime, timezone
import hashlib
from html.parser import HTMLParser
from importlib.metadata import version
import json
import os
from pathlib import Path
import platform
import re
import shlex
import signal
import subprocess
import sys
import time
from urllib.parse import urlsplit
import scrapy
from scrapy.crawler import CrawlerProcess

CONFIG = {'mode': 'Single-page Spider / HTTP11 downloader', 'proxy_mode': 'none', 'login': False, 'captcha_solving': False, 'concurrency': 1, 'gap_seconds': 2, 'application_retries': 0, 'fresh_process_and_session_per_target': True, 'user_agent': 'Scrapy/2.19.0 (+https://scrapy.org)', 'robots': False, 'cookies_enabled': False, 'follow_redirects': True, 'max_redirects': 3, 'meta_refresh': False, 'timeout': 20, 'attempt_timeout_seconds': 35, 'slice_timeout_seconds': 1200, 'max_capture_bytes': 5242880, 'download_maxsize_bytes': 5242880, 'download_size_basis': 'Scrapy handler bounds response download; compression middleware also bounds decompressed body; early wire body and final decompressed body saved privately', 'resources': 'One response chain only; no browser, JavaScript or linked resource loads', 'verify': True, 'http_version': 'HTTP/1.1 stock Twisted handler; no browser TLS impersonation', 'pss_sample_interval_seconds': 0.1, 'min_host_available_mb': 3072, 'max_attempt_tree_pss_mb': 1024, 'pss_scope': 'Isolated Python worker/descendants, including imports, crawler startup, scheduling, downloader, middleware/capture/checks and reactor shutdown; excludes batch parent'}

BLOCK_PHRASES = (
    "verify you are human", "robot or human", "robot check", "access denied",
    "pardon our interruption", "please complete the captcha", "enter the characters",
    "additional verification required", "unusual traffic", "checking your browser",
    "enable javascript and cookies to continue", "just a moment",
)
GATE_PHRASES = ("log in to continue", "sign in to continue", "sign in to linkedin", "consent required", "before you continue")


def utc():
    return datetime.now(timezone.utc).isoformat()


def sha(data):
    return hashlib.sha256(data).hexdigest()


class PageText(HTMLParser):
    """Source-text approximation, not rendered visibility or structured extraction."""
    SKIP = {"head", "script", "style", "template", "noscript"}

    def __init__(self):
        super().__init__(convert_charrefs=True)
        self.hidden = []
        self.parts = []

    def handle_starttag(self, tag, attrs):
        if tag in self.SKIP:
            self.hidden.append(tag)

    def handle_endtag(self, tag):
        if tag in self.hidden:
            self.hidden = self.hidden[:self.hidden.index(tag)]

    def handle_data(self, data):
        if not self.hidden:
            self.parts.append(data)


def evaluate(body, target):
    parser = PageText()
    parser.feed(body.decode("utf-8", errors="replace"))
    text = " ".join(" ".join(parser.parts).split())
    lower = text.lower()
    checks = [
        {"alternatives": group, "matched": [m for m in group if m.lower() in lower]}
        for group in target["required_text_groups"]
    ]
    regex_checks = [
        {**check, "matches": len(re.findall(check["pattern"], text, flags=re.I))}
        for check in target["text_regex_checks"]
    ]
    expected = (
        len(text) >= target["min_text_chars"]
        and all(check["matched"] for check in checks)
        and all(check["matches"] >= check["min_matches"] for check in regex_checks)
    )
    return {
        "text_chars": len(text), "required_text_checks": checks, "text_regex_checks": regex_checks,
        "min_text_chars": target["min_text_chars"], "has_required_content": bool(expected),
        "soft_block_checks": {"phrases_checked": list(BLOCK_PHRASES), "matched": [p for p in BLOCK_PHRASES if p in lower]},
        "gate_checks": {"phrases_checked": list(GATE_PHRASES), "matched": [p for p in GATE_PHRASES if p in lower]},
    }



def result_class(row):
    if row["error"]:
        return row["error"]["classification"]
    if row["navigation_error"]:
        return "navigation_error"
    if row["capture_error"]:
        return "capture_error"
    if row["status"] is None:
        return "status_unobserved"
    if row["status"] in (401, 403, 999):
        return "http_access_denied"
    if row["status"] == 429:
        return "http_rate_limited"
    if row["status"] >= 500:
        return "http_server_error"
    if row["status"] >= 400:
        return "http_client_error"
    if 300 <= row["status"] < 400:
        return "unresolved_redirect"
    if row["checks"]["soft_block_checks"]["matched"]:
        return "soft_block"
    if not row["checks"]["has_required_content"]:
        return "login_or_consent_gate" if row["checks"]["gate_checks"]["matched"] else "missing_required_content"
    return "usable" if row["status"] == 200 and row["capture_complete"] else "incomplete_capture"


def env_direct():
    env = os.environ.copy()
    for key in list(env):
        if key.lower() in ("http_proxy", "https_proxy", "all_proxy"):
            env.pop(key)
    env.update(PYTHONDONTWRITEBYTECODE="1")
    return env


def memory():
    readings = {line.split(":")[0]: int(line.split()[1]) / 1024
                for line in Path("/proc/meminfo").read_text().splitlines() if line.split(":")[0] in ("MemTotal", "MemAvailable", "SwapTotal", "SwapFree")}
    return {**readings, "unit": "MiB", "measured_at": utc()}


def base_row(target):
    return {"target_id": target["id"], "site": target["name"], "category": target["category"],
            "url": target["url"], "original_url": target["url"], "final_url": None,
            "expected_required_content": target["expected_required_content"], "started_at": utc(),
            "request_started_at": None, "status": None, "request_outcome": "not_started",
            "redirect_history": [], "redirect_outcome": "unobserved", "document_responses": [],
            "explicit_request_headers": {}, "response_metadata": {}, "error": None,
            "navigation_error": None, "capture_error": None, "screenshot_error": None,
            "screenshot_private_path": None, "screenshot_sha256": None,
            "capture_complete": False, "raw_response_sha256": None, "raw_response_private_path": None,
            "response_bytes": 0, "downloaded_body_bytes": 0, "body_complete": False, "navigation_seconds": None, "attempt_seconds": None}


STATE = {}


def serialized_headers(headers):
    return [{"name": k.decode("latin1"), "values": [v.decode("latin1") for v in vals]} for k,vals in headers.items()]


class ResponseObserver:
    @classmethod
    def from_crawler(cls, crawler):
        o=cls();o.crawler=crawler;return o

    def process_request(self, request):
        STATE["requests"].append({"url":request.url,"method":request.method,"headers":serialized_headers(request.headers)})
        return None

    def process_response(self, request, response):
        index=len(STATE["responses"])
        wire=STATE["output"] / f"{STATE['target_id']}-response-{index}.wire"
        wire.write_bytes(response.body)
        STATE["responses"].append({"url":response.url,"status":response.status,"protocol":response.protocol,"headers":serialized_headers(response.headers),"wire_body_bytes":len(response.body),"wire_body_sha256":sha(response.body),"wire_file":wire.name,"download_latency":request.meta.get("download_latency")})
        STATE["last_response"]=response
        STATE["write_log"]()
        return response


class SinglePageSpider(scrapy.Spider):
    name="frozen_single_page"

    async def start(self):
        STATE["row"]["request_started_at"]=utc()
        STATE["request_start"]=time.perf_counter()
        STATE["save"]()
        yield scrapy.Request(STATE["url"], callback=self.parse, errback=self.failed, dont_filter=True, meta={"proxy":None})

    def parse(self, response):
        STATE["final_response"]=response
        STATE["row"]["latency_seconds"]=round(time.perf_counter()-STATE["request_start"],3)
        STATE["row"]["request_outcome"]="returned"
        STATE["row"]["capture_complete"]="dataloss" not in response.flags
        return None

    def failed(self, failure):
        message=str(failure.value).splitlines()[0][:500]
        name=failure.type.__name__
        category="redirect_limit_or_loop" if "max redirections" in message.lower() else "timeout" if "timeout" in name.lower() or "took longer" in message.lower() or "timed out" in message.lower() else "transport_error"
        STATE["row"]["error"]={"type":name,"message":message,"classification":category}
        STATE["row"]["request_outcome"]="error"
        STATE["row"]["latency_seconds"]=round(time.perf_counter()-STATE["request_start"],3)
        STATE["failure_request_meta"]=failure.request.meta
        return None


def single(args, target):
    row,capture=base_row(target),b""
    row_path=args.output / f"{target['id']}.json"
    def save():row_path.write_text(json.dumps(row,indent=2)+"\n")
    transport={"requests":[],"responses":[]}
    log=args.output / f"{target['id']}-transport.json"
    def write_log():log.write_text(json.dumps(transport,indent=2)+"\n")
    STATE.update(row=row,save=save,write_log=write_log,output=args.output,target_id=target['id'],url=target['url'],requests=transport['requests'],responses=transport['responses'])
    start=time.perf_counter()
    settings={"USER_AGENT":CONFIG["user_agent"],"CONCURRENT_REQUESTS":1,"CONCURRENT_REQUESTS_PER_DOMAIN":1,"ROBOTSTXT_OBEY":False,"COOKIES_ENABLED":False,"RETRY_ENABLED":False,"RETRY_TIMES":0,"HTTPPROXY_ENABLED":False,"HTTPERROR_ALLOW_ALL":True,"REDIRECT_ENABLED":True,"REDIRECT_MAX_TIMES":3,"METAREFRESH_ENABLED":False,"DOWNLOAD_TIMEOUT":20,"DOWNLOAD_MAXSIZE":5242880,"DOWNLOAD_WARNSIZE":5242880,"DOWNLOAD_VERIFY_CERTIFICATES":True,"DOWNLOAD_FAIL_ON_DATALOSS":True,"HTTPCACHE_ENABLED":False,"TELNETCONSOLE_ENABLED":False,"MEMUSAGE_ENABLED":False,"LOG_LEVEL":"INFO","DEFAULT_REQUEST_HEADERS":{"Accept":"text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8","Accept-Language":"en"},"DOWNLOADER_MIDDLEWARES":{"__main__.ResponseObserver":900,"scrapy.downloadermiddlewares.httpproxy.HttpProxyMiddleware":None},"DOWNLOAD_HANDLERS":{"http":"scrapy.core.downloader.handlers.http11.HTTP11DownloadHandler","https":"scrapy.core.downloader.handlers.http11.HTTP11DownloadHandler"}}
    row["effective_settings"]=settings
    process=CrawlerProcess(settings=settings)
    crawler=process.create_crawler(SinglePageSpider)
    deferred=process.crawl(crawler)
    def setup_failed(failure):
        row["error"]={"type":failure.type.__name__,"message":str(failure.value).splitlines()[0][:500],"classification":"setup_error"}
        save()
    deferred.addErrback(setup_failed)
    save()
    process.start(stop_after_crawl=True,install_signal_handlers=False)
    response=STATE.get("final_response") or STATE.get("last_response")
    if response is not None:
        row["status"]=response.status;row["final_url"]=response.url
        capture=response.body
        row["downloaded_body_bytes"]=len(capture)
        if len(capture)>CONFIG["max_capture_bytes"]:
            capture=capture[:CONFIG["max_capture_bytes"]];row["capture_complete"]=False
            row["capture_error"]={"type":"BodyCaptureLimit","message":"Saved body exceeded 5 MiB; prefix only"}
        dest=args.output / f"{target['id']}.body";dest.write_bytes(capture)
        row.update(raw_response_private_path=str(dest),raw_response_sha256=sha(capture),response_bytes=len(capture))
        meta=response.meta if response.request else STATE.get("failure_request_meta",{})
        urls=meta.get("redirect_urls",[]);reasons=meta.get("redirect_reasons",[])
        row["redirect_history"]=[{"previous":url,"url":urls[i+1] if i+1<len(urls) else row["final_url"],"status":reasons[i]} for i,url in enumerate(urls)]
        row["redirect_outcome"]="limit_or_loop_error" if row["error"] and row["error"]["classification"]=="redirect_limit_or_loop" else "followed" if urls else "none" if response.url==target['url'] else "final_url_changed_history_unavailable"
        row["response_metadata"]={"content_type":response.headers.get(b"Content-Type",b"").decode("latin1"),"content_encoding":response.headers.get(b"Content-Encoding",b"").decode("latin1"),"protocol":response.protocol,"flags":list(response.flags),"body_basis":"Final Spider response after compression middleware" if STATE.get("final_response") else "Last early middleware wire response; terminal downloader error; never usable"}
    else:row["redirect_outcome"]="no_response"
    stats=crawler.stats.get_stats()
    row["scrapy_stats"]={k:v for k,v in stats.items() if isinstance(v,(str,int,float,bool))}
    row["downloader_request_count"]=stats.get("downloader/request_count",0)
    row["observed_request_headers"]=[{"url":req['url'],"method":req['method'],"headers":[h for h in req['headers'] if h['name'].lower() not in ['cookie','authorization','proxy-authorization']]} for req in transport['requests']]
    row["document_responses"]=[{k:r[k] for k in ['url','status','protocol','wire_body_bytes','wire_body_sha256']} for r in transport['responses']]
    write_log();row["transport_log_sha256"]=sha(log.read_bytes())
    row.setdefault("latency_seconds",round(time.perf_counter()-start,3))
    row.update(finished_at=utc(),attempt_seconds=round(time.perf_counter()-start,3),checks=evaluate(capture,target))
    row["empty_content"]=row["checks"]["text_chars"]==0;row["body_complete"]=row["capture_complete"]
    row["classification"]=result_class(row);row["usable_content"]=row["classification"]=="usable"
    save()


def tree_members(root):
    pending, seen, members = [root], set(), {}
    while pending:
        pid = pending.pop()
        if pid in seen:
            continue
        seen.add(pid)
        try:
            for task in Path(f"/proc/{pid}/task").iterdir():
                pending.extend(int(p) for p in (task / "children").read_text().split())
            stat = Path(f"/proc/{pid}/stat").read_text().rsplit(")", 1)[1].split()
            members[pid] = stat[19]  # start time; protects cleanup from PID reuse
        except (OSError, ValueError, IndexError):
            continue
    return members


def pss_members(members):
    total, read = 0, 0
    for pid in members:
        try:
            match = re.search(r"^Pss:\s+(\d+)", Path(f"/proc/{pid}/smaps_rollup").read_text(), re.M)
            if match:
                total += int(match.group(1))
                read += 1
        except OSError:
            continue
    return total / 1024 if read else None


def signal_owned(members, sig):
    # Signal only recorded worker/descendant identities; never other host processes.
    for pid, birth in reversed(list(members.items())):
        try:
            current = Path(f"/proc/{pid}/stat").read_text().rsplit(")", 1)[1].split()[19]
            if current == birth:
                os.kill(pid, sig)
        except (OSError, IndexError):
            pass


async def batch(args):
    manifest_bytes = args.manifest.read_bytes()
    manifest = json.loads(manifest_bytes)
    targets = manifest["targets"]
    assert len(targets) == len({t["id"] for t in targets}) == len({urlsplit(t["url"]).hostname for t in targets}) == 30
    assert version("scrapy") == "2.19.0"
    if args.target:
        targets = [t for t in targets if t["id"] == args.target]
        assert len(targets) == 1
    initial_ram = memory()
    if initial_ram["MemAvailable"] < CONFIG["min_host_available_mb"]:
        raise RuntimeError("Unsafe RAM blocker before batch: less than 3 GiB available")
    os.umask(0o077)
    args.output.mkdir(parents=True, exist_ok=False, mode=0o700)
    run = {"run_id": args.output.name, "tool": "scrapy", "tool_version": version("scrapy"),
           "twisted_version": version("twisted"), "requirements_sha256": sha((Path(__file__).parent/"requirements.txt").read_bytes()), "documentation": json.loads((Path(__file__).parent/"documentation.json").read_text()), "mode": CONFIG["mode"],
           "python_version": platform.python_version(), "platform": platform.platform(),
           "network": "Same local host direct outbound; http_proxy/https_proxy/all_proxy removed case-insensitively; no configured proxies. Different UTC window from prior tools.",
           "manifest_id": manifest["id"], "manifest_sha256": sha(manifest_bytes),
           "script_sha256": sha(Path(__file__).read_bytes()), "config": CONFIG,
           "executed_command": shlex.join([sys.executable, *sys.argv]), "started_at": utc(),
           "host_ram_preflight": initial_ram, "attempts": [],
           "mode_selection":"Documented CrawlerProcess plus one async Spider.start Request. Stock HTTP11 downloader, identified Scrapy User-Agent; parse and errback schedule no further pages. This tests one-page downloader behavior, not Scrapy crawl/scheduling strengths.",
           "body_hash_basis":"Final client-decompressed Spider response bytes, not browser DOM or re-encoded text. Early wire responses, bodies and logs private. Terminal errors retain the last early response where present; incomplete/error state explicit.",
           "latency_basis":"From initial Spider.start Request scheduling to delivered final response or errback; includes redirects, scheduling/middleware and body transfer/decompression. Excludes initial crawler setup, later saving/checks and reactor shutdown. Downloader hop latency is separately recorded privately.",
           "pss_caveat":"100 ms worker-tree samples include interpreter/imports, crawler setup, scheduling, HTTP download, middleware/capture/checks and reactor shutdown. Batch parent excluded; brief peaks may be missed. Different work from direct clients, selectors and browsers.",
           "baseline_sha256":{name:sha((Path(__file__).parent.parent/folder/"results.json").read_bytes()) for name,folder in {"botasaurus":"botasaurus-public-30-20260929","camoufox":"camoufox-public-30-20260929","curl_cffi":"curl-cffi-public-30-20260929","lightpanda":"lightpanda-public-30-20260929","patchright":"patchright-public-30-20260929","scrapling":"scrapling-public-30-20260929","seleniumbase":"seleniumbase-public-30-20260929","wreq":"wreq-public-30-redirects-20260929"}.items()},
           "policy_differences":["Scrapy/2.19.0 bot User-Agent, default Accept/Accept-Language=en and compression headers; stock Twisted HTTP/1.1 with certificate verification enabled, no Chrome TLS impersonation.","Cookies and robots disabled; no robots URL or public-suffix cookie fetch. Fresh isolated process per target. Other tools differ in cookie/header/JS and loaded-resource work.","Normal redirects cap 3; meta-refresh disabled. No per-request handle_httpstatus_all because it disables redirects in this release; global HTTPERROR_ALLOW_ALL retains error pages. Standard redirect Referer handling retained.","No retries, proxy, browser, JavaScript, login, CAPTCHA interaction, link discovery, feeds or HTTP cache. HttpProxyMiddleware disabled and environment proxies removed.","20-second per-download timeout, 35-second whole-worker watchdog, 3 GiB host RAM floor and 1 GiB worker-tree PSS guard. 5 MiB handler/compression limit differs from clients that cap saved bytes only.","One dated URL test measures downloader behavior, not crawling/scheduling scale or reliability. Different time windows and policies establish no global winner or feature cause."]}

    def save_run():
        run["summary"] = {"attempted_distinct_sites": len(run["attempts"]), "usable_results": sum(a["usable_content"] for a in run["attempts"]),
                          "classifications": dict(Counter(a["classification"] for a in run["attempts"]))}
        (args.output / "results.json").write_text(json.dumps(run, indent=2) + "\n")
    slice_started = time.perf_counter()
    for index, target in enumerate(targets):
        if time.perf_counter() - slice_started > CONFIG["slice_timeout_seconds"]:
            run["blocker"] = "Whole-slice deadline before next target"
            save_run()
            raise RuntimeError(run["blocker"])
        if index:
            await asyncio.sleep(CONFIG["gap_seconds"])
        before_ram = memory()
        if before_ram["MemAvailable"] < CONFIG["min_host_available_mb"]:
            run["blocker"] = "Unsafe host RAM before next navigation"
            save_run()
            raise RuntimeError(run["blocker"])
        command = [sys.executable, str(Path(__file__).resolve()), "--manifest", str(args.manifest), "--output", str(args.output), "--single", target["id"]]
        with open(args.output / f"{target['id']}-driver.log", "wb") as log:
            proc = await asyncio.create_subprocess_exec(*command, stdout=log, stderr=log, env=env_direct(), start_new_session=True)
            started = time.perf_counter()
            samples, guard, owned = [], None, {}
            while proc.returncode is None:
                members = tree_members(proc.pid)
                owned.update(members)
                reading = pss_members(members)
                ram = memory()
                if reading is not None:
                    samples.append({"elapsed_seconds": round(time.perf_counter() - started, 3), "pss_mb": round(reading, 3), "host_available_mb": round(ram["MemAvailable"], 3)})
                if ram["MemAvailable"] < CONFIG["min_host_available_mb"] or (reading or 0) > CONFIG["max_attempt_tree_pss_mb"]:
                    guard = "unsafe_memory_guard"
                elif time.perf_counter() - started > CONFIG["attempt_timeout_seconds"]:
                    guard = "attempt_timeout"
                if guard:
                    signal_owned(owned, signal.SIGTERM)
                    await asyncio.sleep(0.2)
                    signal_owned(owned, signal.SIGKILL)
                    break
                await asyncio.sleep(CONFIG["pss_sample_interval_seconds"])
            await proc.wait()
            signal_owned(owned, signal.SIGKILL)
        row_path = args.output / f"{target['id']}.json"
        row = json.loads(row_path.read_text()) if row_path.exists() else base_row(target)
        if guard or proc.returncode != 0:
            row["error"] = {"type": "AttemptGuard" if guard else "AttemptProcessExit", "message": guard or f"Attempt process exit {proc.returncode}", "classification": guard or "attempt_process_error"}
        saved_body = args.output / f"{target['id']}.body"
        row.setdefault("checks", evaluate(saved_body.read_bytes() if saved_body.exists() else b"", target))
        row["empty_content"] = row["checks"]["text_chars"] == 0
        row.setdefault("latency_seconds", round(time.perf_counter() - started, 3))
        row.setdefault("finished_at", utc())
        survivors = tree_members(proc.pid)
        row["cleanup_surviving_owned_pids"] = [pid for pid,birth in survivors.items() if owned.get(pid)==birth]
        row["driver_log_sha256"] = sha((args.output / f"{target['id']}-driver.log").read_bytes())
        row.update(peak_sampled_tree_pss_mb=max((s["pss_mb"] for s in samples), default=None), pss_samples=len(samples),
                   attempt_process_exit=proc.returncode, host_ram_before=before_ram)
        row["classification"] = result_class(row)
        row["usable_content"] = row["classification"] == "usable"
        (args.output / f"{target['id']}-pss.json").write_text(json.dumps(samples) + "\n")
        row_path.write_text(json.dumps(row, indent=2) + "\n")
        run["attempts"].append(row)
        save_run()
        print(f"{target['id']}: HTTP {row['status']} | {row['classification']} | {row['response_bytes']} body bytes | PSS {row['peak_sampled_tree_pss_mb']} MiB", flush=True)
        if row["request_started_at"] is None or guard == "unsafe_memory_guard":
            run["blocker"] = "Setup or unsafe memory blocker: " + str(row["error"])
            save_run()
            raise RuntimeError(run["blocker"])
    run["finished_at"] = utc()
    save_run()
    print(json.dumps(run["summary"]), flush=True)


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--manifest", type=Path, required=True)
    parser.add_argument("--output", type=Path, required=True)
    parser.add_argument("--single")
    parser.add_argument("--target", help="One separate guarded public canary; manifest remains frozen")
    args = parser.parse_args()
    if args.single:
        target = next(t for t in json.loads(args.manifest.read_text())["targets"] if t["id"] == args.single)
        single(args, target)
    else:
        asyncio.run(batch(args))