ligbox-ops-platform/projects/ops-desk/api/app/vm123/openpanel_client.py

"""OpenPanel Community bridge client."""

from __future__ import annotations

import os
from typing import Any

import httpx

BRIDGE_URL = os.getenv("OPENPANEL_BRIDGE_URL", "http://10.10.10.123:18087").rstrip("/")
BRIDGE_TOKEN = os.getenv("OPENPANEL_BRIDGE_TOKEN", "")
OPENADMIN_URL = os.getenv("OPENADMIN_URL", "https://admin.openpanel.ligbox.com.br")
OPENPANEL_URL = os.getenv("OPENPANEL_URL", "https://openpanel.ligbox.com.br")
DEFAULT_PLAN = os.getenv("OPENPANEL_DEFAULT_PLAN", "ligbox-site-cms")


class OpenPanelBridgeError(Exception):
    pass


def bridge_configured() -> bool:
    return bool(BRIDGE_TOKEN)


def _headers() -> dict[str, str]:
    if not bridge_configured():
        raise OpenPanelBridgeError("OPENPANEL_BRIDGE_TOKEN ausente")
    return {"Authorization": f"Bearer {BRIDGE_TOKEN}"}


def autologin_payload(username: str) -> dict[str, Any]:
    """MVP: devolve URL OpenAdmin + instrução CONNECT (Enterprise futuro)."""
    return {
        "username": username,
        "openadmin_url": OPENADMIN_URL,
        "openpanel_url": OPENPANEL_URL,
        "note": "CONNECT autologin requer OpenPanel Enterprise API — use OpenAdmin manualmente",
        "bridge_configured": bridge_configured(),
    }


def health() -> dict[str, Any]:
    if not bridge_configured():
        return {"ok": False, "reason": "OPENPANEL_BRIDGE_TOKEN ausente", "bridge_url": BRIDGE_URL}
    try:
        with httpx.Client(timeout=10.0) as client:
            res = client.get(f"{BRIDGE_URL}/api", headers=_headers())
        body: dict[str, Any] = {}
        try:
            body = res.json()
        except Exception:
            pass
        return {
            "ok": res.status_code < 400,
            "status": res.status_code,
            "bridge_url": BRIDGE_URL,
            "bridge": body.get("bridge"),
        }
    except Exception as exc:
        return {"ok": False, "reason": str(exc), "bridge_url": BRIDGE_URL}


def list_users() -> dict[str, Any]:
    with httpx.Client(timeout=30.0) as client:
        res = client.get(f"{BRIDGE_URL}/api/users", headers=_headers())
    if res.status_code >= 400:
        raise OpenPanelBridgeError(f"bridge list_users HTTP {res.status_code}: {res.text[:300]}")
    return res.json()


def get_user(username: str) -> dict[str, Any]:
    with httpx.Client(timeout=30.0) as client:
        res = client.get(f"{BRIDGE_URL}/api/users/{username}", headers=_headers())
    if res.status_code >= 400:
        raise OpenPanelBridgeError(f"bridge get_user HTTP {res.status_code}: {res.text[:300]}")
    return res.json()


def provision_user(
    *,
    username: str,
    password: str,
    email: str,
    domain: str,
    plan_name: str | None = None,
    metadata: dict[str, Any] | None = None,
) -> dict[str, Any]:
    payload: dict[str, Any] = {
        "username": username.strip().lower(),
        "password": password,
        "email": email,
        "domain": domain.strip().lower(),
        "plan_name": plan_name or DEFAULT_PLAN,
    }
    if metadata:
        payload["metadata"] = metadata
    with httpx.Client(timeout=180.0) as client:
        res = client.post(f"{BRIDGE_URL}/api/users", headers=_headers(), json=payload)
    data = res.json() if res.content else {}
    if res.status_code >= 400 or not data.get("success", True):
        raise OpenPanelBridgeError(data.get("error") or f"bridge provision HTTP {res.status_code}")
    return data


def provision_user_safe(
    *,
    username: str,
    password: str,
    email: str,
    domain: str,
    plan_name: str | None = None,
    metadata: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Idempotente — reutiliza user existente se bridge reportar duplicado."""
    try:
        out = provision_user(
            username=username,
            password=password,
            email=email,
            domain=domain,
            plan_name=plan_name,
            metadata=metadata,
        )
        out["created"] = True
        return out
    except OpenPanelBridgeError as exc:
        msg = str(exc).lower()
        if "exist" in msg or "already" in msg:
            try:
                existing = get_user(username)
                return {
                    "success": True,
                    "created": False,
                    "reused": True,
                    "username": username,
                    "domain": domain,
                    "response": existing,
                }
            except OpenPanelBridgeError:
                raise exc
        raise


def add_domain(*, username: str, domain: str) -> dict[str, Any]:
    payload = {"username": username.strip().lower(), "domain": domain.strip().lower()}
    with httpx.Client(timeout=180.0) as client:
        res = client.post(f"{BRIDGE_URL}/api/domains", headers=_headers(), json=payload)
    data = res.json() if res.content else {}
    if res.status_code >= 400 or not data.get("success", True):
        raise OpenPanelBridgeError(data.get("error") or f"bridge add_domain HTTP {res.status_code}")
    return data


def delete_user(username: str) -> dict[str, Any]:
    with httpx.Client(timeout=120.0) as client:
        res = client.delete(f"{BRIDGE_URL}/api/users/{username.strip().lower()}", headers=_headers())
    data = res.json() if res.content else {}
    if res.status_code >= 400 or not data.get("success", True):
        raise OpenPanelBridgeError(data.get("error") or f"bridge delete HTTP {res.status_code}")
    return data