import base64
import copy
import getpass
import json
import os
import platform
import plistlib
import shutil
import subprocess
import sys
import tempfile
import time
from collections import deque
from dataclasses import dataclass
from pathlib import Path
from textwrap import dedent
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen
from xml.etree import ElementTree

import click

from astrbot.core.utils.astrbot_path import astrbot_paths

DEFAULT_SERVICE_NAME = "astrbot"
DEFAULT_DASHBOARD_PORT = 6185
DEFAULT_STATUS_TIMEOUT_SECONDS = 2.0
DEFAULT_LOG_LINES = 200
MACOS_LABEL_PREFIX = "app.astrbot"
WINDOWS_TASK_XML_NS = "http://schemas.microsoft.com/windows/2004/02/mit/task"


@dataclass(frozen=True)
class ServiceState:
    manager: str
    installed: bool
    state: str
    path: Path | None = None
    enabled: str | None = None
    detail: str | None = None


@dataclass(frozen=True)
class DashboardPort:
    port: int
    detail: str | None = None


@dataclass(frozen=True)
class WebUIStatus:
    url: str
    accessible: bool
    status_code: int | None = None
    detail: str | None = None


@dataclass(frozen=True)
class AppLogConfig:
    enabled: bool
    path: Path
    configured_path: str | None = None


@click.group(name="service")
def service() -> None:
    """Install and manage AstrBot as a background service."""


def _validate_service_name(name: str) -> str:
    allowed_chars = set(
        "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_.-",
    )
    if not name or any(char not in allowed_chars for char in name):
        raise click.ClickException(
            "Service name can only contain letters, numbers, dots, underscores, and hyphens",
        )
    return name


def _get_astrbot_root() -> Path:
    return astrbot_paths.root


def _is_astrbot_root(path: Path) -> bool:
    return path.exists() and path.is_dir() and (path / ".astrbot").exists()


def _resolve_workdir(workdir: Path | None) -> Path:
    astrbot_root = (workdir or _get_astrbot_root()).expanduser().resolve()
    if not _is_astrbot_root(astrbot_root):
        raise click.ClickException(
            f"{astrbot_root} is not a valid AstrBot root directory. "
            "Use 'astrbot init' before installing the service",
        )
    return astrbot_root


def _resolve_astrbot_executable(executable: str | None) -> Path:
    if executable:
        discovered = shutil.which(executable)
        if discovered:
            return Path(discovered).expanduser().absolute()

        explicit_path = Path(executable).expanduser()
        if explicit_path.exists():
            return explicit_path.absolute()

        raise click.ClickException(f"AstrBot executable not found: {executable}")

    discovered = shutil.which("astrbot")
    if discovered:
        return Path(discovered).expanduser().absolute()

    current_argv = Path(sys.argv[0]).expanduser()
    if current_argv.name.startswith("astrbot") and current_argv.exists():
        return current_argv.absolute()

    raise click.ClickException(
        "Cannot find the astrbot executable. Install AstrBot with "
        "'uv tool install astrbot --python 3.12', or pass --executable",
    )


def _run_checked(command: list[str], failure_message: str) -> None:
    try:
        subprocess.run(command, check=True)
    except FileNotFoundError as e:
        raise click.ClickException(f"Command not found: {command[0]}") from e
    except subprocess.CalledProcessError as e:
        raise click.ClickException(f"{failure_message}: {e}") from e


def _run_capture(command: list[str]) -> subprocess.CompletedProcess[str] | None:
    try:
        return subprocess.run(
            command,
            check=False,
            capture_output=True,
            text=True,
        )
    except FileNotFoundError:
        return None


def _quote_systemd_value(value: Path | str) -> str:
    raw = str(value)
    escaped = raw.replace("\\", "\\\\").replace('"', '\\"').replace("%", "%%")
    if any(char.isspace() for char in raw) or any(
        char in raw for char in ['"', "\\", "%", ";"]
    ):
        return f'"{escaped}"'
    return escaped


def _build_systemd_unit(
    service_name: str,
    executable: Path,
    workdir: Path,
) -> str:
    return dedent(
        f"""\
        [Unit]
        Description=AstrBot Service
        Documentation=https://docs.astrbot.app
        After=network-online.target
        Wants=network-online.target

        [Service]
        Type=simple
        WorkingDirectory={_quote_systemd_value(workdir)}
        ExecStart={_quote_systemd_value(executable)} run
        Restart=on-failure
        RestartSec=5
        StandardOutput=journal
        StandardError=journal
        SyslogIdentifier={service_name}
        Environment=PYTHONUNBUFFERED=1

        [Install]
        WantedBy=default.target
        """,
    )


def _systemd_unit_path(service_name: str) -> Path:
    return Path.home() / ".config" / "systemd" / "user" / f"{service_name}.service"


def _systemd_unit_name(service_name: str) -> str:
    return f"{service_name}.service"


def _install_systemd_user_service(
    service_name: str,
    executable: Path,
    workdir: Path,
    *,
    force: bool,
    now: bool,
) -> Path:
    if platform.system() != "Linux":
        raise click.ClickException(
            "systemd service installation is only available on Linux",
        )
    if shutil.which("systemctl") is None:
        raise click.ClickException("systemctl was not found")

    unit_path = _systemd_unit_path(service_name)
    if unit_path.exists() and not force:
        raise click.ClickException(
            f"{unit_path} already exists. Use --force to overwrite",
        )

    unit_path.parent.mkdir(parents=True, exist_ok=True)
    unit_path.write_text(
        _build_systemd_unit(service_name, executable, workdir),
        encoding="utf-8",
    )

    _run_checked(
        ["systemctl", "--user", "daemon-reload"],
        "Failed to reload the systemd user daemon",
    )
    _run_checked(
        ["systemctl", "--user", "enable", unit_path.name],
        "Failed to enable the systemd user service",
    )
    if now:
        _run_checked(
            ["systemctl", "--user", "restart", unit_path.name],
            "Failed to start the systemd user service",
        )

    return unit_path


def _macos_label(service_name: str) -> str:
    return f"{MACOS_LABEL_PREFIX}.{service_name}"


def _launch_agent_path(service_name: str) -> Path:
    return (
        Path.home() / "Library" / "LaunchAgents" / f"{_macos_label(service_name)}.plist"
    )


def _macos_log_dir() -> Path:
    return Path.home() / "Library" / "Logs" / "AstrBot"


def _service_log_paths(service_name: str) -> tuple[Path, Path]:
    system = platform.system()
    if system == "Darwin":
        log_dir = _macos_log_dir()
    elif system == "Windows":
        return _windows_service_log_paths(service_name)
    else:
        log_dir = _get_astrbot_root() / "data" / "logs"
    return log_dir / f"{service_name}.out.log", log_dir / f"{service_name}.err.log"


def _build_launchd_plist(
    service_name: str,
    executable: Path,
    workdir: Path,
    log_dir: Path,
) -> dict:
    label = _macos_label(service_name)
    return {
        "Label": label,
        "ProgramArguments": [str(executable), "run"],
        "WorkingDirectory": str(workdir),
        "RunAtLoad": True,
        "KeepAlive": {"SuccessfulExit": False},
        "StandardOutPath": str(log_dir / f"{service_name}.out.log"),
        "StandardErrorPath": str(log_dir / f"{service_name}.err.log"),
        "EnvironmentVariables": {"PYTHONUNBUFFERED": "1"},
    }


def _install_launch_agent(
    service_name: str,
    executable: Path,
    workdir: Path,
    *,
    force: bool,
    now: bool,
) -> Path:
    if platform.system() != "Darwin":
        raise click.ClickException(
            "launchd service installation is only available on macOS",
        )
    if shutil.which("launchctl") is None:
        raise click.ClickException("launchctl was not found")

    plist_path = _launch_agent_path(service_name)
    if plist_path.exists() and not force:
        raise click.ClickException(
            f"{plist_path} already exists. Use --force to overwrite",
        )

    log_dir = _macos_log_dir()
    log_dir.mkdir(parents=True, exist_ok=True)
    plist_path.parent.mkdir(parents=True, exist_ok=True)
    with plist_path.open("wb") as f:
        plistlib.dump(
            _build_launchd_plist(service_name, executable, workdir, log_dir),
            f,
            sort_keys=False,
        )

    if now:
        if force:
            _stop_launch_agent(service_name, allow_missing=True)
        _start_launch_agent(service_name)

    return plist_path


def _task_element(
    parent: ElementTree.Element,
    name: str,
    text: str | None = None,
    attrib: dict[str, str] | None = None,
) -> ElementTree.Element:
    child = ElementTree.SubElement(
        parent,
        f"{{{WINDOWS_TASK_XML_NS}}}{name}",
        attrib or {},
    )
    if text is not None:
        child.text = text
    return child


def _windows_log_dir() -> Path:
    local_app_data = os.environ.get("LOCALAPPDATA")
    if local_app_data:
        return Path(local_app_data) / "AstrBot" / "Logs"
    return Path.home() / "AppData" / "Local" / "AstrBot" / "Logs"


def _windows_service_log_paths(service_name: str) -> tuple[Path, Path]:
    log_dir = _windows_log_dir()
    return log_dir / f"{service_name}.out.log", log_dir / f"{service_name}.err.log"


def _windows_powershell_executable() -> str:
    return "powershell.exe"


def _quote_windows_cmd_arg(value: Path | str) -> str:
    escaped = str(value).replace('"', '""')
    return f'"{escaped}"'


def _quote_powershell_literal(value: Path | str) -> str:
    escaped = str(value).replace("'", "''")
    return f"'{escaped}'"


def _build_windows_cmd_line(service_name: str, executable: Path) -> str:
    out_log, err_log = _windows_service_log_paths(service_name)
    return (
        f"{_quote_windows_cmd_arg(executable)} run "
        f">> {_quote_windows_cmd_arg(out_log)} "
        f"2>> {_quote_windows_cmd_arg(err_log)}"
    )


def _build_windows_powershell_arguments(
    service_name: str,
    executable: Path,
    workdir: Path,
) -> str:
    script = (
        "$ErrorActionPreference = 'Stop'\n"
        "$env:PYTHONUNBUFFERED = '1'\n"
        "$cmdExe = if ($env:COMSPEC) { $env:COMSPEC } else { 'cmd.exe' }\n"
        f"$astrbotCommand = {_quote_powershell_literal(_build_windows_cmd_line(service_name, executable))}\n"
        "$process = Start-Process "
        "-FilePath $cmdExe "
        "-ArgumentList @('/d', '/c', $astrbotCommand) "
        f"-WorkingDirectory {_quote_powershell_literal(workdir)} "
        "-WindowStyle Hidden "
        "-PassThru "
        "-Wait\n"
        "exit $process.ExitCode\n"
    )
    encoded_script = base64.b64encode(script.encode("utf-16le")).decode("ascii")
    return (
        "-NoLogo -NoProfile -NonInteractive -ExecutionPolicy Bypass "
        f"-WindowStyle Hidden -EncodedCommand {encoded_script}"
    )


def _build_windows_task_xml(
    service_name: str,
    executable: Path,
    workdir: Path,
) -> bytes:
    ElementTree.register_namespace("", WINDOWS_TASK_XML_NS)
    task = ElementTree.Element(
        f"{{{WINDOWS_TASK_XML_NS}}}Task",
        {"version": "1.4"},
    )

    registration_info = _task_element(task, "RegistrationInfo")
    _task_element(registration_info, "Description", "AstrBot Service")

    triggers = _task_element(task, "Triggers")
    logon_trigger = _task_element(triggers, "LogonTrigger")
    _task_element(logon_trigger, "Enabled", "true")

    principals = _task_element(task, "Principals")
    principal = _task_element(principals, "Principal", attrib={"id": "Author"})
    _task_element(principal, "LogonType", "InteractiveToken")
    _task_element(principal, "RunLevel", "LeastPrivilege")

    settings = _task_element(task, "Settings")
    _task_element(settings, "MultipleInstancesPolicy", "IgnoreNew")
    _task_element(settings, "DisallowStartIfOnBatteries", "false")
    _task_element(settings, "StopIfGoingOnBatteries", "false")
    _task_element(settings, "AllowHardTerminate", "true")
    _task_element(settings, "StartWhenAvailable", "true")
    _task_element(settings, "RunOnlyIfNetworkAvailable", "false")
    _task_element(settings, "AllowStartOnDemand", "true")
    _task_element(settings, "Enabled", "true")
    _task_element(settings, "Hidden", "true")
    _task_element(settings, "RunOnlyIfIdle", "false")
    _task_element(settings, "WakeToRun", "false")
    _task_element(settings, "ExecutionTimeLimit", "PT0S")
    _task_element(settings, "Priority", "7")
    restart = _task_element(settings, "RestartOnFailure")
    _task_element(restart, "Interval", "PT1M")
    _task_element(restart, "Count", "3")

    actions = _task_element(task, "Actions", attrib={"Context": "Author"})
    exec_action = _task_element(actions, "Exec")
    _task_element(exec_action, "Command", _windows_powershell_executable())
    _task_element(
        exec_action,
        "Arguments",
        _build_windows_powershell_arguments(service_name, executable, workdir),
    )
    _task_element(exec_action, "WorkingDirectory", str(workdir))

    ElementTree.indent(task, space="  ")
    return ElementTree.tostring(task, encoding="utf-16", xml_declaration=True)


def _windows_task_exists(service_name: str) -> bool:
    result = subprocess.run(
        ["schtasks", "/Query", "/TN", service_name],
        check=False,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )
    return result.returncode == 0


def _install_windows_task(
    service_name: str,
    executable: Path,
    workdir: Path,
    *,
    force: bool,
    now: bool,
) -> None:
    if platform.system() != "Windows":
        raise click.ClickException(
            "Windows scheduled task installation is only available on Windows",
        )
    if shutil.which("schtasks") is None:
        raise click.ClickException("schtasks was not found")
    if _windows_task_exists(service_name) and not force:
        raise click.ClickException(
            f"Scheduled task {service_name} already exists. Use --force to overwrite",
        )

    _windows_log_dir().mkdir(parents=True, exist_ok=True)
    task_xml = _build_windows_task_xml(service_name, executable, workdir)
    temp_path = None
    try:
        with tempfile.NamedTemporaryFile(suffix=".xml", delete=False) as f:
            temp_path = Path(f.name)
            f.write(task_xml)

        command = ["schtasks", "/Create", "/TN", service_name, "/XML", str(temp_path)]
        if force:
            command.append("/F")
        _run_checked(command, "Failed to create the Windows scheduled task")
    finally:
        if temp_path is not None:
            temp_path.unlink(missing_ok=True)

    if now:
        _run_checked(
            ["schtasks", "/Run", "/TN", service_name],
            "Failed to start the Windows scheduled task",
        )


def _first_output_line(result: subprocess.CompletedProcess[str]) -> str | None:
    text = (result.stdout or result.stderr).strip()
    if not text:
        return None
    return text.splitlines()[0].strip()


def _get_systemd_state(service_name: str) -> ServiceState:
    unit_path = _systemd_unit_path(service_name)
    installed = unit_path.exists()
    if shutil.which("systemctl") is None:
        return ServiceState(
            manager="systemd --user",
            installed=installed,
            state="unknown",
            path=unit_path,
            detail="systemctl was not found",
        )

    unit_name = _systemd_unit_name(service_name)
    active_result = _run_capture(["systemctl", "--user", "is-active", unit_name])
    enabled_result = _run_capture(["systemctl", "--user", "is-enabled", unit_name])
    if active_result is None:
        return ServiceState(
            manager="systemd --user",
            installed=installed,
            state="unknown",
            path=unit_path,
            detail="systemctl was not found",
        )

    state = (active_result.stdout or "").strip() or "unknown"
    detail = (
        None if active_result.returncode == 0 else _first_output_line(active_result)
    )
    enabled = None
    if enabled_result is not None:
        enabled = (enabled_result.stdout or "").strip() or None

    if not installed and state in {"inactive", "unknown"}:
        state = "not-installed"

    return ServiceState(
        manager="systemd --user",
        installed=installed,
        state=state,
        path=unit_path,
        enabled=enabled,
        detail=detail,
    )


def _get_launchd_state(service_name: str) -> ServiceState:
    plist_path = _launch_agent_path(service_name)
    installed = plist_path.exists()
    if shutil.which("launchctl") is None:
        return ServiceState(
            manager="launchd",
            installed=installed,
            state="unknown",
            path=plist_path,
            detail="launchctl was not found",
        )

    label = _macos_label(service_name)
    target = f"gui/{os.getuid()}/{label}"
    result = _run_capture(["launchctl", "print", target])
    if result is None:
        return ServiceState(
            manager="launchd",
            installed=installed,
            state="unknown",
            path=plist_path,
            detail="launchctl was not found",
        )

    if result.returncode != 0:
        return ServiceState(
            manager="launchd",
            installed=installed,
            state="not-loaded" if installed else "not-installed",
            path=plist_path,
            detail=_first_output_line(result),
        )

    output = result.stdout or ""
    state = "loaded"
    detail = None
    for line in output.splitlines():
        normalized = line.strip()
        if normalized.startswith("state = "):
            state = normalized.removeprefix("state = ").strip()
        elif normalized.startswith("pid = "):
            detail = normalized

    return ServiceState(
        manager="launchd",
        installed=installed,
        state=state,
        path=plist_path,
        detail=detail,
    )


def _parse_schtasks_field(output: str, field_name: str) -> str | None:
    prefix = f"{field_name}:"
    for line in output.splitlines():
        if line.startswith(prefix):
            return line.removeprefix(prefix).strip()
    return None


def _get_windows_task_state(service_name: str) -> ServiceState:
    if shutil.which("schtasks") is None:
        return ServiceState(
            manager="Task Scheduler",
            installed=False,
            state="unknown",
            detail="schtasks was not found",
        )

    result = _run_capture(
        ["schtasks", "/Query", "/TN", service_name, "/FO", "LIST", "/V"],
    )
    if result is None:
        return ServiceState(
            manager="Task Scheduler",
            installed=False,
            state="unknown",
            detail="schtasks was not found",
        )
    if result.returncode != 0:
        return ServiceState(
            manager="Task Scheduler",
            installed=False,
            state="not-installed",
            detail=_first_output_line(result),
        )

    status = _parse_schtasks_field(result.stdout or "", "Status") or "unknown"
    return ServiceState(
        manager="Task Scheduler",
        installed=True,
        state=status.lower(),
        detail=_parse_schtasks_field(result.stdout or "", "Task To Run"),
    )


def _get_service_state(service_name: str) -> ServiceState:
    system = platform.system()
    if system == "Linux":
        return _get_systemd_state(service_name)
    if system == "Darwin":
        return _get_launchd_state(service_name)
    if system == "Windows":
        return _get_windows_task_state(service_name)
    return ServiceState(
        manager="unknown",
        installed=False,
        state="unsupported",
        detail=f"Unsupported platform: {system}",
    )


def _load_dashboard_port(astrbot_root: Path) -> DashboardPort:
    config_path = astrbot_root / "data" / "cmd_config.json"
    if not config_path.exists():
        return DashboardPort(
            DEFAULT_DASHBOARD_PORT,
            f"{config_path} does not exist; using default port",
        )

    try:
        config = json.loads(config_path.read_text(encoding="utf-8-sig"))
        port = int(config.get("dashboard", {}).get("port", DEFAULT_DASHBOARD_PORT))
    except (OSError, TypeError, ValueError, json.JSONDecodeError) as e:
        return DashboardPort(
            DEFAULT_DASHBOARD_PORT,
            f"Failed to read dashboard port from {config_path}: {e}; using default port",
        )

    if port < 1 or port > 65535:
        return DashboardPort(
            DEFAULT_DASHBOARD_PORT,
            f"Invalid dashboard port {port}; using default port",
        )
    return DashboardPort(port)


def _check_webui(port: int, timeout: float) -> WebUIStatus:
    url = f"http://127.0.0.1:{port}/"
    request = Request(url, headers={"User-Agent": "AstrBot CLI health check"})
    try:
        with urlopen(request, timeout=timeout) as response:
            status_code = response.getcode()
    except HTTPError as e:
        return WebUIStatus(
            url=url,
            accessible=False,
            status_code=e.code,
            detail=f"HTTP {e.code}",
        )
    except URLError as e:
        return WebUIStatus(url=url, accessible=False, detail=str(e.reason))
    except TimeoutError:
        return WebUIStatus(url=url, accessible=False, detail="request timed out")
    except OSError as e:
        return WebUIStatus(url=url, accessible=False, detail=str(e))

    return WebUIStatus(
        url=url,
        accessible=200 <= status_code < 400,
        status_code=status_code,
        detail=f"HTTP {status_code}",
    )


def _is_service_running(service_state: ServiceState) -> bool:
    return service_state.state.lower() in {"active", "running"}


def _health_label(service_state: ServiceState, webui_status: WebUIStatus) -> str:
    service_running = _is_service_running(service_state)
    if service_running and webui_status.accessible:
        return "healthy"
    if service_running or webui_status.accessible:
        return "degraded"
    return "unhealthy"


def _format_yes_no(value: bool) -> str:
    return "yes" if value else "no"


def _control_systemd_service(service_name: str, action: str) -> None:
    if shutil.which("systemctl") is None:
        raise click.ClickException("systemctl was not found")

    unit_path = _systemd_unit_path(service_name)
    if not unit_path.exists():
        raise click.ClickException(
            f"{unit_path} does not exist. Run 'service install' first",
        )

    _run_checked(
        ["systemctl", "--user", action, _systemd_unit_name(service_name)],
        f"Failed to {action} the systemd user service",
    )


def _launchd_target(service_name: str) -> str:
    return f"gui/{os.getuid()}/{_macos_label(service_name)}"


def _launchd_domain() -> str:
    return f"gui/{os.getuid()}"


def _is_launch_agent_loaded(service_name: str) -> bool:
    result = _run_capture(["launchctl", "print", _launchd_target(service_name)])
    return result is not None and result.returncode == 0


def _wait_for_launch_agent_state(
    service_name: str,
    *,
    loaded: bool,
    timeout: float = 5.0,
) -> bool:
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        if _is_launch_agent_loaded(service_name) is loaded:
            return True
        time.sleep(0.1)
    return _is_launch_agent_loaded(service_name) is loaded


def _bootstrap_launch_agent(service_name: str, plist_path: Path) -> None:
    _run_checked(
        ["launchctl", "bootstrap", _launchd_domain(), str(plist_path)],
        f"Failed to load the LaunchAgent from {plist_path}",
    )
    if not _wait_for_launch_agent_state(service_name, loaded=True):
        raise click.ClickException(
            "LaunchAgent was bootstrapped but did not appear in launchd. "
            f"Label: {_macos_label(service_name)}; plist: {plist_path}",
        )


def _enable_launch_agent(service_name: str) -> None:
    _run_checked(
        ["launchctl", "enable", _launchd_target(service_name)],
        f"Failed to enable the LaunchAgent {_macos_label(service_name)}",
    )


def _kickstart_launch_agent(service_name: str) -> None:
    target = _launchd_target(service_name)
    result = _run_capture(["launchctl", "kickstart", "-k", target])
    if result is None:
        raise click.ClickException("launchctl was not found")
    if result.returncode == 0:
        return

    detail = _first_output_line(result)
    message = f"Failed to start the LaunchAgent {target}"
    if detail:
        message = f"{message}: {detail}"
    raise click.ClickException(message)


def _start_launch_agent(service_name: str) -> None:
    if shutil.which("launchctl") is None:
        raise click.ClickException("launchctl was not found")

    plist_path = _launch_agent_path(service_name)
    if not plist_path.exists():
        raise click.ClickException(
            f"{plist_path} does not exist. Run 'service install' first",
        )

    if not _is_launch_agent_loaded(service_name):
        _bootstrap_launch_agent(service_name, plist_path)

    _enable_launch_agent(service_name)
    _kickstart_launch_agent(service_name)


def _stop_launch_agent(service_name: str, *, allow_missing: bool = False) -> None:
    if shutil.which("launchctl") is None:
        raise click.ClickException("launchctl was not found")

    result = _run_capture(["launchctl", "bootout", _launchd_target(service_name)])
    if result is None:
        raise click.ClickException("launchctl was not found")
    if result.returncode != 0 and not allow_missing:
        detail = _first_output_line(result)
        message = "Failed to stop the LaunchAgent"
        if detail:
            message = f"{message}: {detail}"
        raise click.ClickException(message)
    if result.returncode == 0:
        _wait_for_launch_agent_state(service_name, loaded=False)


def _control_windows_task(service_name: str, action: str) -> None:
    if shutil.which("schtasks") is None:
        raise click.ClickException("schtasks was not found")
    if not _windows_task_exists(service_name):
        raise click.ClickException(
            f"Scheduled task {service_name} does not exist. Run 'service install' first",
        )

    match action:
        case "start":
            command = ["schtasks", "/Run", "/TN", service_name]
        case "stop":
            command = ["schtasks", "/End", "/TN", service_name]
        case _:
            raise click.ClickException(f"Unsupported Windows task action: {action}")

    _run_checked(command, f"Failed to {action} the Windows scheduled task")


def _control_service(service_name: str, action: str) -> None:
    system = platform.system()
    if system == "Linux":
        _control_systemd_service(service_name, action)
        return

    if system == "Darwin":
        match action:
            case "start":
                _start_launch_agent(service_name)
            case "stop":
                _stop_launch_agent(service_name)
            case "restart":
                _stop_launch_agent(service_name, allow_missing=True)
                _start_launch_agent(service_name)
            case _:
                raise click.ClickException(f"Unsupported launchd action: {action}")
        return

    if system == "Windows":
        match action:
            case "start" | "stop":
                _control_windows_task(service_name, action)
            case "restart":
                _control_windows_task(service_name, "stop")
                _control_windows_task(service_name, "start")
            case _:
                raise click.ClickException(f"Unsupported Windows task action: {action}")
        return

    raise click.ClickException(f"Unsupported platform: {system}")


def _uninstall_systemd_service(service_name: str) -> Path:
    unit_path = _systemd_unit_path(service_name)
    if not unit_path.exists():
        raise click.ClickException(f"{unit_path} does not exist")

    if shutil.which("systemctl") is not None:
        subprocess.run(
            [
                "systemctl",
                "--user",
                "disable",
                "--now",
                _systemd_unit_name(service_name),
            ],
            check=False,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        )
    unit_path.unlink()
    if shutil.which("systemctl") is not None:
        _run_checked(
            ["systemctl", "--user", "daemon-reload"],
            "Failed to reload the systemd user daemon",
        )
    return unit_path


def _uninstall_launch_agent(service_name: str) -> Path:
    plist_path = _launch_agent_path(service_name)
    if not plist_path.exists():
        raise click.ClickException(f"{plist_path} does not exist")

    _stop_launch_agent(service_name, allow_missing=True)
    plist_path.unlink()
    return plist_path


def _uninstall_windows_task(service_name: str) -> str:
    if shutil.which("schtasks") is None:
        raise click.ClickException("schtasks was not found")
    if not _windows_task_exists(service_name):
        raise click.ClickException(f"Scheduled task {service_name} does not exist")

    _run_checked(
        ["schtasks", "/Delete", "/TN", service_name, "/F"],
        "Failed to delete the Windows scheduled task",
    )
    return service_name


def _uninstall_service(service_name: str) -> Path | str:
    system = platform.system()
    if system == "Linux":
        return _uninstall_systemd_service(service_name)
    if system == "Darwin":
        return _uninstall_launch_agent(service_name)
    if system == "Windows":
        return _uninstall_windows_task(service_name)
    raise click.ClickException(f"Unsupported platform: {system}")


def _resolve_data_path(astrbot_root: Path, configured_path: str | None) -> Path:
    if not configured_path:
        configured_path = "logs/astrbot.log"

    path = Path(configured_path).expanduser()
    if path.is_absolute():
        return path
    return astrbot_root / "data" / path


def _resolve_app_log_path(astrbot_root: Path) -> Path:
    config_path = astrbot_root / "data" / "cmd_config.json"
    if not config_path.exists():
        return _resolve_data_path(astrbot_root, None)

    try:
        config = json.loads(config_path.read_text(encoding="utf-8-sig"))
    except (OSError, json.JSONDecodeError):
        return _resolve_data_path(astrbot_root, None)

    if "log_file" in config:
        log_file_config = config.get("log_file") or {}
        return _resolve_data_path(astrbot_root, log_file_config.get("path"))

    return _resolve_data_path(astrbot_root, config.get("log_file_path"))


def _get_config_path(astrbot_root: Path) -> Path:
    return astrbot_root / "data" / "cmd_config.json"


def _load_or_init_config(astrbot_root: Path) -> dict:
    config_path = _get_config_path(astrbot_root)
    if not config_path.exists():
        from astrbot.core.config.default import DEFAULT_CONFIG

        return copy.deepcopy(DEFAULT_CONFIG)

    try:
        return json.loads(config_path.read_text(encoding="utf-8-sig"))
    except json.JSONDecodeError as e:
        raise click.ClickException(f"Failed to parse config file: {e}") from e


def _save_config(astrbot_root: Path, config: dict) -> None:
    config_path = _get_config_path(astrbot_root)
    config_path.parent.mkdir(parents=True, exist_ok=True)
    config_path.write_text(
        json.dumps(config, ensure_ascii=False, indent=2),
        encoding="utf-8-sig",
    )


def _get_app_log_config(astrbot_root: Path, config: dict) -> AppLogConfig:
    if isinstance(config.get("log_file"), dict):
        log_file_config = config["log_file"]
        configured_path = log_file_config.get("path")
        return AppLogConfig(
            enabled=bool(log_file_config.get("enable", False)),
            path=_resolve_data_path(astrbot_root, configured_path),
            configured_path=configured_path,
        )

    configured_path = config.get("log_file_path")
    return AppLogConfig(
        enabled=bool(config.get("log_file_enable", False)),
        path=_resolve_data_path(astrbot_root, configured_path),
        configured_path=configured_path,
    )


def _set_app_log_config(
    config: dict,
    *,
    enabled: bool,
    path: str | None = None,
) -> None:
    if isinstance(config.get("log_file"), dict):
        config["log_file"]["enable"] = enabled
        if path is not None:
            config["log_file"]["path"] = path
        return

    config["log_file_enable"] = enabled
    if path is not None:
        config["log_file_path"] = path


def _read_last_lines(path: Path, lines: int) -> list[str]:
    with path.open("r", encoding="utf-8", errors="replace") as f:
        return list(deque(f, maxlen=lines))


def _echo_log_line(line: str) -> None:
    click.echo(line.rstrip("\r\n"))


def _show_log_files(paths: list[Path], lines: int, follow: bool) -> None:
    existing_paths = [path for path in paths if path.exists()]
    if not existing_paths:
        joined_paths = ", ".join(str(path) for path in paths)
        raise click.ClickException(f"No log files found: {joined_paths}")

    show_headers = len(existing_paths) > 1
    for index, path in enumerate(existing_paths):
        if show_headers:
            if index:
                click.echo()
            click.echo(f"==> {path} <==")
        for line in _read_last_lines(path, lines):
            _echo_log_line(line)

    if follow:
        _follow_log_files(existing_paths)


def _follow_log_files(paths: list[Path]) -> None:
    positions = {path: path.stat().st_size for path in paths if path.exists()}
    click.echo("Following logs. Press Ctrl+C to stop.")
    try:
        while True:
            for path in paths:
                if not path.exists():
                    continue

                current_size = path.stat().st_size
                previous_position = positions.get(path, 0)
                if current_size < previous_position:
                    previous_position = 0

                with path.open("r", encoding="utf-8", errors="replace") as f:
                    f.seek(previous_position)
                    for line in f:
                        if len(paths) > 1:
                            click.echo(f"[{path.name}] ", nl=False)
                        _echo_log_line(line)
                    positions[path] = f.tell()

            time.sleep(1)
    except KeyboardInterrupt:
        return


def _run_passthrough(command: list[str], failure_message: str) -> None:
    try:
        result = subprocess.run(command, check=False)
    except FileNotFoundError as e:
        raise click.ClickException(f"Command not found: {command[0]}") from e
    except KeyboardInterrupt:
        return

    if result.returncode != 0:
        raise click.ClickException(f"{failure_message}: exit code {result.returncode}")


def _show_journal_logs(service_name: str, lines: int, follow: bool) -> None:
    command = [
        "journalctl",
        "--user",
        "-u",
        _systemd_unit_name(service_name),
        "-n",
        str(lines),
        "--no-pager",
    ]
    if follow:
        command.append("-f")
    _run_passthrough(command, "Failed to read systemd user service logs")


def _show_service_logs(
    service_name: str,
    lines: int,
    follow: bool,
    *,
    include_stderr: bool,
) -> None:
    system = platform.system()
    if system == "Linux":
        _show_journal_logs(service_name, lines, follow)
        return

    if system in {"Darwin", "Windows"}:
        out_log, err_log = _service_log_paths(service_name)
        paths = [out_log]
        if include_stderr:
            paths.append(err_log)
        _show_log_files(paths, lines, follow)
        return

    raise click.ClickException(f"Unsupported platform: {system}")


@service.command(name="install")
@click.option(
    "--name",
    default=DEFAULT_SERVICE_NAME,
    show_default=True,
    help="Service name to install.",
)
@click.option(
    "--workdir",
    type=click.Path(file_okay=False, dir_okay=True, path_type=Path),
    help="AstrBot root directory. Defaults to the current directory.",
)
@click.option(
    "--executable",
    type=str,
    help="Path to the astrbot executable. Defaults to the executable found on PATH.",
)
@click.option("--force", is_flag=True, help="Overwrite an existing service definition.")
@click.option(
    "--now",
    is_flag=True,
    help="Start or restart the service after installing it.",
)
def install(
    name: str,
    workdir: Path | None,
    executable: str | None,
    force: bool,
    now: bool,
) -> None:
    """Install AstrBot as a user-level background service."""
    service_name = _validate_service_name(name)
    astrbot_root = _resolve_workdir(workdir)
    astrbot_executable = _resolve_astrbot_executable(executable)
    system = platform.system()

    if system == "Linux":
        service_path = _install_systemd_user_service(
            service_name,
            astrbot_executable,
            astrbot_root,
            force=force,
            now=now,
        )
        click.echo(f"Installed systemd user service: {service_path}")
        click.echo(f"Manage it with: systemctl --user status {service_path.name}")
        click.echo(
            "To start it at boot before login, enable lingering with: "
            f"loginctl enable-linger {getpass.getuser()}",
        )
        return

    if system == "Darwin":
        plist_path = _install_launch_agent(
            service_name,
            astrbot_executable,
            astrbot_root,
            force=force,
            now=now,
        )
        click.echo(f"Installed LaunchAgent: {plist_path}")
        click.echo(f"LaunchAgent label: {_macos_label(service_name)}")
        return

    if system == "Windows":
        _install_windows_task(
            service_name,
            astrbot_executable,
            astrbot_root,
            force=force,
            now=now,
        )
        click.echo(f"Installed Windows scheduled task: {service_name}")
        click.echo(f"Manage it with: schtasks /Query /TN {service_name}")
        return

    raise click.ClickException(f"Unsupported platform: {system}")


@service.command(name="start")
@click.option(
    "--name",
    default=DEFAULT_SERVICE_NAME,
    show_default=True,
    help="Service name to start.",
)
def start(name: str) -> None:
    """Start the installed background service."""
    service_name = _validate_service_name(name)
    click.echo("Starting service...")
    _control_service(service_name, "start")
    click.echo(f"Started service: {service_name}")


@service.command(name="stop")
@click.option(
    "--name",
    default=DEFAULT_SERVICE_NAME,
    show_default=True,
    help="Service name to stop.",
)
def stop(name: str) -> None:
    """Stop the installed background service."""
    service_name = _validate_service_name(name)
    click.echo("Stopping service...")
    _control_service(service_name, "stop")
    click.echo(f"Stopped service: {service_name}")


@service.command(name="restart")
@click.option(
    "--name",
    default=DEFAULT_SERVICE_NAME,
    show_default=True,
    help="Service name to restart.",
)
def restart(name: str) -> None:
    """Restart the installed background service."""
    service_name = _validate_service_name(name)
    click.echo("Restarting service...")
    _control_service(service_name, "restart")
    click.echo(f"Restarted service: {service_name}")


@service.command(name="uninstall")
@click.option(
    "--name",
    default=DEFAULT_SERVICE_NAME,
    show_default=True,
    help="Service name to uninstall.",
)
@click.option("--force", is_flag=True, help="Do not ask for confirmation.")
def uninstall(name: str, force: bool) -> None:
    """Remove the installed background service."""
    service_name = _validate_service_name(name)
    click.echo("Uninstalling service...")

    if not force:
        click.confirm(
            f"Uninstall AstrBot service {service_name}?",
            default=False,
            abort=True,
        )

    removed = _uninstall_service(service_name)
    click.echo(f"Uninstalled service: {removed}")


@service.group(name="logs", invoke_without_command=True)
@click.pass_context
@click.option(
    "--name",
    default=DEFAULT_SERVICE_NAME,
    show_default=True,
    help="Service name to read logs for.",
)
@click.option(
    "--workdir",
    type=click.Path(file_okay=False, dir_okay=True, path_type=Path),
    help="AstrBot root directory. Required only with --source app.",
)
@click.option(
    "--source",
    type=click.Choice(["service", "app"]),
    default="service",
    show_default=True,
    help="Read service manager output or AstrBot application log file.",
)
@click.option(
    "--lines",
    "-n",
    default=DEFAULT_LOG_LINES,
    show_default=True,
    type=int,
    help="Number of lines to show.",
)
@click.option("--follow", "-f", is_flag=True, help="Follow log output.")
@click.option(
    "--include-stderr",
    is_flag=True,
    help="Also show stderr logs on macOS and Windows.",
)
def logs(
    ctx: click.Context,
    name: str,
    workdir: Path | None,
    source: str,
    lines: int,
    follow: bool,
    include_stderr: bool,
) -> None:
    """View service logs or configure the application log file."""
    if ctx.invoked_subcommand is not None:
        return

    if lines <= 0:
        raise click.ClickException("Lines must be greater than 0")

    service_name = _validate_service_name(name)
    if source == "app":
        astrbot_root = _resolve_workdir(workdir)
        _show_log_files([_resolve_app_log_path(astrbot_root)], lines, follow)
        return

    _show_service_logs(service_name, lines, follow, include_stderr=include_stderr)


@logs.command(name="status")
@click.option(
    "--workdir",
    type=click.Path(file_okay=False, dir_okay=True, path_type=Path),
    help="AstrBot root directory. Defaults to the current directory.",
)
def logs_status(workdir: Path | None) -> None:
    """Show application log file configuration."""
    astrbot_root = _resolve_workdir(workdir)
    config = _load_or_init_config(astrbot_root)
    log_config = _get_app_log_config(astrbot_root, config)

    click.echo("AstrBot application log file")
    click.echo(f"  Enabled: {_format_yes_no(log_config.enabled)}")
    click.echo(f"  Configured path: {log_config.configured_path or 'logs/astrbot.log'}")
    click.echo(f"  Resolved path: {log_config.path}")
    click.echo(f"  Exists: {_format_yes_no(log_config.path.exists())}")


@logs.command(name="enable")
@click.option(
    "--workdir",
    type=click.Path(file_okay=False, dir_okay=True, path_type=Path),
    help="AstrBot root directory. Defaults to the current directory.",
)
@click.option(
    "--path",
    "log_path",
    help="Log file path. Relative paths are resolved from the AstrBot data directory.",
)
def logs_enable(workdir: Path | None, log_path: str | None) -> None:
    """Enable the AstrBot application log file."""
    astrbot_root = _resolve_workdir(workdir)
    config = _load_or_init_config(astrbot_root)
    _set_app_log_config(config, enabled=True, path=log_path)
    _save_config(astrbot_root, config)

    log_config = _get_app_log_config(astrbot_root, config)
    click.echo("Enabled AstrBot application log file.")
    click.echo(f"Log path: {log_config.path}")
    click.echo("Restart AstrBot for this change to take effect.")


@logs.command(name="disable")
@click.option(
    "--workdir",
    type=click.Path(file_okay=False, dir_okay=True, path_type=Path),
    help="AstrBot root directory. Defaults to the current directory.",
)
def logs_disable(workdir: Path | None) -> None:
    """Disable the AstrBot application log file."""
    astrbot_root = _resolve_workdir(workdir)
    config = _load_or_init_config(astrbot_root)
    _set_app_log_config(config, enabled=False)
    _save_config(astrbot_root, config)

    click.echo("Disabled AstrBot application log file.")
    click.echo("Restart AstrBot for this change to take effect.")


@service.command(name="status")
@click.option(
    "--name",
    default=DEFAULT_SERVICE_NAME,
    show_default=True,
    help="Service name to inspect.",
)
@click.option(
    "--workdir",
    type=click.Path(file_okay=False, dir_okay=True, path_type=Path),
    help="AstrBot root directory. Defaults to the current directory.",
)
@click.option(
    "--timeout",
    default=DEFAULT_STATUS_TIMEOUT_SECONDS,
    show_default=True,
    type=float,
    help="WebUI probe timeout in seconds.",
)
def status(name: str, workdir: Path | None, timeout: float) -> None:
    """Check background service state, WebUI health, and port."""
    if timeout <= 0:
        raise click.ClickException("Timeout must be greater than 0")

    service_name = _validate_service_name(name)
    astrbot_root = _resolve_workdir(workdir)
    service_state = _get_service_state(service_name)
    dashboard_port = _load_dashboard_port(astrbot_root)
    webui_status = _check_webui(dashboard_port.port, timeout)
    health = _health_label(service_state, webui_status)

    click.echo("AstrBot service status")
    click.echo(f"  Health: {health}")
    click.echo(f"  Platform: {platform.system()}")
    click.echo(f"  Service name: {service_name}")
    click.echo(f"  Service manager: {service_state.manager}")
    click.echo(f"  Installed: {_format_yes_no(service_state.installed)}")
    if service_state.path is not None:
        click.echo(f"  Definition: {service_state.path}")
    click.echo(f"  Service state: {service_state.state}")
    if service_state.enabled is not None:
        click.echo(f"  Enabled: {service_state.enabled}")
    if service_state.detail:
        click.echo(f"  Service detail: {service_state.detail}")
    click.echo(f"  AstrBot root: {astrbot_root}")
    click.echo(f"  Dashboard port: {dashboard_port.port}")
    if dashboard_port.detail:
        click.echo(f"  Port detail: {dashboard_port.detail}")
    click.echo(f"  WebUI URL: {webui_status.url}")
    click.echo(f"  WebUI accessible: {_format_yes_no(webui_status.accessible)}")
    if webui_status.status_code is not None:
        click.echo(f"  WebUI HTTP status: {webui_status.status_code}")
    if webui_status.detail:
        click.echo(f"  WebUI detail: {webui_status.detail}")