main.py

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)