Create app.py

app.py

@@ -0,0 +1,168 @@
+import os
+import re
+from fastapi import FastAPI, HTTPException, Body
+from fastapi.responses import StreamingResponse
+from pydantic import BaseModel, Field
+import openai # For your custom API
+import google.generativeai as genai # For Gemini API
+
+# --- Configuration ---
+# API keys are now expected to be set as Hugging Face Space secrets.
+# GEMINI_API_KEY for Gemini
+# CUSTOM_API_SECRET_KEY for your custom API
+# CUSTOM_API_BASE_URL for your custom API (can also be a secret if it changes)
+# CUSTOM_API_MODEL for your custom API (can also be a secret)
+
+CUSTOM_API_BASE_URL_DEFAULT = "https://api-q3ieh5raqfuad9o8.aistudio-app.com/v1"
+CUSTOM_API_MODEL_DEFAULT = "gemma3:27b"
+
+# --- Pydantic Models for Request Body ---
+class ChatRequest(BaseModel):
+    message: str
+    api_key: str = Field(None, description="Gemini API Key (optional; will use Space secret 'GEMINI_API_KEY' if not provided)")
+    url: str = Field(None, description="URL (optional, if it's a YouTube or googleusercontent video URL, Gemini will be used)")
+
+# --- FastAPI App Initialization ---
+app = FastAPI(
+    title="Universal Chat API (Secrets Optimized)",
+    description="A Hugging Face Space that routes chat requests to a custom API or Gemini API (for video URLs) using Space secrets for API keys.",
+    version="1.1.0"
+)
+
+# --- Helper Functions ---
+def is_video_url_for_gemini(url: str) -> bool:
+    """
+    Checks if the given URL is a YouTube video URL or a googleusercontent URI
+    suitable for Gemini multimodal processing.
+    """
+    if not url:
+        return False
+    # Regex to match standard YouTube video URLs
+    youtube_regex = (
+        r'(https_?://)?(www\.)?'
+        '(youtube|youtu|youtube-nocookie)\.(com|be)/'
+        '(watch\?v=|embed/|v/|.+\?v=)?([^&=%\?]{11})')
+    
+    # Check for googleusercontent YouTube URIs
+    googleusercontent_youtube_regex = r'https_?://googleusercontent\.com/youtube\.com/\w+'
+    
+    return re.match(youtube_regex, url) is not None or \
+           re.match(googleusercontent_youtube_regex, url) is not None
+
+# --- API Endpoint ---
+@app.post("/chat")
+async def chat_endpoint(request: ChatRequest):
+    """
+    Chat endpoint that routes to Gemini for video URLs or a custom API otherwise.
+    API keys are primarily sourced from Hugging Face Space secrets.
+    """
+    user_message = request.message
+    gemini_api_key_from_request = request.api_key
+    input_url = request.url
+
+    # Determine API keys from secrets, allowing override from request for Gemini
+    gemini_api_key_secret = os.getenv("GEMINI_API_KEY")
+    custom_api_key_secret = os.getenv("CUSTOM_API_SECRET_KEY")
+    
+    # Determine custom API base URL and model from secrets or use defaults
+    custom_api_base_url = os.getenv("CUSTOM_API_BASE_URL", CUSTOM_API_BASE_URL_DEFAULT)
+    custom_api_model = os.getenv("CUSTOM_API_MODEL", CUSTOM_API_MODEL_DEFAULT)
+
+    if input_url and is_video_url_for_gemini(input_url):
+        # --- Gemini API Logic for Video URLs ---
+        gemini_key_to_use = gemini_api_key_from_request or gemini_api_key_secret
+        
+        if not gemini_key_to_use:
+            raise HTTPException(
+                status_code=400, 
+                detail="Gemini API Key is required for video URL processing. "
+                       "Provide it in the request or set 'GEMINI_API_KEY' as a Space secret."
+            )
+        
+        try:
+            genai.configure(api_key=gemini_key_to_use)
+        except Exception as e:
+            raise HTTPException(status_code=500, detail=f"Failed to configure Gemini API: {e}")
+
+        # Prepare the content structure for Gemini multimodal input
+        gemini_contents = [{
+            "parts": [
+                {"text": user_message}, # User's prompt/question
+                {
+                    "file_data": {
+                        "mime_type": "video/youtube", # Common practice, though Gemini might infer
+                        "file_uri": input_url
+                    }
+                }
+            ]
+        }]
+
+        async def gemini_streamer():
+            try:
+                # Model supporting video input, e.g., 'gemini-1.5-flash' or 'gemini-1.5-pro'
+                model = genai.GenerativeModel(model_name="gemini-1.5-flash") 
+                response = await model.generate_content_async(gemini_contents, stream=True)
+                async for chunk in response:
+                    if chunk.text:
+                        yield chunk.text
+            except Exception as e:
+                error_message = f"Error during Gemini API call: {str(e)}"
+                print(error_message)
+                # Check for specific Gemini API errors if needed, e.g., content filtering
+                if "response.prompt_feedback" in str(e).lower(): # Crude check for safety feedback
+                    yield "The response was blocked due to safety settings by the Gemini API."
+                else:
+                    yield error_message
+        return StreamingResponse(gemini_streamer(), media_type="text/plain")
+
+    else:
+        # --- Custom API Logic ---
+        if not custom_api_key_secret:
+            raise HTTPException(
+                status_code=500, 
+                detail="Custom API key ('CUSTOM_API_SECRET_KEY') is not configured in Space secrets."
+            )
+        if not custom_api_base_url:
+             raise HTTPException(
+                status_code=500, 
+                detail="Custom API base URL ('CUSTOM_API_BASE_URL') is not configured in Space secrets or defaults."
+            )
+
+        client = openai.OpenAI(
+            api_key=custom_api_key_secret,
+            base_url=custom_api_base_url
+        )
+
+        async def custom_api_streamer():
+            try:
+                stream = client.chat.completions.create(
+                    model=custom_api_model,
+                    temperature=0.6,
+                    messages=[
+                        {"role": "user", "content": user_message}
+                    ],
+                    stream=True
+                )
+                for chunk in stream:
+                    if hasattr(chunk.choices[0].delta, "reasoning_content") and chunk.choices[0].delta.reasoning_content:
+                        yield chunk.choices[0].delta.reasoning_content
+                    elif chunk.choices[0].delta and chunk.choices[0].delta.content:
+                        yield chunk.choices[0].delta.content
+            except Exception as e:
+                error_message = f"Error during Custom API call: {str(e)}"
+                print(error_message)
+                yield error_message
+        return StreamingResponse(custom_api_streamer(), media_type="text/plain")
+
+# --- Root Endpoint for Health Check (Optional) ---
+@app.get("/")
+async def read_root():
+    return {"message": "Chat API (Secrets Optimized) is running. Use the /chat endpoint to interact."}
+
+# --- To run locally (optional, Hugging Face Spaces handles this) ---
+# if __name__ == "__main__":
+#     import uvicorn
+#     # For local testing, you might want to load .env files if you use them
+#     # from dotenv import load_dotenv
+#     # load_dotenv() 
+#     uvicorn.run(app, host="0.0.0.0", port=8000)