feat(pos): migrate CFDI timbrado from Horux to Facturapi
- Add Facturapi REST service (invoices, customers, orgs, cancel, downloads) - Add JSON payload builder for ingreso/egreso/pago/global invoices - Replace XML queue with Facturapi JSON queue (payload_unsigned, external_id) - Update invoicing blueprint with Facturapi config and download endpoints - Update global invoice service to use Facturapi payloads - Add migration v4.3_facturapi.sql and tenant rollout script - Update invoicing UI: payload preview, PDF/XML downloads, PAC status panel - Add FACTURAPI_USER_KEY to .env.example
This commit is contained in:
329
pos/services/facturapi_service.py
Normal file
329
pos/services/facturapi_service.py
Normal file
@@ -0,0 +1,329 @@
|
||||
# /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_org_status(tenant_config: dict) -> dict:
|
||||
"""Return organization status: configured, has_csd, org_id."""
|
||||
try:
|
||||
api_key = get_api_key(tenant_config)
|
||||
except FacturapiError:
|
||||
return {"configured": False, "has_csd": False, "org_id": None}
|
||||
|
||||
org_id = tenant_config.get("facturapi_org_id")
|
||||
if not org_id:
|
||||
return {"configured": False, "has_csd": False, "org_id": None}
|
||||
|
||||
try:
|
||||
org = get_organization(org_id, api_key)
|
||||
return {
|
||||
"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}
|
||||
|
||||
|
||||
# ─── 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)
|
||||
Reference in New Issue
Block a user