f1e6fba75d
- Modalidad AÉREO (4ª opción en "¿Cómo desea cotizar?"): en la solicitud muestra la sección aérea con el cálculo en vivo P/Vol = (L×A×H cm × bultos)/6000 y el peso a cobrar = max(peso bruto, P/Vol). Fija el transporte en aéreo. - Utilidad compartida crm/common/pricing.py (air_volumetric_kg / air_chargeable_kg, factor internacional 6000). - Motor de costeo (rates): la rama aérea usa el P/Vol por dimensiones si vienen (CostRequest ahora acepta length/width/height_cm); respaldo m³×167 cuando no. - Cotizador: captura por dimensiones (L×A×H + bultos) en modo aéreo y muestra el P/Vol. - Solicitud→Cotización: si es AÉREO, siembra el concepto de flete con cantidad = peso a cobrar (P/Vol) para capturar la tarifa por kg. - Pruebas: test_pricing (ejemplo del doc → 720; max bruto/volumétrico) + cotización aérea desde solicitud. Suite en verde (108). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
401 lines
15 KiB
Python
401 lines
15 KiB
Python
from datetime import date, datetime, timezone
|
|
from decimal import Decimal
|
|
|
|
from fastapi import HTTPException, status
|
|
from sqlalchemy import func
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ..accounts.models import Account
|
|
from ..catalogs.models import CatalogItem
|
|
from ..common.folios import next_folio
|
|
from ..common.pricing import air_chargeable_kg
|
|
from ..service_requests.models import RateRequest, ServiceRequest
|
|
from ..suppliers.models import Supplier
|
|
from .dto import QuoteCreate, QuoteItemCreate, QuoteItemUpdate, QuoteUpdate
|
|
from .models import Quote, QuoteItem
|
|
|
|
|
|
def _exists(db: Session, model, _id: int | None, tenant_id: int, company_id: int) -> bool:
|
|
if _id is None:
|
|
return True
|
|
return (
|
|
db.query(model.id)
|
|
.filter(
|
|
model.id == _id,
|
|
model.tenant_id == tenant_id,
|
|
model.company_id == company_id,
|
|
model.deleted_at.is_(None),
|
|
)
|
|
.first()
|
|
is not None
|
|
)
|
|
|
|
|
|
def _validate_refs(db: Session, data: dict, tenant_id: int, company_id: int) -> None:
|
|
if not _exists(db, Account, data.get("account_id"), tenant_id, company_id):
|
|
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="El cliente asociado no existe")
|
|
if not _exists(db, ServiceRequest, data.get("service_request_id"), tenant_id, company_id):
|
|
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="La solicitud asociada no existe")
|
|
|
|
|
|
def _recompute_totals(db: Session, quote: Quote) -> None:
|
|
"""Recalcula total_cost/total_sale a partir de los conceptos vigentes."""
|
|
cost, sale = (
|
|
db.query(
|
|
func.coalesce(func.sum(QuoteItem.quantity * QuoteItem.unit_cost), 0),
|
|
func.coalesce(func.sum(QuoteItem.quantity * QuoteItem.unit_sale), 0),
|
|
)
|
|
.filter(QuoteItem.quote_id == quote.id, QuoteItem.deleted_at.is_(None))
|
|
.one()
|
|
)
|
|
quote.total_cost = Decimal(cost or 0)
|
|
quote.total_sale = Decimal(sale or 0)
|
|
|
|
|
|
# ----- Quotes -----
|
|
|
|
def get_quotes(
|
|
db: Session,
|
|
tenant_id: int,
|
|
company_id: int,
|
|
search: str | None = None,
|
|
quote_status: str | None = None,
|
|
account_id: int | None = None,
|
|
) -> list[Quote]:
|
|
query = db.query(Quote).filter(
|
|
Quote.tenant_id == tenant_id,
|
|
Quote.company_id == company_id,
|
|
Quote.deleted_at.is_(None),
|
|
)
|
|
if quote_status:
|
|
query = query.filter(Quote.status == quote_status)
|
|
if account_id is not None:
|
|
query = query.filter(Quote.account_id == account_id)
|
|
if search:
|
|
query = query.filter(Quote.reference.ilike(f"%{search}%"))
|
|
return query.order_by(Quote.created_at.desc()).all()
|
|
|
|
|
|
def get_quote(db: Session, quote_id: int, tenant_id: int, company_id: int) -> Quote:
|
|
obj = (
|
|
db.query(Quote)
|
|
.filter(
|
|
Quote.id == quote_id,
|
|
Quote.tenant_id == tenant_id,
|
|
Quote.company_id == company_id,
|
|
Quote.deleted_at.is_(None),
|
|
)
|
|
.first()
|
|
)
|
|
if not obj:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Cotización no encontrada")
|
|
return obj
|
|
|
|
|
|
def _sr_direction(db: Session, service_request_id: int | None) -> str | None:
|
|
"""Dirección impo/expo heredada de la solicitud asociada (para el folio)."""
|
|
if not service_request_id:
|
|
return None
|
|
sr = db.query(ServiceRequest).filter(ServiceRequest.id == service_request_id).first()
|
|
return sr.operation_type if sr else None
|
|
|
|
|
|
def create_quote(
|
|
db: Session, payload: QuoteCreate, tenant_id: int, company_id: int, user_id: str | None = None
|
|
) -> Quote:
|
|
data = payload.model_dump()
|
|
_validate_refs(db, data, tenant_id, company_id)
|
|
obj = Quote(**data, tenant_id=tenant_id, company_id=company_id, created_by=user_id, updated_by=user_id)
|
|
# Fecha de la cotización: por defecto hoy si no se capturó
|
|
if obj.issue_date is None:
|
|
obj.issue_date = date.today()
|
|
# Folio C... auto-generado (mensual), con la dirección heredada de la solicitud
|
|
if not obj.reference:
|
|
obj.reference = next_folio(db, tenant_id, company_id, "C", _sr_direction(db, obj.service_request_id))
|
|
db.add(obj)
|
|
db.commit()
|
|
db.refresh(obj)
|
|
return obj
|
|
|
|
|
|
def create_quotes_from_service_request(
|
|
db: Session, service_request_id: int, tenant_id: int, company_id: int, user_id: str | None = None
|
|
) -> list[Quote]:
|
|
"""Genera cotización(es) a partir de una solicitud de servicio.
|
|
|
|
Si la solicitud es "Ambas" (FCL y LCL), genera **dos** cotizaciones (una por
|
|
variante) para comparar. Cada cotización toma su propio folio C... y hereda la
|
|
dirección impo/expo de la solicitud. Los conceptos se siembran desde las
|
|
solicitudes de tarifa (RateRequest) capturadas en la solicitud.
|
|
"""
|
|
sr = (
|
|
db.query(ServiceRequest)
|
|
.filter(
|
|
ServiceRequest.id == service_request_id,
|
|
ServiceRequest.tenant_id == tenant_id,
|
|
ServiceRequest.company_id == company_id,
|
|
ServiceRequest.deleted_at.is_(None),
|
|
)
|
|
.first()
|
|
)
|
|
if not sr:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Solicitud no encontrada")
|
|
|
|
variants = ["FCL", "LCL"] if (sr.load_type or "").upper() == "AMBAS" else [sr.load_type or None]
|
|
rate_requests = (
|
|
db.query(RateRequest)
|
|
.filter(
|
|
RateRequest.service_request_id == sr.id,
|
|
RateRequest.tenant_id == tenant_id,
|
|
RateRequest.company_id == company_id,
|
|
RateRequest.deleted_at.is_(None),
|
|
)
|
|
.all()
|
|
)
|
|
# Etiquetas legibles de los servicios adicionales (global + tenant) para los conceptos
|
|
service_labels = {
|
|
code: label
|
|
for code, label in db.query(CatalogItem.code, CatalogItem.label).filter(
|
|
CatalogItem.catalog == "servicio_adicional"
|
|
)
|
|
}
|
|
service_costs = sr.additional_service_costs or {}
|
|
|
|
created: list[Quote] = []
|
|
for variant in variants:
|
|
quote = Quote(
|
|
account_id=sr.account_id,
|
|
service_request_id=sr.id,
|
|
currency=sr.currency or "USD",
|
|
load_type=variant,
|
|
status="borrador",
|
|
issue_date=date.today(),
|
|
notes=sr.client_notes or sr.notes,
|
|
owner_user_id=sr.owner_user_id,
|
|
reference=next_folio(db, tenant_id, company_id, "C", sr.operation_type),
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
created_by=user_id,
|
|
updated_by=user_id,
|
|
)
|
|
db.add(quote)
|
|
db.flush()
|
|
for rr in rate_requests:
|
|
amount = rr.rate_amount if rr.rate_amount is not None else Decimal(0)
|
|
db.add(QuoteItem(
|
|
quote_id=quote.id, concept=rr.concept, description=rr.description,
|
|
supplier_id=rr.supplier_id, quantity=Decimal(1),
|
|
unit_cost=amount, unit_sale=amount, currency=rr.currency,
|
|
tenant_id=tenant_id, company_id=company_id,
|
|
))
|
|
# Servicios adicionales marcados en la solicitud → conceptos con su costo estimado
|
|
for code in (sr.additional_services or []):
|
|
amount = Decimal(str(service_costs.get(code) or 0))
|
|
db.add(QuoteItem(
|
|
quote_id=quote.id, concept=code[:60],
|
|
description=service_labels.get(code, "Servicio adicional"),
|
|
quantity=Decimal(1), unit_cost=amount, unit_sale=amount,
|
|
currency=sr.currency, tenant_id=tenant_id, company_id=company_id,
|
|
))
|
|
# Carga aérea: concepto de flete con el peso a cobrar (P/Vol) como cantidad,
|
|
# para que el ejecutivo capture la tarifa por kg.
|
|
if (variant or "").upper() == "AEREO":
|
|
chargeable = air_chargeable_kg(
|
|
sr.weight, sr.length_cm, sr.width_cm, sr.height_cm,
|
|
sr.pallets_count or sr.pieces_count or 1,
|
|
)
|
|
db.add(QuoteItem(
|
|
quote_id=quote.id, concept="flete_internacional",
|
|
description=f"Flete aéreo — peso a cobrar {chargeable.quantize(Decimal('0.01'))} kg (P/Vol)",
|
|
quantity=chargeable, unit_cost=Decimal(0), unit_sale=Decimal(0),
|
|
currency=sr.currency, tenant_id=tenant_id, company_id=company_id,
|
|
))
|
|
db.flush()
|
|
_recompute_totals(db, quote)
|
|
created.append(quote)
|
|
|
|
db.commit()
|
|
for quote in created:
|
|
db.refresh(quote)
|
|
return created
|
|
|
|
|
|
def update_quote(
|
|
db: Session, quote_id: int, payload: QuoteUpdate, tenant_id: int, company_id: int, user_id: str | None = None
|
|
) -> Quote:
|
|
obj = get_quote(db, quote_id, tenant_id, company_id)
|
|
data = payload.model_dump(exclude_unset=True)
|
|
_validate_refs(db, data, tenant_id, company_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_quote(db: Session, quote_id: int, tenant_id: int, company_id: int) -> None:
|
|
obj = get_quote(db, quote_id, tenant_id, company_id)
|
|
obj.deleted_at = datetime.now(timezone.utc)
|
|
db.commit()
|
|
|
|
|
|
def _set_service_request_status(db: Session, quote: Quote, new_status: str) -> None:
|
|
if quote.service_request_id:
|
|
sr = db.query(ServiceRequest).filter(ServiceRequest.id == quote.service_request_id).first()
|
|
if sr:
|
|
sr.status = new_status
|
|
|
|
|
|
def send_quote(db: Session, quote_id: int, tenant_id: int, company_id: int) -> Quote:
|
|
quote = get_quote(db, quote_id, tenant_id, company_id)
|
|
quote.status = "enviada"
|
|
quote.sent_at = datetime.now(timezone.utc)
|
|
_set_service_request_status(db, quote, "cotizada")
|
|
db.commit()
|
|
db.refresh(quote)
|
|
return quote
|
|
|
|
|
|
def accept_quote(db: Session, quote_id: int, tenant_id: int, company_id: int) -> Quote:
|
|
quote = get_quote(db, quote_id, tenant_id, company_id)
|
|
quote.status = "aceptada"
|
|
quote.accepted_at = datetime.now(timezone.utc)
|
|
_set_service_request_status(db, quote, "aceptada")
|
|
db.commit()
|
|
db.refresh(quote)
|
|
return quote
|
|
|
|
|
|
def reject_quote(db: Session, quote_id: int, tenant_id: int, company_id: int) -> Quote:
|
|
quote = get_quote(db, quote_id, tenant_id, company_id)
|
|
quote.status = "rechazada"
|
|
quote.rejected_at = datetime.now(timezone.utc)
|
|
_set_service_request_status(db, quote, "rechazada")
|
|
db.commit()
|
|
db.refresh(quote)
|
|
return quote
|
|
|
|
|
|
def clone_quote(
|
|
db: Session, quote_id: int, tenant_id: int, company_id: int, user_id: str | None = None
|
|
) -> Quote:
|
|
"""Clona una cotización (y sus conceptos) como borrador para re-cotizar (R-C-12).
|
|
|
|
Si la cotización origen fue rechazada, reabre su solicitud a 'en_analisis' para
|
|
cerrar el ciclo de reintento del Diagrama 1.
|
|
"""
|
|
src = get_quote(db, quote_id, tenant_id, company_id)
|
|
new_quote = Quote(
|
|
reference=(f"{src.reference}-R" if src.reference else None),
|
|
service_request_id=src.service_request_id,
|
|
account_id=src.account_id,
|
|
currency=src.currency,
|
|
status="borrador",
|
|
valid_until=src.valid_until,
|
|
notes=src.notes,
|
|
terms=src.terms,
|
|
owner_user_id=src.owner_user_id,
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
created_by=user_id,
|
|
updated_by=user_id,
|
|
)
|
|
db.add(new_quote)
|
|
db.flush()
|
|
src_items = (
|
|
db.query(QuoteItem)
|
|
.filter(QuoteItem.quote_id == src.id, QuoteItem.deleted_at.is_(None))
|
|
.all()
|
|
)
|
|
for it in src_items:
|
|
db.add(QuoteItem(
|
|
quote_id=new_quote.id, concept=it.concept, description=it.description,
|
|
supplier_id=it.supplier_id, quantity=it.quantity, unit_cost=it.unit_cost,
|
|
unit_sale=it.unit_sale, currency=it.currency,
|
|
tenant_id=tenant_id, company_id=company_id,
|
|
))
|
|
db.flush()
|
|
_recompute_totals(db, new_quote)
|
|
# Reabre la solicitud origen para el ciclo de re-cotización
|
|
if src.service_request_id:
|
|
sr = db.query(ServiceRequest).filter(ServiceRequest.id == src.service_request_id).first()
|
|
if sr and sr.status in ("rechazada", "cotizada"):
|
|
sr.status = "en_analisis"
|
|
db.commit()
|
|
db.refresh(new_quote)
|
|
return new_quote
|
|
|
|
|
|
# ----- Quote items -----
|
|
|
|
def get_quote_items(db: Session, quote_id: int, tenant_id: int, company_id: int) -> list[QuoteItem]:
|
|
get_quote(db, quote_id, tenant_id, company_id) # valida scope
|
|
return (
|
|
db.query(QuoteItem)
|
|
.filter(
|
|
QuoteItem.quote_id == quote_id,
|
|
QuoteItem.tenant_id == tenant_id,
|
|
QuoteItem.company_id == company_id,
|
|
QuoteItem.deleted_at.is_(None),
|
|
)
|
|
.order_by(QuoteItem.id.asc())
|
|
.all()
|
|
)
|
|
|
|
|
|
def _get_item(db: Session, item_id: int, tenant_id: int, company_id: int) -> QuoteItem:
|
|
item = (
|
|
db.query(QuoteItem)
|
|
.filter(
|
|
QuoteItem.id == item_id,
|
|
QuoteItem.tenant_id == tenant_id,
|
|
QuoteItem.company_id == company_id,
|
|
QuoteItem.deleted_at.is_(None),
|
|
)
|
|
.first()
|
|
)
|
|
if not item:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Concepto no encontrado")
|
|
return item
|
|
|
|
|
|
def create_quote_item(db: Session, payload: QuoteItemCreate, tenant_id: int, company_id: int) -> QuoteItem:
|
|
quote = get_quote(db, payload.quote_id, tenant_id, company_id)
|
|
if not _exists(db, Supplier, payload.supplier_id, tenant_id, company_id):
|
|
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="El proveedor no existe")
|
|
item = QuoteItem(**payload.model_dump(), tenant_id=tenant_id, company_id=company_id)
|
|
db.add(item)
|
|
db.flush()
|
|
_recompute_totals(db, quote)
|
|
db.commit()
|
|
db.refresh(item)
|
|
return item
|
|
|
|
|
|
def update_quote_item(
|
|
db: Session, item_id: int, payload: QuoteItemUpdate, tenant_id: int, company_id: int
|
|
) -> QuoteItem:
|
|
item = _get_item(db, item_id, tenant_id, company_id)
|
|
data = payload.model_dump(exclude_unset=True)
|
|
if "supplier_id" in data and not _exists(db, Supplier, data["supplier_id"], tenant_id, company_id):
|
|
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="El proveedor no existe")
|
|
for field, value in data.items():
|
|
setattr(item, field, value)
|
|
db.flush()
|
|
quote = get_quote(db, item.quote_id, tenant_id, company_id)
|
|
_recompute_totals(db, quote)
|
|
db.commit()
|
|
db.refresh(item)
|
|
return item
|
|
|
|
|
|
def delete_quote_item(db: Session, item_id: int, tenant_id: int, company_id: int) -> None:
|
|
item = _get_item(db, item_id, tenant_id, company_id)
|
|
quote_id = item.quote_id
|
|
item.deleted_at = datetime.now(timezone.utc)
|
|
db.flush()
|
|
quote = get_quote(db, quote_id, tenant_id, company_id)
|
|
_recompute_totals(db, quote)
|
|
db.commit()
|