Create app.py

app.py

@@ -0,0 +1,194 @@
+from fastapi import FastAPI, HTTPException, Request, Depends, status
+from fastapi.responses import HTMLResponse, JSONResponse
+from fastapi.staticfiles import StaticFiles
+from fastapi.security import HTTPBasic, HTTPBasicCredentials
+from pydantic import BaseModel
+import httpx
+import secrets
+import os
+from typing import List, Dict, Any, Optional
+
+# Initialize FastAPI app
+app = FastAPI(title="AI Chat Application")
+
+# Mount static files
+app.mount("/static", StaticFiles(directory="static"), name="static")
+
+# Security for admin panel
+security = HTTPBasic()
+ADMIN_USERNAME = os.environ.get("ADMIN_USERNAME", "admin")
+ADMIN_PASSWORD = os.environ.get("ADMIN_PASSWORD", "password")  # Change in production!
+
+# Store active chats and settings in memory
+active_chats = {}
+chat_history = {}
+rules = [
+    "Be respectful and helpful",
+    "Do not generate harmful content",
+    "Provide accurate information"
+]
+available_models = ["mistral-large-latest", "gemini", "openai-xlarge"]
+current_model = "mistral-large-latest"
+
+# API endpoint for the chat
+class ChatMessage(BaseModel):
+    message: str
+    chat_id: Optional[str] = None
+    model: Optional[str] = None
+
+class Rule(BaseModel):
+    rule: str
+
+class ModelRequest(BaseModel):
+    model: str
+
+def get_admin_user(credentials: HTTPBasicCredentials = Depends(security)):
+    is_username_correct = secrets.compare_digest(credentials.username, ADMIN_USERNAME)
+    is_password_correct = secrets.compare_digest(credentials.password, ADMIN_PASSWORD)
+    if not (is_username_correct and is_password_correct):
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Incorrect username or password",
+            headers={"WWW-Authenticate": "Basic"},
+        )
+    return credentials.username
+
+@app.get("/", response_class=HTMLResponse)
+async def read_root():
+    with open("index.html", "r") as f:
+        return f.read()
+
+@app.post("/api/chat")
+async def process_chat(chat_request: ChatMessage):
+    # Generate a new chat ID if not provided
+    chat_id = chat_request.chat_id or secrets.token_hex(8)
+    
+    # Initialize chat history if it doesn't exist
+    if chat_id not in chat_history:
+        chat_history[chat_id] = []
+    
+    # Store the message in chat history
+    chat_history[chat_id].append({"role": "user", "content": chat_request.message})
+    
+    # Apply rules to message (simplified implementation)
+    for rule in rules:
+        if rule.startswith("Block:") and rule[6:].lower() in chat_request.message.lower():
+            return {"response": "I cannot respond to that due to content rules.", "chat_id": chat_id}
+    
+    # Select model to use
+    model_to_use = chat_request.model if chat_request.model in available_models else current_model
+    
+    # Prepare request to external API
+    api_request = {
+        "model": model_to_use,
+        "messages": chat_history[chat_id],
+        "temperature": 0.7,
+        "max_tokens": 1000
+    }
+    
+    # Call external API
+    try:
+        async with httpx.AsyncClient(timeout=60.0) as client:
+            response = await client.post(
+                "https://parthsadaria-lokiai.hf.space/chat/completions",
+                json=api_request
+            )
+            response_data = response.json()
+            
+            if "choices" in response_data and len(response_data["choices"]) > 0:
+                ai_message = response_data["choices"][0]["message"]["content"]
+                chat_history[chat_id].append({"role": "assistant", "content": ai_message})
+                return {"response": ai_message, "chat_id": chat_id, "model": model_to_use}
+            else:
+                error_msg = "No response from AI model"
+                chat_history[chat_id].append({"role": "assistant", "content": error_msg})
+                return {"response": error_msg, "chat_id": chat_id}
+    
+    except Exception as e:
+        error_msg = f"Error communicating with AI service: {str(e)}"
+        chat_history[chat_id].append({"role": "assistant", "content": error_msg})
+        return {"response": error_msg, "chat_id": chat_id}
+
+# Admin panel endpoints
+@app.get("/admin", response_class=HTMLResponse)
+async def admin_panel(username: str = Depends(get_admin_user)):
+    html_content = f"""
+    <!DOCTYPE html>
+    <html>
+    <head>
+        <title>Admin Panel</title>
+        <link rel="stylesheet" href="/static/style.css">
+        <script src="/static/script.js"></script>
+    </head>
+    <body class="admin-page">
+        <h1>Admin Panel</h1>
+        
+        <section class="admin-section">
+            <h2>Active Model: <span id="current-model">{current_model}</span></h2>
+            <select id="model-selector">
+                {"".join(f'<option value="{model}">{model}</option>' for model in available_models)}
+            </select>
+            <button onclick="updateModel()">Update Model</button>
+        </section>
+        
+        <section class="admin-section">
+            <h2>Rules</h2>
+            <ul id="rules-list">
+                {"".join(f'<li>{rule} <button onclick="deleteRule({i})">Delete</button></li>' for i, rule in enumerate(rules))}
+            </ul>
+            <div>
+                <input type="text" id="new-rule" placeholder="Add new rule">
+                <button onclick="addRule()">Add Rule</button>
+            </div>
+        </section>
+        
+        <section class="admin-section">
+            <h2>Active Chats</h2>
+            <ul id="active-chats">
+                {"".join(f'<li><a href="#" onclick="viewChat(\'{chat_id}\')">{chat_id}</a></li>' for chat_id in chat_history)}
+            </ul>
+            <div id="chat-viewer"></div>
+        </section>
+    </body>
+    </html>
+    """
+    return html_content
+
+@app.get("/api/rules")
+async def get_rules(username: str = Depends(get_admin_user)):
+    return {"rules": rules}
+
+@app.post("/api/rules")
+async def add_rule(rule_request: Rule, username: str = Depends(get_admin_user)):
+    rules.append(rule_request.rule)
+    return {"status": "success", "rules": rules}
+
+@app.delete("/api/rules/{rule_id}")
+async def delete_rule(rule_id: int, username: str = Depends(get_admin_user)):
+    if 0 <= rule_id < len(rules):
+        rules.pop(rule_id)
+        return {"status": "success", "rules": rules}
+    raise HTTPException(status_code=404, detail="Rule not found")
+
+@app.post("/api/model")
+async def set_model(model_request: ModelRequest, username: str = Depends(get_admin_user)):
+    global current_model
+    if model_request.model in available_models:
+        current_model = model_request.model
+        return {"status": "success", "model": current_model}
+    raise HTTPException(status_code=400, detail="Invalid model")
+
+@app.get("/api/chats")
+async def get_chats(username: str = Depends(get_admin_user)):
+    return {"chats": list(chat_history.keys())}
+
+@app.get("/api/chats/{chat_id}")
+async def get_chat_history(chat_id: str, username: str = Depends(get_admin_user)):
+    if chat_id in chat_history:
+        return {"history": chat_history[chat_id]}
+    raise HTTPException(status_code=404, detail="Chat not found")
+
+# Run the app (for local development)
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=7860)