feat(pos/facturapi): add organization setup flow and detailed status
- get_org_status now returns has_key, has_org_id, pending_steps, error - add find_organization_by_rfc and create_organization helpers - add /facturapi/setup endpoint to link/create Facturapi org - frontend shows detailed PAC status and setup button - support using tenant sk_user_* key when FACTURAPI_USER_KEY env is absent
This commit is contained in:
@@ -186,28 +186,125 @@ def upload_csd(tenant_config: dict, cer_b64: str, key_b64: str, password: str) -
|
||||
return resp.json()
|
||||
|
||||
|
||||
def _get_user_key_for_tenant(tenant_config: dict) -> str:
|
||||
"""Resolve the Facturapi user key to use for organization management.
|
||||
|
||||
Priority:
|
||||
1. FACTURAPI_USER_KEY environment variable
|
||||
2. tenant_config.facturapi_key if it starts with sk_user_
|
||||
"""
|
||||
user_key = _get_user_key()
|
||||
if user_key:
|
||||
return user_key
|
||||
tenant_key = (tenant_config.get("facturapi_key") or "").strip()
|
||||
if tenant_key.startswith("sk_user_"):
|
||||
return tenant_key
|
||||
raise FacturapiError(
|
||||
"FACTURAPI_USER_KEY env or a Facturapi user key (sk_user_*) is required"
|
||||
)
|
||||
|
||||
|
||||
def find_organization_by_rfc(tenant_config: dict) -> Optional[dict]:
|
||||
"""Search for an existing Facturapi organization by tenant RFC.
|
||||
|
||||
Requires a user key (FACTURAPI_USER_KEY env or sk_user_* tenant key).
|
||||
Returns the organization dict or None.
|
||||
"""
|
||||
user_key = _get_user_key_for_tenant(tenant_config)
|
||||
|
||||
rfc = (tenant_config.get("rfc") or "").upper().strip()
|
||||
if not rfc:
|
||||
raise FacturapiError("Tenant RFC is required to search organizations")
|
||||
|
||||
page = 1
|
||||
while True:
|
||||
result = _request("GET", "/organizations", user_key, params={"page": page}, timeout=30)
|
||||
for org in result.get("data", []):
|
||||
legal = org.get("legal", {})
|
||||
if (legal.get("tax_id") or "").upper() == rfc:
|
||||
return org
|
||||
if page >= result.get("total_pages", 1):
|
||||
break
|
||||
page += 1
|
||||
return None
|
||||
|
||||
|
||||
def create_organization(tenant_config: dict) -> dict:
|
||||
"""Create a new Facturapi organization for the tenant and return live key.
|
||||
|
||||
Requires FACTURAPI_USER_KEY env or a user key (sk_user_*) in tenant_config.
|
||||
Uses tenant RFC/razon_social if available.
|
||||
"""
|
||||
user_key = _get_user_key_for_tenant(tenant_config)
|
||||
|
||||
rfc = (tenant_config.get("rfc") or "").upper().strip()
|
||||
name = tenant_config.get("razon_social") or tenant_config.get("name") or rfc or "Nexus"
|
||||
|
||||
# First try to find existing org by RFC
|
||||
existing = find_organization_by_rfc(tenant_config) if rfc else None
|
||||
if existing:
|
||||
org_id = existing["id"]
|
||||
else:
|
||||
payload = {"name": name}
|
||||
org = _request("POST", "/organizations", user_key, json_payload=payload, timeout=60)
|
||||
org_id = org.get("id")
|
||||
if not org_id:
|
||||
raise FacturapiError("Could not create organization: no id returned")
|
||||
|
||||
# Generate live secret key
|
||||
key_resp = _request(
|
||||
"PUT", f"/organizations/{org_id}/apikeys/live", user_key, json_payload={}, timeout=60
|
||||
)
|
||||
live_key = key_resp.get("key") if isinstance(key_resp, dict) else str(key_resp)
|
||||
if not live_key:
|
||||
raise FacturapiError(f"Could not generate live key for org {org_id}")
|
||||
|
||||
return {"org_id": org_id, "api_key": live_key}
|
||||
|
||||
|
||||
def get_org_status(tenant_config: dict) -> dict:
|
||||
"""Return organization status: configured, has_csd, org_id."""
|
||||
result = {
|
||||
"configured": False,
|
||||
"has_key": False,
|
||||
"has_org_id": False,
|
||||
"has_csd": False,
|
||||
"org_id": None,
|
||||
"legal_name": None,
|
||||
"tax_id": None,
|
||||
"pending_steps": [],
|
||||
"error": None,
|
||||
}
|
||||
|
||||
try:
|
||||
api_key = get_api_key(tenant_config)
|
||||
except FacturapiError:
|
||||
return {"configured": False, "has_csd": False, "org_id": None}
|
||||
result["has_key"] = True
|
||||
except FacturapiError as e:
|
||||
result["error"] = str(e)
|
||||
return result
|
||||
|
||||
org_id = tenant_config.get("facturapi_org_id")
|
||||
if not org_id:
|
||||
return {"configured": False, "has_csd": False, "org_id": None}
|
||||
result["error"] = "No Facturapi organization configured"
|
||||
return result
|
||||
|
||||
result["has_org_id"] = True
|
||||
result["org_id"] = org_id
|
||||
|
||||
try:
|
||||
org = get_organization(org_id, api_key)
|
||||
return {
|
||||
legal = org.get("legal", {})
|
||||
cert = org.get("certificate", {})
|
||||
result.update({
|
||||
"configured": True,
|
||||
"org_id": org_id,
|
||||
"has_csd": bool(org.get("certificate", {}).get("has_certificate")),
|
||||
"legal_name": org.get("legal", {}).get("name"),
|
||||
"tax_id": org.get("legal", {}).get("tax_id"),
|
||||
}
|
||||
except FacturapiError:
|
||||
return {"configured": False, "has_csd": False, "org_id": org_id}
|
||||
"has_csd": bool(cert.get("has_certificate")),
|
||||
"legal_name": legal.get("name") or legal.get("legal_name"),
|
||||
"tax_id": legal.get("tax_id"),
|
||||
"pending_steps": org.get("pending_steps", []),
|
||||
})
|
||||
except FacturapiError as e:
|
||||
result["error"] = str(e)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
# ─── Customers ──────────────────────────────────────────────────────────────
|
||||
|
||||
Reference in New Issue
Block a user