142 lines
4.2 KiB
Python
142 lines
4.2 KiB
Python
from fastapi import APIRouter, Depends, Query
|
|
from fastapi.responses import Response
|
|
from fastapi.responses import RedirectResponse
|
|
from typing import Optional
|
|
|
|
from auth.models import TokenPayload
|
|
from auth.dependencies import require_permission
|
|
from manufacturing.models import (
|
|
BatchCreate, BatchResponse,
|
|
DeviceInventoryItem, DeviceInventoryListResponse,
|
|
DeviceStatusUpdate, DeviceAssign,
|
|
ManufacturingStats,
|
|
)
|
|
from manufacturing import service
|
|
from manufacturing import audit
|
|
|
|
router = APIRouter(prefix="/api/manufacturing", tags=["manufacturing"])
|
|
|
|
|
|
@router.get("/stats", response_model=ManufacturingStats)
|
|
def get_stats(
|
|
_user: TokenPayload = Depends(require_permission("manufacturing", "view")),
|
|
):
|
|
return service.get_stats()
|
|
|
|
|
|
@router.get("/audit-log")
|
|
async def get_audit_log(
|
|
limit: int = Query(20, ge=1, le=100),
|
|
_user: TokenPayload = Depends(require_permission("manufacturing", "view")),
|
|
):
|
|
entries = await audit.get_recent(limit=limit)
|
|
return {"entries": entries}
|
|
|
|
|
|
@router.post("/batch", response_model=BatchResponse, status_code=201)
|
|
async def create_batch(
|
|
body: BatchCreate,
|
|
user: TokenPayload = Depends(require_permission("manufacturing", "add")),
|
|
):
|
|
result = service.create_batch(body)
|
|
await audit.log_action(
|
|
admin_user=user.email,
|
|
action="batch_created",
|
|
detail={
|
|
"batch_id": result.batch_id,
|
|
"board_type": result.board_type,
|
|
"board_version": result.board_version,
|
|
"quantity": len(result.serial_numbers),
|
|
},
|
|
)
|
|
return result
|
|
|
|
|
|
@router.get("/devices", response_model=DeviceInventoryListResponse)
|
|
def list_devices(
|
|
status: Optional[str] = Query(None),
|
|
hw_type: Optional[str] = Query(None),
|
|
search: Optional[str] = Query(None),
|
|
limit: int = Query(100, ge=1, le=500),
|
|
offset: int = Query(0, ge=0),
|
|
_user: TokenPayload = Depends(require_permission("manufacturing", "view")),
|
|
):
|
|
items = service.list_devices(
|
|
status=status,
|
|
hw_type=hw_type,
|
|
search=search,
|
|
limit=limit,
|
|
offset=offset,
|
|
)
|
|
return DeviceInventoryListResponse(devices=items, total=len(items))
|
|
|
|
|
|
@router.get("/devices/{sn}", response_model=DeviceInventoryItem)
|
|
def get_device(
|
|
sn: str,
|
|
_user: TokenPayload = Depends(require_permission("manufacturing", "view")),
|
|
):
|
|
return service.get_device_by_sn(sn)
|
|
|
|
|
|
@router.patch("/devices/{sn}/status", response_model=DeviceInventoryItem)
|
|
async def update_status(
|
|
sn: str,
|
|
body: DeviceStatusUpdate,
|
|
user: TokenPayload = Depends(require_permission("manufacturing", "edit")),
|
|
):
|
|
result = service.update_device_status(sn, body)
|
|
await audit.log_action(
|
|
admin_user=user.email,
|
|
action="status_updated",
|
|
serial_number=sn,
|
|
detail={"status": body.status.value, "note": body.note},
|
|
)
|
|
return result
|
|
|
|
|
|
@router.get("/devices/{sn}/nvs.bin")
|
|
async def download_nvs(
|
|
sn: str,
|
|
user: TokenPayload = Depends(require_permission("manufacturing", "view")),
|
|
):
|
|
binary = service.get_nvs_binary(sn)
|
|
await audit.log_action(
|
|
admin_user=user.email,
|
|
action="device_flashed",
|
|
serial_number=sn,
|
|
)
|
|
return Response(
|
|
content=binary,
|
|
media_type="application/octet-stream",
|
|
headers={"Content-Disposition": f'attachment; filename="{sn}_nvs.bin"'},
|
|
)
|
|
|
|
|
|
@router.post("/devices/{sn}/assign", response_model=DeviceInventoryItem)
|
|
async def assign_device(
|
|
sn: str,
|
|
body: DeviceAssign,
|
|
user: TokenPayload = Depends(require_permission("manufacturing", "edit")),
|
|
):
|
|
result = service.assign_device(sn, body)
|
|
await audit.log_action(
|
|
admin_user=user.email,
|
|
action="device_assigned",
|
|
serial_number=sn,
|
|
detail={"customer_email": body.customer_email, "customer_name": body.customer_name},
|
|
)
|
|
return result
|
|
|
|
|
|
@router.get("/devices/{sn}/firmware.bin")
|
|
def redirect_firmware(
|
|
sn: str,
|
|
_user: TokenPayload = Depends(require_permission("manufacturing", "view")),
|
|
):
|
|
"""Redirect to the latest stable firmware binary for this device's hw_type.
|
|
Resolves to GET /api/firmware/{hw_type}/stable/{version}/firmware.bin.
|
|
"""
|
|
url = service.get_firmware_url(sn)
|
|
return RedirectResponse(url=url, status_code=302)
|