"""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 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']} 指定），凭据由该后端自己管理，"
        "不需要也无法在这里登录。"
    )


class PtyProcess:
    """Small Popen-compatible status wrapper for a child made by pty.fork()."""

    def __init__(self, pid: int) -> None:
        self.pid = pid
        self.returncode: int | None = None
        self.stdin = None

    def poll(self) -> int | None:
        if self.returncode is not None:
            return self.returncode
        try:
            waited, status = os.waitpid(self.pid, os.WNOHANG)
        except ChildProcessError:
            return self.returncode
        if waited:
            self.returncode = os.waitstatus_to_exitcode(status)
        return self.returncode


def _start_controlling_pty(argv: list[str], environment: dict[str, str]) -> tuple[PtyProcess, int]:
    """Launch a child with an actual controlling terminal and return its master fd."""
    import pty

    pid, master_fd = pty.fork()
    if pid == 0:
        try:
            os.execvpe(argv[0], argv, environment)
        except OSError:
            os._exit(127)
    return PtyProcess(pid), master_fd

def _record_pty(master_fd: int, output_path: Path) -> None:
    """Copy terminal output while retaining the PTY master for typed input."""
    try:
        with output_path.open("ab") as handle:
            while True:
                try:
                    chunk = os.read(master_fd, 4096)
                except OSError:
                    break
                if not chunk:
                    break
                handle.write(chunk)
                handle.flush()
    finally:
        try:
            os.close(master_fd)
        except OSError:
            pass


@dataclass
class LoginSession:
    provider: str
    started_at: float = field(default_factory=time.time)
    process: subprocess.Popen | None = None
    output_path: Path | None = None
    input_fd: int | 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)
        if provider == "claude" and logged_in and mode["mode"] == "account":
            config_dir = Path(os.environ.get("CLAUDE_CONFIG_DIR") or Path.home() / ".claude")
            try:
                credentials = json.loads((config_dir / ".credentials.json").read_text(encoding="utf-8"))
                oauth = credentials.get("claudeAiOauth", {})
                expiry = oauth.get("expiresAt")
                if isinstance(expiry, (int, float)) and expiry <= time.time() * 1000 and not oauth.get("refreshToken"):
                    logged_in = False
                    text = "Claude OAuth expired; sign in again."
            except (OSError, ValueError, AttributeError):
                pass
        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")
            login_argv = [self._binary(provider), *self.COMMANDS[provider]["login"]]
            if provider == "claude" and USE_POSIX_PTY:
                process, master_fd = _start_controlling_pty(login_argv, child_environment(True))
                threading.Thread(
                    target=_record_pty, args=(master_fd, output_path), daemon=True
                ).start()
                session = LoginSession(
                    provider, process=process, output_path=output_path, input_fd=master_fd
                )
            else:
                handle = output_path.open("ab")
                process = subprocess.Popen(
                    login_argv, stdin=subprocess.PIPE, stdout=handle, stderr=handle,
                    env=child_environment(True),
                    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:
            return False
        try:
            if session.input_fd is not None:
                os.write(session.input_fd, (value + "\r").encode("utf-8"))
            elif session.process.stdin is not None:
                session.process.stdin.write((value + "\r").encode("utf-8"))
                session.process.stdin.flush()
            else:
                return False
        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