142 lines
4.1 KiB
Python
142 lines
4.1 KiB
Python
"""
|
|
Admin export endpoints for inventory snapshot and audit trail exports.
|
|
Supports CSV and Excel formats.
|
|
"""
|
|
|
|
from datetime import datetime
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from fastapi.responses import FileResponse
|
|
from sqlalchemy.orm import Session
|
|
|
|
from backend.database import get_db
|
|
from backend.auth import get_current_admin
|
|
from backend.models import Item, AuditLog
|
|
from backend.services.export_service import (
|
|
InventorySnapshotExporter,
|
|
AuditTrailExporter,
|
|
get_export_filename,
|
|
)
|
|
|
|
router = APIRouter(prefix="/admin/exports", tags=["admin-exports"])
|
|
|
|
|
|
def validate_export_format(format_str: str) -> str:
|
|
"""Validate and normalize export format."""
|
|
format_lower = format_str.lower() if format_str else "csv"
|
|
if format_lower not in ("csv", "xlsx"):
|
|
raise HTTPException(
|
|
status_code=400, detail="Invalid format. Use 'csv' or 'xlsx'."
|
|
)
|
|
return format_lower
|
|
|
|
|
|
@router.post("/inventory-snapshot")
|
|
async def export_inventory_snapshot(
|
|
format: str = Query("csv", description="Export format: csv or xlsx"),
|
|
db: Session = Depends(get_db),
|
|
admin_user=Depends(get_current_admin),
|
|
):
|
|
"""
|
|
Export inventory snapshot in CSV or Excel format.
|
|
|
|
Requires admin authorization.
|
|
Returns file download with timestamp in filename.
|
|
"""
|
|
format_type = validate_export_format(format)
|
|
timestamp = datetime.now().strftime("%Y-%m-%d")
|
|
|
|
# Fetch all items
|
|
items = db.query(Item).all()
|
|
|
|
# Generate export
|
|
if format_type == "csv":
|
|
content = InventorySnapshotExporter.to_csv(items, timestamp)
|
|
media_type = "text/csv; charset=utf-8"
|
|
else:
|
|
content = InventorySnapshotExporter.to_excel(items, timestamp)
|
|
media_type = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
|
|
|
|
filename = get_export_filename("inventory_snapshot", format_type, timestamp)
|
|
|
|
# Log export action
|
|
from backend.models import AuditLog
|
|
audit_entry = AuditLog(
|
|
user_id=admin_user.id,
|
|
action="EXPORT_INVENTORY_SNAPSHOT",
|
|
details=f"Exported in {format_type} format",
|
|
)
|
|
db.add(audit_entry)
|
|
db.commit()
|
|
|
|
# Return file response
|
|
if format_type == "csv":
|
|
# For CSV, use FileResponse with bytes
|
|
import io
|
|
return FileResponse(
|
|
io.BytesIO(content.encode("utf-8")),
|
|
media_type=media_type,
|
|
filename=filename,
|
|
)
|
|
else:
|
|
# For Excel, content is already bytes
|
|
import io
|
|
return FileResponse(
|
|
io.BytesIO(content),
|
|
media_type=media_type,
|
|
filename=filename,
|
|
)
|
|
|
|
|
|
@router.post("/audit-trail")
|
|
async def export_audit_trail(
|
|
format: str = Query("csv", description="Export format: csv or xlsx"),
|
|
db: Session = Depends(get_db),
|
|
admin_user=Depends(get_current_admin),
|
|
):
|
|
"""
|
|
Export audit trail in CSV or Excel format.
|
|
|
|
Requires admin authorization.
|
|
Returns file download with timestamp in filename.
|
|
"""
|
|
format_type = validate_export_format(format)
|
|
timestamp = datetime.now().strftime("%Y-%m-%d")
|
|
|
|
# Fetch all audit logs
|
|
logs = db.query(AuditLog).all()
|
|
|
|
# Generate export
|
|
if format_type == "csv":
|
|
content = AuditTrailExporter.to_csv(logs, timestamp)
|
|
media_type = "text/csv; charset=utf-8"
|
|
else:
|
|
content = AuditTrailExporter.to_excel(logs, timestamp)
|
|
media_type = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
|
|
|
|
filename = get_export_filename("audit_trail", format_type, timestamp)
|
|
|
|
# Log export action
|
|
audit_entry = AuditLog(
|
|
user_id=admin_user.id,
|
|
action="EXPORT_AUDIT_TRAIL",
|
|
details=f"Exported in {format_type} format",
|
|
)
|
|
db.add(audit_entry)
|
|
db.commit()
|
|
|
|
# Return file response
|
|
if format_type == "csv":
|
|
import io
|
|
return FileResponse(
|
|
io.BytesIO(content.encode("utf-8")),
|
|
media_type=media_type,
|
|
filename=filename,
|
|
)
|
|
else:
|
|
import io
|
|
return FileResponse(
|
|
io.BytesIO(content),
|
|
media_type=media_type,
|
|
filename=filename,
|
|
)
|