feat(crm): catálogos Clientes/Prospectos (enriquecido) y Proveedores + direcciones/documentos

Alinea el dominio al spec de catálogos:
- accounts (Clientes/Prospectos): tipo registro/persona, CURP, clasificación
  comercial, bloque fiscal (régimen, CFDI, pago, crédito), auditoría, notas internas
- suppliers (Proveedores): clasificación múltiple, cobertura, países/puertos/
  aeropuertos/aduanas (JSON), fiscal
- addresses y documents: tablas compartidas con FK a cliente o proveedor
- contacts enriquecidos (extensión, whatsapp, área, flags "recibe…", supplier_id)
- migración a7b8c9d0e1f2 (ALTER + CREATE) con downgrade completo
- permisos supplier/address/document; seed de datos actualizado
- 39 tests pytest en verde

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Aduanasoft
2026-07-14 16:19:13 -06:00
parent 2ed6247f1e
commit b12af1a561
31 changed files with 1599 additions and 101 deletions

View File

@@ -0,0 +1,96 @@
from datetime import datetime, timezone
from fastapi import HTTPException, status
from sqlalchemy.orm import Session
from ..accounts.models import Account
from ..suppliers.models import Supplier
from .dto import AddressCreate, AddressUpdate
from .models import Address
def _validate_owner(db: Session, account_id: int | None, supplier_id: int | None, tenant_id: int, company_id: int) -> None:
"""Una dirección debe pertenecer a exactamente un cliente o proveedor existente."""
if (account_id is None) == (supplier_id is None):
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="La dirección debe asociarse a un cliente O a un proveedor",
)
model, _id, msg = (
(Account, account_id, "El cliente asociado no existe")
if account_id is not None
else (Supplier, supplier_id, "El proveedor asociado no existe")
)
exists = (
db.query(model.id)
.filter(
model.id == _id,
model.tenant_id == tenant_id,
model.company_id == company_id,
model.deleted_at.is_(None),
)
.first()
)
if not exists:
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=msg)
def get_addresses(
db: Session,
tenant_id: int,
company_id: int,
account_id: int | None = None,
supplier_id: int | None = None,
) -> list[Address]:
query = db.query(Address).filter(
Address.tenant_id == tenant_id,
Address.company_id == company_id,
Address.deleted_at.is_(None),
)
if account_id is not None:
query = query.filter(Address.account_id == account_id)
if supplier_id is not None:
query = query.filter(Address.supplier_id == supplier_id)
return query.order_by(Address.is_primary.desc(), Address.id.asc()).all()
def get_address(db: Session, address_id: int, tenant_id: int, company_id: int) -> Address:
address = (
db.query(Address)
.filter(
Address.id == address_id,
Address.tenant_id == tenant_id,
Address.company_id == company_id,
Address.deleted_at.is_(None),
)
.first()
)
if not address:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Dirección no encontrada")
return address
def create_address(db: Session, payload: AddressCreate, tenant_id: int, company_id: int) -> Address:
_validate_owner(db, payload.account_id, payload.supplier_id, tenant_id, company_id)
address = Address(**payload.model_dump(), tenant_id=tenant_id, company_id=company_id)
db.add(address)
db.commit()
db.refresh(address)
return address
def update_address(
db: Session, address_id: int, payload: AddressUpdate, tenant_id: int, company_id: int
) -> Address:
address = get_address(db, address_id, tenant_id, company_id)
for field, value in payload.model_dump(exclude_unset=True).items():
setattr(address, field, value)
db.commit()
db.refresh(address)
return address
def delete_address(db: Session, address_id: int, tenant_id: int, company_id: int) -> None:
address = get_address(db, address_id, tenant_id, company_id)
address.deleted_at = datetime.now(timezone.utc)
db.commit()