T2026-08-047 feat(fin): catálogos SAT, conceptos de facturación y datos fiscales del emisor #5
1
backend/api/v1/modules/fin/catalogs/__init__.py
Normal file
1
backend/api/v1/modules/fin/catalogs/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
"""Catálogos oficiales del SAT (schema ``sat``): globales y de solo lectura."""
|
||||
57
backend/api/v1/modules/fin/catalogs/dto.py
Normal file
57
backend/api/v1/modules/fin/catalogs/dto.py
Normal file
@@ -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``."""
|
||||
130
backend/api/v1/modules/fin/catalogs/models.py
Normal file
130
backend/api/v1/modules/fin/catalogs/models.py
Normal file
@@ -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)
|
||||
126
backend/api/v1/modules/fin/catalogs/routes.py
Normal file
126
backend/api/v1/modules/fin/catalogs/routes.py
Normal file
@@ -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)
|
||||
279
backend/api/v1/modules/fin/catalogs/seed_data.py
Normal file
279
backend/api/v1/modules/fin/catalogs/seed_data.py
Normal file
@@ -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
|
||||
101
backend/api/v1/modules/fin/catalogs/service.py
Normal file
101
backend/api/v1/modules/fin/catalogs/service.py
Normal file
@@ -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)
|
||||
Reference in New Issue
Block a user