from flask import Flask, request, jsonify from google.cloud import storage import faiss import pandas as pd import numpy as np from sentence_transformers import SentenceTransformer from sklearn.preprocessing import normalize import os import io import tempfile import logging import time import sys # Added for sys.exit on critical error if needed # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = Flask(__name__) # Global variables for lazy loading index = None metadata = None model = None storage_client = None bucket_name = 'book-api-sujal' # Ensure this is your correct bucket name def initialize_storage_client(): global storage_client # Avoid re-initializing if already done if storage_client is not None: return logger.info("Initializing storage client...") try: storage_client = storage.Client() logger.info("Storage client initialized successfully") except Exception as e: logger.error(f"Error initializing storage client: {e}", exc_info=True) # Depending on your strategy, you might want to raise the error # or allow the application to continue and fail later if storage is needed. # Raising here will prevent the app from starting if GCS access fails. raise def load_faiss_index(): global index # This check is now primarily done within the search route # but kept here for potential direct calls or future use if index is not None: logger.info("FAISS index already loaded.") return index logger.info("Attempting to load FAISS index...") try: if storage_client is None: logger.info("Storage client not initialized, initializing now...") initialize_storage_client() # Add a check if initialization failed and storage_client is still None if storage_client is None: logger.error("Failed to initialize storage client, cannot load FAISS index.") return None # Indicate failure bucket = storage_client.bucket(bucket_name) blob = bucket.blob('book_index.faiss') # Ensure this file exists in the bucket logger.info(f"Checking if index blob exists: gs://{bucket_name}/book_index.faiss") if not blob.exists(): logger.error(f"FAISS index file not found in bucket: gs://{bucket_name}/book_index.faiss") return None # Indicate failure logger.info("Downloading index data...") index_data = blob.download_as_bytes() logger.info(f"Downloaded {len(index_data)} bytes for FAISS index.") # Write bytes to a temporary file # Using 'with' ensures the file descriptor is closed even if errors occur tmp_file_descriptor, tmp_file_name = tempfile.mkstemp(suffix=".faiss") logger.info(f"Writing index data to temporary file: {tmp_file_name}") with os.fdopen(tmp_file_descriptor, 'wb') as tmp_file: tmp_file.write(index_data) logger.info(f"Loading index from temporary file: {tmp_file_name}") index = faiss.read_index(tmp_file_name) # Clean up temporary file os.unlink(tmp_file_name) logger.info("FAISS index loaded successfully") return index except Exception as e: # Log the full traceback for debugging logger.error(f"Error loading FAISS index: {e}", exc_info=True) # Reset global variable on failure index = None # Optionally re-raise or return None to indicate failure return None # Indicate failure to the caller def load_metadata(): global metadata if metadata is not None: logger.info("Metadata already loaded.") return metadata logger.info("Attempting to load metadata...") try: if storage_client is None: logger.info("Storage client not initialized, initializing now...") initialize_storage_client() if storage_client is None: logger.error("Failed to initialize storage client, cannot load metadata.") return None bucket = storage_client.bucket(bucket_name) blob = bucket.blob('book_metadata.pkl') # Ensure this file exists logger.info(f"Checking if metadata blob exists: gs://{bucket_name}/book_metadata.pkl") if not blob.exists(): logger.error(f"Metadata file not found in bucket: gs://{bucket_name}/book_metadata.pkl") return None logger.info("Downloading metadata...") metadata_bytes = blob.download_as_bytes() logger.info(f"Downloaded {len(metadata_bytes)} bytes for metadata.") logger.info("Parsing metadata from bytes...") metadata = pd.read_pickle(io.BytesIO(metadata_bytes)) logger.info(f"Metadata loaded successfully with {len(metadata)} records") return metadata except Exception as e: logger.error(f"Error loading metadata: {e}", exc_info=True) metadata = None # Reset global variable on failure return None # Indicate failure def load_model(): global model if model is not None: logger.info("SentenceTransformer model already loaded.") return model logger.info("Attempting to load SentenceTransformer model ('all-MiniLM-L6-v2')...") try: # This step downloads the model files if not cached locally in the container model = SentenceTransformer('all-MiniLM-L6-v2') logger.info("SentenceTransformer model loaded successfully") return model except Exception as e: logger.error(f"Error loading SentenceTransformer model: {e}", exc_info=True) model = None # Reset global variable on failure return None # Indicate failure # --- REMOVED @app.before_first_request block --- # This decorator caused the AttributeError because it's removed in newer Flask versions @app.route('/', methods=['GET']) def health_check(): """Simple health check endpoint""" # Consider adding checks here to see if resources are loaded, if desired # e.g., is_ready = index is not None and metadata is not None and model is not None return jsonify({ "status": "healthy", # Or dynamically set based on resource status "service": "book-recommender-api", "timestamp": time.time() }) # --- REMOVED /initialize endpoint as lazy loading is preferred --- # If you need manual initialization, you could adapt this to call load functions @app.route('/search', methods=['GET']) def search(): start_time = time.time() logger.info("Search request received") # --- ADDED LAZY LOADING CHECKS --- # Use globals directly now global index, metadata, model try: # Load resources if they haven't been loaded yet if index is None: logger.info("Search route: Triggering FAISS index loading...") loaded_index = load_faiss_index() if loaded_index is None: # Check if loading failed logger.error("Search aborted: Failed to load FAISS index.") # Return 503 Service Unavailable, as the service isn't ready return jsonify({"error": "Service Unavailable", "details": "FAISS index could not be loaded."}), 503 if metadata is None: logger.info("Search route: Triggering metadata loading...") loaded_metadata = load_metadata() if loaded_metadata is None: # Check if loading failed logger.error("Search aborted: Failed to load metadata.") return jsonify({"error": "Service Unavailable", "details": "Metadata could not be loaded."}), 503 if model is None: logger.info("Search route: Triggering model loading...") loaded_model = load_model() if loaded_model is None: # Check if loading failed logger.error("Search aborted: Failed to load SentenceTransformer model.") return jsonify({"error": "Service Unavailable", "details": "SentenceTransformer model could not be loaded."}), 503 # --- END LAZY LOADING CHECKS --- # Expecting query parameters for title, authors, genre, and synopsis title = request.args.get('title') authors = request.args.get('authors') genre = request.args.get('genre') synopsis = request.args.get('synopsis') # Check if all required parameters are provided if not all([title, authors, genre, synopsis]): logger.warning("Missing required parameters") return jsonify({ 'error': 'Please provide title, authors, genre, and synopsis as query parameters.' }), 400 # Create combined query text query_text = f"{title} by {authors}. Genre: {genre}. Synopsis: {synopsis}" logger.info(f"Query text created: {query_text[:50]}...") # Generate embedding for the query and normalize it logger.info("Generating query embedding...") # Model should be loaded by now query_embedding = model.encode([query_text], show_progress_bar=False) query_embedding = normalize(query_embedding, axis=1) # Search the FAISS index for top 10 similar books logger.info("Searching FAISS index...") # Index should be loaded by now distances, indices_result = index.search(np.array(query_embedding).astype('float32'), 10) # Prepare results logger.info("Preparing search results...") results = [] # Metadata should be loaded by now for i, idx in enumerate(indices_result[0]): # Check index bounds against the loaded metadata length if idx >= len(metadata) or idx < 0: logger.warning(f"Index {idx} out of range for metadata (length {len(metadata)})") continue # Use .get() for safer access to DataFrame columns/dictionary keys candidate = metadata.iloc[idx] results.append({ 'title': candidate.get('title', 'N/A'), 'authors': candidate.get('authors', 'N/A'), 'genre': candidate.get('genre', 'N/A'), 'synopsis': str(candidate.get('synopsis', ''))[:200] + "..." if len(str(candidate.get('synopsis', ''))) > 200 else str(candidate.get('synopsis', '')), 'num_ratings': int(candidate.get('num_ratings', 0)), 'num_reviews': int(candidate.get('num_reviews', 0)), 'similarity': float(distances[0][i]) }) elapsed_time = time.time() - start_time logger.info(f"Search completed in {elapsed_time:.2f} seconds with {len(results)} results.") return jsonify({ "results": results, "query": { "title": title, "authors": authors, "genre": genre # Note: Synopsis is usually not returned in the query part }, "execution_time_seconds": elapsed_time }) # Catch specific errors if needed, otherwise fall back to general Exception except Exception as e: # Log the full traceback for server-side debugging logger.error(f"Error processing search request: {e}", exc_info=True) # Return a generic 500 error to the client return jsonify({ "error": "An internal server error occurred during search.", # Optionally include limited details, but avoid exposing sensitive info # "details": str(e) # Be cautious with exposing raw error details }), 500 # This block is mainly for local execution (python app.py) # Gunicorn doesn't use this block directly, it imports the 'app' object if __name__ == '__main__': port = int(os.environ.get('PORT', 8080)) logger.info(f"Starting Flask app directly (not via Gunicorn) on http://0.0.0.0:{port}") # Set debug=True for local development ONLY if needed, NEVER in production/Cloud Run app.run(host='0.0.0.0', port=port, debug=False)