backend/api/v1/modules/fin/concepts/__init__.py

@@ -0,0 +1 @@
+"""Catálogo de conceptos de facturación por empresa."""

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

@@ -0,0 +1,55 @@
+"""Esquemas del catálogo de conceptos de facturación."""
+
+from datetime import datetime
+from decimal import Decimal
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from ..catalogs.dto import ProductServiceResponse, TaxObjectResponse, UnitOfMeasureResponse
+
+
+class ConceptBase(BaseModel):
+    code: str = Field(..., min_length=1, max_length=40, description="Clave interna del concepto")
+    description: str = Field(..., min_length=1, max_length=500)
+    product_service_id: int = Field(..., description="Clave ProdServ del SAT (1:1 por empresa)")
+    unit_of_measure_id: int | None = None
+    tax_object_id: int | None = None
+    unit_price: Decimal | None = Field(None, ge=0, max_digits=14, decimal_places=2)
+    currency: str = Field("MXN", min_length=3, max_length=3)
+    is_active: bool = True
+    notes: str | None = None
+
+
+class ConceptCreate(ConceptBase):
+    pass
+
+
+class ConceptUpdate(BaseModel):
+    """Actualización parcial: solo se tocan los campos enviados."""
+
+    code: str | None = Field(None, min_length=1, max_length=40)
+    description: str | None = Field(None, min_length=1, max_length=500)
+    product_service_id: int | None = None
+    unit_of_measure_id: int | None = None
+    tax_object_id: int | None = None
+    unit_price: Decimal | None = Field(None, ge=0, max_digits=14, decimal_places=2)
+    currency: str | None = Field(None, min_length=3, max_length=3)
+    is_active: bool | None = None
+    notes: str | None = None
+
+
+class ConceptResponse(ConceptBase):
+    """Incluye los objetos del catálogo del SAT ya resueltos, para evitar N+1 en la UI."""
+
+    model_config = ConfigDict(from_attributes=True)
+
+    id: int
+    tenant_id: int
+    company_id: int
+    product_service: ProductServiceResponse | None = None
+    unit_of_measure: UnitOfMeasureResponse | None = None
+    tax_object: TaxObjectResponse | None = None
+    created_by: str | None = None
+    updated_by: str | None = None
+    created_at: datetime
+    updated_at: datetime

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

@@ -0,0 +1,67 @@
+"""Catálogo de conceptos de facturación — ``fin.concepts``.
+
+A diferencia de los catálogos del SAT, este es **propio de cada empresa**: cada
+concepto que la empresa factura (flete internacional, despacho, almacenaje…) se
+registra una vez y queda amarrado a la clave de producto/servicio del SAT que le
+corresponde.
+
+La relación con ``sat.products_services`` es **1:1 por empresa**: si dos conceptos
+compartieran la misma clave ProdServ, al timbrar no habría forma de saber cuál
+descripción corresponde a la clave, así que la unicidad se garantiza por índice y se
+valida además en el service para devolver un 409 con mensaje entendible.
+"""
+
+from sqlalchemy import Boolean, ForeignKey, Index, Integer, Numeric, String, Text, text
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from api.v1.common.base_models import TenantScopedMixin, TimestampMixin
+from core.database import Base
+
+from ..catalogs.models import ProductService, TaxObject, UnitOfMeasure  # noqa: F401  (resuelve las relaciones)
+
+# Los índices son parciales (``WHERE deleted_at IS NULL``): un concepto dado de baja
+# lógica libera su clave y su código para uno nuevo.
+_ALIVE = text("deleted_at IS NULL")
+
+
+class Concept(Base, TenantScopedMixin, TimestampMixin):
+    """Concepto facturable de una empresa, ligado a una clave ProdServ del SAT."""
+
+    __tablename__ = "concepts"
+    __table_args__ = (
+        Index(
+            "uq_fin_concepts_code",
+            "tenant_id", "company_id", "code",
+            unique=True, postgresql_where=_ALIVE, sqlite_where=_ALIVE,
+        ),
+        Index(
+            "uq_fin_concepts_product_service",
+            "tenant_id", "company_id", "product_service_id",
+            unique=True, postgresql_where=_ALIVE, sqlite_where=_ALIVE,
+        ),
+        {"schema": "fin"},
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
+    code: Mapped[str] = mapped_column(String(40), nullable=False)  # clave interna del concepto
+    description: Mapped[str] = mapped_column(String(500), nullable=False)
+    product_service_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("sat.products_services.id"), nullable=False, index=True
+    )
+    unit_of_measure_id: Mapped[int | None] = mapped_column(
+        Integer, ForeignKey("sat.units_of_measure.id"), nullable=True
+    )
+    tax_object_id: Mapped[int | None] = mapped_column(
+        Integer, ForeignKey("sat.tax_objects.id"), nullable=True
+    )
+    unit_price: Mapped[float | None] = mapped_column(Numeric(14, 2), nullable=True)
+    currency: Mapped[str] = mapped_column(String(3), nullable=False, server_default=text("'MXN'"))
+    is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, server_default=text("true"))
+    notes: Mapped[str | None] = mapped_column(Text, nullable=True)
+    created_by: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    updated_by: Mapped[str | None] = mapped_column(String(64), nullable=True)
+
+    # Cargadas con selectinload para que el listado no dispare N+1 consultas.
+    product_service: Mapped["ProductService"] = relationship("ProductService", lazy="selectin")
+    unit_of_measure: Mapped["UnitOfMeasure | None"] = relationship("UnitOfMeasure", lazy="selectin")
+    tax_object: Mapped["TaxObject | None"] = relationship("TaxObject", lazy="selectin")

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

@@ -0,0 +1,96 @@
+"""Endpoints del catálogo de conceptos de facturación (CRUD por empresa)."""
+
+from fastapi import APIRouter, Depends, Query, status
+from sqlalchemy.orm import Session
+
+from api.v1.modules.core.permissions.dependencies import PermissionChecker
+from core.database import get_core_db
+from core.security import get_current_user
+
+from . import service
+from .dto import ConceptCreate, ConceptResponse, ConceptUpdate
+
+router = APIRouter()
+
+
+def _uid(current_user: dict) -> str | None:
+    return current_user.get("sub") or current_user.get("id")
+
+
+@router.get(
+    "/concepts",
+    response_model=list[ConceptResponse],
+    dependencies=[Depends(PermissionChecker(["fin.concept.view"]))],
+)
+def list_concepts(
+    company_id: int = Query(..., description="Company ID"),
+    search: str | None = Query(None, description="Búsqueda por clave o descripción"),
+    active_only: bool | None = Query(None, description="Filtra por conceptos activos o inactivos"),
+    product_service_id: int | None = Query(None, description="Filtra por clave ProdServ del SAT"),
+    current_user: dict = Depends(get_current_user),
+    db: Session = Depends(get_core_db),
+):
+    return service.get_concepts(
+        db, current_user["tenant_id"], company_id, search, active_only, product_service_id
+    )
+
+
+@router.get(
+    "/concepts/{concept_id}",
+    response_model=ConceptResponse,
+    dependencies=[Depends(PermissionChecker(["fin.concept.view"]))],
+)
+def get_concept(
+    concept_id: int,
+    company_id: int = Query(..., description="Company ID"),
+    current_user: dict = Depends(get_current_user),
+    db: Session = Depends(get_core_db),
+):
+    return service.get_concept(db, concept_id, current_user["tenant_id"], company_id)
+
+
+@router.post(
+    "/concepts",
+    response_model=ConceptResponse,
+    status_code=status.HTTP_201_CREATED,
+    dependencies=[Depends(PermissionChecker(["fin.concept.create"]))],
+)
+def create_concept(
+    payload: ConceptCreate,
+    company_id: int = Query(..., description="Company ID"),
+    current_user: dict = Depends(get_current_user),
+    db: Session = Depends(get_core_db),
+):
+    return service.create_concept(db, payload, current_user["tenant_id"], company_id, _uid(current_user))
+
+
+@router.patch(
+    "/concepts/{concept_id}",
+    response_model=ConceptResponse,
+    dependencies=[Depends(PermissionChecker(["fin.concept.edit"]))],
+)
+def update_concept(
+    concept_id: int,
+    payload: ConceptUpdate,
+    company_id: int = Query(..., description="Company ID"),
+    current_user: dict = Depends(get_current_user),
+    db: Session = Depends(get_core_db),
+):
+    return service.update_concept(
+        db, concept_id, payload, current_user["tenant_id"], company_id, _uid(current_user)
+    )
+
+
+@router.delete(
+    "/concepts/{concept_id}",
+    status_code=status.HTTP_204_NO_CONTENT,
+    dependencies=[Depends(PermissionChecker(["fin.concept.delete"]))],
+)
+def delete_concept(
+    concept_id: int,
+    company_id: int = Query(..., description="Company ID"),
+    current_user: dict = Depends(get_current_user),
+    db: Session = Depends(get_core_db),
+):
+    """Baja lógica del concepto (``deleted_at``)."""
+    service.delete_concept(db, concept_id, current_user["tenant_id"], company_id)

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

@@ -0,0 +1,146 @@
+"""Lógica del catálogo de conceptos de facturación.
+
+Todas las consultas filtran por ``tenant_id``, ``company_id`` y ``deleted_at IS NULL``:
+el catálogo es privado de cada empresa dentro de cada tenant.
+"""
+
+from datetime import datetime, timezone
+
+from fastapi import HTTPException, status
+from sqlalchemy import or_
+from sqlalchemy.orm import Session
+
+from ..catalogs.models import ProductService, TaxObject, UnitOfMeasure
+from .dto import ConceptCreate, ConceptUpdate
+from .models import Concept
+
+
+def _check_sat_refs(db: Session, data: dict) -> None:
+    """Verifica que las claves del SAT referidas existan antes de guardar."""
+    for field, model, msg in [
+        ("product_service_id", ProductService, "La clave de producto/servicio del SAT no existe"),
+        ("unit_of_measure_id", UnitOfMeasure, "La unidad de medida del SAT no existe"),
+        ("tax_object_id", TaxObject, "El objeto de impuesto del SAT no existe"),
+    ]:
+        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 _check_unique(
+    db: Session,
+    tenant_id: int,
+    company_id: int,
+    code: str | None,
+    product_service_id: int | None,
+    exclude_id: int | None = None,
+) -> None:
+    """Aplica en el service las mismas reglas que los índices únicos parciales.
+
+    Sin esto el conflicto llegaría al cliente como un IntegrityError crudo; aquí se
+    traduce a un 409 con mensaje en español.
+    """
+    base = db.query(Concept).filter(
+        Concept.tenant_id == tenant_id,
+        Concept.company_id == company_id,
+        Concept.deleted_at.is_(None),
+    )
+    if exclude_id is not None:
+        base = base.filter(Concept.id != exclude_id)
+
+    if code is not None and base.filter(Concept.code == code).first() is not None:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail=f"Ya existe un concepto con la clave '{code}' en esta empresa",
+        )
+    # Regla 1:1 — una clave ProdServ no puede repetirse entre conceptos de la empresa.
+    if product_service_id is not None and base.filter(
+        Concept.product_service_id == product_service_id
+    ).first() is not None:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="La clave de producto/servicio del SAT ya está asignada a otro concepto de esta empresa",
+        )
+
+
+def get_concepts(
+    db: Session,
+    tenant_id: int,
+    company_id: int,
+    search: str | None = None,
+    active_only: bool | None = None,
+    product_service_id: int | None = None,
+) -> list[Concept]:
+    q = db.query(Concept).filter(
+        Concept.tenant_id == tenant_id,
+        Concept.company_id == company_id,
+        Concept.deleted_at.is_(None),
+    )
+    if active_only is not None:
+        q = q.filter(Concept.is_active.is_(active_only))
+    if product_service_id is not None:
+        q = q.filter(Concept.product_service_id == product_service_id)
+    if search:
+        term = f"%{search.strip()}%"
+        q = q.filter(or_(Concept.code.ilike(term), Concept.description.ilike(term)))
+    return q.order_by(Concept.code.asc()).all()
+
+
+def get_concept(db: Session, concept_id: int, tenant_id: int, company_id: int) -> Concept:
+    obj = db.query(Concept).filter(
+        Concept.id == concept_id,
+        Concept.tenant_id == tenant_id,
+        Concept.company_id == company_id,
+        Concept.deleted_at.is_(None),
+    ).first()
+    if not obj:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Concepto no encontrado")
+    return obj
+
+
+def create_concept(
+    db: Session, payload: ConceptCreate, tenant_id: int, company_id: int, user_id: str | None = None
+) -> Concept:
+    data = payload.model_dump()
+    _check_sat_refs(db, data)
+    _check_unique(db, tenant_id, company_id, data["code"], data["product_service_id"])
+    obj = Concept(**data, tenant_id=tenant_id, company_id=company_id, created_by=user_id, updated_by=user_id)
+    db.add(obj)
+    db.commit()
+    db.refresh(obj)
+    return obj
+
+
+def update_concept(
+    db: Session,
+    concept_id: int,
+    payload: ConceptUpdate,
+    tenant_id: int,
+    company_id: int,
+    user_id: str | None = None,
+) -> Concept:
+    obj = get_concept(db, concept_id, tenant_id, company_id)
+    data = payload.model_dump(exclude_unset=True)
+    _check_sat_refs(db, data)
+    _check_unique(
+        db,
+        tenant_id,
+        company_id,
+        data.get("code"),
+        data.get("product_service_id"),
+        exclude_id=obj.id,
+    )
+    for field, value in data.items():
+        setattr(obj, field, value)
+    obj.updated_by = user_id
+    db.commit()
+    db.refresh(obj)
+    return obj
+
+
+def delete_concept(db: Session, concept_id: int, tenant_id: int, company_id: int) -> None:
+    """Baja lógica: libera la clave ProdServ y el código para un concepto nuevo."""
+    obj = get_concept(db, concept_id, tenant_id, company_id)
+    obj.deleted_at = datetime.now(timezone.utc)
+    db.commit()