import os import re import uuid import json import logging import asyncio import time from collections import defaultdict from typing import List, Dict, Any, Optional, AsyncGenerator from datetime import datetime from aiohttp import ClientSession, ClientTimeout, ClientError from fastapi import FastAPI, HTTPException, Request, Depends, Header from fastapi.responses import StreamingResponse from pydantic import BaseModel # Configure logging logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", handlers=[logging.StreamHandler()] ) logger = logging.getLogger(__name__) # Load environment variables API_KEYS = os.getenv('API_KEYS', '').split(',') # Comma-separated API keys RATE_LIMIT_PER_MINUTE = int(os.getenv('RATE_LIMIT_PER_MINUTE', '60')) # Requests per minute per IP if not API_KEYS or API_KEYS == ['']: logger.error("No API keys found. Please set the API_KEYS environment variable.") raise Exception("API_KEYS environment variable not set.") # Simple in-memory rate limiter per IP rate_limit_store_ip = defaultdict(lambda: {"count": 0, "timestamp": time.time()}) async def rate_limiter(request: Request): client_host = request.client.host current_time = time.time() window_start = rate_limit_store_ip[client_host]["timestamp"] if current_time - window_start > 60: rate_limit_store_ip[client_host] = {"count": 1, "timestamp": current_time} else: if rate_limit_store_ip[client_host]["count"] >= RATE_LIMIT_PER_MINUTE: logger.warning(f"Rate limit exceeded for IP: {client_host}") raise HTTPException(status_code=429, detail='Rate limit exceeded.') rate_limit_store_ip[client_host]["count"] += 1 # Custom exception for model not working class ModelNotWorkingException(Exception): def __init__(self, model: str): self.model = model self.message = f"The model '{model}' is currently not working. Please try another model or wait for it to be fixed." super().__init__(self.message) # Mock implementations for ImageResponse and to_data_uri class ImageResponse: def __init__(self, url: str, alt: str): self.url = url self.alt = alt def to_data_uri(image: Any) -> str: return "data:image/png;base64,..." # Replace with actual base64 data class Blackbox: url = "https://www.blackbox.ai" api_endpoint = "https://www.blackbox.ai/api/chat" working = True supports_stream = True supports_system_message = True supports_message_history = True default_model = 'blackboxai' image_models = ['ImageGeneration'] models = [ default_model, 'blackboxai-pro', "llama-3.1-8b", 'llama-3.1-70b', 'llama-3.1-405b', 'gpt-4o', 'gemini-pro', 'gemini-1.5-flash', 'claude-sonnet-3.5', 'PythonAgent', 'JavaAgent', 'JavaScriptAgent', 'HTMLAgent', 'GoogleCloudAgent', 'AndroidDeveloper', 'SwiftDeveloper', 'Next.jsAgent', 'MongoDBAgent', 'PyTorchAgent', 'ReactAgent', 'XcodeAgent', 'AngularJSAgent', *image_models, 'Niansuh', ] agentMode = { 'ImageGeneration': {'mode': True, 'id': "ImageGenerationLV45LJp", 'name': "Image Generation"}, 'Niansuh': {'mode': True, 'id': "NiansuhAIk1HgESy", 'name': "Niansuh"}, } trendingAgentMode = { "blackboxai": {}, "gemini-1.5-flash": {'mode': True, 'id': 'Gemini'}, "llama-3.1-8b": {'mode': True, 'id': "llama-3.1-8b"}, 'llama-3.1-70b': {'mode': True, 'id': "llama-3.1-70b"}, 'llama-3.1-405b': {'mode': True, 'id': "llama-3.1-405b"}, 'blackboxai-pro': {'mode': True, 'id': "BLACKBOXAI-PRO"}, 'PythonAgent': {'mode': True, 'id': "Python Agent"}, 'JavaAgent': {'mode': True, 'id': "Java Agent"}, 'JavaScriptAgent': {'mode': True, 'id': "JavaScript Agent"}, 'HTMLAgent': {'mode': True, 'id': "HTML Agent"}, 'GoogleCloudAgent': {'mode': True, 'id': "Google Cloud Agent"}, 'AndroidDeveloper': {'mode': True, 'id': "Android Developer"}, 'SwiftDeveloper': {'mode': True, 'id': "Swift Developer"}, 'Next.jsAgent': {'mode': True, 'id': "Next.js Agent"}, 'MongoDBAgent': {'mode': True, 'id': "MongoDB Agent"}, 'PyTorchAgent': {'mode': True, 'id': "PyTorch Agent"}, 'ReactAgent': {'mode': True, 'id': "React Agent"}, 'XcodeAgent': {'mode': True, 'id': "Xcode Agent"}, 'AngularJSAgent': {'mode': True, 'id": "AngularJS Agent"}, } userSelectedModel = { "gpt-4o": "gpt-4o", "gemini-pro": "gemini-pro", 'claude-sonnet-3.5': "claude-sonnet-3.5", } model_prefixes = { 'gpt-4o': '@GPT-4o', 'gemini-pro': '@Gemini-PRO', 'claude-sonnet-3.5': '@Claude-Sonnet-3.5', 'PythonAgent': '@Python Agent', 'JavaAgent': '@Java Agent', 'JavaScriptAgent': '@JavaScript Agent', 'HTMLAgent': '@HTML Agent', 'GoogleCloudAgent': '@Google Cloud Agent', 'AndroidDeveloper': '@Android Developer', 'SwiftDeveloper': '@Swift Developer', 'Next.jsAgent': '@Next.js Agent', 'MongoDBAgent': '@MongoDB Agent', 'PyTorchAgent': '@PyTorch Agent', 'ReactAgent': '@React Agent', 'XcodeAgent': '@Xcode Agent', 'AngularJSAgent': '@AngularJS Agent', 'blackboxai-pro': '@BLACKBOXAI-PRO', 'ImageGeneration': '@Image Generation', 'Niansuh': '@Niansuh', } model_referers = { "blackboxai": f"{url}/?model=blackboxai", "gpt-4o": f"{url}/?model=gpt-4o", "gemini-pro": f"{url}/?model=gemini-pro", "claude-sonnet-3.5": f"{url}/?model=claude-sonnet-3.5" } model_aliases = { "gemini-flash": "gemini-1.5-flash", "claude-3.5-sonnet": "claude-sonnet-3.5", "flux": "ImageGeneration", "niansuh": "Niansuh", } @classmethod def get_model(cls, model: str) -> str: if model in cls.models: return model elif model in cls.userSelectedModel: return model elif model in cls.model_aliases: return cls.model_aliases[model] else: return cls.default_model @classmethod async def create_async_generator( cls, model: str, messages: List[Dict[str, str]], proxy: Optional[str] = None, image: Any = None, image_name: Optional[str] = None, webSearchMode: bool = False, **kwargs ) -> AsyncGenerator[Any, None]: model = cls.get_model(model) logger.info(f"Selected model: {model}") if not cls.working or model not in cls.models: logger.error(f"Model {model} is not working or not supported.") raise ModelNotWorkingException(model) headers = { "accept": "*/*", "accept-language": "en-US,en;q=0.9", "cache-control": "no-cache", "content-type": "application/json", "origin": cls.url, "pragma": "no-cache", "priority": "u=1, i", "referer": cls.model_referers.get(model, cls.url), "sec-ch-ua": '"Chromium";v="129", "Not=A?Brand";v="8"', "sec-ch-ua-mobile": "?0", "sec-ch-ua-platform": '"Linux"', "sec-fetch-dest": "empty", "sec-fetch-mode": "cors", "sec-fetch-site": "same-origin", "user-agent": "Mozilla/5.0 (X11; Linux x86_64)", } if model in cls.model_prefixes: prefix = cls.model_prefixes[model] if not messages[0]['content'].startswith(prefix): logger.debug(f"Adding prefix '{prefix}' to the first message.") messages[0]['content'] = f"{prefix} {messages[0]['content']}" random_id = ''.join(random.choices(string.ascii_letters + string.digits, k=7)) messages[-1]['id'] = random_id messages[-1]['role'] = 'user' # Don't log the full message content for privacy logger.debug(f"Generated message ID: {random_id} for model: {model}") if image is not None: messages[-1]['data'] = { 'fileText': '', 'imageBase64': to_data_uri(image), 'title': image_name } messages[-1]['content'] = 'FILE:BB\n$#$\n\n$#$\n' + messages[-1]['content'] logger.debug("Image data added to the message.") data = { "messages": messages, "id": random_id, "previewToken": None, "userId": None, "codeModelMode": True, "agentMode": {}, "trendingAgentMode": {}, "isMicMode": False, "userSystemPrompt": None, "maxTokens": 99999999, "playgroundTopP": 0.9, "playgroundTemperature": 0.5, "isChromeExt": False, "githubToken": None, "clickedAnswer2": False, "clickedAnswer3": False, "clickedForceWebSearch": False, "visitFromDelta": False, "mobileClient": False, "userSelectedModel": None, "webSearchMode": webSearchMode, } if model in cls.agentMode: data["agentMode"] = cls.agentMode[model] elif model in cls.trendingAgentMode: data["trendingAgentMode"] = cls.trendingAgentMode[model] elif model in cls.userSelectedModel: data["userSelectedModel"] = cls.userSelectedModel[model] logger.info(f"Sending request to {cls.api_endpoint} with data (excluding messages).") timeout = ClientTimeout(total=30) # Reduced timeout for faster response retry_attempts = 3 # Reduced retry attempts for faster failure handling for attempt in range(retry_attempts): try: async with ClientSession(headers=headers, timeout=timeout) as session: async with session.post(cls.api_endpoint, json=data, proxy=proxy) as response: response.raise_for_status() logger.info(f"Received response with status {response.status}") if model == 'ImageGeneration': response_text = await response.text() url_match = re.search(r'https://storage\.googleapis\.com/[^\s\)]+', response_text) if url_match: image_url = url_match.group(0) logger.info(f"Image URL found.") yield ImageResponse(image_url, alt=messages[-1]['content']) else: logger.error("Image URL not found in the response.") raise Exception("Image URL not found in the response") else: full_response = "" search_results_json = "" try: async for chunk, _ in response.content.iter_chunks(): if chunk: decoded_chunk = chunk.decode(errors='ignore') decoded_chunk = re.sub(r'\$@\$v=[^$]+\$@\$', '', decoded_chunk) if decoded_chunk.strip(): if '$~~~$' in decoded_chunk: search_results_json += decoded_chunk else: full_response += decoded_chunk yield decoded_chunk logger.info("Finished streaming response chunks.") except Exception as e: logger.exception("Error while iterating over response chunks.") raise e if data["webSearchMode"] and search_results_json: match = re.search(r'\$~~~\$(.*?)\$~~~\$', search_results_json, re.DOTALL) if match: try: search_results = json.loads(match.group(1)) formatted_results = "\n\n**Sources:**\n" for i, result in enumerate(search_results[:5], 1): formatted_results += f"{i}. [{result['title']}]({result['link']})\n" logger.info("Formatted search results.") yield formatted_results except json.JSONDecodeError as je: logger.error("Failed to parse search results JSON.") raise je break # Exit the retry loop if successful except ClientError as ce: logger.error(f"Client error occurred: {ce}. Retrying attempt {attempt + 1}/{retry_attempts}") if attempt == retry_attempts - 1: raise HTTPException(status_code=502, detail="Error communicating with the external API.") except asyncio.TimeoutError: logger.error(f"Request timed out. Retrying attempt {attempt + 1}/{retry_attempts}") if attempt == retry_attempts - 1: raise HTTPException(status_code=504, detail="External API request timed out.") except Exception as e: logger.error(f"Unexpected error: {e}. Retrying attempt {attempt + 1}/{retry_attempts}") if attempt == retry_attempts - 1: raise HTTPException(status_code=500, detail=str(e)) # FastAPI app setup app = FastAPI() # Implement per-IP rate limiting middleware @app.middleware("http") async def rate_limit_middleware(request: Request, call_next): await rate_limiter(request) response = await call_next(request) return response # Pydantic models for OpenAI API class Message(BaseModel): role: str content: str class ChatCompletionRequest(BaseModel): model: str messages: List[Message] temperature: Optional[float] = 1.0 top_p: Optional[float] = 1.0 n: Optional[int] = 1 stream: Optional[bool] = False stop: Optional[Any] = None # Can be a string or list of strings max_tokens: Optional[int] = None presence_penalty: Optional[float] = 0.0 frequency_penalty: Optional[float] = 0.0 logit_bias: Optional[Dict[str, float]] = None user: Optional[str] = None def create_chat_completion_response(content: str, model: str, usage: Dict[str, int]) -> Dict[str, Any]: return { "id": f"chatcmpl-{uuid.uuid4()}", "object": "chat.completion", "created": int(datetime.now().timestamp()), "model": model, "choices": [ { "index": 0, "message": { "role": "assistant", "content": content }, "finish_reason": "stop" } ], "usage": usage } def create_stream_response_chunk(content: str, role: Optional[str] = None, finish_reason: Optional[str] = None): delta = {} if role: delta['role'] = role if content: delta['content'] = content return { "object": "chat.completion.chunk", "created": int(datetime.now().timestamp()), "model": "", # Model name can be added if necessary "choices": [ { "delta": delta, "index": 0, "finish_reason": finish_reason } ] } @app.post("/v1/chat/completions") async def chat_completions(request: ChatCompletionRequest, authorization: str = Header(None)): # Verify API key if not authorization or not authorization.startswith('Bearer '): logger.warning("Invalid authorization header format.") raise HTTPException(status_code=401, detail='Invalid authorization header format.') api_key = authorization[7:] if api_key not in API_KEYS: logger.warning(f"Invalid API key attempted: {api_key}") raise HTTPException(status_code=401, detail='Invalid API key.') logger.info(f"Received chat completion request for model: {request.model}") # Validate model if request.model not in Blackbox.models and request.model not in Blackbox.model_aliases: logger.warning(f"Attempt to use unavailable model: {request.model}") raise HTTPException(status_code=400, detail="The model is not available.") # Process the request try: # Convert messages to dicts messages = [msg.dict() for msg in request.messages] # Check if the user is requesting image generation image_generation_requested = any( re.search(r'\b(generate|create|draw)\b.*\b(image|picture|art)\b', msg['content'], re.IGNORECASE) for msg in messages if msg['role'] == 'user' ) if image_generation_requested: model = 'ImageGeneration' # For image generation, use the last message as prompt prompt = messages[-1]['content'] # Build messages for the Blackbox.create_async_generator messages = [{"role": "user", "content": prompt}] async_generator = Blackbox.create_async_generator( model=model, messages=messages, image=None, image_name=None, webSearchMode=False ) # Collect images images = [] count = 0 async for response in async_generator: if isinstance(response, ImageResponse): images.append(response.url) count += 1 if count >= request.n: break # Build response content with image URLs response_content = "\n".join(f"![Generated Image]({url})" for url in images) completion_tokens = len(response_content.split()) else: # Use the requested model async_generator = Blackbox.create_async_generator( model=request.model, messages=messages, image=None, image_name=None, webSearchMode=False ) # Usage tracking completion_tokens = 0 # Will be updated as we process the response prompt_tokens = sum(len(msg['content'].split()) for msg in messages) if request.stream: async def generate(): nonlocal completion_tokens try: # Initial delta with role initial_chunk = create_stream_response_chunk(content=None, role="assistant") yield f"data: {json.dumps(initial_chunk)}\n\n" async for chunk in async_generator: if isinstance(chunk, str): completion_tokens += len(chunk.split()) response_chunk = create_stream_response_chunk(content=chunk) yield f"data: {json.dumps(response_chunk)}\n\n" elif isinstance(chunk, ImageResponse): content = f"![Generated Image]({chunk.url})" completion_tokens += len(content.split()) response_chunk = create_stream_response_chunk(content=content) yield f"data: {json.dumps(response_chunk)}\n\n" else: pass # Handle other types if necessary # Finish reason final_chunk = create_stream_response_chunk(content=None, finish_reason="stop") yield f"data: {json.dumps(final_chunk)}\n\n" yield "data: [DONE]\n\n" except Exception as e: logger.exception("Error during streaming response generation.") yield f"data: {json.dumps({'error': str(e)})}\n\n" return StreamingResponse(generate(), media_type="text/event-stream") else: response_content = "" async for chunk in async_generator: if isinstance(chunk, str): response_content += chunk elif isinstance(chunk, ImageResponse): response_content += f"![Generated Image]({chunk.url})\n" completion_tokens = len(response_content.split()) usage = { "prompt_tokens": prompt_tokens, "completion_tokens": completion_tokens, "total_tokens": prompt_tokens + completion_tokens } return create_chat_completion_response(response_content, request.model, usage) except ModelNotWorkingException as e: logger.warning(f"Model not working: {e}") raise HTTPException(status_code=503, detail=str(e)) except Exception as e: logger.exception("An unexpected error occurred while processing the chat completions request.") raise HTTPException(status_code=500, detail=str(e)) @app.get("/v1/models") async def get_models(authorization: str = Header(None)): # Verify API key if not authorization or not authorization.startswith('Bearer '): logger.warning("Invalid authorization header format.") raise HTTPException(status_code=401, detail='Invalid authorization header format.') api_key = authorization[7:] if api_key not in API_KEYS: logger.warning(f"Invalid API key attempted: {api_key}") raise HTTPException(status_code=401, detail='Invalid API key.') logger.info("Fetching available models.") # Return models in OpenAI format models_data = [{"id": model, "object": "model", "owned_by": "organization-owner", "permission": []} for model in Blackbox.models] return {"data": models_data, "object": "list"} if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)