Smart Document Analysis Platform

from flask import Flask, request, jsonify, render_template_string from flask_cors import CORS from google import genai from google.genai import types import os import io import httpx import uuid from datetime import datetime, timezone, timedelta from dotenv import load_dotenv import json # Load environment variables from a .env file # This is useful for local development. In production on platforms like Hugging Face, # you'll set these as environment variables directly in the settings. load_dotenv() app = Flask(__name__) CORS(app) # Initialize Gemini client # The API key should be loaded from environment variables api_key = os.getenv('GOOGLE_API_KEY') if not api_key: print("Error: GOOGLE_API_KEY environment variable not set.") # In a real app, you might exit or raise an exception here. # For this example, we'll print an error but allow the app to start; # API calls will fail if the key is missing. # If running locally, make sure you have a .env file with GOOGLE_API_KEY=YOUR_API_KEY pass # Allows the app to run without a key for debugging non-API parts try: client = genai.Client(api_key=api_key) except Exception as e: print(f"Failed to initialize Gemini client: {e}") client = None # Set client to None if initialization fails # In-memory storage for demo (in production, use a database like Redis or PostgreSQL) # Maps our internal cache_id (UUID) to Gemini's cache_name and other info document_caches = {} user_sessions = {} # Not used in this version, but kept from template # HTML template for the web interface HTML_TEMPLATE = """ Smart Document Analysis Platform

📤 Upload PDF Document

📄

Drag and drop your PDF file here, or click to select

Or provide a URL:

💬 Ask Questions

👋 Hello! Upload a PDF document using the panel on the left, and I'll help you analyze it using Gemini API caching!

""" # --- Flask Routes --- @app.route('/') def index(): # Ensure API key is set before rendering, or add a warning to the template if not api_key: # You could modify the template or pass a variable to indicate error state print("Warning: API key not set. API calls will fail.") return render_template_string(HTML_TEMPLATE) @app.route('/health', methods=['GET']) def health_check(): # A simple endpoint to check if the application is running # Can optionally check API client status if needed, but basic 200 is common. if client is None and api_key is not None: # Client failed to initialize despite key being present return jsonify({"status": "unhealthy", "reason": "Gemini client failed to initialize"}), 500 # Note: This doesn't check if the API key is *valid* or if the API is reachable, # just if the Flask app is running and the client object was created. return jsonify({"status": "healthy"}), 200 @app.route('/upload', methods=['POST']) def upload_file(): if client is None or api_key is None: return jsonify({'success': False, 'error': 'API key not configured or Gemini client failed to initialize.'}), 500 try: if 'file' not in request.files: return jsonify({'success': False, 'error': 'No file provided'}) file = request.files['file'] if file.filename == '': return jsonify({'success': False, 'error': 'No file selected'}) # Read file content file_content = file.read() file_io = io.BytesIO(file_content) # --- CORRECTED FILE UPLOAD CALL --- # Upload to Gemini File API using the correct method client.upload_file # Pass the file content as a tuple (filename, file-like object, mime_type) # This replaces the incorrect client.files.upload call document = None # Initialize document variable try: # The mime_type is crucial for the API to correctly process the file. # The filename is used as the display_name by default if not provided. document = client.upload_file( file=(file.filename, file_io, 'application/pdf'), # Use the 'file' argument with tuple format # display_name=file.filename # Optional: explicitly provide a display name ) print(f"File uploaded successfully to Gemini File API: {document.name}") # Log for debugging # Note: client.upload_file returns a google.generativeai.types.File object # which contains the resource name (e.g., 'files/xyz123'). except Exception as upload_error: # Attempt to provide more specific feedback if possible error_msg = str(upload_error) print(f"Error uploading file to Gemini API: {error_msg}") # Check for common upload errors like exceeding file size limits if "file content size exceeds limit" in error_msg.lower(): return jsonify({'success': False, 'error': f'Error uploading file: File size exceeds API limit. {error_msg}'}), 413 # 413 Payload Too Large return jsonify({'success': False, 'error': f'Error uploading file to Gemini API: {error_msg}'}), 500 # --- END CORRECTED FILE UPLOAD CALL --- # Create cache with system instruction cache = None # Initialize cache variable try: system_instruction = "You are an expert document analyzer. Provide detailed, accurate answers based on the uploaded document content. Always be helpful and thorough in your responses." # Use the correct model format as per documentation # Using a specific stable version is recommended for production model = 'models/gemini-2.0-flash-001' print(f"Attempting to create cache for file: {document.name}") # Log cache = client.caches.create( model=model, config=types.CreateCachedContentConfig( display_name=f'pdf document cache: {file.filename}', # Use filename in display_name system_instruction=system_instruction, contents=[document], # contents should be a list of content parts. document is already a File object, which is a valid content part type. ttl="3600s", # 1 hour TTL. Use string format like "300s" or "1h". ) ) print(f"Cache created successfully: {cache.name}") # Log # Store cache info in our in-memory dictionary # We map our internal UUID cache_id to the Gemini API's cache.name (resource name) cache_id = str(uuid.uuid4()) document_caches[cache_id] = { 'gemini_cache_name': cache.name, # Store the Gemini API resource name 'document_name': file.filename, 'gemini_file_name': document.name, # Also store the Gemini File API resource name for cleanup 'created_at': datetime.now().isoformat(), 'expires_at': (datetime.now(timezone.utc) + timedelta(seconds=3600)).isoformat(), # Store expiry time for reference } # Get token count from cache metadata if available # Note: cached_token_count might be available on the cache object after creation token_count = 'Unknown' if hasattr(cache, 'usage_metadata') and cache.usage_metadata: token_count = getattr(cache.usage_metadata, 'cached_token_count', 'Unknown') print(f"Cached token count: {token_count}") return jsonify({ 'success': True, 'cache_id': cache_id, # Return our internal ID 'token_count': token_count }) except Exception as cache_error: error_msg = str(cache_error) print(f"Cache creation failed: {error_msg}") # Log the cache error # If caching fails, attempt to delete the uploaded file to clean up. if document and hasattr(document, 'name'): try: client.files.delete(document.name) print(f"Cleaned up uploaded file {document.name} after caching failure.") except Exception as cleanup_error: print(f"Failed to clean up file {document.name}: {cleanup_error}") # Handle specific cache creation errors # Note: The exact error message for content size can vary or might not be specific # The documentation mentions minimum tokens for caching. if "Cached content is too small" in error_msg or "minimum size" in error_msg.lower() or "tokens required" in error_msg.lower(): return jsonify({ 'success': False, 'error': f'PDF content is too small for caching. Minimum token count varies by model, but is typically 1024+ for Flash. {error_msg}', 'suggestion': 'Try uploading a longer document or combine multiple documents.' }), 400 # 400 Bad Request - client error else: # Re-raise other unexpected errors or return a generic error return jsonify({'success': False, 'error': f'Error creating cache with Gemini API: {error_msg}'}), 500 except Exception as e: print(f"An unexpected error occurred during upload process: {str(e)}") # Log general errors return jsonify({'success': False, 'error': str(e)}), 500 @app.route('/upload-url', methods=['POST']) def upload_from_url(): if client is None or api_key is None: return jsonify({'success': False, 'error': 'API key not configured or Gemini client failed to initialize.'}), 500 try: data = request.get_json() url = data.get('url') if not url: return jsonify({'success': False, 'error': 'No URL provided'}), 400 # 400 Bad Request # Download file from URL response = None try: # Use stream=True for potentially large files, although httpx handles it well. # Add a timeout to prevent hanging on unresponsive URLs. response = httpx.get(url, follow_redirects=True, timeout=30.0) response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx) # Basic check for PDF mime type (optional but good practice) content_type = response.headers.get('Content-Type', '').lower() if 'application/pdf' not in content_type: print(f"Warning: URL content type is not application/pdf: {content_type}") # Decide if you want to block non-PDFs or try to upload anyway # For now, we'll proceed but log a warning. API might reject it. # If strictly PDF required, return an error here: # return jsonify({'success': False, 'error': f'URL does not point to a PDF document (Content-Type: {content_type})'}), 415 # 415 Unsupported Media Type except httpx.HTTPStatusError as e: print(f"HTTP error downloading file from URL {url}: {e.response.status_code} - {e.response.text}") return jsonify({'success': False, 'error': f'HTTP error downloading file from URL: {e.response.status_code} - {e.response.text}'}), e.response.status_code except httpx.RequestError as e: print(f"Error downloading file from URL {url}: {e}") return jsonify({'success': False, 'error': f'Error downloading file from URL: {e}'}), 500 file_io = io.BytesIO(response.content) # --- CORRECTED FILE UPLOAD CALL --- # Upload to Gemini File API using the correct method client.upload_file # Pass the file content as a tuple (filename, file-like object, mime_type) # Use a generic filename for the file-like object if none derived from URL document = None # Initialize document variable try: # Attempt to get filename from URL or headers, otherwise use generic filename = os.path.basename(url) if not filename or '.' not in filename: filename = 'downloaded_document.pdf' # Default generic name # Use the mime type from the response headers if available and looks right mime_type = content_type if 'application/pdf' in content_type else 'application/pdf' document = client.upload_file( file=(filename, file_io, mime_type), # Use parsed filename and mime_type display_name=url # Use the URL as display name in Gemini API ) print(f"File from URL uploaded successfully to Gemini File API: {document.name}") # Log # Note: client.upload_file returns a google.generativeai.types.File object # which contains the resource name (e.g., 'files/xyz123'). except Exception as upload_error: # Attempt to provide more specific feedback if possible error_msg = str(upload_error) print(f"Error uploading file from URL to Gemini API: {error_msg}") # Check for common upload errors like exceeding file size limits if "file content size exceeds limit" in error_msg.lower(): return jsonify({'success': False, 'error': f'Error uploading file: File size exceeds API limit. {error_msg}'}), 413 # 413 Payload Too Large return jsonify({'success': False, 'error': f'Error uploading file from URL to Gemini API: {error_msg}'}), 500 # --- END CORRECTED FILE UPLOAD CALL --- # Create cache with system instruction cache = None # Initialize cache variable try: system_instruction = "You are an expert document analyzer. Provide detailed, accurate answers based on the uploaded document content. Always be helpful and thorough in your responses." # Use the correct model format as per documentation model = 'models/gemini-2.0-flash-001' print(f"Attempting to create cache for file: {document.name}") # Log cache = client.caches.create( model=model, config=types.CreateCachedContentConfig( display_name=f'pdf document cache: {url}', # Use URL in display_name for cache system_instruction=system_instruction, contents=[document], # contents should be a list containing the File object ttl="3600s", # 1 hour TTL. Use string format like "300s" or "1h". ) ) print(f"Cache created successfully: {cache.name}") # Log # Store cache info in our in-memory dictionary # We map our internal UUID cache_id to the Gemini API's cache.name (resource name) cache_id = str(uuid.uuid4()) document_caches[cache_id] = { 'gemini_cache_name': cache.name, # Store the Gemini API resource name 'document_name': url, # Store the URL as the document name 'gemini_file_name': document.name, # Also store the Gemini File API resource name for cleanup 'created_at': datetime.now().isoformat(), 'expires_at': (datetime.now(timezone.utc) + timedelta(seconds=3600)).isoformat(), # Store expiry time for reference } # Get token count from cache metadata if available token_count = 'Unknown' if hasattr(cache, 'usage_metadata') and cache.usage_metadata: token_count = getattr(cache.usage_metadata, 'cached_token_count', 'Unknown') print(f"Cached token count: {token_count}") return jsonify({ 'success': True, 'cache_id': cache_id, # Return our internal ID 'token_count': token_count }) except Exception as cache_error: error_msg = str(cache_error) print(f"Cache creation failed: {error_msg}") # Log the cache error # If caching fails, attempt to delete the uploaded file to clean up. if document and hasattr(document, 'name'): try: client.files.delete(document.name) print(f"Cleaned up uploaded file {document.name} after caching failure.") except Exception as cleanup_error: print(f"Failed to clean up file {document.name}: {cleanup_error}") # Handle specific cache creation errors if "Cached content is too small" in error_msg or "minimum size" in error_msg.lower() or "tokens required" in error_msg.lower(): return jsonify({ 'success': False, 'error': f'PDF content is too small for caching. Minimum token count varies by model, but is typically 1024+ for Flash. {error_msg}', 'suggestion': 'Try uploading a longer document or combine multiple documents.' }), 400 # 400 Bad Request - client error else: # Re-raise other unexpected errors or return a generic error return jsonify({'success': False, 'error': f'Error creating cache with Gemini API: {error_msg}'}), 500 except Exception as e: print(f"An unexpected error occurred during URL upload process: {str(e)}") # Log general errors return jsonify({'success': False, 'error': str(e)}), 500 @app.route('/ask', methods=['POST']) def ask_question(): if client is None or api_key is None: return jsonify({'success': False, 'error': 'API key not configured or Gemini client failed to initialize.'}), 500 try: data = request.get_json() question = data.get('question') cache_id = data.get('cache_id') if not question or not cache_id: return jsonify({'success': False, 'error': 'Missing question or cache_id'}), 400 # 400 Bad Request # --- CORRECTED CACHE LOOKUP --- # Check if our internal cache_id exists in the in-memory dictionary if cache_id not in document_caches: # If not found, it's either an invalid ID, expired, or the server restarted. # For this simple demo, we treat it as unavailable. print(f"Cache ID {cache_id} not found in local storage.") return jsonify({'success': False, 'error': 'Cache not found or expired. Please upload the document again.'}), 404 # 404 Not Found # If found, retrieve the Gemini API cache name cache_info = document_caches[cache_id] gemini_cache_name = cache_info['gemini_cache_name'] print(f"Using Gemini cache name: {gemini_cache_name} for question.") # --- END CORRECTED CACHE LOOKUP --- # Generate response using cached content with correct model format response = client.models.generate_content( model='models/gemini-2.0-flash-001', # Ensure using the model the cache was created with contents=[{'text': question}], # User's question as text content part generation_config=types.GenerateContentConfig( cached_content=gemini_cache_name # Use the retrieved Gemini cache name ) ) # Check if response has parts before accessing .text answer = "Could not generate response from the model." if response and response.candidates: # Handle potential tool_code or other non-text parts if necessary answer_parts = [] for candidate in response.candidates: if candidate.content and candidate.content.parts: for part in candidate.content.parts: if hasattr(part, 'text'): answer_parts.append(part.text) # Add handling for other part types if needed (e.g., tool_code, function_response) # elif hasattr(part, 'tool_code'): # answer_parts.append(f"\n```tool_code\n{part.tool_code.code}\n```\n") # elif hasattr(part, 'function_response'): # answer_parts.append(f"\n```function_response\n{json.dumps(part.function_response, indent=2)}\n```\n") if answer_parts: answer = "".join(answer_parts) else: # Handle cases where candidates exist but have no text parts (e.g., tool calls) answer = "Model returned content without text parts (e.g., tool calls)." print(f"Model returned non-text parts: {response.candidates}") # Log for debugging elif response and response.prompt_feedback and response.prompt_feedback.block_reason: # Handle cases where the prompt was blocked block_reason = response.prompt_feedback.block_reason.name block_message = getattr(response.prompt_feedback, 'block_reason_message', 'No message provided') answer = f"Request blocked by safety filters. Reason: {block_reason}. Message: {block_message}" print(f"Request blocked: {block_reason} - {block_message}") else: # Handle other unexpected response structures print(f"Unexpected response structure from API: {response}") # answer stays as the initial "Could not generate response..." message return jsonify({ 'success': True, 'answer': answer }) except Exception as e: print(f"An error occurred during question asking: {str(e)}") # Log errors # Attempt to provide more specific API error messages error_msg = str(e) if "Resource has been exhausted" in error_msg: error_msg = "API rate limit or quota exceeded. Please try again later." elif "cached_content refers to a resource that has been deleted" in error_msg: error_msg = "The cached document has expired or was deleted from Gemini API. Please upload the document again." # Clean up local entry if API confirms deletion/expiry if cache_id in document_caches: print(f"Removing local entry for cache_id {cache_id} as API confirmed deletion.") del document_caches[cache_id] elif "invalid cached_content value" in error_msg: error_msg = "Invalid cache reference. The cached document might have expired or been deleted. Please upload the document again." # Clean up local entry if API confirms deletion/expiry if cache_id in document_caches: print(f"Removing local entry for cache_id {cache_id} as API confirmed deletion (invalid reference).") del document_caches[cache_id] elif "model does not exist" in error_msg: error_msg = "The specified model is not available." return jsonify({'success': False, 'error': f'Error from Gemini API: {error_msg}'}), 500 # 500 Internal Server Error @app.route('/caches', methods=['GET']) def list_caches(): # Lists caches stored *in this application's memory*. # It does NOT list caches directly from the Gemini API unless you add that logic. try: caches = [] for cache_id, cache_info in list(document_caches.items()): # Use list() to iterate safely if modification occurs during iteration # Optional: Check if the cache still exists in Gemini API before listing # This adds complexity and potential API calls, so skipping for simple demo try: # Attempt to get cache metadata from API to confirm existence/details api_cache_info = client.caches.get(name=cache_info['gemini_cache_name']) # If successful, add to list caches.append({ 'cache_id': cache_id, # Our internal ID 'document_name': cache_info['document_name'], 'gemini_cache_name': cache_info['gemini_cache_name'], # Include Gemini name 'created_at': cache_info['created_at'], 'expires_at': getattr(api_cache_info, 'expire_time', 'Unknown'), # Get actual expiry from API 'cached_token_count': getattr(api_cache_info.usage_metadata, 'cached_token_count', 'Unknown') if hasattr(api_cache_info, 'usage_metadata') else 'Unknown' }) except Exception as e: # If API lookup fails (e.g., cache expired/deleted), remove from our local map print(f"Gemini cache {cache_info['gemini_cache_name']} for local ID {cache_id} not found via API. Removing from local storage. Error: {e}") del document_caches[cache_id] # Don't add it to the list of active caches return jsonify({'success': True, 'caches': caches}) except Exception as e: print(f"An error occurred listing caches: {str(e)}") return jsonify({'success': False, 'error': str(e)}) @app.route('/cache/', methods=['DELETE']) def delete_cache(cache_id): if client is None or api_key is None: return jsonify({'success': False, 'error': 'API key not configured or Gemini client failed to initialize.'}), 500 try: if cache_id not in document_caches: return jsonify({'success': False, 'error': 'Cache not found'}), 404 # 404 Not Found cache_info = document_caches[cache_id] gemini_cache_name_to_delete = cache_info['gemini_cache_name'] gemini_file_name_to_delete = cache_info['gemini_file_name'] # Delete from Gemini API Cache Service try: client.caches.delete(gemini_cache_name_to_delete) print(f"Gemini cache deleted: {gemini_cache_name_to_delete}") # Log except Exception as delete_error: error_msg = str(delete_error) print(f"Error deleting Gemini cache {gemini_cache_name_to_delete}: {error_msg}") # Log # Handle case where the cache was already gone (e.g. expired) if "Resource not found" in error_msg: print(f"Gemini cache {gemini_cache_name_to_delete} already gone from API.") else: # For other errors, you might want to stop and return the error return jsonify({'success': False, 'error': f'Failed to delete cache from API: {error_msg}'}), 500 # Also delete the associated file from Gemini File API to free up storage if gemini_file_name_to_delete: try: client.files.delete(gemini_file_name_to_delete) print(f"Associated Gemini file deleted: {gemini_file_name_to_delete}") # Log except Exception as file_delete_error: error_msg = str(file_delete_error) print(f"Error deleting Gemini file {gemini_file_name_to_delete}: {error_msg}") # Log if "Resource not found" in error_msg: print(f"Gemini file {gemini_file_name_to_delete} already gone from API.") else: # Log but continue, deleting the cache is the primary goal pass # Remove from local storage *after* attempting API deletion del document_caches[cache_id] print(f"Local cache entry deleted for ID: {cache_id}") # Log return jsonify({'success': True, 'message': 'Cache and associated file deleted successfully'}) except Exception as e: print(f"An unexpected error occurred during cache deletion process: {str(e)}") # Log return jsonify({'success': False, 'error': str(e)}), 500 if __name__ == '__main__': import os port = int(os.environ.get("PORT", 7860)) print(f"Starting Flask app on port {port}") # Log start # In production, set debug=False # Use threaded=True or a production WSGI server (like Gunicorn) for concurrent requests app.run(debug=True, host='0.0.0.0', port=port, threaded=True)

📚 Smart Document Analysis Platform

Or provide a URL:

✅ Document Cached Successfully!