afe659e56a
- Tabla crm.cases (expediente) con folio EXP2026-08-001 (next_folio entidad EXP,
sin dirección). Nace al crear la Oportunidad y se hereda vía case_id a
solicitud → cotización → operación → factura. advance_stage solo avanza.
- case_id (FK a crm.cases) en crm.opportunities/service_requests/quotes,
ops.shipments y fin.invoices; propagación en sus create_*. Migración
d4e5f6a7b8c9 reversible.
- Endpoints GET /v1/crm/cases, /cases/{id}, /cases/by-ref/{ref} con timeline
(historia completa para UI y otros sistemas).
- Frontend: casesAPI, ruta /dashboard/crm/expedientes (lista + timeline vertical),
chip "📁 Expediente" en solicitud/cotización, "Expedientes" en el sidebar.
- Consecutivo de folios sin tope (soporta >10,000,000/mes).
- 4 pruebas de expediente (minteo, propagación, timeline, no-retroceso). Suite en verde (113).
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
110 lines
4.2 KiB
Python
110 lines
4.2 KiB
Python
"""Lógica del Expediente: minteo del folio, avance de etapa y armado del timeline."""
|
|
|
|
from fastapi import HTTPException, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ..common.folios import next_folio
|
|
from .models import Case
|
|
|
|
# Orden de etapas (solo se avanza, nunca retrocede)
|
|
STAGE_ORDER = ["oportunidad", "solicitud", "cotizacion", "operacion", "facturacion", "cerrado"]
|
|
|
|
|
|
def create_case(
|
|
db: Session, tenant_id: int, company_id: int, *, account_id: int | None = None,
|
|
title: str | None = None, stage: str = "oportunidad", user_id: str | None = None,
|
|
) -> Case:
|
|
"""Mintea un expediente con folio EXP... (sin commit; lo confirma quien lo invoca)."""
|
|
case = Case(
|
|
reference=next_folio(db, tenant_id, company_id, "EXP", None, with_direction=False),
|
|
account_id=account_id, title=title, stage=stage, status="abierto",
|
|
tenant_id=tenant_id, company_id=company_id, created_by=user_id, updated_by=user_id,
|
|
)
|
|
db.add(case)
|
|
db.flush()
|
|
return case
|
|
|
|
|
|
def advance_stage(db: Session, case_id: int | None, stage: str) -> None:
|
|
"""Avanza la etapa del expediente si la nueva es posterior a la actual."""
|
|
if not case_id or stage not in STAGE_ORDER:
|
|
return
|
|
case = db.query(Case).filter(Case.id == case_id).first()
|
|
if not case:
|
|
return
|
|
current = case.stage if case.stage in STAGE_ORDER else "oportunidad"
|
|
if STAGE_ORDER.index(stage) > STAGE_ORDER.index(current):
|
|
case.stage = stage
|
|
|
|
|
|
def get_cases(
|
|
db: Session, tenant_id: int, company_id: int, search: str | None = None,
|
|
account_id: int | None = None, stage: str | None = None,
|
|
) -> list[Case]:
|
|
q = db.query(Case).filter(
|
|
Case.tenant_id == tenant_id, Case.company_id == company_id, Case.deleted_at.is_(None),
|
|
)
|
|
if account_id is not None:
|
|
q = q.filter(Case.account_id == account_id)
|
|
if stage:
|
|
q = q.filter(Case.stage == stage)
|
|
if search:
|
|
q = q.filter(Case.reference.ilike(f"%{search}%"))
|
|
return q.order_by(Case.created_at.desc()).all()
|
|
|
|
|
|
def get_case(db: Session, case_id: int, tenant_id: int, company_id: int) -> Case:
|
|
obj = (
|
|
db.query(Case)
|
|
.filter(Case.id == case_id, Case.tenant_id == tenant_id, Case.company_id == company_id, Case.deleted_at.is_(None))
|
|
.first()
|
|
)
|
|
if not obj:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Expediente no encontrado")
|
|
return obj
|
|
|
|
|
|
def get_case_by_reference(db: Session, reference: str, tenant_id: int, company_id: int) -> Case:
|
|
obj = (
|
|
db.query(Case)
|
|
.filter(Case.reference == reference, Case.tenant_id == tenant_id, Case.company_id == company_id,
|
|
Case.deleted_at.is_(None))
|
|
.first()
|
|
)
|
|
if not obj:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Expediente no encontrado")
|
|
return obj
|
|
|
|
|
|
def build_timeline(db: Session, case: Case) -> list[dict]:
|
|
"""Devuelve la historia del expediente: todas las entidades ligadas por case_id,
|
|
en orden cronológico. Un único lookup para la UI y para otros sistemas."""
|
|
# Import local para evitar ciclos de importación entre módulos.
|
|
from ..opportunities.models import Opportunity
|
|
from ..quotes.models import Quote
|
|
from ..service_requests.models import ServiceRequest
|
|
from api.v1.modules.fin.invoices.models import Invoice
|
|
from api.v1.modules.ops.shipments.models import Shipment
|
|
|
|
events: list[dict] = []
|
|
specs = [
|
|
("oportunidad", Opportunity, "/dashboard/crm/oportunidades"),
|
|
("solicitud", ServiceRequest, "/dashboard/crm/solicitudes"),
|
|
("cotizacion", Quote, "/dashboard/crm/cotizaciones"),
|
|
("operacion", Shipment, "/dashboard/ops/embarques"),
|
|
("factura", Invoice, "/dashboard/fin/facturas"),
|
|
]
|
|
for kind, model, base_url in specs:
|
|
rows = db.query(model).filter(model.case_id == case.id, model.deleted_at.is_(None)).all()
|
|
for r in rows:
|
|
events.append({
|
|
"kind": kind,
|
|
"id": r.id,
|
|
"reference": getattr(r, "reference", None),
|
|
"status": getattr(r, "status", None),
|
|
"created_at": r.created_at,
|
|
"url": f"{base_url}/{r.id}",
|
|
})
|
|
events.sort(key=lambda e: e["created_at"])
|
|
return events
|