consultoria-as
27358312dc
- 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
427 lines
14 KiB
Python
427 lines
14 KiB
Python
# /home/Autopartes/pos/services/facturapi_service.py
|
|
"""Facturapi integration for Nexus POS.
|
|
|
|
Uses Facturapi REST API directly (requests + Basic Auth) so it is safe for
|
|
multi-tenant use. Each call receives the API key explicitly, avoiding the
|
|
global client used by the official facturapi Python library.
|
|
|
|
Authentication modes:
|
|
1. User key (FACTURAPI_USER_KEY env): creates/verifies organizations per tenant.
|
|
2. Secret key per tenant (tenant_config.facturapi_secret_key): uses existing org.
|
|
|
|
Reference: https://docs.facturapi.io/
|
|
"""
|
|
|
|
import os
|
|
import base64
|
|
import logging
|
|
from decimal import Decimal
|
|
from typing import Optional
|
|
|
|
import requests
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
BASE_URL = "https://www.facturapi.io/v2"
|
|
USER_KEY = os.environ.get("FACTURAPI_USER_KEY", "")
|
|
|
|
|
|
class FacturapiError(Exception):
|
|
def __init__(self, message: str, status_code: int = 0, response_body: str = ""):
|
|
super().__init__(message)
|
|
self.status_code = status_code
|
|
self.response_body = response_body
|
|
|
|
|
|
# ─── HTTP helpers ───────────────────────────────────────────────────────────
|
|
|
|
def _request(method: str, endpoint: str, api_key: str, json_payload=None, params=None,
|
|
extra_headers=None, timeout=60):
|
|
"""Make a request to Facturapi REST API with Basic Auth."""
|
|
url = f"{BASE_URL}{endpoint}"
|
|
headers = {"Content-Type": "application/json"}
|
|
if extra_headers:
|
|
headers.update(extra_headers)
|
|
|
|
try:
|
|
resp = requests.request(
|
|
method,
|
|
url,
|
|
auth=(api_key, ""),
|
|
headers=headers,
|
|
json=json_payload,
|
|
params=params,
|
|
timeout=timeout,
|
|
)
|
|
except requests.RequestException as e:
|
|
raise FacturapiError(f"Connection error: {e}", status_code=0)
|
|
|
|
if not resp.ok:
|
|
raise FacturapiError(
|
|
f"Facturapi {method.upper()} {endpoint} failed: {resp.status_code} {resp.text[:500]}",
|
|
status_code=resp.status_code,
|
|
response_body=resp.text,
|
|
)
|
|
|
|
if resp.status_code == 204 or not resp.content:
|
|
return {}
|
|
return resp.json()
|
|
|
|
|
|
def _download(method: str, endpoint: str, api_key: str, params=None, timeout=60) -> bytes:
|
|
"""Download binary content (XML/PDF)."""
|
|
url = f"{BASE_URL}{endpoint}"
|
|
resp = requests.request(
|
|
method,
|
|
url,
|
|
auth=(api_key, ""),
|
|
params=params,
|
|
timeout=timeout,
|
|
)
|
|
if not resp.ok:
|
|
raise FacturapiError(
|
|
f"Download failed: {resp.status_code} {resp.text[:500]}",
|
|
status_code=resp.status_code,
|
|
)
|
|
return resp.content
|
|
|
|
|
|
# ─── Tenant config helpers ──────────────────────────────────────────────────
|
|
|
|
def _get_secret_key(tenant_config: dict) -> Optional[str]:
|
|
for key in ("facturapi_key", "facturapi_secret_key"):
|
|
val = (tenant_config.get(key) or "").strip()
|
|
if val:
|
|
return val
|
|
return None
|
|
|
|
|
|
def _get_user_key() -> Optional[str]:
|
|
return USER_KEY.strip() or None
|
|
|
|
|
|
def _is_user_key_mode(tenant_config: dict) -> bool:
|
|
return bool(_get_user_key()) and not _get_secret_key(tenant_config)
|
|
|
|
|
|
def get_api_key(tenant_config: dict) -> str:
|
|
"""Resolve the API key to use for a tenant.
|
|
|
|
Priority:
|
|
1. tenant_config.facturapi_secret_key (manual override)
|
|
2. FACTURAPI_USER_KEY env (auto-org mode)
|
|
"""
|
|
secret = _get_secret_key(tenant_config)
|
|
if secret:
|
|
return secret
|
|
user = _get_user_key()
|
|
if user:
|
|
return user
|
|
raise FacturapiError(
|
|
"Facturapi not configured. Set FACTURAPI_USER_KEY env or tenant_config.facturapi_secret_key"
|
|
)
|
|
|
|
|
|
# ─── Organizations ──────────────────────────────────────────────────────────
|
|
|
|
def create_organization(tenant_config: dict) -> dict:
|
|
"""Create a new Facturapi organization for the tenant.
|
|
|
|
Requires FACTURAPI_USER_KEY.
|
|
Returns dict with id, api_key.
|
|
"""
|
|
user_key = _get_user_key()
|
|
if not user_key:
|
|
raise FacturapiError("FACTURAPI_USER_KEY is required to create organizations")
|
|
|
|
payload = {"name": tenant_config.get("razon_social", tenant_config.get("name", "Nexus"))}
|
|
legal = tenant_config.get("legal_name") or tenant_config.get("razon_social")
|
|
if legal:
|
|
payload["legal"] = {"name": legal}
|
|
if tenant_config.get("rfc"):
|
|
payload["legal"] = payload.get("legal", {})
|
|
payload["legal"]["tax_id"] = tenant_config["rfc"]
|
|
|
|
org = _request("POST", "/organizations", user_key, json_payload=payload)
|
|
org_id = org.get("id")
|
|
|
|
# Generate live secret key
|
|
key_resp = _request("PUT", f"/organizations/{org_id}/apikeys/live", user_key, json_payload={})
|
|
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_organization(org_id: str, api_key: str) -> dict:
|
|
return _request("GET", f"/organizations/{org_id}", api_key)
|
|
|
|
|
|
def upload_csd(tenant_config: dict, cer_b64: str, key_b64: str, password: str) -> dict:
|
|
"""Upload CSD (Certificado de Sello Digital) to Facturapi.
|
|
|
|
cer_b64 and key_b64 are base64-encoded strings.
|
|
"""
|
|
api_key = get_api_key(tenant_config)
|
|
org_id = tenant_config.get("facturapi_org_id")
|
|
if not org_id:
|
|
raise FacturapiError("No Facturapi organization configured for tenant")
|
|
|
|
cer_bytes = base64.b64decode(cer_b64)
|
|
key_bytes = base64.b64decode(key_b64)
|
|
|
|
url = f"{BASE_URL}/organizations/{org_id}/certificate"
|
|
files = {
|
|
"certificate": ("certificate.cer", cer_bytes, "application/octet-stream"),
|
|
"private_key": ("private_key.key", key_bytes, "application/octet-stream"),
|
|
"secret": (None, password),
|
|
}
|
|
resp = requests.post(url, auth=(api_key, ""), files=files, timeout=60)
|
|
if not resp.ok:
|
|
raise FacturapiError(
|
|
f"CSD upload failed: {resp.status_code} {resp.text[:500]}",
|
|
status_code=resp.status_code,
|
|
)
|
|
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:
|
|
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)
|
|
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:
|
|
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)
|
|
legal = org.get("legal", {})
|
|
cert = org.get("certificate", {})
|
|
result.update({
|
|
"configured": True,
|
|
"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 ──────────────────────────────────────────────────────────────
|
|
|
|
def create_or_update_customer(tenant_config: dict, customer_data: dict) -> str:
|
|
"""Create or update a customer in Facturapi and return its id.
|
|
|
|
customer_data: {
|
|
legal_name: str,
|
|
tax_id: str,
|
|
tax_system: str,
|
|
email: str,
|
|
zip: str,
|
|
country: str (optional, ISO 3166 alpha-3),
|
|
}
|
|
"""
|
|
api_key = get_api_key(tenant_config)
|
|
tax_id = (customer_data.get("tax_id") or "").upper().strip()
|
|
if not tax_id:
|
|
raise FacturapiError("Customer tax_id is required")
|
|
|
|
# Try to find existing customer
|
|
existing_id = None
|
|
try:
|
|
result = _request("GET", "/customers", api_key, params={"search": tax_id})
|
|
for c in result.get("data", []):
|
|
if (c.get("tax_id") or "").upper() == tax_id:
|
|
existing_id = c.get("id")
|
|
break
|
|
except FacturapiError as e:
|
|
logger.warning("Failed to search Facturapi customer: %s", e)
|
|
|
|
is_foreign = bool(customer_data.get("country")) and customer_data["country"] != "MEX"
|
|
|
|
payload = {
|
|
"legal_name": customer_data.get("legal_name", ""),
|
|
"email": customer_data.get("email"),
|
|
"address": {
|
|
"zip": customer_data.get("zip", "00000"),
|
|
},
|
|
}
|
|
if is_foreign:
|
|
payload["tax_id"] = tax_id
|
|
payload["address"]["country"] = customer_data["country"]
|
|
else:
|
|
payload["tax_id"] = tax_id
|
|
if customer_data.get("tax_system"):
|
|
payload["tax_system"] = customer_data["tax_system"]
|
|
|
|
if existing_id:
|
|
_request("PUT", f"/customers/{existing_id}", api_key, json_payload=payload)
|
|
return existing_id
|
|
|
|
new_customer = _request("POST", "/customers", api_key, json_payload=payload)
|
|
return new_customer.get("id")
|
|
|
|
|
|
# ─── Invoices ───────────────────────────────────────────────────────────────
|
|
|
|
def create_invoice(tenant_config: dict, payload: dict) -> dict:
|
|
"""Create and stamp an invoice in Facturapi.
|
|
|
|
Returns the Facturapi invoice object.
|
|
"""
|
|
api_key = get_api_key(tenant_config)
|
|
return _request("POST", "/invoices", api_key, json_payload=payload, timeout=90)
|
|
|
|
|
|
def cancel_invoice(tenant_config: dict, invoice_id: str, motive: str,
|
|
replacement_uuid: Optional[str] = None) -> dict:
|
|
"""Cancel an invoice in Facturapi.
|
|
|
|
Motive codes:
|
|
01: errores con relacion (requires replacement_uuid)
|
|
02: errores sin relacion
|
|
03: no se llevo a cabo la operacion
|
|
04: operacion nominativa relacionada en factura global
|
|
"""
|
|
api_key = get_api_key(tenant_config)
|
|
params = {"motive": motive}
|
|
if replacement_uuid:
|
|
params["replacement"] = replacement_uuid
|
|
return _request("DELETE", f"/invoices/{invoice_id}", api_key, params=params, timeout=60)
|
|
|
|
|
|
def download_xml(tenant_config: dict, invoice_id: str) -> bytes:
|
|
api_key = get_api_key(tenant_config)
|
|
return _download("GET", f"/invoices/{invoice_id}/xml", api_key)
|
|
|
|
|
|
def download_pdf(tenant_config: dict, invoice_id: str) -> bytes:
|
|
api_key = get_api_key(tenant_config)
|
|
return _download("GET", f"/invoices/{invoice_id}/pdf", api_key)
|
|
|
|
|
|
# ─── Helpers ─────────────────────────────────────────────────────────────────
|
|
|
|
def is_lco_rejection(message: str) -> bool:
|
|
"""Detect SAT LCO rejection (CSD not yet propagated)."""
|
|
if not message:
|
|
return False
|
|
msg = message.lower()
|
|
return any(
|
|
pattern in msg
|
|
for pattern in [
|
|
"lco",
|
|
"no se encontro el rfc",
|
|
"rfc no registrado",
|
|
"lista de contribuyentes obligados",
|
|
"csd no registrado",
|
|
]
|
|
)
|
|
|
|
|
|
def to_cents(amount) -> int:
|
|
"""Convert Decimal/float/None to integer cents for Facturapi."""
|
|
if amount is None:
|
|
return 0
|
|
return int(Decimal(str(amount)).quantize(Decimal("0.01")) * 100)
|