"""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, }