Create app.py

app.py

@@ -0,0 +1,232 @@
+from fastapi import FastAPI, HTTPException, Request
+from fastapi.responses import JSONResponse, StreamingResponse
+from fastapi.middleware.cors import CORSMiddleware
+import httpx
+import json
+import os
+import time
+import asyncio
+
+app = FastAPI()
+
+# CORS Configuration
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# Environment Configuration
+STATUS_URL = os.environ.get("STATUS_URL", "https://duckduckgo.com/duckchat/v1/status")
+CHAT_URL = os.environ.get("CHAT_URL", "https://duckduckgo.com/duckchat/v1/chat")
+REFERER = os.environ.get("REFERER", "https://duckduckgo.com/")
+ORIGIN = os.environ.get("ORIGIN", "https://duckduckgo.com")
+USER_AGENT = os.environ.get("USER_AGENT", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/127.0.0.0 Safari/537.36")
+COOKIE = os.environ.get("COOKIE", "dcm=3; s=l; bf=1")
+
+DEFAULT_HEADERS = {
+    "User-Agent": USER_AGENT,
+    "Accept": "text/event-stream",
+    "Accept-Language": "en-US,en;q=0.5",
+    "Referer": REFERER,
+    "Content-Type": "application/json",
+    "Origin": ORIGIN,
+    "Connection": "keep-alive",
+    "Cookie": COOKIE,
+    "Sec-Fetch-Dest": "empty",
+    "Sec-Fetch-Mode": "cors",
+    "Sec-Fetch-Site": "same-origin",
+    "Pragma": "no-cache",
+    "TE": "trailers",
+}
+
+SUPPORTED_MODELS = ["o3-mini", "gpt-4o-mini", "claude-3-haiku-20240307", "meta-llama/Llama-3.3-70B-Instruct-Turbo"]
+TIMEOUT = 30.0  # Seconds
+
+async def get_vqd():
+    """Fetch DuckDuckGo authentication token."""
+    try:
+        async with httpx.AsyncClient() as client:
+            response = await client.get(
+                STATUS_URL,
+                headers={**DEFAULT_HEADERS, "x-vqd-accept": "1"},
+                timeout=10.0
+            )
+            response.raise_for_status()
+            vqd = response.headers.get("x-vqd-4")
+            if not vqd:
+                raise ValueError("Missing x-vqd-4 header in response")
+            return vqd
+    except httpx.HTTPStatusError as e:
+        raise HTTPException(status_code=e.response.status_code, detail=f"VQD fetch failed: {str(e)}")
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"VQD error: {str(e)}")
+
+async def duckduckgo_chat_stream(model: str, messages: list):
+    """Handle streaming chat response."""
+    try:
+        x_vqd_4 = await get_vqd()
+        chat_headers = {
+            **DEFAULT_HEADERS,
+            "x-vqd-4": x_vqd_4,
+            "Accept": "text/event-stream",
+        }
+
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                CHAT_URL,
+                headers=chat_headers,
+                json={"model": model, "messages": messages},
+                timeout=TIMEOUT
+            )
+            response.raise_for_status()
+
+            async def event_generator():
+                try:
+                    async for chunk in response.aiter_bytes():
+                        decoded_chunk = chunk.decode('utf-8')
+                        for line in decoded_chunk.split('\n'):
+                            line = line.strip()
+                            if line.startswith("data: "):
+                                try:
+                                    data = json.loads(line[5:])
+                                    if "error" in data:
+                                        yield f"data: {json.dumps({'error': data['error']})}\n\n"
+                                        return
+
+                                    message = data.get("message", "")
+                                    if not message:
+                                        continue
+
+                                    yield format_openai_chunk(message, model)
+                                    await asyncio.sleep(0.001)  # Rate limit
+                                except json.JSONDecodeError as e:
+                                    yield f"data: {json.dumps({'error': f'JSON error: {str(e)}'})}\n\n"
+                                    return
+                except Exception as e:
+                    yield f"data: {json.dumps({'error': f'Stream error: {str(e)}'})}\n\n"
+                finally:
+                    yield "data: [DONE]\n\n"
+
+            return StreamingResponse(event_generator(), media_type="text/event-stream")
+
+    except httpx.HTTPStatusError as e:
+        raise HTTPException(status_code=e.response.status_code, detail=f"Chat error: {str(e)}")
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Chat error: {str(e)}")
+
+async def duckduckgo_chat_non_stream(model: str, messages: list):
+    """Handle non-streaming chat response."""
+    try:
+        x_vqd_4 = await get_vqd()
+        chat_headers = {
+            **DEFAULT_HEADERS,
+            "x-vqd-4": x_vqd_4,
+        }
+
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                CHAT_URL,
+                headers=chat_headers,
+                json={"model": model, "messages": messages},
+                timeout=TIMEOUT
+            )
+            response.raise_for_status()
+
+            full_response = []
+            async for chunk in response.aiter_bytes():
+                decoded_chunk = chunk.decode('utf-8')
+                for line in decoded_chunk.split('\n'):
+                    line = line.strip()
+                    if line.startswith("data: "):
+                        try:
+                            data = json.loads(line[5:])
+                            full_response.append(data.get("message", ""))
+                        except json.JSONDecodeError:
+                            continue
+
+            return "".join(full_response)
+
+    except httpx.HTTPStatusError as e:
+        raise HTTPException(status_code=e.response.status_code, detail=f"Chat error: {str(e)}")
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Chat error: {str(e)}")
+
+def format_openai_chunk(content: str, model: str):
+    """Format response chunk in OpenAI style."""
+    return json.dumps({
+        "id": f"chatcmpl-{int(time.time()*1000)}",
+        "object": "chat.completion.chunk",
+        "created": int(time.time()),
+        "model": model,
+        "choices": [{
+            "delta": {"content": content},
+            "index": 0,
+            "finish_reason": None
+        }]
+    }) + "\n\n"
+
+@app.post("/v1/chat/completions")
+async def chat_completions(request: Request):
+    try:
+        data = await request.json()
+        model = data.get("model", "o3-mini")
+        messages = data.get("messages", [])
+        stream = data.get("stream", False)
+
+        # Validation
+        if model not in SUPPORTED_MODELS:
+            raise HTTPException(400, f"Unsupported model: {model}")
+        if not messages:
+            raise HTTPException(400, "Empty messages list")
+
+        # Process messages
+        system_message = next((m for m in messages if m["role"] == "system"), None)
+        history = "\n".join(
+            f"{m['role']}: {m['content']}" 
+            for m in messages 
+            if m["role"] != "system" and m != messages[-1]
+        )
+        current_query = messages[-1]["content"] if messages else ""
+
+        combined_content = f"{system_message['content']}\n{history}\nUser: {current_query}" if system_message else f"{history}\nUser: {current_query}"
+        payload = [{"role": "user", "content": combined_content}]
+
+        if stream:
+            return await duckduckgo_chat_stream(model, payload)
+        else:
+            response_text = await duckduckgo_chat_non_stream(model, payload)
+            return JSONResponse({
+                "id": f"chatcmpl-{int(time.time()*1000)}",
+                "object": "chat.completion",
+                "created": int(time.time()),
+                "model": model,
+                "choices": [{
+                    "message": {"role": "assistant", "content": response_text},
+                    "finish_reason": "stop",
+                    "index": 0
+                }],
+                "usage": {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}
+            })
+
+    except HTTPException as e:
+        raise e
+    except Exception as e:
+        raise HTTPException(500, f"Server error: {str(e)}")
+
+@app.get("/")
+async def health_check():
+    return {"status": "healthy", "timestamp": int(time.time())}
+
+@app.exception_handler(HTTPException)
+async def http_error_handler(request: Request, exc: HTTPException):
+    return JSONResponse(
+        status_code=exc.status_code,
+        content={"error": exc.detail},
+    )
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=7860)