from fastapi import HTTPException, Header, Depends from fastapi.security import APIKeyHeader from typing import Optional from app.config import settings import app.vertex.config as config import os import json from app.utils.logging import vertex_log # API Key security scheme api_key_header = APIKeyHeader(name="Authorization", auto_error=False) # Function to validate API key def validate_api_key(api_key_to_validate: str) -> bool: return True # Dependency for API key validation async def get_api_key(authorization: Optional[str] = Header(None)): if authorization is None: raise HTTPException( status_code=401, detail="Missing API key. Please include 'Authorization: Bearer YOUR_API_KEY' header." ) # Check if the header starts with "Bearer " if not authorization.startswith("Bearer "): raise HTTPException( status_code=401, detail="Invalid API key format. Use 'Authorization: Bearer YOUR_API_KEY'" ) # Extract the API key api_key = authorization.replace("Bearer ", "") # Validate the API key if not validate_api_key(api_key): # Call local validate_api_key raise HTTPException( status_code=401, detail="Invalid API key" ) return api_key def validate_settings(): """Validate settings for Vertex API access.""" # 检查API key api_key = None if hasattr(settings, 'API_KEY') and settings.API_KEY: api_key = settings.API_KEY else: api_key = config.API_KEY if not api_key: vertex_log('warning', "API key is not set. Some functionality may be limited.") # 检查Google credentials JSON google_credentials_json = None if hasattr(settings, 'GOOGLE_CREDENTIALS_JSON') and settings.GOOGLE_CREDENTIALS_JSON: google_credentials_json = settings.GOOGLE_CREDENTIALS_JSON else: google_credentials_json = config.GOOGLE_CREDENTIALS_JSON if google_credentials_json: try: # 尝试解析JSON确保其有效 json.loads(google_credentials_json) vertex_log('info', "Google Credentials JSON is valid") except json.JSONDecodeError: vertex_log('error', "Google Credentials JSON is not valid JSON. Please check the format.") return False # 检查project ID project_id = None if hasattr(settings, 'PROJECT_ID') and settings.PROJECT_ID: project_id = settings.PROJECT_ID else: project_id = config.PROJECT_ID if not project_id: vertex_log('warning', "Vertex AI Project ID is not set. Required for non-API key methods.") # 检查location location = None if hasattr(settings, 'LOCATION') and settings.LOCATION: location = settings.LOCATION else: location = config.LOCATION if not location: vertex_log('warning', "Vertex AI Location is not set, using default: us-central1") # 验证凭证目录 credentials_dir = None if hasattr(settings, 'CREDENTIALS_DIR') and settings.CREDENTIALS_DIR: credentials_dir = settings.CREDENTIALS_DIR else: credentials_dir = config.CREDENTIALS_DIR if not os.path.exists(credentials_dir): try: os.makedirs(credentials_dir, exist_ok=True) vertex_log('info', f"Created credentials directory at: {credentials_dir}") except Exception as e: vertex_log('error', f"Failed to create credentials directory: {e}") return False return True