import logging import re import time import random from google import genai from google.genai import types from openai import OpenAI import requests logger = logging.getLogger(__name__) GEMINI_FALLBACK_MODELS = ['gemini-2.5-flash', 'gemini-2.0-flash', 'gemini-1.5-flash', 'gemini-1.5-pro'] # Increase retries and use exponential backoff with jitter for transient errors MAX_RETRIES = 5 RETRY_DELAY_SECONDS = 2 def _is_rate_limit_error(error: Exception) -> bool: if hasattr(error, 'code'): try: if error.code in [429, 503]: return True if error.code in [400, 401, 403, 404]: return False except Exception: pass error_text = str(error).lower() # Broad detection of temporary/unavailable/rate-limit/quota messages # Avoid raw 'rate' to prevent matching 'generate' keywords = ['503', 'unavailable', '429', 'rate limit', 'rate-limit', 'overloaded', 'busy', 'quota', 'resource_exhausted', 'resource exhausted'] return any(k in error_text for k in keywords) def _fetch_google_sheet_csv(url: str) -> str: """Fetches public Google Sheet data as CSV and returns it as a formatted string context.""" if not url: return "" try: import requests import csv from io import StringIO import re # Convert /edit or /view URL to /export?format=csv csv_url = re.sub(r'/(edit|view).*$', '/export?format=csv', url) if '/export?format=csv' not in csv_url: csv_url = url + '/export?format=csv' # Fetch CSV with a tight timeout to prevent slowing down chat responses response = requests.get(csv_url, timeout=3.0) response.raise_for_status() # Parse CSV (up to 150 rows to limit context size) csv_text = response.text f = StringIO(csv_text) reader = csv.reader(f) lines = [] for i, row in enumerate(reader): if i > 150: break if any(cell.strip() for cell in row): lines.append(" | ".join(cell.strip() for cell in row)) if lines: return "\n\n--- GOOGLE SHEET LIVE DATA ---\n" + "\n".join(lines) + "\n------------------------------\n" return "" except Exception as e: print(f"Failed to fetch Google Sheet CSV: {e}") return "" def _get_language_instruction(prompt_text: str) -> str: if re.search(r'[\u0980-\u09FF]', prompt_text): return ( "\n[LANGUAGE INSTRUCTION] The user is communicating in Bengali (বাংলা). " "You MUST respond entirely in Bengali using proper Bengali script." ) return "" def _download_media(url: str): try: if url.startswith('/'): import os import mimetypes from django.conf import settings # Remove chatpro prefix if present for local resolution local_path = url if local_path.startswith('/chatpro/media/'): local_path = local_path.replace('/chatpro/media/', '/media/', 1) if local_path.startswith('/media/'): # Map URL to filesystem path file_path = os.path.join(settings.BASE_DIR, local_path.lstrip('/')) if os.path.exists(file_path): with open(file_path, 'rb') as f: content = f.read() content_type, _ = mimetypes.guess_type(file_path) return content, content_type or 'application/octet-stream' logger.info(f"Skipping download for unresolvable local url: {url}") return None, None response = requests.get(url, timeout=10) response.raise_for_status() content_type = response.headers.get('Content-Type', '') return response.content, content_type except Exception as e: logger.error(f"Failed to download media from {url}: {e}") return None, None def _call_gemini(client, model: str, contents: list, system_prompt: str): config = types.GenerateContentConfig( system_instruction=system_prompt if system_prompt else None, temperature=0.7, ) response = client.models.generate_content( model=model, contents=contents, config=config ) if not response.text: raise RuntimeError("Received an empty response from Gemini API.") return response.text def get_gemini_response(api_key: str, model_name: str, system_prompt: str, prompt_text: str, message_type: str = 'text', attachment_url: str = None, history: list = None, google_sheet_url: str = None) -> str: """ Generates a response from Google Gemini using the new google-genai SDK. Retries on 503/429 errors and falls back to alternate models. """ if not api_key: raise ValueError("Gemini API key is missing. Please configure it in the chatbot settings.") sheet_context = _fetch_google_sheet_csv(google_sheet_url) language_instruction = _get_language_instruction(prompt_text) effective_prompt = (system_prompt or "") + sheet_context + language_instruction primary_model = model_name or 'gemini-2.0-flash' models_to_try = [primary_model] for fallback in ['gemini-1.5-flash', 'gemini-1.5-pro']: if fallback not in models_to_try: models_to_try.append(fallback) last_error = None client = genai.Client(api_key=api_key) contents = [] if history: for msg in history: role = 'user' if msg.get('role') == 'user' else 'model' parts = [] if msg.get('content'): parts.append(types.Part.from_text(text=msg['content'])) # Skip downloading media files for old chat history to prevent HTTP request delays. # Instead, append a placeholder indicating that a file was sent. if msg.get('attachment_url'): msg_type = msg.get('message_type') or 'attachment' placeholder = f" [{msg_type.upper()}]" if parts: existing_text = parts[0].text parts[0] = types.Part.from_text(text=existing_text + placeholder) else: parts.append(types.Part.from_text(text=placeholder.strip())) if parts: contents.append(types.Content(role=role, parts=parts)) current_parts = [] if prompt_text: current_parts.append(types.Part.from_text(text=prompt_text)) elif not prompt_text and attachment_url: current_parts.append(types.Part.from_text(text="Please analyze the attached media.")) if attachment_url: media_bytes, mime_type = _download_media(attachment_url) if media_bytes: if not mime_type: if message_type == 'image': mime_type = 'image/jpeg' elif message_type == 'voice': mime_type = 'audio/mp3' else: mime_type = 'application/octet-stream' if 'audio/mp4' in mime_type or 'video/mp4' in mime_type: if message_type == 'voice': mime_type = 'audio/mp4' current_parts.append(types.Part.from_bytes(data=media_bytes, mime_type=mime_type)) if current_parts: contents.append(types.Content(role='user', parts=current_parts)) if not contents: contents.append(types.Content(role='user', parts=[types.Part.from_text(text="Hello")])) for model in models_to_try: for attempt in range(MAX_RETRIES): try: return _call_gemini(client, model, contents, effective_prompt) except Exception as e: last_error = e logger.warning(f"Gemini attempt {attempt + 1}/{MAX_RETRIES} failed for model {model}: {e}") # If transient rate/overload error, retry with exponential backoff + jitter if _is_rate_limit_error(e) and attempt < MAX_RETRIES - 1: backoff = RETRY_DELAY_SECONDS * (2 ** attempt) jitter = random.uniform(0, 1) sleep_for = backoff + jitter logger.info(f"Transient Gemini error detected, retrying after {sleep_for:.1f}s (attempt {attempt+2})") time.sleep(sleep_for) continue # non-retryable or last attempt -> break to try next model break error_msg = str(last_error) if _is_rate_limit_error(last_error): raise RuntimeError( "Google Gemini is temporarily overloaded. Please wait a moment and try again. " "(জেমিনি সার্ভার এখন ব্যস্ত। কিছুক্ষণ পর আবার চেষ্টা করুন।)" ) logger.error(f"Gemini API Error: {error_msg}") raise RuntimeError(f"Google Gemini Error: {error_msg}") def get_openai_response(api_key: str, model_name: str, system_prompt: str, prompt_text: str, history=None, message_type: str = 'text', attachment_url: str = None, google_sheet_url: str = None) -> str: """ Generates a response from OpenAI. """ if not api_key: raise ValueError("OpenAI API key is missing. Please configure it in the chatbot settings.") try: client = OpenAI(api_key=api_key) sheet_context = _fetch_google_sheet_csv(google_sheet_url) # Audio handling (Whisper) if message_type == 'voice' and attachment_url: media_bytes, _ = _download_media(attachment_url) if media_bytes: import tempfile import os # whisper requires a file-like object with a name ending in an audio extension with tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") as tmp: tmp.write(media_bytes) tmp_path = tmp.name try: with open(tmp_path, "rb") as audio_file: transcript = client.audio.transcriptions.create( model="whisper-1", file=audio_file ) prompt_text = f"{prompt_text}\n\n[User Voice Message Transcription]: {transcript.text}".strip() except Exception as ex: logger.error(f"Whisper transcription failed: {ex}") finally: if os.path.exists(tmp_path): os.remove(tmp_path) attachment_url = None # Audio is handled, don't pass URL to chat completion messages = [] effective_prompt = (system_prompt or "") + sheet_context if effective_prompt.strip(): messages.append({"role": "system", "content": effective_prompt}) if history: for msg in history: role = 'user' if msg.get('role') == 'user' else 'assistant' if msg.get('attachment_url') and msg.get('message_type') == 'image': content_arr = [] if msg.get('content'): content_arr.append({"type": "text", "text": msg['content']}) content_arr.append({ "type": "image_url", "image_url": {"url": msg['attachment_url']} }) messages.append({"role": role, "content": content_arr}) elif msg.get('content'): messages.append({"role": role, "content": msg['content']}) language_instruction = _get_language_instruction(prompt_text) if language_instruction: messages.append({"role": "system", "content": language_instruction.strip()}) # Image handling if attachment_url and message_type == 'image': messages.append({ "role": "user", "content": [ {"type": "text", "text": prompt_text if prompt_text else "Please analyze this image."}, { "type": "image_url", "image_url": { "url": attachment_url, }, }, ] }) else: messages.append({"role": "user", "content": prompt_text}) model = model_name or 'gpt-4o-mini' response = client.chat.completions.create( model=model, messages=messages, temperature=0.7 ) return response.choices[0].message.content except Exception as e: logger.error(f"OpenAI API Error: {str(e)}") raise RuntimeError(f"OpenAI Error: {str(e)}")