Nexus/server/application/services/btpanel_service.py

"""Baota panel orchestration — proxy to remote panel API or SSH."""

from __future__ import annotations

import json
import logging
from typing import Any

from sqlalchemy.ext.asyncio import AsyncSession

from server.domain.models import AuditLog, Server
from server.infrastructure.btpanel.client import BtPanelApiError, BtPanelClient
from server.infrastructure.btpanel.credentials import (
    BtPanelCredentials,
    get_bt_panel_base_url,
    merge_bt_panel_extra,
    public_bt_panel_status,
    read_bt_panel_credentials,
    bt_panel_configured,
)
from server.infrastructure.btpanel.bootstrap_state import (
    STATE_DISABLED,
    STATE_INSTALLING,
    STATE_PENDING,
    apply_bootstrap_failure,
    apply_bootstrap_success,
    merge_bootstrap_fields,
    _iso,
    _utcnow,
)
from server.infrastructure.btpanel.bootstrap_lock import bootstrap_lock
from server.infrastructure.btpanel.source_ip import get_bt_panel_source_ip
from server.infrastructure.btpanel.ssh_bootstrap import BootstrapRemoteError, ssh_bootstrap_panel
from server.infrastructure.btpanel.ssh_login import (
    detect_bt_panel_installed,
    normalize_temp_login_url,
    ssh_temp_login_url,
)
from server.infrastructure.database.audit_log_repo import AuditLogRepositoryImpl
from server.infrastructure.database.server_repo import ServerRepositoryImpl

logger = logging.getLogger("nexus.btpanel.service")


class BtPanelService:
    def __init__(self, db: AsyncSession):
        self.db = db
        self.servers = ServerRepositoryImpl(db)
        self.audit = AuditLogRepositoryImpl(db)

    async def get_server(self, server_id: int) -> Server:
        server = await self.servers.get_by_id(server_id)
        if not server:
            raise ValueError("服务器不存在")
        return server

    def _client(self, server: Server) -> BtPanelClient:
        creds = read_bt_panel_credentials(server)
        if not creds:
            raise ValueError("未配置宝塔面板地址或 API 密钥")
        return BtPanelClient(creds, server.id)

    async def list_server_statuses(self) -> list[dict[str, Any]]:
        servers = await self.servers.get_all()
        out: list[dict[str, Any]] = []
        for s in servers:
            out.append({
                "id": s.id,
                "name": s.name,
                "domain": s.domain,
                "category": s.category,
                **public_bt_panel_status(s),
            })
        return out

    async def get_config(self, server_id: int, *, refresh_bt_installed: bool = False) -> dict[str, Any]:
        server = await self.get_server(server_id)
        bt_installed = await self._resolve_bt_installed(server, refresh=refresh_bt_installed)
        return {**public_bt_panel_status(server), "bt_installed": bt_installed}

    async def _resolve_bt_installed(self, server: Server, *, refresh: bool = False) -> bool | None:
        from server.infrastructure.btpanel.credentials import get_bt_panel_block

        block = get_bt_panel_block(server)
        cached = block.get("bt_installed")
        state = block.get("bootstrap_state")

        if not refresh:
            if bt_panel_configured(server):
                return True if cached is None else bool(cached)
            if cached is not None and state not in (STATE_PENDING, STATE_INSTALLING):
                return bool(cached)
            if state not in (STATE_PENDING, STATE_INSTALLING):
                return bool(cached) if cached is not None else None

        detected = await detect_bt_panel_installed(server)
        server.extra_attrs = merge_bootstrap_fields(
            server.extra_attrs if isinstance(server.extra_attrs, dict) else {},
            bt_installed=detected,
            bt_installed_checked_at=_iso(_utcnow()),
        )
        await self.servers.update(server)
        return detected

    async def get_global_settings(self) -> dict[str, Any]:
        from server.config import settings

        return {
            "bt_panel_source_ip": (getattr(settings, "BT_PANEL_SOURCE_IP", None) or "").strip(),
            "bt_panel_auto_bootstrap_enabled": self._auto_bootstrap_enabled(),
            "resolved_center_ip": get_bt_panel_source_ip(),
        }

    async def update_global_settings(
        self,
        *,
        bt_panel_source_ip: str | None,
        bt_panel_auto_bootstrap_enabled: bool | None,
        admin_username: str,
        ip_address: str | None,
    ) -> dict[str, Any]:
        from server.infrastructure.btpanel.source_ip import _is_ip as _is_valid_source_ip

        if bt_panel_source_ip is not None:
            ip = bt_panel_source_ip.strip()
            if ip and not _is_valid_source_ip(ip):
                raise ValueError("bt_panel_source_ip 必须是合法 IPv4/IPv6 地址")
            await self._persist_setting("bt_panel_source_ip", ip)
        if bt_panel_auto_bootstrap_enabled is not None:
            await self._persist_setting(
                "bt_panel_auto_bootstrap_enabled",
                "true" if bt_panel_auto_bootstrap_enabled else "false",
            )
        await self.audit.create(AuditLog(
            admin_username=admin_username,
            action="bt_panel_global_settings",
            target_type="setting",
            target_id=0,
            detail=json.dumps({
                "bt_panel_source_ip": bt_panel_source_ip,
                "bt_panel_auto_bootstrap_enabled": bt_panel_auto_bootstrap_enabled,
            }, ensure_ascii=False),
            ip_address=ip_address,
        ))
        return await self.get_global_settings()

    async def _persist_setting(self, key: str, value: str) -> None:
        from server.config import settings
        from server.infrastructure.database.setting_repo import SettingRepositoryImpl
        from server.infrastructure.settings_broadcast import publish_setting_change

        repo = SettingRepositoryImpl(self.db)
        await repo.set(key, value)
        if settings.apply_db_override(key, value):
            await publish_setting_change(key, value)

    @staticmethod
    def _auto_bootstrap_enabled() -> bool:
        from server.application.services.btpanel_bootstrap_schedule import is_auto_bootstrap_enabled

        return is_auto_bootstrap_enabled()

    async def update_config(
        self,
        server_id: int,
        *,
        base_url: str | None,
        api_key: str | None,
        verify_ssl: bool | None,
        auto_bootstrap: bool | None,
        admin_username: str,
        ip_address: str | None,
    ) -> dict[str, Any]:
        server = await self.get_server(server_id)
        extra = merge_bt_panel_extra(
            server.extra_attrs if isinstance(server.extra_attrs, dict) else {},
            base_url=base_url,
            api_key=api_key,
            verify_ssl=verify_ssl,
        )
        if auto_bootstrap is not None:
            if auto_bootstrap:
                from datetime import datetime, timezone
                from server.infrastructure.btpanel.bootstrap_state import _iso

                now = _iso(datetime.now(timezone.utc))
                extra = merge_bootstrap_fields(
                    extra,
                    auto_bootstrap=True,
                    bootstrap_state=STATE_PENDING,
                    bootstrap_next_attempt_at=now,
                    bootstrap_last_error=None,
                )
            else:
                extra = merge_bootstrap_fields(
                    extra,
                    auto_bootstrap=False,
                    bootstrap_state=STATE_DISABLED,
                    bootstrap_next_attempt_at=None,
                )
        server.extra_attrs = extra
        await self.servers.update(server)
        await self.audit.create(AuditLog(
            admin_username=admin_username,
            action="bt_panel_config_update",
            target_type="server",
            target_id=server_id,
            detail=json.dumps({"base_url": base_url, "auto_bootstrap": auto_bootstrap}, ensure_ascii=False),
            ip_address=ip_address,
        ))
        return await self.get_config(server_id)

    async def bootstrap_server(
        self,
        server_id: int,
        *,
        source: str,
        admin_username: str,
        ip_address: str | None,
        force: bool = False,
    ) -> dict[str, Any]:
        async with bootstrap_lock(server_id):
            return await self._bootstrap_server_locked(
                server_id,
                source=source,
                admin_username=admin_username,
                ip_address=ip_address,
                force=force,
            )

    async def _bootstrap_server_locked(
        self,
        server_id: int,
        *,
        source: str,
        admin_username: str,
        ip_address: str | None,
        force: bool,
    ) -> dict[str, Any]:
        from datetime import datetime, timezone

        from server.infrastructure.btpanel.bootstrap_state import is_eligible_for_background_bootstrap

        server = await self.get_server(server_id)

        if bt_panel_configured(server) and source == "background":
            return {**public_bt_panel_status(server), "skipped": True, "reason": "already_configured"}

        if source == "background" and not is_eligible_for_background_bootstrap(server):
            return {**public_bt_panel_status(server), "skipped": True, "reason": "not_due"}

        if force or source in ("manual", "batch", "immediate"):
            now = _iso(datetime.now(timezone.utc))
            from server.infrastructure.btpanel.credentials import get_bt_panel_block

            block = get_bt_panel_block(server)
            fields: dict[str, Any] = {
                "bootstrap_state": STATE_PENDING,
                "bootstrap_next_attempt_at": now,
                "bootstrap_last_error": None,
            }
            # 手动「立即获取」不覆盖用户已关闭的自动获取
            if source != "manual" or block.get("auto_bootstrap") is not False:
                fields["auto_bootstrap"] = True
            server.extra_attrs = merge_bootstrap_fields(
                server.extra_attrs if isinstance(server.extra_attrs, dict) else {},
                **fields,
            )
            await self.servers.update(server)

        center_ip = get_bt_panel_source_ip()
        if not center_ip:
            server.extra_attrs = apply_bootstrap_failure(
                server.extra_attrs if isinstance(server.extra_attrs, dict) else {},
                error_code="missing_center_ip",
                permanent=False,
            )
            await self.servers.update(server)
            await self._audit_bootstrap(server_id, admin_username, ip_address, source, ok=False, error="missing_center_ip")
            return {**public_bt_panel_status(server), "ok": False, "error": "missing_center_ip"}

        try:
            data = await ssh_bootstrap_panel(server, center_ip=center_ip)
        except BootstrapRemoteError as exc:
            permanent = exc.code == "ssh_auth_failed"
            server.extra_attrs = apply_bootstrap_failure(
                server.extra_attrs if isinstance(server.extra_attrs, dict) else {},
                error_code=exc.code,
                permanent=permanent,
            )
            await self.servers.update(server)
            await self._audit_bootstrap(
                server_id, admin_username, ip_address, source, ok=False, error=exc.code,
            )
            return {**public_bt_panel_status(server), "ok": False, "error": exc.code}

        base_url = str(data.get("base_url") or "").strip()
        api_key = str(data.get("api_key") or "").strip()
        if not base_url or not api_key:
            server.extra_attrs = apply_bootstrap_failure(
                server.extra_attrs if isinstance(server.extra_attrs, dict) else {},
                error_code="bootstrap_empty_credentials",
                permanent=False,
            )
            await self.servers.update(server)
            await self._audit_bootstrap(
                server_id, admin_username, ip_address, source, ok=False, error="bootstrap_empty_credentials",
            )
            return {**public_bt_panel_status(server), "ok": False, "error": "bootstrap_empty_credentials"}

        server.extra_attrs = apply_bootstrap_success(
            server.extra_attrs if isinstance(server.extra_attrs, dict) else {},
            base_url=base_url,
            api_key=api_key,
            verify_ssl=bool(data.get("verify_ssl")),
            actions=list(data.get("actions") or []),
        )
        await self.servers.update(server)
        await self._audit_bootstrap(
            server_id,
            admin_username,
            ip_address,
            source,
            ok=True,
            actions=list(data.get("actions") or []),
        )
        return {**public_bt_panel_status(server), "ok": True}

    async def batch_bootstrap(
        self,
        server_ids: list[int],
        *,
        all_unconfigured: bool = False,
        admin_username: str,
        ip_address: str | None,
    ) -> dict[str, Any]:
        from server.application.services.server_batch_service import start_batch_job

        if all_unconfigured:
            statuses = await self.list_server_statuses()
            ids = [int(s["id"]) for s in statuses if not s.get("configured")]
        else:
            ids = [int(x) for x in server_ids if x]
        if not ids:
            raise ValueError("没有可批量初始化的服务器（均已配置或未录入）")
        return await start_batch_job(
            op="bt-panel-bootstrap",
            server_ids=ids,
            operator=admin_username,
        )

    async def _audit_bootstrap(
        self,
        server_id: int,
        admin_username: str,
        ip_address: str | None,
        source: str,
        *,
        ok: bool,
        error: str | None = None,
        actions: list[str] | None = None,
    ) -> None:
        await self.audit.create(AuditLog(
            admin_username=admin_username,
            action="bt_panel_ssh_bootstrap",
            target_type="server",
            target_id=server_id,
            detail=json.dumps({
                "source": source,
                "ok": ok,
                "error": error,
                "actions": actions or [],
            }, ensure_ascii=False)[:500],
            ip_address=ip_address,
        ))

    async def create_login_url(
        self,
        server_id: int,
        *,
        admin_username: str,
        ip_address: str | None,
    ) -> dict[str, str]:
        server = await self.get_server(server_id)
        url: str | None = None
        method = "api"

        creds = read_bt_panel_credentials(server)
        if creds:
            try:
                url = await self._login_url_via_api(
                    creds,
                    server.id,
                    prefer_host=server.domain or "",
                )
            except BtPanelApiError as exc:
                logger.warning("bt panel api temp login failed server=%s: %s", server_id, exc)
                method = "ssh_fallback"

        if not url:
            method = "ssh"
            panel_base = (creds.base_url if creds else None) or get_bt_panel_base_url(server)
            url = await ssh_temp_login_url(server, base_url=panel_base)

        await self.audit.create(AuditLog(
            admin_username=admin_username,
            action="bt_panel_login_url",
            target_type="server",
            target_id=server_id,
            detail=json.dumps({"method": method}, ensure_ascii=False),
            ip_address=ip_address,
        ))
        return {"url": url, "method": method}

    async def _login_url_via_api(
        self,
        creds: BtPanelCredentials,
        server_id: int,
        *,
        prefer_host: str = "",
    ) -> str:
        client = BtPanelClient(creds, server_id)
        data = await client.post("/config?action=set_temp_login", {})
        if isinstance(data, dict):
            for key in ("url", "login_url", "msg"):
                val = data.get(key)
                if isinstance(val, str) and "tmp_token=" in val:
                    return normalize_temp_login_url(
                        val.strip(),
                        base_url=creds.base_url,
                        prefer_host=prefer_host,
                    )
            if data.get("status") and isinstance(data.get("data"), dict):
                inner = data["data"]
                for key in ("url", "login_url"):
                    val = inner.get(key)
                    if isinstance(val, str) and "tmp_token=" in val:
                        return normalize_temp_login_url(
                            val.strip(),
                            base_url=creds.base_url,
                            prefer_host=prefer_host,
                        )
        raise BtPanelApiError("set_temp_login 未返回登录链接")

    async def proxy(
        self,
        server_id: int,
        path: str,
        data: dict[str, Any] | None = None,
    ) -> Any:
        server = await self.get_server(server_id)
        client = self._client(server)
        return await client.post(path, data)

    # ── B1–B3 ──
    async def system_total(self, server_id: int) -> Any:
        return await self.proxy(server_id, "/system?action=GetSystemTotal")

    async def system_disk(self, server_id: int) -> Any:
        return await self.proxy(server_id, "/system?action=GetDiskInfo")

    async def system_network(self, server_id: int) -> Any:
        return await self.proxy(server_id, "/system?action=GetNetWork")

    # ── C1,C6 ──
    async def list_sites(self, server_id: int, *, limit: int = 200) -> Any:
        return await self.proxy(server_id, "/data?action=getData", {
            "table": "sites",
            "limit": limit,
            "p": 1,
            "order": "id desc",
        })

    async def site_start(self, server_id: int, site_id: int, name: str) -> Any:
        return await self.proxy(server_id, "/site?action=SiteStart", {"id": site_id, "name": name})

    async def site_stop(self, server_id: int, site_id: int, name: str) -> Any:
        return await self.proxy(server_id, "/site?action=SiteStop", {"id": site_id, "name": name})

    # ── C4 ──
    async def php_versions(self, server_id: int) -> Any:
        return await self.proxy(server_id, "/site?action=GetPHPVersion")

    async def create_site(self, server_id: int, payload: dict[str, Any], *, admin_username: str, ip_address: str | None) -> Any:
        result = await self.proxy(server_id, "/site?action=AddSite", payload)
        await self.audit.create(AuditLog(
            admin_username=admin_username,
            action="bt_panel_site_create",
            target_type="server",
            target_id=server_id,
            detail=json.dumps({"webname": payload.get("webname")}, ensure_ascii=False)[:500],
            ip_address=ip_address,
        ))
        return result

    # ── C10 ──
    async def list_domains(self, server_id: int, site_id: int) -> Any:
        return await self.proxy(server_id, "/data?action=getData", {
            "table": "domain",
            "search": str(site_id),
            "list": "true",
        })

    async def add_domain(self, server_id: int, payload: dict[str, Any]) -> Any:
        return await self.proxy(server_id, "/site?action=AddDomain", payload)

    async def delete_domain(self, server_id: int, payload: dict[str, Any]) -> Any:
        return await self.proxy(server_id, "/site?action=DelDomain", payload)

    # ── C15 ──
    async def list_ssl_sites(self, server_id: int) -> Any:
        return await self.list_sites(server_id, limit=500)

    async def set_ssl(self, server_id: int, payload: dict[str, Any], *, admin_username: str, ip_address: str | None) -> Any:
        result = await self.proxy(server_id, "/site?action=SetSSL", payload)
        await self.audit.create(AuditLog(
            admin_username=admin_username,
            action="bt_panel_ssl_set",
            target_type="server",
            target_id=server_id,
            detail=json.dumps({"siteName": payload.get("siteName")}, ensure_ascii=False)[:300],
            ip_address=ip_address,
        ))
        return result

    async def apply_letsencrypt(self, server_id: int, payload: dict[str, Any]) -> Any:
        return await self.proxy(server_id, "/acme?action=apply_cert", payload)

    # ── D1–D4 ──
    async def list_databases(self, server_id: int, *, limit: int = 200) -> Any:
        return await self.proxy(server_id, "/data?action=getData", {
            "table": "databases",
            "limit": limit,
            "p": 1,
        })

    async def create_database(self, server_id: int, payload: dict[str, Any], *, admin_username: str, ip_address: str | None) -> Any:
        result = await self.proxy(server_id, "/database?action=AddDatabase", payload)
        await self.audit.create(AuditLog(
            admin_username=admin_username,
            action="bt_panel_db_create",
            target_type="server",
            target_id=server_id,
            detail=json.dumps({"name": payload.get("name")}, ensure_ascii=False)[:200],
            ip_address=ip_address,
        ))
        return result

    async def reset_database_password(self, server_id: int, payload: dict[str, Any]) -> Any:
        return await self.proxy(server_id, "/database?action=ResDatabasePassword", payload)

    async def backup_database(self, server_id: int, db_id: int) -> Any:
        return await self.proxy(server_id, "/database?action=ToBackup", {"id": db_id})

    # ── F1 ──
    async def list_crontab(self, server_id: int) -> Any:
        return await self.proxy(server_id, "/crontab?action=GetCrontab")

    # ── H1 ──
    async def service_admin(self, server_id: int, name: str, op: str, *, admin_username: str, ip_address: str | None) -> Any:
        result = await self.proxy(server_id, "/system?action=ServiceAdmin", {"name": name, "type": op})
        await self.audit.create(AuditLog(
            admin_username=admin_username,
            action="bt_panel_service_admin",
            target_type="server",
            target_id=server_id,
            detail=json.dumps({"name": name, "type": op}, ensure_ascii=False),
            ip_address=ip_address,
        ))
        return result