diff --git a/backend/main.py b/backend/main.py index 25b6cdc8..d25b27e7 100644 --- a/backend/main.py +++ b/backend/main.py @@ -9,7 +9,7 @@ from slowapi.util import get_remote_address from . import models from .database import engine from .routers import items, operations, users, auth, sync, categories -from .routers.admin import backups, ai_config, db_config +from .routers.admin import backups, ai_config, db_config, exports from .logger import log from .scheduler import scheduler, sync_scheduler_config from .services.image_storage import ensure_image_directories, IMAGES_ROOT @@ -173,6 +173,7 @@ app.include_router(categories.router) app.include_router(backups.router) app.include_router(ai_config.router) app.include_router(db_config.router) +app.include_router(exports.router) # [STATIC FILES] Mount /images/ directory for serving uploaded photos # Ensure directory exists before mounting (StaticFiles requires pre-existing directory) diff --git a/backend/routers/admin/exports.py b/backend/routers/admin/exports.py new file mode 100644 index 00000000..617fea45 --- /dev/null +++ b/backend/routers/admin/exports.py @@ -0,0 +1,141 @@ +""" +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, + )