"""Provider sign-in, driven from the UI but completed by the operator.

The orchestrator never handles a password or a token. It starts the CLI's own device-code
login, shows the verification URL and code the CLI prints, and waits. The operator opens
that URL in their own browser; the CLI writes the credential into the service account's
home directory itself. Nothing secret passes through this process or its logs.
"""

from __future__ import annotations

import json
import os
import re
import shlex
import shutil
import subprocess
import threading
import time
from dataclasses import dataclass, field
from pathlib import Path

from .infra.process import child_environment, is_running, terminate_group

#: CLIs colour their output. Left in, the escape sequence becomes part of the link the
#: operator is told to open, and the verification code stops matching entirely.
ANSI = re.compile(r"\x1b\[[0-9;?]*[A-Za-z]|\x1b\][^\x07\x1b]*(?:\x07|\x1b\\)|\x1b[()][AB012]")


def strip_ansi(text: str) -> str:
    return ANSI.sub("", text)


#: Where the verification URL and the code appear in what the CLIs print.
URL_PATTERN = re.compile(r"https?://[^\s\"'<>\[\]]+")
URL_CONTINUATION = re.compile(r"^[A-Za-z0-9._~:/?#\[\]@!$&'()*+,;=%-]+$")
CODE_PATTERN = re.compile(r"\b([A-Z0-9]{3,6}-[A-Z0-9]{3,6}|[A-Z0-9]{6,10})\b")

#: Words printed next to the one-time code, used to prefer it over other codes on screen.
CODE_HINTS = ("one-time code", "验证码", "enter this", "code:")


def find_urls(text: str) -> list[str]:
    """Return URLs, rejoining pseudo-terminal line wraps in query strings."""
    lines = text.splitlines()
    urls = []
    for index, line in enumerate(lines):
        for match in URL_PATTERN.finditer(line):
            url = match.group()
            if "?" in url:
                continuation_index = index + 1
                while continuation_index < len(lines):
                    # ``script`` can put a blank record between physical terminal
                    # wraps. Skip those records only when the next non-empty line
                    # is still a URL-safe continuation; never bridge into prose.
                    next_index = continuation_index
                    while next_index < len(lines) and not lines[next_index].strip():
                        next_index += 1
                    if next_index >= len(lines):
                        break
                    fragment = lines[next_index].strip()
                    if not URL_CONTINUATION.fullmatch(fragment):
                        break
                    url += fragment
                    continuation_index = next_index + 1
            urls.append(url)
    return urls


def pick_code(text: str) -> str:
    """The one-time code, preferring one printed near the words that announce it."""
    lines = text.splitlines()
    for index, line in enumerate(lines):
        if any(hint in line.lower() for hint in CODE_HINTS):
            for candidate in lines[index : index + 3]:
                found = CODE_PATTERN.findall(candidate)
                if found:
                    return found[0]
    found = CODE_PATTERN.findall(text)
    return found[0] if found else ""


#: Lines that mean the CLI is telling us we are not signed in.
SIGNED_OUT = ("not logged in", "please run", "login required", "unauthorized", "no credentials")

#: How an account can be signed in, most specific first. An enterprise backend supplies its
#: own credentials, so a device-code login neither applies to it nor would help.
MODES = {
    "claude": (
        ("bedrock", ("CLAUDE_CODE_USE_BEDROCK",), "AWS Bedrock"),
        ("vertex", ("CLAUDE_CODE_USE_VERTEX",), "Google Vertex"),
        ("foundry", ("CLAUDE_CODE_USE_FOUNDRY",), "Azure Foundry"),
        ("gateway", ("ANTHROPIC_BASE_URL",), "自建网关"),
        ("api_key", ("ANTHROPIC_API_KEY", "ANTHROPIC_AUTH_TOKEN"), "API key"),
    ),
    "codex": (
        ("gateway", ("OPENAI_BASE_URL",), "自建网关"),
        ("api_key", ("OPENAI_API_KEY", "CODEX_ACCESS_TOKEN"), "API key"),
    ),
}

#: Enterprise backends authenticate elsewhere; offering a device login there misleads.
INTERACTIVE_MODES = {"account", "api_key"}

# Claude setup-token renders its verification URL only when attached to a terminal.
USE_POSIX_PTY = os.name != "nt"


def detect_mode(provider: str, environment: dict[str, str] | None = None) -> dict:
    """Which credential route this provider is configured for, and where it came from."""
    source = dict(os.environ if environment is None else environment)
    for name, variables, label in MODES.get(provider, ()):
        present = [v for v in variables if source.get(v)]
        if present:
            # The variable name is reported; its value never is.
            return {"mode": name, "label": label, "from": present[0]}
    return {"mode": "account", "label": "个人账号登录", "from": ""}


def _why(status: dict, exit_code: int | None, output: str) -> str:
    """One sentence on where a login stands, from what the CLI actually reported."""
    if status.get("logged_in"):
        return "已登录。"
    if exit_code is None:
        return "登录进程仍在等待你在浏览器里完成授权。"
    if exit_code == 0:
        return (
            "登录进程报告成功，但 status 仍显示未登录。"
            "多半是它把凭据写到了另一个家目录,确认服务账号与执行登录的账号是同一个。"
        )
    tail = [line for line in output.splitlines() if line.strip()][-1:] if output else []
    return f"登录进程以退出码 {exit_code} 结束。" + (f"最后一行：{tail[0][:160]}" if tail else "")


def _hint(provider: str, mode: dict) -> str:
    """What the operator has to do, given how this provider is meant to authenticate."""
    binary = "codex" if provider == "codex" else "claude"
    if mode["mode"] == "account":
        return f"点「登录」用设备码登录，或在服务器上以服务账号运行 {binary} 登录。"
    if mode["mode"] == "api_key":
        if provider == "codex":
            return (
                "已配置 API key。若仍未登录，在服务器上以服务账号执行："
                "printenv OPENAI_API_KEY | codex login --with-api-key"
            )
        return "已配置 API key，Claude 会直接使用，无需登录步骤。"
    return (
        f"使用{mode['label']}（由 {mode['from']} 指定），凭据由该后端自己管理，"
        "不需要也无法在这里登录。"
    )


@dataclass
class LoginSession:
    provider: str
    started_at: float = field(default_factory=time.time)
    process: subprocess.Popen | None = None
    output_path: Path | None = None

    def output(self) -> str:
        try:
            return self.output_path.read_text(encoding="utf-8", errors="replace") if self.output_path else ""
        except OSError:
            return ""

    def verification(self) -> dict:
        """The URL and code to hand the operator, as soon as the CLI prints them."""
        text = strip_ansi(self.output())
        urls = [u.rstrip(".,;") for u in find_urls(text) if "api." not in u]
        return {
            "url": urls[0] if urls else "",
            "code": pick_code(text),
            "output": text[-2000:],
            # None while it is still waiting; 0 once the CLI reports it finished.
            "exit_code": self.process.poll() if self.process else None,
        }

    def running(self) -> bool:
        return self.process is not None and self.process.poll() is None


class ProviderAuth:
    """Check sign-in state and run the CLIs' device-code login."""

    #: How each CLI reports and performs sign-in, headlessly.
    COMMANDS = {
        "codex": {"status": ["login", "status"], "login": ["login", "--device-auth"]},
        "claude": {"status": ["auth", "status"], "login": ["setup-token"]},
    }

    def __init__(
        self,
        adapters: dict,
        log_root: str | Path,
        timeout_s: float = 600,
        environment: dict[str, str] | None = None,
    ) -> None:
        #: name -> adapter, used only for the binary path each was configured with.
        self.adapters = adapters
        self.log_root = Path(log_root)
        self.timeout_s = timeout_s
        #: Overridable so a test can describe an enterprise host without setting real vars.
        self.environment = environment
        self._sessions: dict[str, LoginSession] = {}
        self._guard = threading.Lock()

    def _binary(self, provider: str) -> str:
        adapter = self.adapters.get(provider)
        return getattr(adapter, "binary", provider) or provider

    def status(self, provider: str) -> dict:
        """Whether this provider is signed in for the account running the service."""
        spec = self.COMMANDS.get(provider)
        if spec is None:
            return {"provider": provider, "known": False, "logged_in": False, "detail": "未知的 provider"}
        try:
            result = subprocess.run(
                [self._binary(provider), *spec["status"]],
                capture_output=True, text=True, errors="replace", timeout=20,
                env=child_environment(True),
            )
        except (OSError, subprocess.TimeoutExpired) as error:
            return {"provider": provider, "known": True, "logged_in": False, "detail": str(error)}
        text = ((result.stdout or "") + (result.stderr or "")).strip()
        signed_out = any(marker in text.lower() for marker in SIGNED_OUT)
        if provider == "claude":
            try:
                status_payload = json.loads(result.stdout or "{}")
            except (json.JSONDecodeError, TypeError):
                status_payload = {}
            logged_in = result.returncode == 0 and status_payload.get("loggedIn") is True
        else:
            logged_in = result.returncode == 0 and not signed_out
        mode = detect_mode(provider, self.environment)
        return {
            "provider": provider,
            "known": True,
            "logged_in": logged_in,
            "detail": text.splitlines()[0][:200] if text else "",
            **mode,
            # A device-code login only makes sense for the routes that have one.
            "can_login_here": mode["mode"] in INTERACTIVE_MODES,
            "hint": _hint(provider, mode),
        }

    def status_all(self) -> dict:
        return {name: self.status(name) for name in self.COMMANDS}

    def begin(self, provider: str) -> dict:
        """Start the CLI's device-code login and return what the operator must open."""
        if provider not in self.COMMANDS:
            raise ValueError(f"未知的 provider: {provider}")
        mode = detect_mode(provider, self.environment)
        if mode["mode"] not in INTERACTIVE_MODES:
            raise ValueError(
                f"该 provider 配置为{mode['label']}（{mode['from']}），凭据由该后端管理，"
                "不能在这里登录。"
            )
        with self._guard:
            existing = self._sessions.get(provider)
            if existing and existing.running():
                return {"provider": provider, "already_running": True, **existing.verification()}

            self.log_root.mkdir(parents=True, exist_ok=True)
            output_path = self.log_root / f"login-{provider}.log"
            output_path.write_text("", encoding="utf-8")
            handle = output_path.open("ab")
            login_argv = [self._binary(provider), *self.COMMANDS[provider]["login"]]
            if provider == "claude" and USE_POSIX_PTY:
                script_binary = shutil.which("script")
                if script_binary:
                    login_argv = [script_binary, "-qefc", shlex.join(login_argv), "/dev/null"]
            process = subprocess.Popen(
                login_argv,
                stdin=subprocess.PIPE, stdout=handle, stderr=handle,
                env=child_environment(True),
                # Its own group, so cancelling reaches the whole login, exactly as it
                # does for an agent run.
                creationflags=subprocess.CREATE_NEW_PROCESS_GROUP if os.name == "nt" else 0,
                start_new_session=os.name != "nt",
            )
            session = LoginSession(provider, process=process, output_path=output_path)
            self._sessions[provider] = session

        # Give the CLI a moment to print its verification URL before answering.
        deadline = time.time() + 15
        while time.time() < deadline:
            found = session.verification()
            if found["url"]:
                return {"provider": provider, "already_running": False, **found}
            if not session.running():
                break
            time.sleep(0.4)
        return {"provider": provider, "already_running": False, **session.verification()}

    def submit_code(self, provider: str, code: str) -> bool:
        """Forward a browser callback code to the still-waiting provider CLI."""
        value = code.strip()
        if not value or "\r" in value or "\n" in value:
            raise ValueError("授权码不能为空且必须为单行")
        session = self._sessions.get(provider)
        if session is None or not session.running() or session.process is None or session.process.stdin is None:
            return False
        try:
            session.process.stdin.write((value + "\r").encode("utf-8"))
            session.process.stdin.flush()
        except (BrokenPipeError, OSError):
            return False
        return True
    def progress(self, provider: str) -> dict:
        """Where a started login has got to, and whether it finished."""
        session = self._sessions.get(provider)
        if session is None:
            return {"provider": provider, "running": False, "started": False, **self.status(provider)}
        if session.running() and time.time() - session.started_at > self.timeout_s:
            self.cancel(provider)
        found = session.verification()
        status = self.status(provider)
        finished = found["exit_code"]
        return {
            "provider": provider,
            "started": True,
            "running": session.running(),
            **found,
            **status,
            # What the CLI itself concluded, which is the only thing that explains a
            # login that looks complete in the browser but is not recorded on disk.
            "finished_ok": finished == 0,
            "why": _why(status, finished, found["output"]),
        }

    def cancel(self, provider: str) -> bool:
        session = self._sessions.get(provider)
        if session is None or session.process is None:
            return False
        pid = session.process.pid
        if is_running(pid):
            terminate_group(pid)
        return True