821675ab4a
Specs stay at repo root (cross-VM). Move deploy and code into logical projects with README per domain, updated manifest.yaml, and symlinks at legacy paths for VM122 backward compatibility.
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,
|
|
}
|