""" 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, )