feat(fin,crm): catálogo c_UsoCFDI y claves fiscales del receptor

Cierra las decisiones pendientes 1 y 5. Agrega sat.cfdi_uses con su endpoint de
solo lectura y amarra la ficha del cliente a los catálogos del SAT con
crm.accounts.tax_regime_id y cfdi_use_id.

Las columnas de texto libre tax_regime y cfdi_use se conservan intactas: la
migración hace un backfill conservador que solo resuelve lo inequívoco (la clave
del catálogo, o la descripción exacta sin distinguir mayúsculas ni espacios) y
deja en NULL lo que no case, porque deducir el régimen de un receptor a partir
de texto libre provoca CFDI rechazados. La UI muestra el texto anterior junto al
selector para que el usuario elija la clave que corresponde.

El selector de régimen se acota al tipo de persona de la cuenta, y el service
valida ambas claves contra el catálogo.

sync_catalogs ahora omite los catálogos cuya tabla todavía no existe: al correr
el historial desde cero, la migración anterior la invoca antes de que se creen
los catálogos agregados después.

Las claves de c_UsoCFDI quedan pendientes de validación con el área Fiscal antes
de producción, igual que el subset de c_ClaveProdServ; no se cargaron las
banderas de persona física/moral ni la compatibilidad por régimen.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

backend/api/v1/modules/crm/accounts/dto.py

@@ -25,6 +25,9 @@ class AccountBase(BaseModel):
     # Fiscal
     tax_regime: str | None = Field(None, max_length=120)
     cfdi_use: str | None = Field(None, max_length=60)
+    # Claves contra los catálogos del SAT; sustituyen al texto libre de arriba al timbrar.
+    tax_regime_id: int | None = Field(None, description="c_RegimenFiscal del receptor")
+    cfdi_use_id: int | None = Field(None, description="c_UsoCFDI del receptor")
     payment_method: str | None = Field(None, max_length=60)
     payment_form: str | None = Field(None, max_length=60)
     currency: str | None = Field(None, max_length=3)
@@ -65,6 +68,8 @@ class AccountUpdate(BaseModel):
     website: str | None = Field(None, max_length=255)
     tax_regime: str | None = Field(None, max_length=120)
     cfdi_use: str | None = Field(None, max_length=60)
+    tax_regime_id: int | None = None
+    cfdi_use_id: int | None = None
     payment_method: str | None = Field(None, max_length=60)
     payment_form: str | None = Field(None, max_length=60)
     currency: str | None = Field(None, max_length=3)

backend/api/v1/modules/crm/accounts/models.py

@@ -1,9 +1,10 @@
 from decimal import Decimal
 
-from sqlalchemy import Integer, Numeric, String, Text, text
+from sqlalchemy import ForeignKey, Integer, Numeric, String, Text, text
 from sqlalchemy.orm import Mapped, mapped_column
 
 from api.v1.common.base_models import TenantScopedMixin, TimestampMixin
+from api.v1.modules.fin.catalogs.models import CfdiUse, TaxRegime  # noqa: F401  (resuelve las FK)
 from core.database import Base
 
 
@@ -47,8 +48,17 @@ class Account(Base, TenantScopedMixin, TimestampMixin):
     website: Mapped[str | None] = mapped_column(String(255), nullable=True)
 
     # ----- Información fiscal -----
+    # Régimen fiscal y uso de CFDI en texto libre: se conservan como capturó el usuario
+    # para no perder lo ya registrado, pero lo que vale al timbrar son las FK de abajo.
     tax_regime: Mapped[str | None] = mapped_column(String(120), nullable=True)  # régimen fiscal
     cfdi_use: Mapped[str | None] = mapped_column(String(60), nullable=True)  # uso de CFDI
+    # Claves del receptor contra los catálogos del SAT (c_RegimenFiscal y c_UsoCFDI).
+    tax_regime_id: Mapped[int | None] = mapped_column(
+        Integer, ForeignKey("sat.tax_regimes.id"), nullable=True
+    )
+    cfdi_use_id: Mapped[int | None] = mapped_column(
+        Integer, ForeignKey("sat.cfdi_uses.id"), nullable=True
+    )
     payment_method: Mapped[str | None] = mapped_column(String(60), nullable=True)  # método de pago
     payment_form: Mapped[str | None] = mapped_column(String(60), nullable=True)  # forma de pago
     currency: Mapped[str | None] = mapped_column(String(3), nullable=True)  # moneda

backend/api/v1/modules/crm/accounts/service.py

@@ -3,10 +3,24 @@ from datetime import datetime, timezone
 from fastapi import HTTPException, status
 from sqlalchemy.orm import Session
 
+from api.v1.modules.fin.catalogs.models import CfdiUse, TaxRegime
+
 from .dto import AccountCreate, AccountUpdate
 from .models import Account
 
 
+def _validate_sat_refs(db: Session, data: dict) -> None:
+    """Verifica las claves del SAT del receptor antes de guardar la cuenta."""
+    for field, model, msg in [
+        ("tax_regime_id", TaxRegime, "El régimen fiscal indicado no existe en el catálogo del SAT"),
+        ("cfdi_use_id", CfdiUse, "El uso de CFDI indicado no existe en el catálogo del SAT"),
+    ]:
+        value = data.get(field)
+        if field in data and value is not None:
+            if db.query(model.id).filter(model.id == value).first() is None:
+                raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=msg)
+
+
 def get_accounts(
     db: Session,
     tenant_id: int,
@@ -53,8 +67,10 @@ def get_account(db: Session, account_id: int, tenant_id: int, company_id: int) -
 def create_account(
     db: Session, payload: AccountCreate, tenant_id: int, company_id: int, user_id: str | None = None
 ) -> Account:
+    data = payload.model_dump()
+    _validate_sat_refs(db, data)
     account = Account(
-        **payload.model_dump(),
+        **data,
         tenant_id=tenant_id,
         company_id=company_id,
         created_by=user_id,
@@ -75,7 +91,9 @@ def update_account(
     user_id: str | None = None,
 ) -> Account:
     account = get_account(db, account_id, tenant_id, company_id)
-    for field, value in payload.model_dump(exclude_unset=True).items():
+    data = payload.model_dump(exclude_unset=True)
+    _validate_sat_refs(db, data)
+    for field, value in data.items():
         setattr(account, field, value)
     account.updated_by = user_id
     db.commit()

backend/api/v1/modules/fin/catalogs/dto.py

@@ -55,3 +55,7 @@ class PaymentMethodResponse(SatCatalogItem):
 
 class TaxObjectResponse(SatCatalogItem):
     """``c_ObjetoImp``."""
+
+
+class CfdiUseResponse(SatCatalogItem):
+    """``c_UsoCFDI``."""

backend/api/v1/modules/fin/catalogs/models.py

@@ -128,3 +128,16 @@ class TaxObject(Base, SatCatalogMixin):
     __table_args__ = {"schema": "sat"}
 
     code: Mapped[str] = mapped_column(String(2), nullable=False, unique=True, index=True)
+
+
+class CfdiUse(Base, SatCatalogMixin):
+    """``c_UsoCFDI`` — uso que el receptor le dará al comprobante.
+
+    Lo declara el receptor, no el emisor, y el SAT lo valida contra su régimen
+    fiscal: por eso vive en la ficha del cliente (``crm.accounts.cfdi_use_id``).
+    """
+
+    __tablename__ = "cfdi_uses"
+    __table_args__ = {"schema": "sat"}
+
+    code: Mapped[str] = mapped_column(String(4), nullable=False, unique=True, index=True)

backend/api/v1/modules/fin/catalogs/routes.py

@@ -18,6 +18,7 @@ from core.security import get_current_user
 
 from . import service
 from .dto import (
+    CfdiUseResponse,
     PaymentFormResponse,
     PaymentMethodResponse,
     ProductServiceResponse,
@@ -124,3 +125,14 @@ def list_tax_objects(
 ):
     """``c_ObjetoImp`` — si la partida es objeto de impuesto."""
     return service.get_tax_objects(db, search, active_only)
+
+
+@router.get("/catalogs/cfdi-uses", response_model=list[CfdiUseResponse])
+def list_cfdi_uses(
+    search: str | None = _SEARCH,
+    active_only: bool = _ACTIVE_ONLY,
+    current_user: dict = Depends(get_current_user),
+    db: Session = Depends(get_core_db),
+):
+    """``c_UsoCFDI`` — uso que el receptor le dará al comprobante."""
+    return service.get_cfdi_uses(db, search, active_only)

backend/api/v1/modules/fin/catalogs/seed_data.py

@@ -49,6 +49,7 @@ products_services_table = _catalog_table("products_services")
 voucher_types_table = _catalog_table("voucher_types")
 payment_methods_table = _catalog_table("payment_methods")
 tax_objects_table = _catalog_table("tax_objects")
+cfdi_uses_table = _catalog_table("cfdi_uses")
 
 
 # ---------------------------------------------------------------------------
@@ -234,6 +235,48 @@ PRODUCTS_SERVICES: list[dict] = [
 ]
 
 
+# ---------------------------------------------------------------------------
+# c_UsoCFDI
+# ---------------------------------------------------------------------------
+# Catálogo del uso que el receptor da al comprobante. Se siembran clave y
+# descripción; **no** se cargan las banderas de persona física/moral ni la
+# compatibilidad por régimen fiscal, porque esa matriz cambia entre versiones del
+# catálogo y equivocarla provoca rechazos al timbrar.
+#
+# Pendiente validación con área Fiscal antes de producción, igual que el subset de
+# c_ClaveProdServ.
+
+CFDI_USES: list[dict] = [
+    {"code": code, "description": description, "is_active": True}
+    for code, description in [
+        ("G01", "Adquisición de mercancías"),
+        ("G02", "Devoluciones, descuentos o bonificaciones"),
+        ("G03", "Gastos en general"),
+        ("I01", "Construcciones"),
+        ("I02", "Mobiliario y equipo de oficina por inversiones"),
+        ("I03", "Equipo de transporte"),
+        ("I04", "Equipo de cómputo y accesorios"),
+        ("I05", "Dados, troqueles, moldes, matrices y herramental"),
+        ("I06", "Comunicaciones telefónicas"),
+        ("I07", "Comunicaciones satelitales"),
+        ("I08", "Otra maquinaria y equipo"),
+        ("D01", "Honorarios médicos, dentales y gastos hospitalarios"),
+        ("D02", "Gastos médicos por incapacidad o discapacidad"),
+        ("D03", "Gastos funerales"),
+        ("D04", "Donativos"),
+        ("D05", "Intereses reales efectivamente pagados por créditos hipotecarios (casa habitación)"),
+        ("D06", "Aportaciones voluntarias al SAR"),
+        ("D07", "Primas por seguros de gastos médicos"),
+        ("D08", "Gastos de transportación escolar obligatoria"),
+        ("D09", "Depósitos en cuentas para el ahorro, primas que tengan como base planes de pensiones"),
+        ("D10", "Pagos por servicios educativos (colegiaturas)"),
+        ("S01", "Sin efectos fiscales"),
+        ("CP01", "Pagos"),
+        ("CN01", "Nómina"),
+    ]
+]
+
+
 # Orden estable de sincronización: (tabla, filas).
 CATALOGS: list[tuple[sa.Table, list[dict]]] = [
     (tax_regimes_table, TAX_REGIMES),
@@ -244,6 +287,7 @@ CATALOGS: list[tuple[sa.Table, list[dict]]] = [
     (voucher_types_table, VOUCHER_TYPES),
     (payment_methods_table, PAYMENT_METHODS),
     (tax_objects_table, TAX_OBJECTS),
+    (cfdi_uses_table, CFDI_USES),
 ]
 
 
@@ -257,11 +301,24 @@ def sync_catalogs(connection) -> dict[str, int]:
     Devuelve un resumen ``{"sat.tabla": filas_insertadas}`` útil para la bitácora de
     la migración.
 
+    Los catálogos cuya tabla todavía no existe se omiten: al correr el historial de
+    migraciones desde cero, una migración antigua invoca esta misma función cuando los
+    catálogos agregados después aún no se han creado. Cada uno se siembra en la
+    migración que lo crea.
+
     Se usa contra el ``connection`` que da ``op.get_bind()`` en Alembic, o contra la
     conexión de una sesión en pruebas.
     """
+    inspector = sa.inspect(connection)
+    # La inspección no aplica el schema_translate_map (las pruebas mapean sat -> None
+    # sobre SQLite), así que se resuelve el schema efectivo a mano.
+    schema_map = connection.get_execution_options().get("schema_translate_map") or {}
+
     inserted: dict[str, int] = {}
     for table, rows in CATALOGS:
+        effective_schema = schema_map.get(table.schema, table.schema)
+        if not inspector.has_table(table.name, schema=effective_schema):
+            continue
         key = f"sat.{table.name}"
         inserted[key] = 0
         for row in rows:

backend/api/v1/modules/fin/catalogs/service.py

@@ -8,6 +8,7 @@ from sqlalchemy import or_
 from sqlalchemy.orm import Session
 
 from .models import (
+    CfdiUse,
     PaymentForm,
     PaymentMethod,
     ProductService,
@@ -99,3 +100,7 @@ def get_payment_methods(db: Session, search=None, active_only=True) -> list[Paym
 
 def get_tax_objects(db: Session, search=None, active_only=True) -> list[TaxObject]:
     return search_catalog(db, TaxObject, search, active_only)
+
+
+def get_cfdi_uses(db: Session, search=None, active_only=True) -> list[CfdiUse]:
+    return search_catalog(db, CfdiUse, search, active_only)