409 lines
16 KiB
Python
409 lines
16 KiB
Python
"""Ticket CRUD endpoints, photo uploads, SLA checks, and category listing."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import uuid
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Annotated
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, UploadFile, status
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from app.core.config import settings
|
|
from app.core.database import get_db
|
|
from app.core.security import get_current_user, require_roles
|
|
from app.models.category import Category
|
|
from app.models.ticket import Ticket, TicketPhoto
|
|
from app.models.unit import Unit
|
|
from app.models.user import User
|
|
from app.schemas.ticket import (
|
|
CategoryOut,
|
|
CategoryTreeOut,
|
|
SLAStatusOut,
|
|
TicketBrief,
|
|
TicketCreate,
|
|
TicketListResponse,
|
|
TicketOut,
|
|
TicketPhotoOut,
|
|
TicketUpdate,
|
|
UnitOut,
|
|
)
|
|
from app.services import ticket as ticket_service
|
|
from app.services.sla import get_sla_status
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(prefix="/api/tickets", tags=["tickets"])
|
|
|
|
# Ensure uploads directory exists
|
|
UPLOADS_DIR = settings.BASE_DIR / "uploads"
|
|
UPLOADS_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
# ── Units ──────────────────────────────────────────────────────────
|
|
@router.get("/units", response_model=list[UnitOut])
|
|
async def list_units(
|
|
db: Annotated[AsyncSession, Depends(get_db)],
|
|
property_filter: str | None = Query(None, alias="property"),
|
|
building: str | None = Query(None),
|
|
) -> list[Unit]:
|
|
"""List all units, optionally filtered by property (East/West) and building."""
|
|
query = select(Unit).order_by(Unit.apartment_code)
|
|
if property_filter:
|
|
query = query.where(Unit.property == property_filter)
|
|
if building:
|
|
query = query.where(Unit.building == building)
|
|
result = await db.execute(query)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
@router.get("/units/grouped")
|
|
async def list_units_grouped(
|
|
db: Annotated[AsyncSession, Depends(get_db)],
|
|
property_filter: str | None = Query(None, alias="property"),
|
|
) -> dict[str, dict[str, list[dict]]]:
|
|
"""Return units grouped as ``{property: {building: [units]}}``.
|
|
|
|
Additive convenience variant of ``GET /api/tickets/units`` for the
|
|
Property → Building → Apartment cascade. Units without a building are
|
|
grouped under an empty-string key.
|
|
"""
|
|
query = select(Unit).order_by(Unit.property, Unit.building, Unit.apartment_code)
|
|
if property_filter:
|
|
query = query.where(Unit.property == property_filter)
|
|
result = await db.execute(query)
|
|
grouped: dict[str, dict[str, list[dict]]] = {}
|
|
for unit in result.scalars().all():
|
|
prop = unit.property or ""
|
|
building = unit.building or ""
|
|
grouped.setdefault(prop, {}).setdefault(building, []).append(
|
|
UnitOut.model_validate(unit).model_dump()
|
|
)
|
|
return grouped
|
|
|
|
|
|
# ── Categories ──────────────────────────────────────────────────────
|
|
async def _build_category_tree(db: AsyncSession, parent_id: int | None = None, include_hidden: bool = False) -> list[CategoryTreeOut]:
|
|
"""Build a nested category tree."""
|
|
query = select(Category).where(Category.parent_id == parent_id).order_by(Category.name)
|
|
if not include_hidden:
|
|
query = query.where(Category.show_in_form.is_(True))
|
|
result = await db.execute(query)
|
|
categories = result.scalars().all()
|
|
tree = []
|
|
for cat in categories:
|
|
children = await _build_category_tree(db, cat.id, include_hidden=include_hidden)
|
|
tree.append(CategoryTreeOut(
|
|
id=cat.id,
|
|
type=cat.type,
|
|
name=cat.name,
|
|
parent_id=cat.parent_id,
|
|
sla_urgency=cat.sla_urgency,
|
|
show_in_form=cat.show_in_form,
|
|
children=children,
|
|
))
|
|
return tree
|
|
|
|
|
|
@router.get("/categories", response_model=list[CategoryTreeOut])
|
|
async def list_categories(
|
|
db: Annotated[AsyncSession, Depends(get_db)],
|
|
type_filter: str | None = Query(None, alias="type"),
|
|
include_hidden: bool = Query(False, description="Include alert-only categories (e.g. Gas Leak)"),
|
|
) -> list[CategoryTreeOut]:
|
|
"""Return the full category tree, optionally filtered by type (maintenance, cs, emergency).
|
|
|
|
Alert-only categories (``show_in_form=False``, e.g. Gas Leak) are excluded
|
|
unless ``include_hidden=true`` is passed (used by the emergency quick path).
|
|
"""
|
|
if type_filter:
|
|
# Return flat list of top-level categories of the given type
|
|
query = select(Category).where(Category.parent_id.is_(None), Category.type == type_filter).order_by(Category.name)
|
|
if not include_hidden:
|
|
query = query.where(Category.show_in_form.is_(True))
|
|
result = await db.execute(query)
|
|
parents = result.scalars().all()
|
|
tree = []
|
|
for parent in parents:
|
|
children_query = select(Category).where(Category.parent_id == parent.id).order_by(Category.name)
|
|
if not include_hidden:
|
|
children_query = children_query.where(Category.show_in_form.is_(True))
|
|
children_result = await db.execute(children_query)
|
|
children = children_result.scalars().all()
|
|
tree.append(CategoryTreeOut(
|
|
id=parent.id,
|
|
type=parent.type,
|
|
name=parent.name,
|
|
parent_id=parent.parent_id,
|
|
sla_urgency=parent.sla_urgency,
|
|
show_in_form=parent.show_in_form,
|
|
children=[CategoryOut.model_validate(c) for c in children],
|
|
))
|
|
return tree
|
|
return await _build_category_tree(db, include_hidden=include_hidden)
|
|
|
|
|
|
@router.get("/categories/flat", response_model=list[CategoryOut])
|
|
async def list_categories_flat(
|
|
db: Annotated[AsyncSession, Depends(get_db)],
|
|
type_filter: str | None = Query(None, alias="type"),
|
|
include_hidden: bool = Query(False, description="Include alert-only categories (e.g. Gas Leak)"),
|
|
) -> list[CategoryOut]:
|
|
"""Return a flat list of all categories (no nesting), optionally filtered by type.
|
|
|
|
Alert-only categories are excluded by default; children of alert-only parents
|
|
are excluded too so no orphaned picker entries leak through.
|
|
"""
|
|
query = select(Category).order_by(Category.type, Category.name)
|
|
if type_filter:
|
|
query = query.where(Category.type == type_filter)
|
|
result = await db.execute(query)
|
|
categories = result.scalars().all()
|
|
if include_hidden:
|
|
return [CategoryOut.model_validate(c) for c in categories]
|
|
# Exclude alert-only categories and any children whose parent is alert-only.
|
|
hidden_ids = {
|
|
c.id
|
|
for c in categories
|
|
if not c.show_in_form and c.parent_id is None
|
|
}
|
|
visible = [
|
|
c
|
|
for c in categories
|
|
if c.show_in_form and (c.parent_id is None or c.parent_id not in hidden_ids)
|
|
]
|
|
return [CategoryOut.model_validate(c) for c in visible]
|
|
|
|
|
|
# ── Ticket CRUD ──────────────────────────────────────────────────────
|
|
@router.post("", response_model=TicketOut, status_code=status.HTTP_201_CREATED)
|
|
async def create_ticket(
|
|
body: TicketCreate,
|
|
db: Annotated[AsyncSession, Depends(get_db)],
|
|
current_user: Annotated[User, Depends(get_current_user)],
|
|
) -> Ticket:
|
|
"""Create a new ticket. Auto-generates ticket number PAV-YYYY-NNNNN."""
|
|
ticket = await ticket_service.create_ticket(db, body.model_dump(exclude_none=True), user=current_user)
|
|
# Reload with relationships
|
|
ticket = await ticket_service.get_ticket(db, ticket.id)
|
|
sla = await get_sla_status(ticket)
|
|
ticket.sla_status = sla # type: ignore[attr-defined]
|
|
return ticket
|
|
|
|
|
|
@router.get("", response_model=TicketListResponse)
|
|
async def list_tickets(
|
|
db: Annotated[AsyncSession, Depends(get_db)],
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(50, ge=1, le=200),
|
|
status: str | None = Query(None),
|
|
priority: str | None = Query(None),
|
|
property: str | None = Query(None),
|
|
building: str | None = Query(None),
|
|
unit_id: int | None = Query(None),
|
|
category_id: int | None = Query(None),
|
|
assigned_to: int | None = Query(None),
|
|
date_from: datetime | None = Query(None),
|
|
date_to: datetime | None = Query(None),
|
|
) -> TicketListResponse:
|
|
"""List tickets with optional filtering and pagination."""
|
|
tickets, total = await ticket_service.list_tickets(
|
|
db,
|
|
status_filter=status,
|
|
priority_filter=priority,
|
|
property_filter=property,
|
|
building_filter=building,
|
|
unit_id=unit_id,
|
|
category_id=category_id,
|
|
assigned_to=assigned_to,
|
|
date_from=date_from,
|
|
date_to=date_to,
|
|
page=page,
|
|
page_size=page_size,
|
|
)
|
|
items = [TicketBrief.model_validate(t) for t in tickets]
|
|
return TicketListResponse(items=items, total=total, page=page, page_size=page_size)
|
|
|
|
|
|
@router.get("/{ticket_id}/transitions")
|
|
async def get_ticket_transitions(
|
|
ticket_id: int,
|
|
db: Annotated[AsyncSession, Depends(get_db)],
|
|
) -> dict:
|
|
"""Return the valid next statuses for a ticket's current status."""
|
|
ticket = await ticket_service.get_ticket(db, ticket_id)
|
|
return {
|
|
"current_status": ticket.status,
|
|
"transitions": ticket_service.VALID_TRANSITIONS.get(ticket.status, []),
|
|
}
|
|
|
|
|
|
@router.get("/{ticket_id}", response_model=TicketOut)
|
|
async def get_ticket(
|
|
ticket_id: int,
|
|
db: Annotated[AsyncSession, Depends(get_db)],
|
|
) -> Ticket:
|
|
"""Get a single ticket by ID with full details, timeline, photos, and SLA status."""
|
|
ticket = await ticket_service.get_ticket(db, ticket_id)
|
|
sla = await get_sla_status(ticket)
|
|
ticket.sla_status = sla # type: ignore[attr-defined]
|
|
return ticket
|
|
|
|
|
|
@router.patch("/{ticket_id}", response_model=TicketOut)
|
|
async def update_ticket(
|
|
ticket_id: int,
|
|
body: TicketUpdate,
|
|
db: Annotated[AsyncSession, Depends(get_db)],
|
|
current_user: Annotated[User, Depends(get_current_user)],
|
|
) -> Ticket:
|
|
"""Update a ticket. Validates status transitions and logs timeline."""
|
|
ticket = await ticket_service.update_ticket(db, ticket_id, body.model_dump(exclude_none=True), user=current_user)
|
|
sla = await get_sla_status(ticket)
|
|
ticket.sla_status = sla # type: ignore[attr-defined]
|
|
return ticket
|
|
|
|
|
|
@router.delete("/{ticket_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
async def delete_ticket(
|
|
ticket_id: int,
|
|
db: Annotated[AsyncSession, Depends(get_db)],
|
|
current_user: Annotated[User, Depends(require_roles("Admin/Jerome", "Admin/Wahab"))],
|
|
) -> None:
|
|
"""Delete a ticket and its children (timeline, photos, escalations).
|
|
|
|
Admin-only: intended for removing test/scratch tickets from the demo
|
|
database, never for routine workflow use.
|
|
"""
|
|
ticket = await ticket_service.delete_ticket(db, ticket_id)
|
|
# Remove orphaned photo files from disk after the DB rows are gone.
|
|
for photo in ticket.photos:
|
|
if photo.photo_url:
|
|
name = photo.photo_url.rsplit("/", 1)[-1]
|
|
try:
|
|
(UPLOADS_DIR / name).unlink(missing_ok=True)
|
|
except OSError:
|
|
logger.warning(
|
|
"Could not remove orphaned photo file %s for ticket %s",
|
|
name,
|
|
ticket_id,
|
|
)
|
|
|
|
|
|
# ── Status Transitions (convenience endpoints) ───────────────────────
|
|
@router.post("/{ticket_id}/status", response_model=TicketOut)
|
|
async def change_ticket_status(
|
|
ticket_id: int,
|
|
body: dict,
|
|
db: Annotated[AsyncSession, Depends(get_db)],
|
|
current_user: Annotated[User, Depends(get_current_user)],
|
|
) -> Ticket:
|
|
"""Change a ticket's status with optional note.
|
|
|
|
Request body:
|
|
```json
|
|
{"status": "Logged", "note": "Verified with customer"}
|
|
```
|
|
"""
|
|
status_val = body.get("status")
|
|
if not status_val:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="status is required")
|
|
update_data = {"status": status_val}
|
|
if "note" in body:
|
|
update_data["note"] = body["note"]
|
|
|
|
ticket = await ticket_service.update_ticket(db, ticket_id, update_data, user=current_user)
|
|
sla = await get_sla_status(ticket)
|
|
ticket.sla_status = sla # type: ignore[attr-defined]
|
|
return ticket
|
|
|
|
|
|
# ── SLA ──────────────────────────────────────────────────────────────
|
|
@router.get("/{ticket_id}/sla", response_model=SLAStatusOut)
|
|
async def check_sla(
|
|
ticket_id: int,
|
|
db: Annotated[AsyncSession, Depends(get_db)],
|
|
) -> dict:
|
|
"""Check SLA status for a ticket — response deadline, resolution deadline, breach flags."""
|
|
ticket = await ticket_service.get_ticket(db, ticket_id)
|
|
return await get_sla_status(ticket)
|
|
|
|
|
|
# ── Photo Uploads ────────────────────────────────────────────────────
|
|
ALLOWED_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".webp"}
|
|
ALLOWED_MIME_TYPES = {"image/jpeg", "image/png", "image/gif", "image/webp"}
|
|
|
|
|
|
@router.post("/{ticket_id}/photos", response_model=list[TicketPhotoOut], status_code=status.HTTP_201_CREATED)
|
|
async def upload_photos(
|
|
ticket_id: int,
|
|
db: Annotated[AsyncSession, Depends(get_db)],
|
|
current_user: Annotated[User, Depends(get_current_user)],
|
|
files: list[UploadFile],
|
|
is_before: bool = True,
|
|
) -> list[TicketPhoto]:
|
|
"""Upload photos for a ticket (before/after). Stores files under uploads/."""
|
|
# Verify ticket exists
|
|
await ticket_service.get_ticket(db, ticket_id)
|
|
|
|
# Validate all files before any write operations
|
|
for file in files:
|
|
if file.content_type not in ALLOWED_MIME_TYPES:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"Unsupported content type '{file.content_type}'. Allowed: image/jpeg, image/png, image/gif, image/webp",
|
|
)
|
|
ext = Path(file.filename or "photo.jpg").suffix.lower() if file.filename else ".jpg"
|
|
if ext not in ALLOWED_EXTENSIONS:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"Unsupported file extension '{ext}'. Allowed: .jpg, .jpeg, .png, .gif, .webp",
|
|
)
|
|
|
|
created_photos: list[TicketPhoto] = []
|
|
file_data: list[tuple[Path, bytes]] = []
|
|
for file in files:
|
|
# Generate unique filename
|
|
ext = Path(file.filename or "photo.jpg").suffix.lower() if file.filename else ".jpg"
|
|
unique_name = f"{uuid.uuid4().hex}{ext}"
|
|
file_path = UPLOADS_DIR / unique_name
|
|
|
|
content = await file.read()
|
|
file_data.append((file_path, content))
|
|
|
|
# Database record (no file write yet)
|
|
photo = TicketPhoto(
|
|
ticket_id=ticket_id,
|
|
photo_url=f"/uploads/{unique_name}",
|
|
is_before=is_before,
|
|
)
|
|
db.add(photo)
|
|
created_photos.append(photo)
|
|
|
|
await db.flush()
|
|
for p in created_photos:
|
|
await db.refresh(p)
|
|
|
|
# Write files to disk only after successful DB flush
|
|
for file_path, content in file_data:
|
|
file_path.write_bytes(content)
|
|
|
|
return created_photos
|
|
|
|
|
|
@router.get("/{ticket_id}/photos", response_model=list[TicketPhotoOut])
|
|
async def list_photos(
|
|
ticket_id: int,
|
|
db: Annotated[AsyncSession, Depends(get_db)],
|
|
) -> list[TicketPhoto]:
|
|
"""List all photos for a ticket."""
|
|
await ticket_service.get_ticket(db, ticket_id)
|
|
result = await db.execute(
|
|
select(TicketPhoto)
|
|
.where(TicketPhoto.ticket_id == ticket_id)
|
|
.order_by(TicketPhoto.id)
|
|
)
|
|
return list(result.scalars().all())
|