# /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)