118 lines
4.4 KiB
Python
118 lines
4.4 KiB
Python
"""
|
|
Helper functions for API endpoints
|
|
"""
|
|
import uuid
|
|
from typing import Any, Type
|
|
from fastapi import HTTPException, status
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import Query
|
|
from datetime import datetime, timedelta
|
|
from app.models.user import User
|
|
from app.models.ticket import Ticket
|
|
from app.models.category import Category
|
|
from app.services.audit_service import AuditService
|
|
|
|
|
|
def validate_uuid_param(value: str, param_name: str = "ID") -> uuid.UUID:
|
|
"""Valida y convierte string a UUID"""
|
|
try:
|
|
return uuid.UUID(value)
|
|
except ValueError:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"Invalid {param_name} format"
|
|
)
|
|
|
|
|
|
def apply_client_permissions(query: Query, model: Type, current_user: User) -> Query:
|
|
"""Aplica filtros de tenant y permisos de cliente"""
|
|
query = query.where(model.tenant_id == current_user.tenant_id)
|
|
if current_user.role in ["CLIENT_USER", "CLIENT_ADMIN"]:
|
|
query = query.where(model.created_by == current_user.id)
|
|
return query
|
|
|
|
|
|
def apply_enum_filter(query: Query, model_field: Any, filter_value: str,
|
|
enum_class: Type, filter_name: str) -> Query:
|
|
"""Aplica filtro de enum genérico"""
|
|
if filter_value:
|
|
try:
|
|
enum_val = enum_class[filter_value.upper()]
|
|
return query.where(model_field == enum_val)
|
|
except KeyError:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"Invalid {filter_name}: {filter_value}"
|
|
)
|
|
return query
|
|
|
|
|
|
async def safe_audit_log(db: AsyncSession, **kwargs):
|
|
"""Registra en auditoría sin fallar la operación principal"""
|
|
try:
|
|
await AuditService.log(db=db, **kwargs)
|
|
await db.commit()
|
|
except Exception:
|
|
pass # Silent fail para audit logs
|
|
|
|
|
|
async def generate_next_ticket_number(db: AsyncSession, tenant_id: uuid.UUID) -> str:
|
|
"""Genera el siguiente número de ticket único para el tenant"""
|
|
result = await db.execute(
|
|
select(Ticket.ticket_number)
|
|
.where(Ticket.tenant_id == tenant_id)
|
|
.order_by(Ticket.ticket_number.desc())
|
|
.limit(1)
|
|
)
|
|
last_ticket_number = result.scalar_one_or_none()
|
|
|
|
if last_ticket_number:
|
|
last_number = int(last_ticket_number.split('-')[1])
|
|
next_number = last_number + 1
|
|
else:
|
|
next_number = 1
|
|
|
|
return f"TK-{next_number:06d}"
|
|
|
|
|
|
def calculate_sla_deadlines(category: Category = None) -> tuple[datetime, datetime]:
|
|
"""Calcula SLA response y resolution deadlines"""
|
|
if not category:
|
|
return None, None
|
|
|
|
now = datetime.utcnow()
|
|
sla_response_due = now + timedelta(hours=category.sla_response_hours)
|
|
sla_resolution_due = now + timedelta(hours=category.sla_resolution_hours)
|
|
return sla_response_due, sla_resolution_due
|
|
|
|
|
|
def ticket_to_dict(ticket: Ticket) -> dict:
|
|
"""Convierte un modelo Ticket a diccionario de respuesta"""
|
|
return {
|
|
"id": str(ticket.id),
|
|
"ticket_number": ticket.ticket_number,
|
|
"subject": ticket.subject,
|
|
"title": ticket.subject,
|
|
"description": ticket.description,
|
|
"status": ticket.status.value,
|
|
"priority": ticket.priority.value,
|
|
"category_id": str(ticket.category_id) if ticket.category_id else None,
|
|
"category_name": ticket.category.name if ticket.category else None,
|
|
"affected_system_id": str(ticket.affected_system_id) if ticket.affected_system_id else None,
|
|
"system_id": str(ticket.affected_system_id) if ticket.affected_system_id else None,
|
|
"affected_system_name": ticket.affected_system.name if ticket.affected_system else None,
|
|
"contact_email": None,
|
|
"contact_phone": None,
|
|
"created_by": str(ticket.created_by),
|
|
"assigned_to": str(ticket.assigned_to) if ticket.assigned_to else None,
|
|
"assigned_to_name": f"{ticket.assigned_to_user.first_name} {ticket.assigned_to_user.last_name}" if ticket.assigned_to_user else None,
|
|
"created_at": ticket.created_at,
|
|
"updated_at": ticket.updated_at,
|
|
"sla_response_due": ticket.sla_response_due,
|
|
"sla_resolution_due": ticket.sla_resolution_due,
|
|
"first_response_at": ticket.first_response_at,
|
|
"resolved_at": ticket.resolved_at,
|
|
"tenant_id": str(ticket.tenant_id)
|
|
}
|