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>
542 lines
21 KiB
Python
542 lines
21 KiB
Python
"""Lógica del módulo Tarifario: CRUD, importación por Excel y motor de costeo."""
|
||
|
||
import io
|
||
from datetime import date
|
||
from decimal import Decimal
|
||
from typing import Any
|
||
|
||
from fastapi import HTTPException, status
|
||
from sqlalchemy import and_, or_
|
||
from sqlalchemy.orm import Session
|
||
|
||
from .dto import (
|
||
CostChargeLine,
|
||
CostOption,
|
||
CostRequest,
|
||
ImportConfirm,
|
||
ImportPreview,
|
||
ImportPreviewRow,
|
||
RateLaneCreate,
|
||
RateSheetCreate,
|
||
RateSheetUpdate,
|
||
)
|
||
from ..common.pricing import air_volumetric_kg
|
||
from .models import RateBreak, RateCharge, RateLane, RateSheet
|
||
|
||
# Factor volumétrico aéreo: 1 m³ = 167 kg (equivale a 6000 cm³/kg).
|
||
# Respaldo cuando solo se conoce el volumen en m³ (sin dimensiones cm).
|
||
AIR_VOLUMETRIC_FACTOR = Decimal("167")
|
||
|
||
|
||
# ============================================================ CRUD tarifarios
|
||
def _sheet_query(db: Session, tenant_id: int, company_id: int):
|
||
return db.query(RateSheet).filter(
|
||
RateSheet.tenant_id == tenant_id,
|
||
RateSheet.company_id == company_id,
|
||
RateSheet.deleted_at.is_(None),
|
||
)
|
||
|
||
|
||
def list_sheets(db: Session, tenant_id: int, company_id: int, mode: str | None = None,
|
||
supplier_id: int | None = None) -> list[RateSheet]:
|
||
q = _sheet_query(db, tenant_id, company_id)
|
||
if mode:
|
||
q = q.filter(RateSheet.mode == mode)
|
||
if supplier_id:
|
||
q = q.filter(RateSheet.supplier_id == supplier_id)
|
||
return q.order_by(RateSheet.created_at.desc()).all()
|
||
|
||
|
||
def lane_count(db: Session, tenant_id: int, sheet_id: int) -> int:
|
||
return (
|
||
db.query(RateLane)
|
||
.filter(RateLane.rate_sheet_id == sheet_id, RateLane.tenant_id == tenant_id,
|
||
RateLane.deleted_at.is_(None))
|
||
.count()
|
||
)
|
||
|
||
|
||
def get_sheet(db: Session, tenant_id: int, company_id: int, sheet_id: int) -> RateSheet:
|
||
sheet = _sheet_query(db, tenant_id, company_id).filter(RateSheet.id == sheet_id).first()
|
||
if not sheet:
|
||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Tarifario no encontrado")
|
||
return sheet
|
||
|
||
|
||
def create_sheet(db: Session, tenant_id: int, company_id: int, data: RateSheetCreate,
|
||
user_id: str | None) -> RateSheet:
|
||
sheet = RateSheet(
|
||
tenant_id=tenant_id, company_id=company_id,
|
||
**data.model_dump(),
|
||
created_by=user_id, updated_by=user_id,
|
||
)
|
||
db.add(sheet)
|
||
db.commit()
|
||
db.refresh(sheet)
|
||
return sheet
|
||
|
||
|
||
def update_sheet(db: Session, tenant_id: int, company_id: int, sheet_id: int,
|
||
data: RateSheetUpdate, user_id: str | None) -> RateSheet:
|
||
sheet = get_sheet(db, tenant_id, company_id, sheet_id)
|
||
for field, value in data.model_dump(exclude_unset=True).items():
|
||
setattr(sheet, field, value)
|
||
sheet.updated_by = user_id
|
||
db.commit()
|
||
db.refresh(sheet)
|
||
return sheet
|
||
|
||
|
||
def delete_sheet(db: Session, tenant_id: int, company_id: int, sheet_id: int) -> None:
|
||
from sqlalchemy import func
|
||
sheet = get_sheet(db, tenant_id, company_id, sheet_id)
|
||
sheet.deleted_at = func.now()
|
||
db.commit()
|
||
|
||
|
||
# ============================================================ Rutas (lanes)
|
||
def list_lanes(db: Session, tenant_id: int, sheet_id: int) -> list[RateLane]:
|
||
return (
|
||
db.query(RateLane)
|
||
.filter(RateLane.rate_sheet_id == sheet_id, RateLane.tenant_id == tenant_id,
|
||
RateLane.deleted_at.is_(None))
|
||
.order_by(RateLane.region, RateLane.destination)
|
||
.all()
|
||
)
|
||
|
||
|
||
def breaks_of(db: Session, lane_id: int) -> list[RateBreak]:
|
||
return (
|
||
db.query(RateBreak)
|
||
.filter(RateBreak.rate_lane_id == lane_id, RateBreak.deleted_at.is_(None))
|
||
.order_by(RateBreak.from_qty)
|
||
.all()
|
||
)
|
||
|
||
|
||
def _add_lane(db: Session, tenant_id: int, company_id: int, sheet_id: int,
|
||
lane_data: RateLaneCreate) -> RateLane:
|
||
payload = lane_data.model_dump(exclude={"breaks"})
|
||
lane = RateLane(tenant_id=tenant_id, company_id=company_id, rate_sheet_id=sheet_id, **payload)
|
||
db.add(lane)
|
||
db.flush() # id
|
||
for br in lane_data.breaks:
|
||
db.add(RateBreak(
|
||
tenant_id=tenant_id, company_id=company_id, rate_lane_id=lane.id,
|
||
from_qty=br.from_qty, rate=br.rate,
|
||
))
|
||
return lane
|
||
|
||
|
||
def create_lane(db: Session, tenant_id: int, company_id: int, sheet_id: int,
|
||
lane_data: RateLaneCreate) -> RateLane:
|
||
get_sheet(db, tenant_id, company_id, sheet_id) # valida pertenencia
|
||
lane = _add_lane(db, tenant_id, company_id, sheet_id, lane_data)
|
||
db.commit()
|
||
db.refresh(lane)
|
||
return lane
|
||
|
||
|
||
def delete_lane(db: Session, tenant_id: int, sheet_id: int, lane_id: int) -> None:
|
||
from sqlalchemy import func
|
||
lane = (
|
||
db.query(RateLane)
|
||
.filter(RateLane.id == lane_id, RateLane.rate_sheet_id == sheet_id,
|
||
RateLane.tenant_id == tenant_id)
|
||
.first()
|
||
)
|
||
if not lane:
|
||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Ruta no encontrada")
|
||
lane.deleted_at = func.now()
|
||
db.commit()
|
||
|
||
|
||
# ============================================================ Cargos adicionales
|
||
def list_charges(db: Session, tenant_id: int, sheet_id: int) -> list[RateCharge]:
|
||
return (
|
||
db.query(RateCharge)
|
||
.filter(RateCharge.rate_sheet_id == sheet_id, RateCharge.tenant_id == tenant_id,
|
||
RateCharge.deleted_at.is_(None))
|
||
.order_by(RateCharge.concept)
|
||
.all()
|
||
)
|
||
|
||
|
||
def create_charge(db: Session, tenant_id: int, company_id: int, sheet_id: int, data) -> RateCharge:
|
||
get_sheet(db, tenant_id, company_id, sheet_id)
|
||
ch = RateCharge(
|
||
tenant_id=tenant_id, company_id=company_id, rate_sheet_id=sheet_id,
|
||
rate_lane_id=data.rate_lane_id, concept=data.concept, charge_type=data.charge_type,
|
||
value=data.value, condition=data.condition,
|
||
)
|
||
db.add(ch)
|
||
db.commit()
|
||
db.refresh(ch)
|
||
return ch
|
||
|
||
|
||
def update_charge(db: Session, tenant_id: int, sheet_id: int, charge_id: int, data) -> RateCharge:
|
||
ch = (
|
||
db.query(RateCharge)
|
||
.filter(RateCharge.id == charge_id, RateCharge.rate_sheet_id == sheet_id,
|
||
RateCharge.tenant_id == tenant_id)
|
||
.first()
|
||
)
|
||
if not ch:
|
||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Cargo no encontrado")
|
||
for field, value in data.model_dump(exclude_unset=True).items():
|
||
setattr(ch, field, value)
|
||
db.commit()
|
||
db.refresh(ch)
|
||
return ch
|
||
|
||
|
||
def delete_charge(db: Session, tenant_id: int, sheet_id: int, charge_id: int) -> None:
|
||
from sqlalchemy import func
|
||
ch = (
|
||
db.query(RateCharge)
|
||
.filter(RateCharge.id == charge_id, RateCharge.rate_sheet_id == sheet_id,
|
||
RateCharge.tenant_id == tenant_id)
|
||
.first()
|
||
)
|
||
if not ch:
|
||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Cargo no encontrado")
|
||
ch.deleted_at = func.now()
|
||
db.commit()
|
||
|
||
|
||
# ============================================================ Importación Excel
|
||
# Plantillas por modo: encabezados esperados (orden libre, se detectan por nombre).
|
||
TEMPLATES: dict[str, list[str]] = {
|
||
"aereo": ["Region", "Origen", "Destino", "IATA", "Min", "100", "300", "500", "1000"],
|
||
"maritimo_fcl": ["Origen", "Destino", "Tipo contenedor", "Tarifa", "Transito", "Notas"],
|
||
"maritimo_lcl": ["Origen", "Destino", "Tarifa W/M", "Minimo", "Notas"],
|
||
"terrestre": ["Origen", "Destino", "Tarifa", "Transito", "Notas"],
|
||
}
|
||
|
||
|
||
def build_template(mode: str) -> bytes:
|
||
"""Genera un .xlsx con los encabezados del modo + una fila de ejemplo."""
|
||
import openpyxl
|
||
|
||
if mode not in TEMPLATES:
|
||
raise HTTPException(status_code=400, detail=f"Modo '{mode}' no válido")
|
||
wb = openpyxl.Workbook()
|
||
ws = wb.active
|
||
ws.title = mode
|
||
headers = TEMPLATES[mode]
|
||
ws.append(headers)
|
||
examples = {
|
||
"aereo": ["EUROPA", "NLU", "Frankfurt", "FRA", 190, 1.00, 1.00, 0.95, 0.90],
|
||
"maritimo_fcl": ["MXZLO", "CNSHA", "40HC", 2500, 28, "THC no incluido"],
|
||
"maritimo_lcl": ["MXZLO", "USLAX", 45, 80, "1 W/M = 1 ton o 1 m3"],
|
||
"terrestre": ["Monterrey", "Laredo", 850, 1, ""],
|
||
}
|
||
ws.append(examples[mode])
|
||
buf = io.BytesIO()
|
||
wb.save(buf)
|
||
return buf.getvalue()
|
||
|
||
|
||
def _num(v: Any) -> Decimal | None:
|
||
if v is None or v == "":
|
||
return None
|
||
try:
|
||
return Decimal(str(v).replace("$", "").replace(",", "").strip())
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def parse_excel(mode: str, content: bytes) -> ImportPreview:
|
||
"""Lee el Excel y devuelve una vista previa con validaciones (no persiste)."""
|
||
import openpyxl
|
||
|
||
if mode not in TEMPLATES:
|
||
raise HTTPException(status_code=400, detail=f"Modo '{mode}' no válido")
|
||
try:
|
||
wb = openpyxl.load_workbook(io.BytesIO(content), data_only=True, read_only=True)
|
||
except Exception:
|
||
raise HTTPException(status_code=400, detail="No se pudo leer el archivo Excel")
|
||
ws = wb.active
|
||
rows_iter = ws.iter_rows(values_only=True)
|
||
header = next(rows_iter, None)
|
||
if not header:
|
||
raise HTTPException(status_code=400, detail="El archivo está vacío")
|
||
cols = [str(c).strip() if c is not None else "" for c in header]
|
||
idx = {name.lower(): i for i, name in enumerate(cols)}
|
||
|
||
def cell(row, name):
|
||
i = idx.get(name.lower())
|
||
return row[i] if i is not None and i < len(row) else None
|
||
|
||
preview_rows: list[ImportPreviewRow] = []
|
||
valid = 0
|
||
for n, row in enumerate(rows_iter, start=2):
|
||
if row is None or all(c is None or str(c).strip() == "" for c in row):
|
||
continue
|
||
errors: list[str] = []
|
||
warnings: list[str] = []
|
||
data: dict = {}
|
||
if mode == "aereo":
|
||
data = {
|
||
"region": cell(row, "Region"),
|
||
"origin": cell(row, "Origen"),
|
||
"destination": cell(row, "Destino") or cell(row, "IATA"),
|
||
"iata": cell(row, "IATA"),
|
||
"min_charge": _num(cell(row, "Min")),
|
||
"breaks": {b: _num(cell(row, b)) for b in ("100", "300", "500", "1000")},
|
||
}
|
||
if not data["destination"]:
|
||
errors.append("Falta destino/IATA")
|
||
if not any(v is not None for v in data["breaks"].values()):
|
||
errors.append("Sin tarifas por quiebre")
|
||
elif mode == "maritimo_fcl":
|
||
data = {
|
||
"origin": cell(row, "Origen"),
|
||
"destination": cell(row, "Destino"),
|
||
"equipment_type": cell(row, "Tipo contenedor"),
|
||
"flat_rate": _num(cell(row, "Tarifa")),
|
||
"transit_days": _num(cell(row, "Transito")),
|
||
"notes": cell(row, "Notas"),
|
||
}
|
||
if data["flat_rate"] is None:
|
||
errors.append("Falta la tarifa")
|
||
if not data["equipment_type"]:
|
||
warnings.append("Sin tipo de contenedor")
|
||
elif mode == "maritimo_lcl":
|
||
data = {
|
||
"origin": cell(row, "Origen"),
|
||
"destination": cell(row, "Destino"),
|
||
"wm_rate": _num(cell(row, "Tarifa W/M")),
|
||
"min_charge": _num(cell(row, "Minimo")),
|
||
"notes": cell(row, "Notas"),
|
||
}
|
||
if data["wm_rate"] is None:
|
||
errors.append("Falta la tarifa W/M")
|
||
else: # terrestre
|
||
data = {
|
||
"origin": cell(row, "Origen"),
|
||
"destination": cell(row, "Destino"),
|
||
"flat_rate": _num(cell(row, "Tarifa")),
|
||
"transit_days": _num(cell(row, "Transito")),
|
||
"notes": cell(row, "Notas"),
|
||
}
|
||
if data["flat_rate"] is None:
|
||
errors.append("Falta la tarifa")
|
||
if not data.get("destination"):
|
||
errors.append("Falta destino")
|
||
ok = not errors
|
||
if ok:
|
||
valid += 1
|
||
preview_rows.append(ImportPreviewRow(row=n, data=_jsonable(data), ok=ok,
|
||
warnings=warnings, errors=errors))
|
||
return ImportPreview(mode=mode, total=len(preview_rows), valid=valid,
|
||
rows=preview_rows, columns=cols)
|
||
|
||
|
||
def _jsonable(d: dict) -> dict:
|
||
out = {}
|
||
for k, v in d.items():
|
||
if isinstance(v, Decimal):
|
||
out[k] = float(v)
|
||
elif isinstance(v, dict):
|
||
out[k] = {kk: (float(vv) if isinstance(vv, Decimal) else vv) for kk, vv in v.items()}
|
||
else:
|
||
out[k] = v
|
||
return out
|
||
|
||
|
||
def _rows_to_lanes(mode: str, rows: list[ImportPreviewRow], default_origin: str | None) -> list[RateLaneCreate]:
|
||
lanes: list[RateLaneCreate] = []
|
||
for r in rows:
|
||
if not r.ok:
|
||
continue
|
||
d = r.data
|
||
origin = d.get("origin") or default_origin
|
||
if mode == "aereo":
|
||
breaks = [
|
||
{"from_qty": Decimal(b), "rate": Decimal(str(v))}
|
||
for b, v in (d.get("breaks") or {}).items() if v is not None
|
||
]
|
||
lanes.append(RateLaneCreate(
|
||
origin=str(origin) if origin else None,
|
||
destination=str(d.get("destination")),
|
||
region=d.get("region"), rate_unit="per_kg",
|
||
min_charge=_num(d.get("min_charge")),
|
||
breaks=breaks, # type: ignore[arg-type]
|
||
))
|
||
elif mode == "maritimo_fcl":
|
||
lanes.append(RateLaneCreate(
|
||
origin=str(origin) if origin else None, destination=str(d.get("destination")),
|
||
equipment_type=d.get("equipment_type"), rate_unit="per_container",
|
||
flat_rate=_num(d.get("flat_rate")),
|
||
transit_days=int(d["transit_days"]) if d.get("transit_days") else None,
|
||
notes=d.get("notes"),
|
||
))
|
||
elif mode == "maritimo_lcl":
|
||
lanes.append(RateLaneCreate(
|
||
origin=str(origin) if origin else None, destination=str(d.get("destination")),
|
||
rate_unit="per_wm", min_charge=_num(d.get("min_charge")),
|
||
breaks=[{"from_qty": Decimal(0), "rate": Decimal(str(d["wm_rate"]))}], # type: ignore[arg-type]
|
||
notes=d.get("notes"),
|
||
))
|
||
else:
|
||
lanes.append(RateLaneCreate(
|
||
origin=str(origin) if origin else None, destination=str(d.get("destination")),
|
||
rate_unit="flat", flat_rate=_num(d.get("flat_rate")),
|
||
transit_days=int(d["transit_days"]) if d.get("transit_days") else None,
|
||
notes=d.get("notes"),
|
||
))
|
||
return lanes
|
||
|
||
|
||
def confirm_import(db: Session, tenant_id: int, company_id: int, data: ImportConfirm,
|
||
user_id: str | None) -> RateSheet:
|
||
"""Crea el tarifario + rutas a partir de la vista previa confirmada."""
|
||
sheet = RateSheet(
|
||
tenant_id=tenant_id, company_id=company_id,
|
||
supplier_id=data.supplier_id, mode=data.mode, name=data.name,
|
||
currency=data.currency, valid_from=data.valid_from, valid_to=data.valid_to,
|
||
default_origin=data.default_origin, status=data.status or "borrador",
|
||
notes=data.notes, created_by=user_id, updated_by=user_id,
|
||
)
|
||
db.add(sheet)
|
||
db.flush()
|
||
for lane in data.lanes:
|
||
_add_lane(db, tenant_id, company_id, sheet.id, lane)
|
||
db.commit()
|
||
db.refresh(sheet)
|
||
return sheet
|
||
|
||
|
||
def import_from_excel(db: Session, tenant_id: int, company_id: int, mode: str,
|
||
content: bytes, header: RateSheetCreate, user_id: str | None) -> RateSheet:
|
||
"""Atajo: parsea el Excel y crea el tarifario en un solo paso."""
|
||
preview = parse_excel(mode, content)
|
||
lanes = _rows_to_lanes(mode, preview.rows, header.default_origin)
|
||
return confirm_import(
|
||
db, tenant_id, company_id,
|
||
ImportConfirm(**header.model_dump(), lanes=lanes), user_id,
|
||
)
|
||
|
||
|
||
# ============================================================ Motor de costeo
|
||
def _volumetric_kg(volume_m3: Decimal | None) -> Decimal:
|
||
return (volume_m3 or Decimal(0)) * AIR_VOLUMETRIC_FACTOR
|
||
|
||
|
||
def _rate_for(breaks: list[RateBreak], qty: Decimal) -> Decimal | None:
|
||
"""Tarifa aplicable al peso/wm 'qty' (mayor quiebre cuyo umbral <= qty)."""
|
||
if not breaks:
|
||
return None
|
||
applicable = None
|
||
for b in breaks:
|
||
if b.from_qty <= qty:
|
||
applicable = b.rate
|
||
if applicable is None:
|
||
applicable = breaks[0].rate # por debajo del primer quiebre → tarifa base (gobierna el mínimo)
|
||
return applicable
|
||
|
||
|
||
def _best_break_cost(breaks: list[RateBreak], qty: Decimal) -> Decimal:
|
||
"""Costo base con optimización de quiebre (declarar peso mayor si conviene)."""
|
||
base_rate = _rate_for(breaks, qty)
|
||
base = (qty * base_rate) if base_rate is not None else Decimal(0)
|
||
for b in breaks:
|
||
if b.from_qty > qty:
|
||
candidate = b.from_qty * b.rate
|
||
if candidate < base:
|
||
base = candidate
|
||
return base
|
||
|
||
|
||
def _apply_charges(db: Session, sheet: RateSheet, lane: RateLane, base: Decimal,
|
||
chargeable: Decimal, quantity: int, dangerous: bool) -> list[CostChargeLine]:
|
||
charges = (
|
||
db.query(RateCharge)
|
||
.filter(
|
||
RateCharge.deleted_at.is_(None),
|
||
or_(RateCharge.rate_sheet_id == sheet.id, RateCharge.rate_lane_id == lane.id),
|
||
)
|
||
.all()
|
||
)
|
||
lines: list[CostChargeLine] = []
|
||
for c in charges:
|
||
if c.concept == "dgr" and not dangerous:
|
||
continue
|
||
v = c.value or Decimal(0)
|
||
if c.charge_type == "fijo" or c.charge_type == "por_guia":
|
||
amt = v
|
||
elif c.charge_type == "por_kg":
|
||
amt = v * chargeable
|
||
elif c.charge_type == "por_contenedor":
|
||
amt = v * quantity
|
||
elif c.charge_type == "porcentaje":
|
||
amt = base * v / Decimal(100)
|
||
else:
|
||
amt = v
|
||
lines.append(CostChargeLine(concept=c.concept, amount=amt))
|
||
return lines
|
||
|
||
|
||
def quote_cost(db: Session, tenant_id: int, company_id: int, req: CostRequest) -> list[CostOption]:
|
||
on_date = req.on_date or date.today()
|
||
sheets = _sheet_query(db, tenant_id, company_id).filter(
|
||
RateSheet.mode == req.mode,
|
||
RateSheet.status == "activo",
|
||
or_(RateSheet.valid_from.is_(None), RateSheet.valid_from <= on_date),
|
||
or_(RateSheet.valid_to.is_(None), RateSheet.valid_to >= on_date),
|
||
).all()
|
||
|
||
gross = req.gross_weight_kg or Decimal(0)
|
||
options: list[CostOption] = []
|
||
for sheet in sheets:
|
||
lanes_q = db.query(RateLane).filter(
|
||
RateLane.rate_sheet_id == sheet.id, RateLane.deleted_at.is_(None),
|
||
)
|
||
if req.destination:
|
||
lanes_q = lanes_q.filter(RateLane.destination == req.destination)
|
||
for lane in lanes_q.all():
|
||
# Origen: match exacto o el default del tarifario.
|
||
lane_origin = lane.origin or sheet.default_origin
|
||
if req.origin and lane_origin and lane_origin != req.origin:
|
||
continue
|
||
if req.mode == "maritimo_fcl":
|
||
if req.equipment_type and lane.equipment_type and lane.equipment_type != req.equipment_type:
|
||
continue
|
||
chargeable = Decimal(req.quantity)
|
||
base = (lane.flat_rate or Decimal(0)) * req.quantity
|
||
detail = f"{req.quantity} x {lane.equipment_type or 'contenedor'}"
|
||
elif req.mode == "terrestre":
|
||
chargeable = Decimal(req.quantity)
|
||
base = (lane.flat_rate or Decimal(0)) * req.quantity
|
||
detail = "tarifa por ruta"
|
||
elif req.mode == "maritimo_lcl":
|
||
tons = gross / Decimal(1000)
|
||
wm = max(tons, req.volume_m3 or Decimal(0))
|
||
brks = breaks_of(db, lane.id)
|
||
base = _best_break_cost(brks, wm) if brks else Decimal(0)
|
||
chargeable = wm
|
||
base = max(base, lane.min_charge or Decimal(0))
|
||
detail = f"W/M {wm.quantize(Decimal('0.01'))}"
|
||
else: # aereo
|
||
# P/Vol por dimensiones (L×A×H×cant / 6000); si no hay dimensiones,
|
||
# respaldo con el volumen en m³ × 167.
|
||
vol_by_dims = air_volumetric_kg(req.length_cm, req.width_cm, req.height_cm, req.quantity)
|
||
volumetric = vol_by_dims if vol_by_dims > 0 else _volumetric_kg(req.volume_m3)
|
||
chargeable = max(gross, volumetric)
|
||
brks = breaks_of(db, lane.id)
|
||
base = _best_break_cost(brks, chargeable)
|
||
base = max(base, lane.min_charge or Decimal(0))
|
||
detail = f"facturable {chargeable.quantize(Decimal('0.01'))} kg (P/Vol)"
|
||
|
||
charge_lines = _apply_charges(db, sheet, lane, base, chargeable, req.quantity, req.dangerous)
|
||
total = base + sum((c.amount for c in charge_lines), Decimal(0))
|
||
options.append(CostOption(
|
||
rate_sheet_id=sheet.id, rate_sheet_name=sheet.name, supplier_id=sheet.supplier_id,
|
||
currency=sheet.currency, chargeable=chargeable, base_cost=base,
|
||
charges=charge_lines, total_cost=total, transit_days=lane.transit_days, detail=detail,
|
||
))
|
||
options.sort(key=lambda o: o.total_cost)
|
||
return options
|