3ee63b3018
New /api/v1/infra/stack/status probes all stack apps/APIs/SW; Infra UI groups proc-cards by VM; wire vm123 router; menu INFRA COD and Serviços IaaS · Infra as Code labels. Co-authored-by: Cursor <cursoragent@cursor.com>
98 lines
3.2 KiB
Python
98 lines
3.2 KiB
Python
"""Cliente FOSSBilling Admin API."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import secrets
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from app.vm123.role_map import FOSS_GROUP_BY_ROLE
|
|
|
|
FOSS_BASE = os.getenv("FOSSBILLING_URL", "https://financeiro.ligbox.com.br").rstrip("/")
|
|
FOSS_ADMIN_USER = os.getenv("FOSS_ADMIN_USER", "admin")
|
|
FOSS_ADMIN_API_KEY = os.getenv("FOSS_ADMIN_API_KEY", os.getenv("FOSS_API_KEY", ""))
|
|
FOSS_PUBLIC_ADMIN = os.getenv("FOSS_PUBLIC_ADMIN_URL", f"{FOSS_BASE}/admin")
|
|
|
|
|
|
class FossConfigError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def _configured() -> bool:
|
|
return bool(FOSS_ADMIN_API_KEY)
|
|
|
|
|
|
def _auth():
|
|
if not _configured():
|
|
raise FossConfigError("FOSS_ADMIN_API_KEY não configurado no Desk")
|
|
return (FOSS_ADMIN_USER, FOSS_ADMIN_API_KEY)
|
|
|
|
|
|
def _post(path: str, payload: dict) -> dict[str, Any]:
|
|
url = f"{FOSS_BASE}/api/admin/{path.lstrip('/')}"
|
|
with httpx.Client(timeout=20.0) as client:
|
|
res = client.post(url, json=payload, auth=_auth())
|
|
if res.status_code >= 400:
|
|
raise RuntimeError(f"FOSS {path} HTTP {res.status_code}: {res.text[:300]}")
|
|
try:
|
|
return res.json()
|
|
except Exception:
|
|
return {"raw": res.text}
|
|
|
|
|
|
def find_client_by_email(email: str) -> dict[str, Any] | None:
|
|
data = _post("client/get_list", {"per_page": 50, "search": email.strip()})
|
|
items = data.get("result", {}).get("list") if isinstance(data.get("result"), dict) else data.get("list")
|
|
if not items:
|
|
return None
|
|
needle = email.strip().lower()
|
|
for item in items:
|
|
if str(item.get("email", "")).lower() == needle:
|
|
return item
|
|
return items[0] if items else None
|
|
|
|
|
|
def find_client_by_domain(domain: str) -> dict[str, Any] | None:
|
|
dom = domain.strip().lower()
|
|
data = _post("client/get_list", {"per_page": 100})
|
|
items = data.get("result", {}).get("list") if isinstance(data.get("result"), dict) else data.get("list") or []
|
|
for item in items:
|
|
for field in ("company", "company_vat", "email"):
|
|
val = str(item.get(field, "")).lower()
|
|
if dom in val:
|
|
return item
|
|
return None
|
|
|
|
|
|
def staff_group_name_for_role(desk_role: str) -> str | None:
|
|
return FOSS_GROUP_BY_ROLE.get(desk_role)
|
|
|
|
|
|
def create_staff(*, email: str, name: str, desk_role: str, password: str | None = None) -> dict[str, Any]:
|
|
"""Cria staff FOSS — grupo staff deve existir no Admin (manual v1)."""
|
|
group_name = staff_group_name_for_role(desk_role)
|
|
if not group_name:
|
|
return {"skipped": True, "reason": f"role {desk_role} sem grupo FOSS"}
|
|
pwd = password or secrets.token_urlsafe(14)
|
|
payload: dict[str, Any] = {
|
|
"email": email.strip().lower(),
|
|
"name": name,
|
|
"password": pwd,
|
|
"status": "active",
|
|
"admin_group_id": group_name,
|
|
}
|
|
try:
|
|
result = _post("staff/create", payload)
|
|
except RuntimeError as exc:
|
|
if "admin_group" in str(exc).lower() or "group" in str(exc).lower():
|
|
return {"skipped": True, "reason": str(exc), "group": group_name}
|
|
raise
|
|
return {
|
|
"foss_staff_id": result.get("id") or result.get("result"),
|
|
"email": email,
|
|
"group": group_name,
|
|
"admin_url": FOSS_PUBLIC_ADMIN,
|
|
"created": True,
|
|
}
|