import asyncio
import json
import os
import secrets
import time
import uuid

from fastapi import Depends, FastAPI, HTTPException, Request
from fastapi.responses import FileResponse, JSONResponse, Response
from fastapi.security import HTTPBasic, HTTPBasicCredentials
from pydantic import BaseModel
from playwright.async_api import TimeoutError as PlaywrightTimeoutError

from camoufox.async_api import AsyncCamoufox

app = FastAPI(title="Stealth Browser API", version="1.1")

BASE_DIR = os.path.dirname(os.path.abspath(__file__))
DATA_DIR = os.getenv("DATA_DIR", "/data")
LOG_FILE = os.path.join(DATA_DIR, "logs.jsonl")

PROXY = os.getenv("PROXY", "")
ADMIN_USER = os.getenv("ADMIN_USER", "admin")
ADMIN_PASS = os.getenv("ADMIN_PASS", "")  # empty = no auth (LAN mode)
CAPTCHA_WAIT_MS = int(os.getenv("CAPTCHA_WAIT_MS", "4000"))
SOLVE_TIMEOUT_S = int(os.getenv("SOLVE_TIMEOUT_S", "900"))

CAPTCHA_SELECTORS = (
    'iframe[src*="challenges.cloudflare.com"],'
    'iframe[src*="hcaptcha.com"],'
    'iframe[src*="recaptcha"],'
    'iframe[src*="arkoselabs"],'
    'iframe[src*="funcaptcha"],'
    'iframe[src*="turnstile"],'
    'div[style*="captcha"]'
)
MAX_CONCURRENT = int(os.getenv("MAX_CONCURRENT", "3"))
_semaphore = asyncio.Semaphore(MAX_CONCURRENT)

# ---------------------------------------------------------------- auth
security = HTTPBasic(auto_error=False)


def require_auth(creds: HTTPBasicCredentials | None = Depends(security)):
    if not ADMIN_PASS:
        return None  # open mode
    if not creds or not secrets.compare_digest(creds.username, ADMIN_USER) or not secrets.compare_digest(creds.password, ADMIN_PASS):
        raise HTTPException(401, "Invalid or missing credentials")
    return creds


# ---------------------------------------------------------------- browser helper
def browser_kwargs() -> dict:
    kwargs = {"headless": True}
    if PROXY:
        kwargs["proxy"] = {"server": PROXY}
    return kwargs


# ---------------------------------------------------------------- logging
_log_lock = asyncio.Lock()


async def log(entry: dict):
    os.makedirs(DATA_DIR, exist_ok=True)
    entry["ts"] = time.strftime("%Y-%m-%d %H:%M:%S")
    async with _log_lock:
        with open(LOG_FILE, "a") as f:
            f.write(json.dumps(entry) + "\n")


async def read_logs(limit: int = 200):
    if not os.path.exists(LOG_FILE):
        return []
    lines = []
    async with _log_lock:
        with open(LOG_FILE) as f:
            lines = f.readlines()
    out = []
    for line in lines[-limit:]:
        try:
            out.append(json.loads(line))
        except json.JSONDecodeError:
            pass
    return list(reversed(out))


# ---------------------------------------------------------------- jobs
class Job:
    def __init__(self, url: str, extract: str, selector: str | None, pause: bool):
        self.id = uuid.uuid4().hex[:8]
        self.url = url
        self.extract = extract
        self.selector = selector
        self.pause = pause
        self.created = time.time()
        self.status = "queued"  # queued|loading|captcha|solving|done|failed|cancelled|timeout
        self.outcome = None
        self.error = None
        self.title = None
        self.status_code = None
        self.final_url = None
        self.text = None
        self.viewport = {"width": 1280, "height": 720}
        self.page = None
        self.solved = asyncio.Event()
        self.cancelled = False
        self.action_lock = asyncio.Lock()
        self.finished = asyncio.Event()

    def to_dict(self):
        return {
            "id": self.id, "url": self.url, "status": self.status,
            "outcome": self.outcome, "created": self.created,
            "title": self.title, "status_code": self.status_code,
            "error": self.error,
        }


jobs: dict[str, Job] = {}


class FetchRequest(BaseModel):
    url: str
    extract: str = "text"  # text | html
    selector: str | None = None
    pause: bool = False  # always open admin WebUI for manual check


# ---------------------------------------------------------------- fetch flow
async def detect_captcha(page) -> bool:
    try:
        await page.wait_for_selector(CAPTCHA_SELECTORS, state="attached", timeout=CAPTCHA_WAIT_MS)
        return True
    except PlaywrightTimeoutError:
        return False
    except Exception:
        return False


async def run_fetch(job: Job):
    t0 = time.time()
    async with _semaphore:
        try:
            async with AsyncCamoufox(**browser_kwargs()) as browser:
                page = await browser.new_page()
                job.page = page
                job.status = "loading"
                await log({"job": job.id, "url": job.url, "event": "load_start"})
                resp = await page.goto(job.url, wait_until="domcontentloaded", timeout=45_000)
                job.status_code = resp.status if resp else None
                job.final_url = page.url
                job.title = await page.title()

                has_captcha = job.pause or await detect_captcha(page)
                if has_captcha and not job.cancelled:
                    job.status = "captcha"
                    await log({"job": job.id, "url": job.url, "event": "captcha_detected"})
                    try:
                        await asyncio.wait_for(job.solved.wait(), timeout=SOLVE_TIMEOUT_S)
                    except asyncio.TimeoutError:
                        job.status = "timeout"
                        job.outcome = "captcha_timeout"
                        job.error = "No human solved the captcha in time"
                        await log({"job": job.id, "url": job.url, "event": "captcha_timeout"})
                        return
                    if job.cancelled:
                        return

                if job.cancelled:
                    return
                job.status = "solving"
                try:
                    await page.wait_for_load_state("networkidle", timeout=15_000)
                except Exception:
                    pass
                job.title = await page.title()
                if job.selector:
                    el = page.locator(job.selector)
                    job.text = await el.first.inner_text() if await el.count() else ""
                elif job.extract == "html":
                    job.text = await page.content()
                else:
                    job.text = await page.inner_text("body")
                job.status = "done"
                job.outcome = "captcha_solved" if has_captcha else "ok"
                await log({
                    "job": job.id, "url": job.url, "event": "done",
                    "outcome": job.outcome, "status_code": job.status_code,
                    "duration_ms": int((time.time() - t0) * 1000), "title": job.title,
                })
        except Exception as e:
            job.status = "failed"
            job.error = str(e)
            job.outcome = "failed"
            await log({
                "job": job.id, "url": job.url, "event": "failed",
                "error": str(e)[:500], "duration_ms": int((time.time() - t0) * 1000),
            })
        finally:
            job.page = None
            if job.outcome == "cancelled":
                await log({"job": job.id, "url": job.url, "event": "cancelled", "duration_ms": int((time.time() - t0) * 1000)})
            job.finished.set()


@app.post("/fetch", dependencies=[Depends(require_auth)])
async def fetch(req: FetchRequest):
    if not req.url.startswith(("http://", "https://")):
        raise HTTPException(400, "url must start with http(s)://")
    job = Job(req.url, req.extract, req.selector, req.pause)
    jobs[job.id] = job
    asyncio.create_task(run_fetch(job))
    await job.finished.wait()
    if job.status in ("done",):
        return {"status": job.status_code, "title": job.title, "url": job.final_url, "text": (job.text or "")[:200_000], "outcome": job.outcome, "job_id": job.id}
    raise HTTPException(502, f"{job.outcome or job.status}: {job.error or 'unknown error'}")


# ---------------------------------------------------------------- admin / WebUI endpoints
class Action(BaseModel):
    type: str  # click|type|key|scroll|goto|reload|back
    x: float | None = None
    y: float | None = None
    img_w: float | None = None
    text: str | None = None
    key: str | None = None
    delta: int | None = None
    url: str | None = None


@app.get("/", include_in_schema=False)
async def index():
    return FileResponse(os.path.join(BASE_DIR, "static", "index.html"))


@app.get("/health")
async def health():
    return {"ok": True, "version": app.version, "active": sum(1 for j in jobs.values() if j.status in ("loading", "captcha", "solving"))}


@app.get("/api/jobs", dependencies=[Depends(require_auth)])
async def list_jobs():
    return [j.to_dict() for j in jobs.values()][-50:][::-1]


@app.get("/api/jobs/{job_id}", dependencies=[Depends(require_auth)])
async def job_detail(job_id: str):
    job = jobs.get(job_id)
    if not job:
        raise HTTPException(404, "job not found")
    d = job.to_dict()
    d["text"] = (job.text or "")[:50_000]
    return d


@app.get("/api/jobs/{job_id}/screenshot", dependencies=[Depends(require_auth)])
async def screenshot(job_id: str):
    job = jobs.get(job_id)
    if not job or job.page is None:
        raise HTTPException(404, "no live page for this job")
    png = await job.page.screenshot(type="jpeg", quality=60)
    return Response(content=png, media_type="image/jpeg")


async def run_action(job: Job, action: Action):
    page = job.page
    if page is None:
        raise HTTPException(404, "page closed")
    a = action
    if a.type == "click":
        vw, vh = job.viewport["width"], job.viewport["height"]
        img_w = a.img_w or vw
        x = a.x * vw / img_w
        y = a.y * vh / img_w  # same scale
        await page.mouse.click(x, y)
    elif a.type == "type":
        await page.keyboard.type(a.text or "", delay=25)
    elif a.type == "key":
        await page.keyboard.press(a.key or "Enter")
    elif a.type == "scroll":
        await page.mouse.wheel(0, a.delta or 300)
    elif a.type == "goto":
        await page.goto(a.url, wait_until="domcontentloaded", timeout=45_000)
    elif a.type == "reload":
        await page.reload(wait_until="domcontentloaded")
    elif a.type == "back":
        await page.go_back(wait_until="domcontentloaded")
    else:
        raise HTTPException(400, f"unknown action {a.type}")
    await log({"job": job.id, "event": "action", "type": a.type})
    return job.to_dict()


@app.post("/api/jobs/{job_id}/action", dependencies=[Depends(require_auth)])
async def do_action(job_id: str, action: Action):
    job = jobs.get(job_id)
    if not job:
        raise HTTPException(404)
    async with job.action_lock:
        return await run_action(job, action)


@app.post("/api/jobs/{job_id}/continue", dependencies=[Depends(require_auth)])
async def continue_job(job_id: str):
    job = jobs.get(job_id)
    if not job:
        raise HTTPException(404)
    job.status = "solving"
    job.solved.set()
    return {"ok": True}


@app.post("/api/jobs/{job_id}/cancel", dependencies=[Depends(require_auth)])
async def cancel_job(job_id: str):
    job = jobs.get(job_id)
    if not job:
        raise HTTPException(404)
    job.cancelled = True
    job.outcome = "cancelled"
    job.status = "cancelled"
    job.solved.set()
    return {"ok": True}


@app.get("/api/logs", dependencies=[Depends(require_auth)])
async def get_logs():
    return await read_logs()


# periodic log of health to admin logs
@app.on_event("startup")
async def startup():
    await log({"event": "server_start", "version": app.version, "proxy": bool(PROXY)})
    print(f"[startup] auth={'on' if ADMIN_PASS else 'OFF (LAN open)'} captcha_wait={CAPTCHA_WAIT_MS}ms solve_timeout={SOLVE_TIMEOUT_S}s", flush=True)