"""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 os
import re
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\"'<>\[\]]+")
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 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"}


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 URL_PATTERN.findall(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": ["--version"], "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)
        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")
            process = subprocess.Popen(
                [self._binary(provider), *self.COMMANDS[provider]["login"]],
                stdin=subprocess.DEVNULL, 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 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