151 lines
4.8 KiB
Python
151 lines
4.8 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any, Dict
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy.orm import Session
|
|
|
|
from core.celery_app import celery_app
|
|
from core.database import get_core_db
|
|
from core.security import get_current_user, get_tenant_from_token, validate_access_to_resource
|
|
|
|
from api.v1.modules.core.tasks_tracking import track_and_dispatch
|
|
|
|
from .schemas import (
|
|
CoveEligibilityResponse,
|
|
FacturaCoveResponse,
|
|
GenerateCoveFromInvoiceRequest,
|
|
)
|
|
from .service import FacturaCoveDomainService
|
|
from .tasks import factura_cove_generate
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.post(
|
|
"/invoices/{invoice_id}/cove",
|
|
response_model=FacturaCoveResponse,
|
|
summary="Generar COVE a partir de una factura (asíncrono)",
|
|
)
|
|
def trigger_cove_for_invoice(
|
|
invoice_id: int,
|
|
body: GenerateCoveFromInvoiceRequest,
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Dispara la tarea Celery `factura_cove_generate` para una factura específica.
|
|
|
|
- Valida acceso a la compañía.
|
|
- Registra la tarea en el tracker de tareas.
|
|
- Retorna el `task_id` para hacer polling de estado desde el frontend.
|
|
"""
|
|
tenant_id = validate_access_to_resource(db, body.company_id, current_user)
|
|
|
|
# Validación mínima de existencia/propiedad de la factura (el dominio hará validaciones más profundas).
|
|
from api.v1.modules.a76.invoices.models import InvoiceHeader
|
|
|
|
invoice: InvoiceHeader | None = db.get(InvoiceHeader, invoice_id)
|
|
if not invoice:
|
|
raise HTTPException(status_code=404, detail=f"Factura {invoice_id} no encontrada.")
|
|
if invoice.company_id != body.company_id or invoice.tenant_id != tenant_id:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="La factura no pertenece a la compañía o tenant actuales.",
|
|
)
|
|
|
|
task = track_and_dispatch(
|
|
db=db,
|
|
task=factura_cove_generate,
|
|
tenant_id=tenant_id,
|
|
company_id=body.company_id,
|
|
requested_by_user=(
|
|
current_user.get("email")
|
|
or current_user.get("preferred_username")
|
|
or current_user.get("username")
|
|
or "system"
|
|
),
|
|
task_name="factura_cove_generate",
|
|
task_group="factura_cove",
|
|
task_origin="a76/factura_cove/invoices/cove",
|
|
args=[invoice_id, int(tenant_id), body.company_id, body.recipient_email],
|
|
)
|
|
|
|
return FacturaCoveResponse(
|
|
task_id=task.id,
|
|
status="queued",
|
|
message="Tarea de validación/generación de COVE encolada.",
|
|
)
|
|
|
|
|
|
@router.get(
|
|
"/invoices/cove/{task_id}/status",
|
|
summary="Estado de tarea de COVE para factura",
|
|
)
|
|
def get_cove_status(task_id: str) -> Dict[str, Any]:
|
|
"""
|
|
Consulta el estado de una tarea Celery de generación de COVE.
|
|
|
|
Retorna:
|
|
- state: 'PROCESSING' | 'SUCCESS' | 'FAILURE'
|
|
- info: { current: int, status: str } (cuando state == 'PROCESSING')
|
|
- result: dict (cuando state == 'SUCCESS' o 'FAILURE')
|
|
"""
|
|
task_result = celery_app.AsyncResult(task_id)
|
|
|
|
if task_result.state in ("PENDING", "STARTED"):
|
|
return {
|
|
"state": "PROCESSING",
|
|
"info": {"current": 0, "status": "Iniciando generación de COVE..."},
|
|
}
|
|
|
|
if task_result.state == "PROGRESS":
|
|
return {
|
|
"state": "PROCESSING",
|
|
"info": task_result.info or {"current": 0, "status": "Procesando COVE..."},
|
|
}
|
|
|
|
if task_result.state == "SUCCESS":
|
|
return {
|
|
"state": "SUCCESS",
|
|
"result": task_result.result,
|
|
}
|
|
|
|
error_info = task_result.result
|
|
if isinstance(error_info, Exception):
|
|
error_msg = str(error_info)
|
|
else:
|
|
error_msg = str(error_info) if error_info else "Error desconocido"
|
|
|
|
return {
|
|
"state": "FAILURE",
|
|
"result": error_msg,
|
|
}
|
|
|
|
|
|
@router.get(
|
|
"/invoices/{invoice_id}/cove/eligibility",
|
|
response_model=CoveEligibilityResponse,
|
|
summary="Verifica si una factura puede generar COVE",
|
|
)
|
|
def check_cove_eligibility(
|
|
invoice_id: int,
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Evalúa si la factura tiene todos los datos necesarios (VU, factura, partidas)
|
|
para poder generar un COVE. No dispara la tarea Celery.
|
|
"""
|
|
tenant_id = get_tenant_from_token(current_user)
|
|
if not tenant_id:
|
|
raise HTTPException(status_code=400, detail="Tenant ID not found in token")
|
|
|
|
tenant_id_int = int(tenant_id)
|
|
|
|
service = FacturaCoveDomainService(db)
|
|
eligibility = service.check_eligibility(invoice_id=invoice_id, tenant_id=tenant_id_int, company_id=company_id)
|
|
return eligibility
|
|
|