#!/usr/bin/env python3
"""Small local HTTP worker for the YYB admin integration.

The worker owns per-job sessions and invokes the already verified protocol
scripts. It intentionally exposes QR images and state only; cookies and raw
payment links never leave the worker API.
"""
from __future__ import annotations

import argparse
import base64
import importlib.util
import json
import os
import re
import subprocess
import sys
import threading
import time
import uuid
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import urlparse

ROOT = Path(__file__).resolve().parent.parent
DEFAULT_DATA = ROOT / "config" / "worker-jobs"
WORKER_KEY = os.environ.get("YYB_WORKER_KEY", "")
_jobs: dict[str, dict] = {}
_lock = threading.Lock()


def _load_selector():
    path = ROOT / "scripts" / "select-peace-elite.py"
    spec = importlib.util.spec_from_file_location("yyb_worker_selector", path)
    if not spec or not spec.loader:
        raise RuntimeError("无法加载和平精英选择器")
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def _job_dir(job_id: str) -> Path:
    return Path(_jobs[job_id]["directory"])


def _safe_log(job: dict, line: str) -> None:
    # Do not persist cookies, payment URI, or long opaque tokens in the worker API.
    clean = re.sub(r"weixin://\S+", "[付款链接已隐藏]", line)
    clean = re.sub(r"HOLD_[A-Za-z0-9_-]+", "[订单已隐藏]", clean)
    clean = re.sub(r"(订单\s*[:：]\s*)\S+", r"\1[订单已隐藏]", clean)
    clean = re.sub(r"(?:token|openid|openkey|cookie)=\S+", "[敏感字段已隐藏]", clean, flags=re.I)
    clean = re.sub(r"(?:pay_token|web_token|anti_token|session_id|sessionid)\s*[:= ]\s*\S+",
                   "[敏感字段已隐藏]", clean, flags=re.I)
    with _lock:
        job["logs"] = (job.get("logs", []) + [clean.strip()])[-100:]


def _run_process(job_id: str, command: list[str], phase: str) -> None:
    job = _jobs[job_id]
    environment = os.environ.copy()
    environment.pop("NODE_OPTIONS", None)
    with _lock:
        job["phase"] = phase
        job["status"] = "running"
    try:
        process = subprocess.Popen(command, cwd=ROOT, stdout=subprocess.PIPE,
                                   stderr=subprocess.STDOUT, text=True,
                                   bufsize=1, env=environment)
        with _lock:
            job["process_pid"] = process.pid
        assert process.stdout is not None
        for line in process.stdout:
            _safe_log(job, line)
        code = process.wait()
        with _lock:
            job["process_pid"] = None
            if code != 0:
                job["status"] = "failed"
                job["phase"] = phase
                job["message"] = f"{phase}失败（退出码 {code}）"
            elif phase == "login":
                job["status"] = "ready"
                job["phase"] = "selection"
                job["message"] = "登录成功，请选择平台、点券、区服和角色"
            else:
                job["status"] = "success"
                job["phase"] = "completed"
                job["message"] = "付款流程已完成"
    except Exception as exc:  # noqa: BLE001
        with _lock:
            job["status"] = "failed"
            job["message"] = str(exc)


def _start_login(job_id: str, provider: str, timeout: int) -> None:
    job = _jobs[job_id]
    directory = _job_dir(job_id)
    session = directory / "mall-session.json"
    qr = directory / ("qq-login.jpg" if provider == "qq" else "wechat-login.jpg")
    command = [sys.executable, f"scripts/login-{provider}.py", "--session", str(session),
               "--qr", str(qr), "--timeout", str(timeout)]
    with _lock:
        job["provider"] = provider
        job["qr_path"] = str(qr)
        job["session_path"] = str(session)
        job["status"] = "waiting_login"
        job["phase"] = "login"
    threading.Thread(target=_run_process, args=(job_id, command, "login"), daemon=True).start()


def _selection_options(job_id: str, platform: str, points: int | None, zone_id: str | None = None) -> dict:
    if job_id not in _jobs:
        raise ValueError("任务不存在")
    selector = _load_selector()
    session = json.loads((_job_dir(job_id) / "mall-session.json").read_text(encoding="utf-8"))
    cookies = session.get("cookies", {})
    products = selector.product_options(cookies, platform)
    if points is not None and not any(int(item["points"]) == points for item in products):
        raise ValueError("当前登录态不支持该点券档位")
    product = next((item for item in products if int(item["points"]) == points), None) if points else None
    if product is None:
        product = products[0]
    cmall = selector.Cmall(cookies, str(product["offer_id"]), platform)
    zones = cmall.zones()
    selected_zone = next((zone for zone in zones if str(zone["zone_id"]) == str(zone_id)), None)
    if zone_id and selected_zone is None:
        raise ValueError("区服不存在")
    selected_zone = selected_zone or (zones[0] if zones else None)
    roles = cmall.roles(selected_zone["zone_id"]) if selected_zone else []
    return {"products": products, "zones": zones, "roles": roles,
            "default_product": product, "default_zone": selected_zone}


def _start_payment(job_id: str, selection: dict) -> None:
    directory = _job_dir(job_id)
    session = directory / "mall-session.json"
    response = directory / "mall-order-response.json"
    output = directory / "jsdom-order"
    command = [sys.executable, "main.py", "mall", "auto", "--session", str(session),
               "--order-template", str(ROOT / "config" / "mall-order-template.json"),
               "--quantity", "1", "--product-id", str(selection["product_id"]),
               "--offer-id", str(selection["offer_id"]),
               "--role-id", str(selection["role_id"]), "--role-name", str(selection["role_name"]),
               "--zone-id", str(selection["zone_id"]), "--zone-name", str(selection["zone_name"]),
               "--output", str(response)]
    if selection.get("order_pf"):
        command.extend(["--pf", str(selection["order_pf"])])
    def run() -> None:
        _run_process(job_id, command, "order")
        job = _jobs[job_id]
        if job.get("status") != "success" or not response.exists():
            return
        pay_command = [sys.executable, "scripts/jsdom-pay.py", "--session", str(session),
                       "--mall-response", str(response), "--out-dir", str(output),
                       "--zone-id", str(selection["zone_id"]),
                       "--pf", str(selection.get("order_pf", "")),
                       "--amount-fen", str(selection["price_fen"])]
        _run_process(job_id, pay_command, "payment")
    threading.Thread(target=run, daemon=True).start()


def _stop_job(job_id: str) -> None:
    if job_id not in _jobs:
        raise ValueError("任务不存在")
    pid = _jobs[job_id].get("process_pid")
    if pid:
        try:
            os.kill(int(pid), 15)
        except ProcessLookupError:
            pass
    with _lock:
        _jobs[job_id]["status"] = "failed"
        _jobs[job_id]["phase"] = "stopped"
        _jobs[job_id]["message"] = "任务已停止"


def _public_job(job_id: str) -> dict:
    job = _jobs[job_id]
    result = {key: value for key, value in job.items()
              if key not in {"directory", "session_path", "process_pid"}}
    qr_path = job.get("qr_path", "")
    if qr_path and Path(qr_path).exists():
        qr_bytes = Path(qr_path).read_bytes()
        result["qr_data"] = base64.b64encode(qr_bytes).decode("ascii")
        result["qr_mime_type"] = "image/jpeg" if qr_bytes.startswith(b"\xff\xd8\xff") else "image/png"
    output = Path(job["directory"]) / "jsdom-order"
    for name in ("wechat-pay.png", "payment-status.json"):
        path = None
        if output.exists():
            direct = output / name
            path = direct if direct.exists() else next(output.glob(f"*/{name}"), None)
        if path and name.endswith(".png"):
            payment_qr_bytes = path.read_bytes()
            result["payment_qr_data"] = base64.b64encode(payment_qr_bytes).decode("ascii")
            result["payment_qr_mime_type"] = "image/jpeg" if payment_qr_bytes.startswith(b"\xff\xd8\xff") else "image/png"
        elif path:
            try:
                status = json.loads(path.read_text(encoding="utf-8"))
                matched = status.get("matched_completion") if isinstance(status, dict) else None
                result["payment_status"] = {
                    "checked_at": status.get("checked_at") if isinstance(status, dict) else None,
                    "matched_completion": {
                        "is_finished": matched.get("is_finished"),
                        "status": matched.get("status"),
                    } if isinstance(matched, dict) else None,
                }
            except (OSError, json.JSONDecodeError):
                pass
    return result


class Handler(BaseHTTPRequestHandler):
    server_version = "YYBWorker/1"

    def _json(self, status: int, value: dict) -> None:
        body = json.dumps(value, ensure_ascii=False).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _body(self) -> dict:
        length = int(self.headers.get("Content-Length", "0"))
        return json.loads(self.rfile.read(length) or b"{}")

    def _authorized(self) -> bool:
        if not WORKER_KEY:
            return True
        value = self.headers.get("Authorization", "")
        return value == f"Bearer {WORKER_KEY}"

    def do_POST(self) -> None:  # noqa: N802
        if not self._authorized():
            return self._json(401, {"detail": "未授权"})
        path = urlparse(self.path).path.strip("/").split("/")
        try:
            if path == ["v1", "jobs"]:
                job_id = uuid.uuid4().hex[:16]
                directory = DEFAULT_DATA / job_id
                directory.mkdir(parents=True, exist_ok=True)
                _jobs[job_id] = {"job_id": job_id, "status": "created", "phase": "login",
                                 "logs": [], "directory": str(directory), "created_at": int(time.time())}
                return self._json(201, _public_job(job_id))
            if len(path) == 4 and path[:2] == ["v1", "jobs"] and path[3] == "login":
                job_id = path[2]
                body = self._body()
                if job_id not in _jobs or body.get("provider") not in {"qq", "wechat"}:
                    return self._json(400, {"detail": "无效任务或登录方式"})
                _start_login(job_id, body["provider"], int(body.get("timeout", 600)))
                return self._json(202, _public_job(job_id))
            if len(path) == 4 and path[:2] == ["v1", "jobs"] and path[3] == "selection-options":
                job_id = path[2]
                body = self._body()
                options = _selection_options(job_id, str(body.get("platform", "android")), body.get("points"), body.get("zone_id"))
                _jobs[job_id]["selection_options"] = options
                return self._json(200, options)
            if len(path) == 4 and path[:2] == ["v1", "jobs"] and path[3] == "selection":
                job_id = path[2]
                body = self._body()
                required = ("platform", "points", "product_id", "role_id", "role_name", "zone_id")
                if job_id not in _jobs or any(not body.get(key) for key in required):
                    return self._json(400, {"detail": "选择参数不完整"})
                selector = _load_selector()
                if body["platform"] not in selector.PLATFORMS:
                    return self._json(400, {"detail": "不支持的平台"})
                session_path = _job_dir(job_id) / "mall-session.json"
                cookies = json.loads(session_path.read_text(encoding="utf-8")).get("cookies", {})
                product = next((item for item in selector.product_options(cookies, body["platform"])
                                if str(item["product_id"]) == str(body["product_id"])
                                and int(item["points"]) == int(body["points"])), None)
                if product is None:
                    return self._json(400, {"detail": "商品已失效，请重新选择"})
                cmall = selector.Cmall(cookies, str(product["offer_id"]), body["platform"])
                zone = next((item for item in cmall.zones()
                             if str(item["zone_id"]) == str(body["zone_id"])), None)
                if zone is None:
                    return self._json(400, {"detail": "区服已失效，请重新选择"})
                role = next((item for item in cmall.roles(zone["zone_id"])
                             if str(item["role_id"]) == str(body["role_id"])), None)
                if role is None or role.get("ban_status") == "1":
                    return self._json(400, {"detail": "角色不可充值，请重新选择"})
                _jobs[job_id]["selection"] = {
                    "platform": body["platform"], "points": product["points"],
                    "price_fen": product["price_fen"],
                    "product_id": product["product_id"], "offer_id": product["offer_id"],
                    "zone_id": zone["zone_id"], "zone_name": zone["name"],
                    "role_id": role["role_id"], "role_name": role["name"],
                    "order_pf": selector.PLATFORMS[body["platform"]]["order_pf"],
                }
                _jobs[job_id]["phase"] = "payment"
                return self._json(200, _public_job(job_id))
            if len(path) == 4 and path[:2] == ["v1", "jobs"] and path[3] == "payment":
                job_id = path[2]
                if job_id not in _jobs or not _jobs[job_id].get("selection"):
                    return self._json(400, {"detail": "请先完成角色选择"})
                _start_payment(job_id, _jobs[job_id]["selection"])
                return self._json(202, _public_job(job_id))
            if len(path) == 4 and path[:2] == ["v1", "jobs"] and path[3] == "stop":
                _stop_job(path[2])
                return self._json(200, _public_job(path[2]))
            return self._json(404, {"detail": "接口不存在"})
        except Exception as exc:  # noqa: BLE001
            return self._json(500, {"detail": str(exc)})

    def do_GET(self) -> None:  # noqa: N802
        if urlparse(self.path).path == "/health":
            return self._json(200, {"status": "ok"})
        if not self._authorized():
            return self._json(401, {"detail": "未授权"})
        path = urlparse(self.path).path.strip("/").split("/")
        if len(path) == 3 and path[:2] == ["v1", "jobs"] and path[2] in _jobs:
            return self._json(200, _public_job(path[2]))
        return self._json(404, {"detail": "接口不存在"})

    def log_message(self, fmt: str, *args) -> None:
        return


def main() -> int:
    global DEFAULT_DATA, WORKER_KEY
    parser = argparse.ArgumentParser(description="YYB admin worker")
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=8810)
    parser.add_argument("--data-dir", type=Path, default=DEFAULT_DATA)
    parser.add_argument("--key", default=None, help="HTTP Bearer 鉴权密钥；默认读取 YYB_WORKER_KEY")
    args = parser.parse_args()
    DEFAULT_DATA = args.data_dir
    if args.key is not None:
        WORKER_KEY = args.key
    if args.host not in {"127.0.0.1", "localhost", "::1"} and not WORKER_KEY:
        parser.error("监听非本机地址时必须设置 --key 或 YYB_WORKER_KEY")
    DEFAULT_DATA.mkdir(parents=True, exist_ok=True)
    server = ThreadingHTTPServer((args.host, args.port), Handler)
    print(f"YYB worker listening on {args.host}:{args.port}", flush=True)
    server.serve_forever()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())