From e24435c74beea86686b042f96508400aea41672e Mon Sep 17 00:00:00 2001 From: Jair Cedillo Date: Fri, 7 Aug 2026 16:57:26 -0500 Subject: [PATCH] =?UTF-8?q?feat(fin):=20cat=C3=A1logos=20SAT=20en=20schema?= =?UTF-8?q?=20sat=20con=20seeds=20idempotentes?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Agrega los 8 catálogos oficiales del SAT (c_RegimenFiscal, c_Impuesto, c_FormaPago, c_ClaveUnidad, c_ClaveProdServ, c_TipoDeComprobante, c_MetodoPago y c_ObjetoImp) como tablas globales de solo lectura en el schema sat: sin tenant_id, sin CRUD y sin baja física (las claves retiradas se desactivan para no romper los CFDI históricos). Las semillas viven en catalogs/seed_data.py, no dentro de una migración, para que corregir un dato del catálogo no exija escribir una migración nueva. sync_catalogs() hace upsert por clave: inserta lo que falta, actualiza descripción y banderas, y nunca borra. El subset de c_ClaveProdServ (11 claves de logística) queda pendiente de validación con el área Fiscal antes de producción. Co-Authored-By: Claude Opus 5 (1M context) --- .../api/v1/modules/fin/catalogs/__init__.py | 1 + backend/api/v1/modules/fin/catalogs/dto.py | 57 ++++ backend/api/v1/modules/fin/catalogs/models.py | 130 ++++++++ backend/api/v1/modules/fin/catalogs/routes.py | 126 ++++++++ .../api/v1/modules/fin/catalogs/seed_data.py | 279 ++++++++++++++++++ .../api/v1/modules/fin/catalogs/service.py | 101 +++++++ 6 files changed, 694 insertions(+) create mode 100644 backend/api/v1/modules/fin/catalogs/__init__.py create mode 100644 backend/api/v1/modules/fin/catalogs/dto.py create mode 100644 backend/api/v1/modules/fin/catalogs/models.py create mode 100644 backend/api/v1/modules/fin/catalogs/routes.py create mode 100644 backend/api/v1/modules/fin/catalogs/seed_data.py create mode 100644 backend/api/v1/modules/fin/catalogs/service.py diff --git a/backend/api/v1/modules/fin/catalogs/__init__.py b/backend/api/v1/modules/fin/catalogs/__init__.py new file mode 100644 index 0000000..0778b59 --- /dev/null +++ b/backend/api/v1/modules/fin/catalogs/__init__.py @@ -0,0 +1 @@ +"""Catálogos oficiales del SAT (schema ``sat``): globales y de solo lectura.""" diff --git a/backend/api/v1/modules/fin/catalogs/dto.py b/backend/api/v1/modules/fin/catalogs/dto.py new file mode 100644 index 0000000..0a80605 --- /dev/null +++ b/backend/api/v1/modules/fin/catalogs/dto.py @@ -0,0 +1,57 @@ +"""Esquemas de respuesta de los catálogos del SAT (solo lectura).""" + +from pydantic import BaseModel, ConfigDict + + +class SatCatalogItem(BaseModel): + """Forma común de todo catálogo del SAT: clave + descripción.""" + + model_config = ConfigDict(from_attributes=True) + + id: int + code: str + description: str + is_active: bool + + +class TaxRegimeResponse(SatCatalogItem): + """``c_RegimenFiscal``: incluye a qué tipo de persona aplica el régimen.""" + + applies_to_individual: bool # persona física + applies_to_legal_entity: bool # persona moral + + +class TaxResponse(SatCatalogItem): + """``c_Impuesto``: indica si el impuesto puede retenerse o trasladarse.""" + + is_withholding: bool + is_transferred: bool + is_local: bool + + +class UnitOfMeasureResponse(SatCatalogItem): + """``c_ClaveUnidad``: nombre corto, símbolo y nota larga del catálogo.""" + + description: str | None = None + name: str + symbol: str | None = None + + +class PaymentFormResponse(SatCatalogItem): + """``c_FormaPago``.""" + + +class ProductServiceResponse(SatCatalogItem): + """``c_ClaveProdServ``.""" + + +class VoucherTypeResponse(SatCatalogItem): + """``c_TipoDeComprobante``.""" + + +class PaymentMethodResponse(SatCatalogItem): + """``c_MetodoPago``.""" + + +class TaxObjectResponse(SatCatalogItem): + """``c_ObjetoImp``.""" diff --git a/backend/api/v1/modules/fin/catalogs/models.py b/backend/api/v1/modules/fin/catalogs/models.py new file mode 100644 index 0000000..3c19338 --- /dev/null +++ b/backend/api/v1/modules/fin/catalogs/models.py @@ -0,0 +1,130 @@ +"""Modelos de los catálogos oficiales del SAT — schema ``sat``. + +Son catálogos **globales**: los publica el SAT, valen igual para cualquier tenant y +compañía, por eso no heredan ``TenantScopedMixin``. Tampoco se borran: cuando el SAT +retira una clave, el registro se marca ``is_active = false`` para que las facturas +históricas que la usan sigan resolviendo su descripción (de ahí que se use +``BaseTimestampMixin``, sin ``deleted_at``). + +La API los expone únicamente en modo lectura; el alta y la actualización pasan por +``seed_data.sync_catalogs()``. +""" + +from sqlalchemy import Boolean, Integer, String, text +from sqlalchemy.orm import Mapped, mapped_column + +from api.v1.common.base_models import BaseTimestampMixin +from core.database import Base + + +class SatCatalogMixin(BaseTimestampMixin): + """Campos comunes a todo catálogo del SAT. + + ``code`` (la clave oficial) se declara en cada modelo porque su longitud + cambia de catálogo en catálogo. + """ + + id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True) + description: Mapped[str] = mapped_column(String(500), nullable=False) + is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, server_default=text("true")) + + +class TaxRegime(Base, SatCatalogMixin): + """``c_RegimenFiscal`` — régimen fiscal del emisor y del receptor del CFDI. + + Las banderas indican a qué tipo de persona aplica el régimen: una persona física + no puede declararse en el 601 (General de Ley Personas Morales) y viceversa. + """ + + __tablename__ = "tax_regimes" + __table_args__ = {"schema": "sat"} + + code: Mapped[str] = mapped_column(String(3), nullable=False, unique=True, index=True) + applies_to_individual: Mapped[bool] = mapped_column( # persona física + Boolean, nullable=False, server_default=text("false") + ) + applies_to_legal_entity: Mapped[bool] = mapped_column( # persona moral + Boolean, nullable=False, server_default=text("false") + ) + + +class Tax(Base, SatCatalogMixin): + """``c_Impuesto`` — impuestos federales que pueden trasladarse o retenerse.""" + + __tablename__ = "taxes" + __table_args__ = {"schema": "sat"} + + code: Mapped[str] = mapped_column(String(3), nullable=False, unique=True, index=True) + is_withholding: Mapped[bool] = mapped_column( # puede retenerse + Boolean, nullable=False, server_default=text("false") + ) + is_transferred: Mapped[bool] = mapped_column( # puede trasladarse + Boolean, nullable=False, server_default=text("false") + ) + # Los impuestos locales (ISH y similares) viajan en el complemento "Impuestos + # Locales" con claves ajenas a c_Impuesto; la bandera queda disponible para + # cuando el negocio defina ese catálogo. + is_local: Mapped[bool] = mapped_column(Boolean, nullable=False, server_default=text("false")) + + +class PaymentForm(Base, SatCatalogMixin): + """``c_FormaPago`` — con qué se pagó (efectivo, transferencia, tarjeta…).""" + + __tablename__ = "payment_forms" + __table_args__ = {"schema": "sat"} + + code: Mapped[str] = mapped_column(String(2), nullable=False, unique=True, index=True) + + +class UnitOfMeasure(Base, SatCatalogMixin): + """``c_ClaveUnidad`` — unidad de medida de la partida. + + Único catálogo que separa nombre corto y definición: ``name`` es lo que se + muestra al capturar y ``description`` la nota larga del SAT, que puede venir + vacía. + """ + + __tablename__ = "units_of_measure" + __table_args__ = {"schema": "sat"} + + code: Mapped[str] = mapped_column(String(20), nullable=False, unique=True, index=True) + name: Mapped[str] = mapped_column(String(255), nullable=False) + symbol: Mapped[str | None] = mapped_column(String(20), nullable=True) + # Se redeclara para permitir NULL: aquí la descripción es la nota del catálogo. + description: Mapped[str | None] = mapped_column(String(500), nullable=True) + + +class ProductService(Base, SatCatalogMixin): + """``c_ClaveProdServ`` — clave de producto o servicio de la partida.""" + + __tablename__ = "products_services" + __table_args__ = {"schema": "sat"} + + code: Mapped[str] = mapped_column(String(8), nullable=False, unique=True, index=True) + + +class VoucherType(Base, SatCatalogMixin): + """``c_TipoDeComprobante`` — I ingreso, E egreso, T traslado, N nómina, P pago.""" + + __tablename__ = "voucher_types" + __table_args__ = {"schema": "sat"} + + code: Mapped[str] = mapped_column(String(1), nullable=False, unique=True, index=True) + + +class PaymentMethod(Base, SatCatalogMixin): + """``c_MetodoPago`` — PUE (una sola exhibición) o PPD (parcialidades/diferido).""" + + __tablename__ = "payment_methods" + __table_args__ = {"schema": "sat"} + + code: Mapped[str] = mapped_column(String(3), nullable=False, unique=True, index=True) + + +class TaxObject(Base, SatCatalogMixin): + """``c_ObjetoImp`` — si la partida es o no objeto de impuesto.""" + + __tablename__ = "tax_objects" + __table_args__ = {"schema": "sat"} + + code: Mapped[str] = mapped_column(String(2), nullable=False, unique=True, index=True) diff --git a/backend/api/v1/modules/fin/catalogs/routes.py b/backend/api/v1/modules/fin/catalogs/routes.py new file mode 100644 index 0000000..8396ab8 --- /dev/null +++ b/backend/api/v1/modules/fin/catalogs/routes.py @@ -0,0 +1,126 @@ +"""Endpoints de los catálogos del SAT — **solo lectura**. + +No se exponen POST/PUT/PATCH/DELETE a propósito: son catálogos fijos publicados por +el SAT y se mantienen con ``seed_data.sync_catalogs()``, no por API. + +Nota: aunque los catálogos son globales, el router del módulo exige ``fin.access``, +permiso que se resuelve sobre una compañía; por eso las peticiones siguen llevando +``company_id`` en la query string. +""" + +from typing import Literal + +from fastapi import APIRouter, Depends, Query +from sqlalchemy.orm import Session + +from core.database import get_core_db +from core.security import get_current_user + +from . import service +from .dto import ( + PaymentFormResponse, + PaymentMethodResponse, + ProductServiceResponse, + TaxObjectResponse, + TaxRegimeResponse, + TaxResponse, + UnitOfMeasureResponse, + VoucherTypeResponse, +) + +router = APIRouter() + +_SEARCH = Query(None, description="Búsqueda por clave o descripción") +_ACTIVE_ONLY = Query(True, description="Solo claves vigentes") + + +@router.get("/catalogs/tax-regimes", response_model=list[TaxRegimeResponse]) +def list_tax_regimes( + search: str | None = _SEARCH, + active_only: bool = _ACTIVE_ONLY, + person_type: Literal["fisica", "moral"] | None = Query( + None, description="Acota al régimen de persona física o moral" + ), + current_user: dict = Depends(get_current_user), + db: Session = Depends(get_core_db), +): + """``c_RegimenFiscal`` — régimen fiscal del emisor/receptor del CFDI.""" + return service.get_tax_regimes(db, search, active_only, person_type) + + +@router.get("/catalogs/taxes", response_model=list[TaxResponse]) +def list_taxes( + search: str | None = _SEARCH, + active_only: bool = _ACTIVE_ONLY, + current_user: dict = Depends(get_current_user), + db: Session = Depends(get_core_db), +): + """``c_Impuesto`` — impuestos federales trasladados y retenidos.""" + return service.get_taxes(db, search, active_only) + + +@router.get("/catalogs/payment-forms", response_model=list[PaymentFormResponse]) +def list_payment_forms( + search: str | None = _SEARCH, + active_only: bool = _ACTIVE_ONLY, + current_user: dict = Depends(get_current_user), + db: Session = Depends(get_core_db), +): + """``c_FormaPago`` — medio con el que se liquidó el comprobante.""" + return service.get_payment_forms(db, search, active_only) + + +@router.get("/catalogs/units-of-measure", response_model=list[UnitOfMeasureResponse]) +def list_units_of_measure( + search: str | None = _SEARCH, + active_only: bool = _ACTIVE_ONLY, + current_user: dict = Depends(get_current_user), + db: Session = Depends(get_core_db), +): + """``c_ClaveUnidad`` — unidad de medida de la partida.""" + return service.get_units_of_measure(db, search, active_only) + + +@router.get("/catalogs/products-services", response_model=list[ProductServiceResponse]) +def list_products_services( + search: str | None = _SEARCH, + active_only: bool = _ACTIVE_ONLY, + limit: int = Query(50, ge=1, le=200, description="Máximo de claves devueltas"), + current_user: dict = Depends(get_current_user), + db: Session = Depends(get_core_db), +): + """``c_ClaveProdServ`` — clave de producto/servicio; pensado para autocompletado.""" + return service.get_products_services(db, search, active_only, limit) + + +@router.get("/catalogs/voucher-types", response_model=list[VoucherTypeResponse]) +def list_voucher_types( + search: str | None = _SEARCH, + active_only: bool = _ACTIVE_ONLY, + current_user: dict = Depends(get_current_user), + db: Session = Depends(get_core_db), +): + """``c_TipoDeComprobante`` — ingreso, egreso, traslado, nómina o pago.""" + return service.get_voucher_types(db, search, active_only) + + +@router.get("/catalogs/payment-methods", response_model=list[PaymentMethodResponse]) +def list_payment_methods( + search: str | None = _SEARCH, + active_only: bool = _ACTIVE_ONLY, + current_user: dict = Depends(get_current_user), + db: Session = Depends(get_core_db), +): + """``c_MetodoPago`` — PUE o PPD.""" + return service.get_payment_methods(db, search, active_only) + + +@router.get("/catalogs/tax-objects", response_model=list[TaxObjectResponse]) +def list_tax_objects( + search: str | None = _SEARCH, + active_only: bool = _ACTIVE_ONLY, + current_user: dict = Depends(get_current_user), + db: Session = Depends(get_core_db), +): + """``c_ObjetoImp`` — si la partida es objeto de impuesto.""" + return service.get_tax_objects(db, search, active_only) diff --git a/backend/api/v1/modules/fin/catalogs/seed_data.py b/backend/api/v1/modules/fin/catalogs/seed_data.py new file mode 100644 index 0000000..60ebe35 --- /dev/null +++ b/backend/api/v1/modules/fin/catalogs/seed_data.py @@ -0,0 +1,279 @@ +"""Datos semilla de los catálogos del SAT y su sincronización idempotente. + +Los catálogos viven aquí y no dentro de una migración concreta a propósito: cuando el +SAT corrige una descripción o publica una clave nueva, basta editar estas listas y +volver a correr :func:`sync_catalogs`, sin escribir una migración de esquema. + +Las tablas se describen con ``sa.Table`` ligeros sobre un ``MetaData`` propio (no con +los modelos ORM) para que la migración pueda importar este módulo sin acoplarse a la +definición ORM, que sigue evolucionando. +""" + +import sqlalchemy as sa + +_metadata = sa.MetaData() + + +def _catalog_table(name: str, *extra_columns: sa.Column) -> sa.Table: + """Tabla mínima de catálogo: las columnas que toca el upsert, nada más.""" + return sa.Table( + name, + _metadata, + sa.Column("id", sa.Integer, primary_key=True), + sa.Column("code", sa.String, nullable=False), + sa.Column("description", sa.String), + sa.Column("is_active", sa.Boolean), + *extra_columns, + schema="sat", + ) + + +tax_regimes_table = _catalog_table( + "tax_regimes", + sa.Column("applies_to_individual", sa.Boolean), + sa.Column("applies_to_legal_entity", sa.Boolean), +) +taxes_table = _catalog_table( + "taxes", + sa.Column("is_withholding", sa.Boolean), + sa.Column("is_transferred", sa.Boolean), + sa.Column("is_local", sa.Boolean), +) +payment_forms_table = _catalog_table("payment_forms") +units_of_measure_table = _catalog_table( + "units_of_measure", + sa.Column("name", sa.String), + sa.Column("symbol", sa.String), +) +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") + + +# --------------------------------------------------------------------------- +# c_RegimenFiscal (CFDI 4.0) +# --------------------------------------------------------------------------- + +def _regime(code: str, description: str, individual: bool, legal_entity: bool) -> dict: + return { + "code": code, + "description": description, + "applies_to_individual": individual, + "applies_to_legal_entity": legal_entity, + "is_active": True, + } + + +TAX_REGIMES: list[dict] = [ + _regime("601", "General de Ley Personas Morales", False, True), + _regime("603", "Personas Morales con Fines no Lucrativos", False, True), + _regime("605", "Sueldos y Salarios e Ingresos Asimilados a Salarios", True, False), + _regime("606", "Arrendamiento", True, False), + _regime("607", "Régimen de Enajenación o Adquisición de Bienes", True, False), + _regime("608", "Demás ingresos", True, False), + _regime("610", "Residentes en el Extranjero sin Establecimiento Permanente en México", True, True), + _regime("611", "Ingresos por Dividendos (socios y accionistas)", True, False), + _regime("612", "Personas Físicas con Actividades Empresariales y Profesionales", True, False), + _regime("614", "Ingresos por intereses", True, False), + _regime("615", "Régimen de los ingresos por obtención de premios", True, False), + _regime("616", "Sin obligaciones fiscales", True, False), + _regime("620", "Sociedades Cooperativas de Producción que optan por diferir sus ingresos", False, True), + _regime("621", "Incorporación Fiscal", True, False), + _regime("622", "Actividades Agrícolas, Ganaderas, Silvícolas y Pesqueras", False, True), + _regime("623", "Opcional para Grupos de Sociedades", False, True), + _regime("624", "Coordinados", False, True), + _regime("625", "Régimen de las Actividades Empresariales con ingresos a través de Plataformas Tecnológicas", True, False), + _regime("626", "Régimen Simplificado de Confianza", True, True), +] + + +# --------------------------------------------------------------------------- +# c_Impuesto +# --------------------------------------------------------------------------- +# is_local queda en false para los tres: los impuestos locales (ISH y similares) +# se declaran en el complemento "Impuestos Locales" con claves que no pertenecen +# a c_Impuesto. No se siembran registros locales inventados. + +TAXES: list[dict] = [ + {"code": "001", "description": "ISR", "is_withholding": True, "is_transferred": False, "is_local": False, "is_active": True}, + {"code": "002", "description": "IVA", "is_withholding": True, "is_transferred": True, "is_local": False, "is_active": True}, + {"code": "003", "description": "IEPS", "is_withholding": True, "is_transferred": True, "is_local": False, "is_active": True}, +] + + +# --------------------------------------------------------------------------- +# c_FormaPago +# --------------------------------------------------------------------------- + +PAYMENT_FORMS: list[dict] = [ + {"code": code, "description": description, "is_active": True} + for code, description in [ + ("01", "Efectivo"), + ("02", "Cheque nominativo"), + ("03", "Transferencia electrónica de fondos"), + ("04", "Tarjeta de crédito"), + ("05", "Monedero electrónico"), + ("06", "Dinero electrónico"), + ("08", "Vales de despensa"), + ("12", "Dación en pago"), + ("13", "Pago por subrogación"), + ("14", "Pago por consignación"), + ("15", "Condonación"), + ("17", "Compensación"), + ("23", "Novación"), + ("24", "Confusión"), + ("25", "Remisión de deuda"), + ("26", "Prescripción o caducidad"), + ("27", "A satisfacción del acreedor"), + ("28", "Tarjeta de débito"), + ("29", "Tarjeta de servicios"), + ("30", "Aplicación de anticipos"), + ("31", "Intermediario pagos"), + ("99", "Por definir"), + ] +] + + +# --------------------------------------------------------------------------- +# c_TipoDeComprobante +# --------------------------------------------------------------------------- + +VOUCHER_TYPES: list[dict] = [ + {"code": code, "description": description, "is_active": True} + for code, description in [ + ("I", "Ingreso"), + ("E", "Egreso"), + ("T", "Traslado"), + ("N", "Nómina"), + ("P", "Pago"), + ] +] + + +# --------------------------------------------------------------------------- +# c_MetodoPago +# --------------------------------------------------------------------------- + +PAYMENT_METHODS: list[dict] = [ + {"code": "PUE", "description": "Pago en una sola exhibición", "is_active": True}, + {"code": "PPD", "description": "Pago en parcialidades o diferido", "is_active": True}, +] + + +# --------------------------------------------------------------------------- +# c_ObjetoImp +# --------------------------------------------------------------------------- +# Versiones posteriores del catálogo incorporan las claves 05–07; no se siembran +# hasta que el área Fiscal confirme la versión vigente (ver PENDIENTE DECISIÓN). + +TAX_OBJECTS: list[dict] = [ + {"code": "01", "description": "No objeto de impuesto", "is_active": True}, + {"code": "02", "description": "Sí objeto de impuesto", "is_active": True}, + {"code": "03", "description": "Sí objeto del impuesto y no obligado al desglose", "is_active": True}, + {"code": "04", "description": "Sí objeto del impuesto y no causa impuesto", "is_active": True}, +] + + +# --------------------------------------------------------------------------- +# c_ClaveUnidad — subset operativo +# --------------------------------------------------------------------------- +# description queda en NULL: es la nota larga del catálogo, que aquí no aporta. + +UNITS_OF_MEASURE: list[dict] = [ + {"code": code, "name": name, "symbol": symbol, "description": None, "is_active": True} + for code, name, symbol in [ + ("H87", "Pieza", "pz"), + ("E48", "Unidad de servicio", None), + ("ACT", "Actividad", None), + ("C62", "Uno", None), + ("KGM", "Kilogramo", "kg"), + ("TNE", "Tonelada métrica", "t"), + ("GRM", "Gramo", "g"), + ("LTR", "Litro", "l"), + ("MTR", "Metro", "m"), + ("MTK", "Metro cuadrado", "m²"), + ("MTQ", "Metro cúbico", "m³"), + ("KMT", "Kilómetro", "km"), + ("CMT", "Centímetro", "cm"), + ("DAY", "Día", "d"), + ("HUR", "Hora", "h"), + ("MON", "Mes", None), + ("XBX", "Caja", None), + ("XPK", "Paquete", None), + ("XPX", "Paleta / tarima", None), + ("XLT", "Lote", None), + ("E51", "Trabajo", None), + ] +] + + +# --------------------------------------------------------------------------- +# c_ClaveProdServ — subset de logística +# --------------------------------------------------------------------------- +# Subset inicial de c_ClaveProdServ para agente de carga — pendiente validación con +# área Fiscal antes de producción. El catálogo completo son ~52,000 claves; aquí solo +# se siembran las del giro. Si falta una clave para un caso de uso, se documenta como +# PENDIENTE DECISIÓN: no se deduce ni se inventa. + +PRODUCTS_SERVICES: list[dict] = [ + {"code": code, "description": description, "is_active": True} + for code, description in [ + ("78101500", "Transporte de carga por carretera"), + ("78101600", "Transporte de carga marítimo"), + ("78101700", "Transporte de carga por ferrocarril"), + ("78101800", "Transporte de carga aérea"), + ("78102200", "Servicios postales de paqueteo y courrier"), + ("78121600", "Embalaje"), + ("78131600", "Almacenaje"), + ("78141500", "Servicios de planificación logística"), + ("78141600", "Servicios de expedición de fletes"), + ("84131500", "Seguros de vida, salud y accidentes / seguros de carga"), + ("80101500", "Servicios de consultoría de negocios y administración corporativa"), + ] +] + + +# Orden estable de sincronización: (tabla, filas). +CATALOGS: list[tuple[sa.Table, list[dict]]] = [ + (tax_regimes_table, TAX_REGIMES), + (taxes_table, TAXES), + (payment_forms_table, PAYMENT_FORMS), + (units_of_measure_table, UNITS_OF_MEASURE), + (products_services_table, PRODUCTS_SERVICES), + (voucher_types_table, VOUCHER_TYPES), + (payment_methods_table, PAYMENT_METHODS), + (tax_objects_table, TAX_OBJECTS), +] + + +def sync_catalogs(connection) -> dict[str, int]: + """Sincroniza los catálogos del SAT contra la base, de forma idempotente. + + Inserta las claves que faltan y actualiza descripción y banderas de las que ya + existen. **Nunca borra**: una clave retirada por el SAT se desactiva a mano para + no romper los CFDI históricos que la referencian. + + Devuelve un resumen ``{"sat.tabla": filas_insertadas}`` útil para la bitácora de + la migración. + + Se usa contra el ``connection`` que da ``op.get_bind()`` en Alembic, o contra la + conexión de una sesión en pruebas. + """ + inserted: dict[str, int] = {} + for table, rows in CATALOGS: + key = f"sat.{table.name}" + inserted[key] = 0 + for row in rows: + existing = connection.execute( + sa.select(table.c.id).where(table.c.code == row["code"]) + ).scalar() + values = {k: v for k, v in row.items() if k != "code"} + if existing is None: + connection.execute(table.insert().values(code=row["code"], **values)) + inserted[key] += 1 + else: + connection.execute( + table.update().where(table.c.id == existing).values(**values) + ) + return inserted diff --git a/backend/api/v1/modules/fin/catalogs/service.py b/backend/api/v1/modules/fin/catalogs/service.py new file mode 100644 index 0000000..6393fe0 --- /dev/null +++ b/backend/api/v1/modules/fin/catalogs/service.py @@ -0,0 +1,101 @@ +"""Consultas de los catálogos del SAT. + +Son globales (sin tenant_id / company_id) y de solo lectura: aquí no hay altas, +cambios ni bajas, únicamente búsqueda para llenar los selectores de captura. +""" + +from sqlalchemy import or_ +from sqlalchemy.orm import Session + +from .models import ( + PaymentForm, + PaymentMethod, + ProductService, + Tax, + TaxObject, + TaxRegime, + UnitOfMeasure, + VoucherType, +) + +# Catálogos que además del código y la descripción buscan por nombre corto. +_SEARCHABLE_EXTRA_FIELDS = {UnitOfMeasure: ("name",)} + + +def search_catalog( + db: Session, + model, + search: str | None = None, + active_only: bool = True, + limit: int | None = None, +) -> list: + """Devuelve las claves de un catálogo, filtradas por texto libre. + + ``search`` compara contra la clave o la descripción sin distinguir mayúsculas. + """ + q = db.query(model) + if active_only: + q = q.filter(model.is_active.is_(True)) + if search: + term = f"%{search.strip()}%" + fields = [model.code, model.description] + for extra in _SEARCHABLE_EXTRA_FIELDS.get(model, ()): + fields.append(getattr(model, extra)) + q = q.filter(or_(*[f.ilike(term) for f in fields])) + q = q.order_by(model.code.asc()) + if limit is not None: + q = q.limit(limit) + return q.all() + + +def get_tax_regimes( + db: Session, + search: str | None = None, + active_only: bool = True, + person_type: str | None = None, +) -> list[TaxRegime]: + """``c_RegimenFiscal``, opcionalmente acotado al tipo de persona. + + ``person_type='fisica'`` deja solo los regímenes que puede usar una persona + física; ``'moral'``, los de persona moral. + """ + q = db.query(TaxRegime) + if active_only: + q = q.filter(TaxRegime.is_active.is_(True)) + if search: + term = f"%{search.strip()}%" + q = q.filter(or_(TaxRegime.code.ilike(term), TaxRegime.description.ilike(term))) + if person_type == "fisica": + q = q.filter(TaxRegime.applies_to_individual.is_(True)) + elif person_type == "moral": + q = q.filter(TaxRegime.applies_to_legal_entity.is_(True)) + return q.order_by(TaxRegime.code.asc()).all() + + +def get_taxes(db: Session, search=None, active_only=True) -> list[Tax]: + return search_catalog(db, Tax, search, active_only) + + +def get_payment_forms(db: Session, search=None, active_only=True) -> list[PaymentForm]: + return search_catalog(db, PaymentForm, search, active_only) + + +def get_units_of_measure(db: Session, search=None, active_only=True) -> list[UnitOfMeasure]: + return search_catalog(db, UnitOfMeasure, search, active_only) + + +def get_products_services(db: Session, search=None, active_only=True, limit=50) -> list[ProductService]: + """``c_ClaveProdServ``. Va paginado porque alimenta un autocompletado.""" + return search_catalog(db, ProductService, search, active_only, limit=limit) + + +def get_voucher_types(db: Session, search=None, active_only=True) -> list[VoucherType]: + return search_catalog(db, VoucherType, search, active_only) + + +def get_payment_methods(db: Session, search=None, active_only=True) -> list[PaymentMethod]: + return search_catalog(db, PaymentMethod, search, active_only) + + +def get_tax_objects(db: Session, search=None, active_only=True) -> list[TaxObject]: + return search_catalog(db, TaxObject, search, active_only)