""" FastAPI Server for Audit Checklist Application ============================================= This server provides a REST API for managing audit checklists with MongoDB integration. It supports CRUD operations for checklist data and user management. Features: - MongoDB Atlas integration with async Motor driver - CORS support for React frontend - Comprehensive error handling - Health check endpoint - Environment-based configuration Author: Audit Checklist Team Version: 1.0.0 """ import os import sys import base64 from datetime import datetime from typing import Optional, List, Dict, Any, Union import logging # FastAPI imports from fastapi import FastAPI, HTTPException, status, UploadFile, File from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse # Pydantic for data validation from pydantic import BaseModel, Field # MongoDB async driver import motor.motor_asyncio from bson import ObjectId # Environment variables from dotenv import load_dotenv # Load environment variables from .env file load_dotenv('mongodb.env') # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # ============================================================================= # CONFIGURATION AND INITIALIZATION # ============================================================================= # Environment variables with fallback defaults MONGODB_URI = os.getenv('MONGODB_URI', 'mongodb://localhost:27017/audit_checklist') PORT = int(os.getenv('PORT', 8000)) # Hugging Face Spaces uses port 8000 CORS_ORIGIN = os.getenv('CORS_ORIGIN', '*') # Allow all origins by default for mobile apps # Initialize FastAPI application app = FastAPI( title="Audit Checklist API", description="REST API for managing audit checklists with MongoDB integration", version="1.0.0", docs_url="/docs", # Swagger UI documentation redoc_url="/redoc" # ReDoc documentation ) # Configure CORS middleware # Allow all origins for Hugging Face Spaces deployment cors_origins = [CORS_ORIGIN] if CORS_ORIGIN != "*" else ["*"] app.add_middleware( CORSMiddleware, allow_origins=cors_origins, allow_credentials=True, allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"], allow_headers=["*"], ) # MongoDB connection setup client = None db = None async def connect_to_mongodb(): """ Establish connection to MongoDB Atlas This function creates an async connection to MongoDB using the Motor driver. It handles connection errors gracefully and logs the status. """ global client, db try: client = motor.motor_asyncio.AsyncIOMotorClient(MONGODB_URI) db = client.audit_checklist # Test the connection await client.admin.command('ping') logger.info("Successfully connected to MongoDB Atlas") # Create indexes for better performance await create_database_indexes() except Exception as e: logger.error(f"Failed to connect to MongoDB: {e}") raise e async def create_database_indexes(): """ Create database indexes for optimal query performance This function creates indexes on frequently queried fields to improve database performance and query speed. """ try: # Index on userId for faster user-specific queries await db.checklists.create_index("userId", unique=False) # Index on createdAt for time-based queries await db.checklists.create_index("createdAt", unique=False) logger.info("Database indexes created successfully") except Exception as e: logger.warning(f"Could not create indexes: {e}") async def close_mongodb_connection(): """Close MongoDB connection gracefully""" global client if client: client.close() logger.info("MongoDB connection closed") # ============================================================================= # PYDANTIC MODELS FOR DATA VALIDATION # ============================================================================= class ChecklistItem(BaseModel): """ Model for individual checklist items Attributes: id: Unique identifier for the item requirement: Description of the requirement to be checked compliance: Compliance status (N/A, Compliant, Non-Compliant) deviation: Description of any deviation found action: Action taken to address the issue checkedAt: Timestamp when the item was checked checkedBy: Name of the person who checked the item """ id: str = Field(..., description="Unique identifier for the checklist item") requirement: str = Field(..., description="Description of the requirement") compliance: str = Field(default="N/A", description="Compliance status: N/A, Compliant, or Non-Compliant") deviation: str = Field(default="", description="Description of any deviation found") action: str = Field(default="", description="Action taken to address the issue") checkedAt: Optional[datetime] = Field(default=None, description="Timestamp when item was checked") checkedBy: str = Field(default="", description="Name of the person who checked the item") class ChecklistSection(BaseModel): """ Model for checklist sections containing multiple items Attributes: id: Unique identifier for the section title: Display title of the section icon: Icon identifier for UI display items: List of checklist items in this section """ id: str = Field(..., description="Unique identifier for the section") title: str = Field(..., description="Display title of the section") icon: str = Field(..., description="Icon identifier for UI display") items: List[ChecklistItem] = Field(default=[], description="List of checklist items") class Metadata(BaseModel): """ Metadata model for checklist information """ userName: Optional[str] = Field(default=None, description="Name of the user who saved the checklist") savedAt: Optional[str] = Field(default=None, description="ISO timestamp when saved") savedAtFormatted: Optional[str] = Field(default=None, description="Formatted timestamp when saved") userId: Optional[str] = Field(default=None, description="User ID") version: Optional[str] = Field(default="1.0", description="Version of the checklist") class ChecklistData(BaseModel): """ Complete checklist data model Attributes: userId: Unique identifier for the user title: Title of the checklist sections: List of sections in the checklist totalItems: Total number of items across all sections completedItems: Number of completed items nonCompliantItems: Number of non-compliant items complianceScore: Calculated compliance percentage verificationDate: Date when the checklist was verified createdAt: Timestamp when the checklist was created updatedAt: Timestamp when the checklist was last updated imageData: Base64 encoded image data (stored when compliance < 100%) imageType: MIME type of the image (e.g., image/jpeg, image/png) """ userId: str = Field(..., description="Unique identifier for the user") title: str = Field(..., description="Title of the checklist") sections: List[ChecklistSection] = Field(default=[], description="List of checklist sections") totalItems: Optional[int] = Field(default=0, description="Total number of items") completedItems: Optional[int] = Field(default=0, description="Number of completed items") nonCompliantItems: Optional[int] = Field(default=0, description="Number of non-compliant items") complianceScore: Optional[float] = Field(default=0.0, description="Compliance percentage score") verificationDate: Optional[datetime] = Field(default=None, description="Date of verification") createdAt: Optional[datetime] = Field(default=None, description="Creation timestamp") updatedAt: Optional[datetime] = Field(default=None, description="Last update timestamp") metadata: Optional[Metadata] = Field(default=None, description="Additional metadata including user information") imageData: Optional[str] = Field(default=None, description="Base64 encoded image data for non-compliant checklists") imageType: Optional[str] = Field(default=None, description="MIME type of the image") class Config: # Allow extra fields that might be sent from frontend extra = "allow" class ChecklistResponse(BaseModel): """ Standard API response model for checklist operations Attributes: success: Boolean indicating if the operation was successful data: The checklist data (if successful) message: Optional message describing the result error: Error message (if unsuccessful) """ success: bool = Field(..., description="Whether the operation was successful") data: Optional[Union[ChecklistData, Dict[str, Any]]] = Field(default=None, description="Checklist data") message: Optional[str] = Field(default=None, description="Success message") error: Optional[str] = Field(default=None, description="Error message") class HealthResponse(BaseModel): """Health check response model""" status: str = Field(..., description="Health status") timestamp: datetime = Field(..., description="Current timestamp") database: str = Field(..., description="Database connection status") version: str = Field(default="1.0.0", description="API version") # ============================================================================= # UTILITY FUNCTIONS # ============================================================================= def calculate_checklist_metrics(checklist_data: Dict[str, Any]) -> Dict[str, Any]: """ Calculate compliance metrics for a checklist Args: checklist_data: Dictionary containing checklist data Returns: Dictionary with calculated metrics (totalItems, completedItems, nonCompliantItems, complianceScore) """ total_items = 0 completed_items = 0 non_compliant_items = 0 # Iterate through all sections and items for section in checklist_data.get('sections', []): for item in section.get('items', []): total_items += 1 # Count completed items (those that are not N/A) if item.get('compliance') != 'N/A': completed_items += 1 # Count non-compliant items if item.get('compliance') == 'Non-Compliant': non_compliant_items += 1 # Calculate compliance score compliance_score = (completed_items / total_items * 100) if total_items > 0 else 0 return { 'totalItems': total_items, 'completedItems': completed_items, 'nonCompliantItems': non_compliant_items, 'complianceScore': round(compliance_score, 2) } def serialize_checklist(checklist_doc: Dict[str, Any]) -> Dict[str, Any]: """ Serialize MongoDB document to JSON-serializable format Args: checklist_doc: MongoDB document Returns: Dictionary with ObjectId converted to string """ if checklist_doc and '_id' in checklist_doc: checklist_doc['_id'] = str(checklist_doc['_id']) # Convert datetime objects to ISO strings for field in ['createdAt', 'updatedAt', 'verificationDate']: if field in checklist_doc and checklist_doc[field]: # Only convert if it's a datetime object, not a string if hasattr(checklist_doc[field], 'isoformat'): checklist_doc[field] = checklist_doc[field].isoformat() # Convert datetime objects in items for section in checklist_doc.get('sections', []): for item in section.get('items', []): if 'checkedAt' in item and item['checkedAt']: # Only convert if it's a datetime object, not a string if hasattr(item['checkedAt'], 'isoformat'): item['checkedAt'] = item['checkedAt'].isoformat() return checklist_doc # ============================================================================= # API ENDPOINTS # ============================================================================= @app.on_event("startup") async def startup_event(): """ Application startup event handler This function is called when the FastAPI application starts up. It initializes the MongoDB connection and sets up the database. """ logger.info("Starting Audit Checklist API server...") await connect_to_mongodb() logger.info(f"Server ready on port {PORT}") @app.on_event("shutdown") async def shutdown_event(): """ Application shutdown event handler This function is called when the FastAPI application shuts down. It gracefully closes the MongoDB connection. """ logger.info("Shutting down Audit Checklist API server...") await close_mongodb_connection() @app.get("/", response_model=Dict[str, str]) async def root(): """ Root endpoint providing basic API information Returns: Dictionary with API name and version """ return { "message": "Audit Checklist API", "version": "1.0.0", "docs": "/docs", "health": "/health" } @app.get("/health", response_model=HealthResponse) async def health_check(): """ Health check endpoint for monitoring and load balancers This endpoint checks the status of the API server and database connection. It's useful for health monitoring, load balancers, and DevOps automation. Returns: HealthResponse with current status information """ try: # Test database connection await client.admin.command('ping') db_status = "connected" except Exception as e: logger.error(f"Database health check failed: {e}") db_status = "disconnected" return HealthResponse( status="healthy" if db_status == "connected" else "unhealthy", timestamp=datetime.utcnow(), database=db_status, version="1.0.0" ) @app.get("/api/checklist/{user_id}", response_model=ChecklistResponse) async def get_checklist(user_id: str): """ Retrieve checklist data for a specific user This endpoint fetches the checklist data for a given user ID. If no checklist exists for the user, it creates a new one with the default structure. Args: user_id: Unique identifier for the user Returns: ChecklistResponse containing the checklist data Raises: HTTPException: If database operation fails """ try: logger.info(f"Fetching checklist for user: {user_id}") # Query the database for the user's checklist # Find the most recent checklist for the user checklist_doc = await db.checklists.find_one( {"userId": user_id}, sort=[("createdAt", -1)] # Get the most recent one ) if not checklist_doc: # Create new checklist if none exists logger.info(f"No checklist found for user {user_id}, creating new one") # Default checklist structure - Complete 38-item audit checklist default_checklist = { "userId": user_id, "title": "Checklist di Audit Operativo (38 Controlli)", "sections": [ { "id": "S1", "title": "1. PERSONALE E IGIENE (6 Controlli)", "icon": "Users", "items": [ {"id": "I1.1", "requirement": "Indumenti da lavoro puliti (divisa, grembiule).", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""}, {"id": "I1.2", "requirement": "Scarpe antinfortunistiche / Calzature pulite.", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""}, {"id": "I1.3", "requirement": "Cuffie e/o Retine per capelli indossate correttamente.", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""}, {"id": "I1.4", "requirement": "Assenza di gioielli, piercing visibili, unghie lunghe.", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""}, {"id": "I1.5", "requirement": "Lavaggio mani documentato all'ingresso/reingresso.", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""}, {"id": "I1.6", "requirement": "Assenza di cibo/bevande non autorizzate nelle aree produttive.", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""} ] }, { "id": "S2", "title": "2. STRUTTURE E IMPIANTI (6 Controlli)", "icon": "Building", "items": [ {"id": "I2.1", "requirement": "Illuminazione adeguata e funzionante in tutte le aree.", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""}, {"id": "I2.2", "requirement": "Porte esterne/interne in buone condizioni e chiuse correttamente (sigillatura).", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""}, {"id": "I2.3", "requirement": "Integrità di pavimenti, pareti e soffitti (assenza di crepe/danni).", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""}, {"id": "I2.4", "requirement": "Controllo vetri / lampade protette o anti-frantumazione.", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""}, {"id": "I2.5", "requirement": "Condizioni igieniche dei servizi igienici e spogliatoi (pulizia e dotazioni).", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""}, {"id": "I2.6", "requirement": "Ventilazione e aspirazione funzionanti, pulite e non ostruite.", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""} ] }, { "id": "S3", "title": "3. GESTIONE E IGIENE AMBIENTALE (4 Controlli)", "icon": "Package", "items": [ {"id": "I3.1", "requirement": "Contenitori dei rifiuti puliti, chiusi e identificati.", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""}, {"id": "I3.2", "requirement": "Separazione corretta dei rifiuti (es. umido, secco, plastica).", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""}, {"id": "I3.3", "requirement": "Area di stoccaggio rifiuti (interna ed esterna) ordinata e sanificata.", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""}, {"id": "I3.4", "requirement": "Frequenza di rimozione dei rifiuti adeguata a prevenire accumuli.", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""} ] }, { "id": "S4", "title": "4. CONTROLLO PROCESSO E QUALITÀ (6 Controlli)", "icon": "Settings", "items": [ {"id": "I4.1", "requirement": "Monitoraggio e registrazione dei Punti Critici di Controllo (CCP).", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""}, {"id": "I4.2", "requirement": "Procedure di Buona Fabbricazione (GMP) rispettate (es. pulizia prima dell'avvio).", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""}, {"id": "I4.3", "requirement": "Verifica e calibrazione del Metal Detector eseguita e registrata.", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""}, {"id": "I4.4", "requirement": "Corretta identificazione dei lotti di materie prime e semilavorati.", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""}, {"id": "I4.5", "requirement": "Rispettate le temperature di stoccaggio dei prodotti finiti e intermedi.", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""}, {"id": "I4.6", "requirement": "Corretta gestione, etichettatura e isolamento dei prodotti non conformi.", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""} ] }, { "id": "S5", "title": "5. CONTROLLO INFESTANTI (5 Controlli)", "icon": "Shield", "items": [ {"id": "I5.1", "requirement": "Assenza di tracce visibili di roditori, insetti o uccelli.", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""}, {"id": "I5.2", "requirement": "Stazioni di monitoraggio/trappole numerate, integre e registrate.", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""}, {"id": "I5.3", "requirement": "Mappe e registri delle trappole aggiornati e disponibili.", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""}, {"id": "I5.4", "requirement": "Barriere fisiche (es. reti, spazzole) anti-intrusione in buone condizioni.", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""}, {"id": "I5.5", "requirement": "Prodotti chimici di controllo infestanti stoccati in modo sicuro e separato.", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""} ] }, { "id": "S6", "title": "6. MANUTENZIONE E VETRI (6 Controlli)", "icon": "Settings", "items": [ {"id": "I6.1", "requirement": "Piano di manutenzione preventiva e correttiva rispettato e registrato.", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""}, {"id": "I6.2", "requirement": "Lubrificanti e oli utilizzati autorizzati per uso alimentare (se richiesto).", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""}, {"id": "I6.3", "requirement": "Registri di calibrazione/verifica degli strumenti di misura critici aggiornati.", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""}, {"id": "I6.4", "requirement": "Gestione dei vetri rotti (registri rotture) aggiornata e conforme alla procedura.", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""}, {"id": "I6.5", "requirement": "Assenza di perdite, gocciolamenti o cavi scoperti da impianti/macchinari.", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""}, {"id": "I6.6", "requirement": "Area di deposito utensili e pezzi di ricambio ordinata e pulita.", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""} ] }, { "id": "S7", "title": "7. DOCUMENTAZIONE E FORMAZIONE (5 Controlli)", "icon": "FileText", "items": [ {"id": "I7.1", "requirement": "Documentazione di processo (procedure, istruzioni) aggiornata e disponibile.", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""}, {"id": "I7.2", "requirement": "Registri di formazione del personale aggiornati e verificabili.", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""}, {"id": "I7.3", "requirement": "Certificazioni e attestati del personale validi e aggiornati.", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""}, {"id": "I7.4", "requirement": "Controllo documenti (versioni, distribuzione, obsolescenza) conforme.", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""}, {"id": "I7.5", "requirement": "Archiviazione e conservazione documenti secondo i requisiti normativi.", "compliance": "N/A", "deviation": "", "action": "", "checkedAt": None, "checkedBy": ""} ] } ], "totalItems": 38, "completedItems": 0, "nonCompliantItems": 0, "complianceScore": 0.0, "verificationDate": None, "createdAt": datetime.utcnow(), "updatedAt": datetime.utcnow() } # Insert the new checklist result = await db.checklists.insert_one(default_checklist) checklist_doc = await db.checklists.find_one({"_id": result.inserted_id}) logger.info(f"Created new checklist for user {user_id}") # Serialize the document serialized_checklist = serialize_checklist(checklist_doc) return ChecklistResponse( success=True, data=ChecklistData(**serialized_checklist), message="Checklist retrieved successfully" ) except Exception as e: logger.error(f"Error retrieving checklist for user {user_id}: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to retrieve checklist: {str(e)}" ) @app.put("/api/checklist/{user_id}", response_model=ChecklistResponse) async def save_checklist(user_id: str, checklist_data: dict): """ Save or update checklist data for a specific user This endpoint saves the checklist data to the database. It calculates compliance metrics and updates the timestamp. Args: user_id: Unique identifier for the user checklist_data: The checklist data to save Returns: ChecklistResponse confirming the save operation Raises: HTTPException: If database operation fails """ try: logger.info(f"Saving checklist for user: {user_id}") # Use the dictionary directly (already converted from JSON) checklist_dict = checklist_data.copy() # Calculate compliance metrics metrics = calculate_checklist_metrics(checklist_dict) checklist_dict.update(metrics) # Handle image data for non-compliant checklists compliance_score = metrics.get('complianceScore', 0.0) if compliance_score < 100.0 and 'collectedImages' in checklist_dict and checklist_dict['collectedImages']: # Process collected images from individual items collected_images = checklist_dict['collectedImages'] logger.info(f"Storing {len(collected_images)} images for non-compliant checklist (score: {compliance_score}%)") # Store the collected images in the checklist data checklist_dict['imageData'] = collected_images checklist_dict['imageType'] = 'multiple_items' elif compliance_score < 100.0: logger.info(f"No image data provided for non-compliant checklist (score: {compliance_score}%)") # Update timestamps checklist_dict['updatedAt'] = datetime.utcnow() # Check if this is an update to existing checklist or new save existing_checklist = await db.checklists.find_one({"userId": user_id}) if existing_checklist: # Update existing checklist checklist_dict['createdAt'] = existing_checklist.get('createdAt', datetime.utcnow()) # Remove _id if it exists to prevent immutable field error if '_id' in checklist_dict: del checklist_dict['_id'] result = await db.checklists.update_one( {"userId": user_id}, {"$set": checklist_dict} ) if result.modified_count > 0: logger.info(f"Updated existing checklist for user {user_id}") message = "Checklist updated successfully" else: logger.error(f"Failed to update checklist for user {user_id}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to update checklist" ) else: # Create new checklist checklist_dict['createdAt'] = datetime.utcnow() # Remove _id if it exists to let MongoDB generate a new one if '_id' in checklist_dict: del checklist_dict['_id'] result = await db.checklists.insert_one(checklist_dict) if result.inserted_id: logger.info(f"Created new checklist for user {user_id}") message = "Checklist created successfully" else: logger.error(f"Failed to create checklist for user {user_id}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to save checklist" ) # Retrieve the updated/created checklist if existing_checklist: # For updates, get the updated document updated_checklist = await db.checklists.find_one({"userId": user_id}) serialized_checklist = serialize_checklist(updated_checklist) else: # For new checklists, get the newly created document created_checklist = await db.checklists.find_one({"_id": result.inserted_id}) serialized_checklist = serialize_checklist(created_checklist) return ChecklistResponse( success=True, data=serialized_checklist, # Return as dict instead of Pydantic model message=message ) except Exception as e: logger.error(f"Error saving checklist for user {user_id}: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to save checklist: {str(e)}" ) @app.get("/api/checklists/by-user/{user_name}", response_model=Dict[str, Any]) async def get_checklists_by_user_name(user_name: str): """ Retrieve all checklists for a specific user name Args: user_name: The name of the user to retrieve checklists for Returns: Dictionary containing list of checklists for the user Raises: HTTPException: If database operation fails """ try: logger.info(f"Retrieving checklists for user: {user_name}") # Find all checklists where metadata.userName matches cursor = db.checklists.find({"metadata.userName": user_name}) checklists = [] async for checklist_doc in cursor: serialized_checklist = serialize_checklist(checklist_doc) checklists.append(serialized_checklist) return { "success": True, "data": checklists, "count": len(checklists), "user_name": user_name } except Exception as e: logger.error(f"Error retrieving checklists for user {user_name}: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to retrieve checklists for user: {str(e)}" ) @app.post("/api/checklist/{user_id}/image") async def upload_checklist_image(user_id: str, image: UploadFile = File(...)): """ Upload an image for a checklist (for non-compliant cases) Args: user_id: Unique identifier for the user image: The image file to upload Returns: Dictionary containing the base64 encoded image data and metadata """ try: logger.info(f"Uploading image for user: {user_id}") # Validate image file type allowed_types = ["image/jpeg", "image/jpg", "image/png", "image/gif", "image/webp"] if image.content_type not in allowed_types: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Invalid image type. Allowed types: {', '.join(allowed_types)}" ) # Read image data image_data = await image.read() # Encode to base64 base64_data = base64.b64encode(image_data).decode('utf-8') # Create data URL data_url = f"data:{image.content_type};base64,{base64_data}" logger.info(f"Image uploaded successfully for user {user_id}, size: {len(image_data)} bytes") return { "success": True, "data": { "imageData": base64_data, "imageType": image.content_type, "dataUrl": data_url, "size": len(image_data), "filename": image.filename }, "message": "Image uploaded successfully" } except HTTPException: raise except Exception as e: logger.error(f"Error uploading image for user {user_id}: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to upload image: {str(e)}" ) async def ensure_default_template_exists(): """ Ensure the default template exists in the database. Creates it if it doesn't exist. """ try: logger.info("Checking if default template exists...") # Check if template exists (check both old and new IDs) template = await db.checklist_templates.find_one({ "$or": [ {"templateId": "default"}, {"templateId": "default-audit-checklist"} ] }) if not template: logger.info("Default template not found, creating it now...") # Create the default template default_template = { "templateId": "default", "title": "Checklist di Audit Operativo", "sections": [ { "id": "S1", "title": "1. PERSONALE E IGIENE", "icon": "Users", "items": [ {"id": "I1.1", "requirement": "Indumenti da lavoro puliti (divisa, grembiule)."}, {"id": "I1.2", "requirement": "Scarpe antinfortunistiche / Calzature pulite."}, {"id": "I1.3", "requirement": "Cuffie e/o Retine per capelli indossate correttamente."}, {"id": "I1.4", "requirement": "Assenza di gioielli, piercing visibili, unghie lunghe."}, {"id": "I1.5", "requirement": "Lavaggio mani documentato all'ingresso/reingresso."}, {"id": "I1.6", "requirement": "Assenza di cibo/bevande non autorizzate nelle aree produttive."}, ], }, { "id": "S2", "title": "2. STRUTTURE E IMPIANTI", "icon": "Building", "items": [ {"id": "I2.1", "requirement": "Illuminazione adeguata e funzionante in tutte le aree."}, {"id": "I2.2", "requirement": "Porte esterne/interne in buone condizioni e chiuse correttamente (sigillatura)."}, {"id": "I2.3", "requirement": "Integrità di pavimenti, pareti e soffitti (assenza di crepe/danni)."}, {"id": "I2.4", "requirement": "Controllo vetri / lampade protette o anti-frantumazione."}, {"id": "I2.5", "requirement": "Condizioni igieniche dei servizi igienici e spogliatoi (pulizia e dotazioni)."}, {"id": "I2.6", "requirement": "Ventilazione e aspirazione funzionanti, pulite e non ostruite."}, ], }, { "id": "S3", "title": "3. GESTIONE E IGIENE AMBIENTALE", "icon": "Package", "items": [ {"id": "I3.1", "requirement": "Contenitori dei rifiuti puliti, chiusi e identificati."}, {"id": "I3.2", "requirement": "Separazione corretta dei rifiuti (es. umido, secco, plastica)."}, {"id": "I3.3", "requirement": "Area di stoccaggio rifiuti (interna ed esterna) ordinata e sanificata."}, {"id": "I3.4", "requirement": "Frequenza di rimozione dei rifiuti adeguata a prevenire accumuli."}, ], }, { "id": "S4", "title": "4. CONTROLLO PROCESSO E QUALITÀ", "icon": "Settings", "items": [ {"id": "I4.1", "requirement": "Monitoraggio e registrazione dei Punti Critici di Controllo (CCP)."}, {"id": "I4.2", "requirement": "Procedure di Buona Fabbricazione (GMP) rispettate (es. pulizia prima dell'avvio)."}, {"id": "I4.3", "requirement": "Verifica e calibrazione del Metal Detector eseguita e registrata."}, {"id": "I4.4", "requirement": "Corretta identificazione dei lotti di materie prime e semilavorati."}, {"id": "I4.5", "requirement": "Rispettate le temperature di stoccaggio dei prodotti finiti e intermedi."}, {"id": "I4.6", "requirement": "Corretta gestione, etichettatura e isolamento dei prodotti non conformi."}, ], }, { "id": "S5", "title": "5. CONTROLLO INFESTANTI", "icon": "Shield", "items": [ {"id": "I5.1", "requirement": "Assenza di tracce visibili di roditori, insetti o uccelli."}, {"id": "I5.2", "requirement": "Stazioni di monitoraggio/trappole numerate, integre e registrate."}, {"id": "I5.3", "requirement": "Mappe e registri delle trappole aggiornati e disponibili."}, {"id": "I5.4", "requirement": "Barriere fisiche (es. reti, spazzole) anti-intrusione in buone condizioni."}, {"id": "I5.5", "requirement": "Prodotti chimici di controllo infestanti stoccati in modo sicuro e separato."}, ], }, { "id": "S6", "title": "6. MANUTENZIONE E VETRI", "icon": "Settings", "items": [ {"id": "I6.1", "requirement": "Piano di manutenzione preventiva e correttiva rispettato e registrato."}, {"id": "I6.2", "requirement": "Lubrificanti e oli utilizzati autorizzati per uso alimentare (se richiesto)."}, {"id": "I6.3", "requirement": "Registri di calibrazione/verifica degli strumenti di misura critici aggiornati."}, {"id": "I6.4", "requirement": "Gestione dei vetri rotti (registri rotture) aggiornata e conforme alla procedura."}, {"id": "I6.5", "requirement": "Assenza di perdite, gocciolamenti o cavi scoperti da impianti/macchinari."}, {"id": "I6.6", "requirement": "Area di deposito utensili e pezzi di ricambio ordinata e pulita."}, ], }, { "id": "S7", "title": "7. DOCUMENTAZIONE E FORMAZIONE", "icon": "FileText", "items": [ {"id": "I7.1", "requirement": "Documentazione di processo (procedure, istruzioni) aggiornata e disponibile."}, {"id": "I7.2", "requirement": "Registri di formazione del personale aggiornati e verificabili."}, {"id": "I7.3", "requirement": "Certificazioni e attestati del personale validi e aggiornati."}, {"id": "I7.4", "requirement": "Controllo documenti (versioni, distribuzione, obsolescenza) conforme."}, {"id": "I7.5", "requirement": "Archiviazione e conservazione documenti secondo i requisiti normativi."}, ], }, ], "metadata": { "createdAt": datetime.utcnow().isoformat(), "updatedAt": datetime.utcnow().isoformat(), } } result = await db.checklist_templates.insert_one(default_template) logger.info(f"✅ Successfully created default template with ID: {result.inserted_id}") # Fetch the newly created template to return with _id template = await db.checklist_templates.find_one({"templateId": "default"}) return template else: logger.info("✅ Default template already exists in database") return template except Exception as e: logger.error(f"❌ Error ensuring default template exists: {e}", exc_info=True) return None @app.get("/api/checklist-template/default", response_model=Dict[str, Any]) async def get_default_checklist_template(): """ Get the default checklist template Returns: Dictionary containing the default checklist template structure Raises: HTTPException: If template not found or database operation fails """ try: logger.info("📋 GET /api/checklist-template/default - Retrieving default checklist template") # Ensure template exists (creates if not found) template_doc = await ensure_default_template_exists() logger.info(f"Template check result: {'Found' if template_doc else 'Not Found'}") if not template_doc: # Fallback to looking for old template ID logger.info("Checking for old template ID: default-audit-checklist") template_doc = await db.checklist_templates.find_one({"templateId": "default-audit-checklist"}) if not template_doc: # If template doesn't exist, create it from the hardcoded structure logger.info("Default template not found, creating from hardcoded structure") # Use the same structure as in get_checklist but mark as template default_template = { "templateId": "default-audit-checklist", "title": "Checklist di Audit Operativo (38 Controlli)", "description": "Template per audit operativo in ambiente di produzione alimentare", "version": "1.0", "isTemplate": True, "createdAt": datetime.utcnow(), "updatedAt": datetime.utcnow(), "sections": [ { "id": "S1", "title": "1. PERSONALE E IGIENE", "icon": "Users", "items": [ {"id": "I1.1", "requirement": "Indumenti da lavoro puliti (divisa, grembiule).", "compliance": "N/A", "deviation": "", "action": "", "imageData": None}, {"id": "I1.2", "requirement": "Scarpe antinfortunistiche / Calzature pulite.", "compliance": "N/A", "deviation": "", "action": "", "imageData": None}, {"id": "I1.3", "requirement": "Cuffie e/o Retine per capelli indossate correttamente.", "compliance": "N/A", "deviation": "", "action": "", "imageData": None}, {"id": "I1.4", "requirement": "Assenza di gioielli, piercing visibili, unghie lunghe.", "compliance": "N/A", "deviation": "", "action": "", "imageData": None}, {"id": "I1.5", "requirement": "Lavaggio mani documentato all'ingresso/reingresso.", "compliance": "N/A", "deviation": "", "action": "", "imageData": None}, {"id": "I1.6", "requirement": "Assenza di cibo/bevande non autorizzate nelle aree produttive.", "compliance": "N/A", "deviation": "", "action": "", "imageData": None} ] }, { "id": "S2", "title": "2. STRUTTURE E IMPIANTI", "icon": "Building", "items": [ {"id": "I2.1", "requirement": "Illuminazione adeguata e funzionante in tutte le aree.", "compliance": "N/A", "deviation": "", "action": "", "imageData": None}, {"id": "I2.2", "requirement": "Porte esterne/interne in buone condizioni e chiuse correttamente (sigillatura).", "compliance": "N/A", "deviation": "", "action": "", "imageData": None}, {"id": "I2.3", "requirement": "Integrità di pavimenti, pareti e soffitti (assenza di crepe/danni).", "compliance": "N/A", "deviation": "", "action": "", "imageData": None}, {"id": "I2.4", "requirement": "Controllo vetri / lampade protette o anti-frantumazione.", "compliance": "N/A", "deviation": "", "action": "", "imageData": None}, {"id": "I2.5", "requirement": "Condizioni igieniche dei servizi igienici e spogliatoi (pulizia e dotazioni).", "compliance": "N/A", "deviation": "", "action": "", "imageData": None}, {"id": "I2.6", "requirement": "Ventilazione e aspirazione funzionanti, pulite e non ostruite.", "compliance": "N/A", "deviation": "", "action": "", "imageData": None} ] }, { "id": "S3", "title": "3. GESTIONE E IGIENE AMBIENTALE", "icon": "Package", "items": [ {"id": "I3.1", "requirement": "Contenitori dei rifiuti puliti, chiusi e identificati.", "compliance": "N/A", "deviation": "", "action": "", "imageData": None}, {"id": "I3.2", "requirement": "Separazione corretta dei rifiuti (es. umido, secco, plastica).", "compliance": "N/A", "deviation": "", "action": "", "imageData": None}, {"id": "I3.3", "requirement": "Area di stoccaggio rifiuti (interna ed esterna) ordinata e sanificata.", "compliance": "N/A", "deviation": "", "action": "", "imageData": None}, {"id": "I3.4", "requirement": "Frequenza di rimozione dei rifiuti adeguata a prevenire accumuli.", "compliance": "N/A", "deviation": "", "action": "", "imageData": None} ] }, { "id": "S4", "title": "4. CONTROLLO PROCESSO E QUALITÀ", "icon": "Settings", "items": [ {"id": "I4.1", "requirement": "Monitoraggio e registrazione dei Punti Critici di Controllo (CCP).", "compliance": "N/A", "deviation": "", "action": "", "imageData": None}, {"id": "I4.2", "requirement": "Procedure di Buona Fabbricazione (GMP) rispettate (es. pulizia prima dell'avvio).", "compliance": "N/A", "deviation": "", "action": "", "imageData": None}, {"id": "I4.3", "requirement": "Verifica e calibrazione del Metal Detector eseguita e registrata.", "compliance": "N/A", "deviation": "", "action": "", "imageData": None}, {"id": "I4.4", "requirement": "Corretta identificazione dei lotti di materie prime e semilavorati.", "compliance": "N/A", "deviation": "", "action": "", "imageData": None}, {"id": "I4.5", "requirement": "Rispettate le temperature di stoccaggio dei prodotti finiti e intermedi.", "compliance": "N/A", "deviation": "", "action": "", "imageData": None}, {"id": "I4.6", "requirement": "Corretta gestione, etichettatura e isolamento dei prodotti non conformi.", "compliance": "N/A", "deviation": "", "action": "", "imageData": None} ] }, { "id": "S5", "title": "5. CONTROLLO INFESTANTI", "icon": "Shield", "items": [ {"id": "I5.1", "requirement": "Assenza di tracce visibili di roditori, insetti o uccelli.", "compliance": "N/A", "deviation": "", "action": "", "imageData": None}, {"id": "I5.2", "requirement": "Stazioni di monitoraggio/trappole numerate, integre e registrate.", "compliance": "N/A", "deviation": "", "action": "", "imageData": None}, {"id": "I5.3", "requirement": "Mappe e registri delle trappole aggiornati e disponibili.", "compliance": "N/A", "deviation": "", "action": "", "imageData": None}, {"id": "I5.4", "requirement": "Barriere fisiche (es. reti, spazzole) anti-intrusione in buone condizioni.", "compliance": "N/A", "deviation": "", "action": "", "imageData": None}, {"id": "I5.5", "requirement": "Prodotti chimici di controllo infestanti stoccati in modo sicuro e separato.", "compliance": "N/A", "deviation": "", "action": "", "imageData": None} ] }, { "id": "S6", "title": "6. MANUTENZIONE E VETRI", "icon": "Settings", "items": [ {"id": "I6.1", "requirement": "Piano di manutenzione preventiva e correttiva rispettato e registrato.", "compliance": "N/A", "deviation": "", "action": "", "imageData": None}, {"id": "I6.2", "requirement": "Lubrificanti e oli utilizzati autorizzati per uso alimentare (se richiesto).", "compliance": "N/A", "deviation": "", "action": "", "imageData": None}, {"id": "I6.3", "requirement": "Registri di calibrazione/verifica degli strumenti di misura critici aggiornati.", "compliance": "N/A", "deviation": "", "action": "", "imageData": None}, {"id": "I6.4", "requirement": "Gestione dei vetri rotti (registri rotture) aggiornata e conforme alla procedura.", "compliance": "N/A", "deviation": "", "action": "", "imageData": None}, {"id": "I6.5", "requirement": "Assenza di perdite, gocciolamenti o cavi scoperti da impianti/macchinari.", "compliance": "N/A", "deviation": "", "action": "", "imageData": None}, {"id": "I6.6", "requirement": "Area di deposito utensili e pezzi di ricambio ordinata e pulita.", "compliance": "N/A", "deviation": "", "action": "", "imageData": None} ] }, { "id": "S7", "title": "7. DOCUMENTAZIONE E FORMAZIONE", "icon": "FileText", "items": [ {"id": "I7.1", "requirement": "Documentazione di processo (procedure, istruzioni) aggiornata e disponibile.", "compliance": "N/A", "deviation": "", "action": "", "imageData": None}, {"id": "I7.2", "requirement": "Registri di formazione del personale aggiornati e verificabili.", "compliance": "N/A", "deviation": "", "action": "", "imageData": None}, {"id": "I7.3", "requirement": "Certificazioni e attestati del personale validi e aggiornati.", "compliance": "N/A", "deviation": "", "action": "", "imageData": None}, {"id": "I7.4", "requirement": "Controllo documenti (versioni, distribuzione, obsolescenza) conforme.", "compliance": "N/A", "deviation": "", "action": "", "imageData": None}, {"id": "I7.5", "requirement": "Archiviazione e conservazione documenti secondo i requisiti normativi.", "compliance": "N/A", "deviation": "", "action": "", "imageData": None} ] } ], "totalItems": 38, "completedItems": 0, "nonCompliantItems": 0, "complianceScore": 0.0 } # Insert the template await db.checklist_templates.insert_one(default_template) template_doc = default_template # Serialize the document serialized_template = serialize_checklist(template_doc) return { "success": True, "data": serialized_template, "message": "Default checklist template retrieved successfully" } except Exception as e: logger.error(f"Error retrieving default checklist template: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to retrieve checklist template: {str(e)}" ) @app.post("/api/checklist-template/item", response_model=Dict[str, Any]) async def add_template_item(item_data: dict): """ Add a new item to the shared checklist template Args: item_data: Dictionary containing item information (sectionId, requirement, etc.) Returns: Dictionary confirming the item was added Raises: HTTPException: If database operation fails """ try: logger.info(f"Adding new item to shared template") # Ensure template exists first and get it template = await ensure_default_template_exists() if not template: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Template not found" ) template_id = template.get("templateId", "default") # Find the section and add the new item section_id = item_data.get('sectionId') section_index = next((i for i, s in enumerate(template['sections']) if s['id'] == section_id), None) if section_index is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Section {section_id} not found" ) new_item = { "id": item_data.get('id', f"I{section_id[-1]}.{len(template['sections'][section_index]['items'])+1}"), "requirement": item_data.get('requirement', ''), } # Update the template result = await db.checklist_templates.update_one( {"templateId": template_id, "sections.id": section_id}, {"$push": {"sections.$.items": new_item}} ) if result.modified_count > 0: logger.info(f"Successfully added item to template") return { "success": True, "message": "Item added successfully", "data": new_item } else: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to add item" ) except HTTPException: raise except Exception as e: logger.error(f"Error adding item to template: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to add item: {str(e)}" ) @app.put("/api/checklist-template/item/{item_id}", response_model=Dict[str, Any]) async def update_template_item(item_id: str, item_data: dict): """ Update an existing item in the shared checklist template Args: item_id: ID of the item to update item_data: Dictionary containing updated item information Returns: Dictionary confirming the item was updated Raises: HTTPException: If database operation fails """ try: logger.info(f"Updating item {item_id} in shared template") # Ensure template exists first and get it template = await ensure_default_template_exists() template_id = template.get("templateId") if template else "default" # Prepare update fields update_fields = {} if 'requirement' in item_data: update_fields["sections.$[].items.$[item].requirement"] = item_data['requirement'] if not update_fields: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="No valid fields to update" ) # Update the item in template (use the template ID we found) result = await db.checklist_templates.update_one( {"templateId": template_id}, {"$set": update_fields}, array_filters=[{"item.id": item_id}] ) if result.modified_count > 0: logger.info(f"Successfully updated item {item_id} in template") return { "success": True, "message": "Item updated successfully", "data": item_data } else: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Item not found or no changes made" ) except HTTPException: raise except Exception as e: logger.error(f"Error updating item {item_id} in template: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to update item: {str(e)}" ) @app.delete("/api/checklist-template/item/{item_id}", response_model=Dict[str, Any]) async def delete_template_item(item_id: str): """ Delete an item from the shared checklist template Args: item_id: ID of the item to delete Returns: Dictionary confirming the item was deleted Raises: HTTPException: If database operation fails """ try: logger.info(f"Deleting item {item_id} from shared template") # Ensure template exists first and get it template = await ensure_default_template_exists() template_id = template.get("templateId") if template else "default" # Remove the item from the template result = await db.checklist_templates.update_one( {"templateId": template_id}, {"$pull": {"sections.$[].items": {"id": item_id}}} ) if result.modified_count > 0: logger.info(f"Successfully deleted item {item_id} from template") return { "success": True, "message": "Item deleted successfully" } else: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Item not found" ) except HTTPException: raise except Exception as e: logger.error(f"Error deleting item {item_id} from template: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to delete item: {str(e)}" ) @app.post("/api/checklist-template/section", response_model=Dict[str, Any]) async def add_template_section(section_data: dict): """ Add a new section to the shared checklist template Args: section_data: Dictionary containing section information Returns: Dictionary confirming the section was added Raises: HTTPException: If database operation fails """ try: logger.info(f"Adding new section to shared template") # Ensure template exists first and get it template = await ensure_default_template_exists() if not template: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Template not found" ) template_id = template.get("templateId", "default") # Create new section new_section = { "id": section_data.get('id', f"S{len(template['sections'])+1}"), "title": section_data.get('title', 'New Section'), "icon": section_data.get('icon', 'Settings'), "items": [] } # Add the new section result = await db.checklist_templates.update_one( {"templateId": template_id}, {"$push": {"sections": new_section}} ) if result.modified_count > 0: logger.info(f"Successfully added section to template") return { "success": True, "message": "Section added successfully", "data": new_section } else: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to add section" ) except HTTPException: raise except Exception as e: logger.error(f"Error adding section to template: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to add section: {str(e)}" ) @app.delete("/api/checklist-template/section/{section_id}", response_model=Dict[str, Any]) async def delete_template_section(section_id: str): """ Delete a section from the shared checklist template Args: section_id: ID of the section to delete Returns: Dictionary confirming the section was deleted Raises: HTTPException: If database operation fails """ try: logger.info(f"Deleting section {section_id} from shared template") # Ensure template exists first and get it template = await ensure_default_template_exists() template_id = template.get("templateId") if template else "default" # Remove the section from the template result = await db.checklist_templates.update_one( {"templateId": template_id}, {"$pull": {"sections": {"id": section_id}}} ) if result.modified_count > 0: logger.info(f"Successfully deleted section {section_id} from template") return { "success": True, "message": "Section deleted successfully" } else: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Section not found" ) except HTTPException: raise except Exception as e: logger.error(f"Error deleting section {section_id} from template: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to delete section: {str(e)}" ) @app.post("/api/checklist/{user_id}/item", response_model=Dict[str, Any]) async def add_checklist_item(user_id: str, item_data: dict): """ Add a new item to a checklist Args: user_id: Unique identifier for the user item_data: Dictionary containing item information (sectionId, requirement, etc.) Returns: Dictionary confirming the item was added Raises: HTTPException: If database operation fails """ try: logger.info(f"Adding new item to checklist for user: {user_id}") # Get the user's checklist checklist_doc = await db.checklists.find_one({"userId": user_id}) if not checklist_doc: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Checklist not found for user" ) # Find the section and add the new item section_id = item_data.get('sectionId') new_item = { "id": item_data.get('id', f"I{section_id}.{len(checklist_doc['sections'][int(section_id[-1])-1]['items'])+1}"), "requirement": item_data.get('requirement', ''), "compliance": "N/A", "deviation": "", "action": "", "imageData": None, "checkedAt": None, "checkedBy": "" } # Update the checklist result = await db.checklists.update_one( {"userId": user_id, "sections.id": section_id}, {"$push": {f"sections.$.items": new_item}} ) if result.modified_count > 0: logger.info(f"Successfully added item to checklist for user {user_id}") return { "success": True, "message": "Item added successfully", "data": new_item } else: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to add item" ) except HTTPException: raise except Exception as e: logger.error(f"Error adding item for user {user_id}: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to add item: {str(e)}" ) @app.put("/api/checklist/{user_id}/item/{item_id}", response_model=Dict[str, Any]) async def update_checklist_item(user_id: str, item_id: str, item_data: dict): """ Update an existing checklist item Args: user_id: Unique identifier for the user item_id: ID of the item to update item_data: Dictionary containing updated item information Returns: Dictionary confirming the item was updated Raises: HTTPException: If database operation fails """ try: logger.info(f"Updating item {item_id} for user: {user_id}") # Check if checklist exists checklist = await db.checklists.find_one({"userId": user_id}) if not checklist: logger.warning(f"No checklist found for user {user_id}") raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Checklist not found for user {user_id}. Please create a checklist first." ) # Prepare update fields update_fields = {} for field in ['requirement', 'compliance', 'deviation', 'action']: if field in item_data: update_fields[f"sections.$[].items.$[item].{field}"] = item_data[field] if not update_fields: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="No valid fields to update" ) # Update the item result = await db.checklists.update_one( {"userId": user_id}, {"$set": update_fields}, array_filters=[{"item.id": item_id}] ) if result.modified_count > 0: logger.info(f"Successfully updated item {item_id} for user {user_id}") return { "success": True, "message": "Item updated successfully", "data": item_data } else: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Item not found or no changes made" ) except HTTPException: raise except Exception as e: logger.error(f"Error updating item {item_id} for user {user_id}: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to update item: {str(e)}" ) @app.delete("/api/checklist/{user_id}/item/{item_id}", response_model=Dict[str, Any]) async def delete_checklist_item(user_id: str, item_id: str): """ Delete a checklist item Args: user_id: Unique identifier for the user item_id: ID of the item to delete Returns: Dictionary confirming the item was deleted Raises: HTTPException: If database operation fails """ try: logger.info(f"Deleting item {item_id} for user: {user_id}") # Check if checklist exists checklist = await db.checklists.find_one({"userId": user_id}) if not checklist: logger.warning(f"No checklist found for user {user_id}") raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Checklist not found for user {user_id}. Please create a checklist first." ) # Remove the item from the checklist result = await db.checklists.update_one( {"userId": user_id}, {"$pull": {"sections.$[].items": {"id": item_id}}} ) if result.modified_count > 0: logger.info(f"Successfully deleted item {item_id} for user {user_id}") return { "success": True, "message": "Item deleted successfully" } else: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Item not found" ) except HTTPException: raise except Exception as e: logger.error(f"Error deleting item {item_id} for user {user_id}: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to delete item: {str(e)}" ) @app.post("/api/checklist/{user_id}/section", response_model=Dict[str, Any]) async def add_checklist_section(user_id: str, section_data: dict): """ Add a new section to a checklist Args: user_id: Unique identifier for the user section_data: Dictionary containing section information Returns: Dictionary confirming the section was added Raises: HTTPException: If database operation fails """ try: logger.info(f"Adding new section to checklist for user: {user_id}") # Get the user's checklist checklist_doc = await db.checklists.find_one({"userId": user_id}) if not checklist_doc: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Checklist not found for user" ) # Create new section new_section = { "id": section_data.get('id', f"S{len(checklist_doc['sections'])+1}"), "title": section_data.get('title', 'New Section'), "icon": section_data.get('icon', 'Settings'), "items": [] } # Add the new section result = await db.checklists.update_one( {"userId": user_id}, {"$push": {"sections": new_section}} ) if result.modified_count > 0: logger.info(f"Successfully added section to checklist for user {user_id}") return { "success": True, "message": "Section added successfully", "data": new_section } else: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to add section" ) except HTTPException: raise except Exception as e: logger.error(f"Error adding section for user {user_id}: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to add section: {str(e)}" ) @app.delete("/api/checklist/{user_id}/section/{section_id}", response_model=Dict[str, Any]) async def delete_checklist_section(user_id: str, section_id: str): """ Delete a checklist section and all its items Args: user_id: Unique identifier for the user section_id: ID of the section to delete Returns: Dictionary confirming the section was deleted Raises: HTTPException: If database operation fails """ try: logger.info(f"Deleting section {section_id} for user: {user_id}") # Check if checklist exists checklist = await db.checklists.find_one({"userId": user_id}) if not checklist: logger.warning(f"No checklist found for user {user_id}") raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Checklist not found for user {user_id}. Please create a checklist first." ) # Remove the section from the checklist result = await db.checklists.update_one( {"userId": user_id}, {"$pull": {"sections": {"id": section_id}}} ) if result.modified_count > 0: logger.info(f"Successfully deleted section {section_id} for user {user_id}") return { "success": True, "message": "Section deleted successfully" } else: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Section not found" ) except HTTPException: raise except Exception as e: logger.error(f"Error deleting section {section_id} for user {user_id}: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to delete section: {str(e)}" ) @app.get("/api/checklists", response_model=Dict[str, Any]) async def get_all_checklists(): """ Retrieve all checklists (admin endpoint) This endpoint returns all checklists in the database. It's intended for administrative purposes and should be protected with proper authentication in a production environment. Returns: Dictionary containing list of all checklists Raises: HTTPException: If database operation fails """ try: logger.info("Retrieving all checklists") # Find all checklists cursor = db.checklists.find({}) checklists = [] async for checklist_doc in cursor: serialized_checklist = serialize_checklist(checklist_doc) checklists.append(serialized_checklist) return { "success": True, "data": checklists, "count": len(checklists), "message": f"Retrieved {len(checklists)} checklists" } except Exception as e: logger.error(f"Error retrieving all checklists: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to retrieve checklists: {str(e)}" ) # ============================================================================= # ERROR HANDLERS # ============================================================================= @app.exception_handler(HTTPException) async def http_exception_handler(request, exc): """Handle HTTP exceptions with consistent error format""" logger.error(f"HTTP Exception: {exc.detail}") return JSONResponse( status_code=exc.status_code, content={ "success": False, "error": exc.detail, "status_code": exc.status_code } ) @app.exception_handler(Exception) async def general_exception_handler(request, exc): """Handle general exceptions with logging""" logger.error(f"Unhandled exception: {exc}", exc_info=True) return JSONResponse( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content={ "success": False, "error": "Internal server error", "detail": str(exc) if os.getenv("DEBUG", "false").lower() == "true" else "An unexpected error occurred" } ) # ============================================================================= # MAIN EXECUTION # ============================================================================= if __name__ == "__main__": """ Main execution block for running the server This block is executed when the script is run directly (not imported). It starts the Uvicorn ASGI server with the configured settings. """ import uvicorn logger.info("Starting FastAPI server...") # Run the server uvicorn.run( "main:app", host="0.0.0.0", port=PORT, reload=True, # Enable auto-reload for development log_level="info" )