Upload 9 files

app.py

@@ -1,170 +1,164 @@
 import gradio as gr
-import threading
+from fastapi import FastAPI, Request, HTTPException
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel
+from typing import List, Optional
+import asyncio
+import random
 from datetime import datetime
+import threading
+import uvicorn
 
-from src.config import get_api_keys, get_scan_interval_hours
+from src.config import get_api_keys
 from src.model_tester import ModelTester
 from src.scheduler import Scheduler
 from src.utils import format_timestamp
 
 
-class AppState:
-    def __init__(self):
-        self.model_tester = ModelTester()
-        self.scheduler = Scheduler(task_callback=self.run_scan)
-        self.scan_result = {
-            "available_models": [],
-            "available_free_models": [],
-            "total_available": 0,
-            "free_available": 0,
-            "timestamp": None
-        }
-        self.last_update_time = None
-        self._initial_scan_done = False
-
-    def run_scan(self):
-        result = self.model_tester.scan_all_models()
-        self.scan_result = result
-        self.last_update_time = format_timestamp()
-        self._initial_scan_done = True
+model_tester = ModelTester()
+scheduler = Scheduler(task_callback=lambda: model_tester.scan_all_models())
 
 
-app_state = AppState()
+class Message(BaseModel):
+    role: str
+    content: str
 
 
-def get_available_models():
-    models = app_state.scan_result.get("available_models", [])
-    return "\n".join(models) if models else "No models available yet"
+class ChatCompletionRequest(BaseModel):
+    model: Optional[str] = None
+    messages: List[Message]
+    temperature: Optional[float] = 1.0
+    max_tokens: Optional[int] = None
+    stream: Optional[bool] = False
 
 
-def get_available_free_models():
-    models = app_state.scan_result.get("available_free_models", [])
-    return "\n".join(models) if models else "No free models available yet"
+fastapi_app = FastAPI(title="OpenRouter Free API")
 
 
-def get_models_api(free_only: bool = False):
-    """API endpoint to get available models as JSON"""
-    if free_only:
-        return app_state.scan_result.get("available_free_models", [])
-    return app_state.scan_result.get("available_models", [])
+@fastapi_app.on_event("startup")
+async def startup_event():
+    threading.Thread(target=lambda: scheduler.start(), daemon=True).start()
 
 
-def scan_and_get_models():
-    """Trigger scan and return models"""
-    app_state.scheduler.run_now()
-    import time
-    time.sleep(2)
-    return app_state.scan_result.get("available_models", [])
+@fastapi_app.get("/v1/models")
+async def list_models():
+    available = model_tester.get_available_models(free_only=False)
+    available_free = model_tester.get_available_models(free_only=True)
+    
+    models = []
+    for model_id in available:
+        is_free = model_id in available_free
+        models.append({
+            "id": model_id,
+            "object": "model",
+            "created": 1677610602,
+            "owned_by": "openrouter",
+            "permission": [],
+            "root": model_id,
+            "parent": None,
+            "free": is_free
+        })
+    
+    return {
+        "object": "list",
+        "data": models
+    }
 
 
-def chat(prompt: str, model: str = None):
-    """Chat endpoint - tries direct model first, then falls back to list match"""
-    result = app_state.model_tester.chat_completion_sync(prompt, model)
+@fastapi_app.post("/v1/chat/completions")
+async def chat_completions(request: ChatCompletionRequest):
+    prompt = request.messages[-1].content if request.messages else ""
+    model_hint = request.model
     
-    if result.get("success"):
-        return {
-            "success": True,
-            "response": result.get("response"),
-            "method": result.get("method"),
-            "model": result.get("model")
-        }
-    else:
-        return {
-            "success": False,
-            "error": result.get("error", "Unknown error"),
-            "method": result.get("method")
+    result = await model_tester.chat_completion(prompt, model_hint)
+    
+    if not result.get("success"):
+        raise HTTPException(status_code=400, detail=result.get("error", "Request failed"))
+    
+    response_data = result.get("response", {})
+    
+    content = ""
+    if "choices" in response_data and response_data["choices"]:
+        content = response_data["choices"][0].get("message", {}).get("content", "")
+    
+    return {
+        "id": response_data.get("id", f"chatcmpl-{random.randint(100000, 999999)}"),
+        "object": "chat.completion",
+        "created": int(datetime.now().timestamp()),
+        "model": result.get("model", request.model or "unknown"),
+        "choices": [
+            {
+                "index": 0,
+                "message": {
+                    "role": "assistant",
+                    "content": content
+                },
+                "finish_reason": "stop"
+            }
+        ],
+        "usage": {
+            "prompt_tokens": len(prompt),
+            "completion_tokens": 10,
+            "total_tokens": len(prompt) + 10
         }
+    }
 
 
-def get_scan_status():
-    total = app_state.scan_result.get("total_available", 0)
-    free = app_state.scan_result.get("free_available", 0)
-    timestamp = app_state.last_update_time or "Never"
-    return f"Free: {free} | Total: {total} | Last scan: {timestamp}"
-
-
-def trigger_scan():
-    app_state.scheduler.run_now()
-    return "Scan triggered! Results will update shortly..."
+@fastapi_app.get("/health")
+async def health():
+    return {"status": "ok"}
 
 
-def get_api_keys_status():
-    try:
-        keys = get_api_keys()
-        return f"{len(keys)} keys configured"
-    except ValueError as e:
-        return f"Error: {str(e)}"
+def get_available_models():
+    return model_tester.get_available_models(free_only=False)
 
 
-def get_interval_info():
-    return f"Scan interval: every {get_scan_interval_hours()} hour(s)"
+def get_available_free_models():
+    return model_tester.get_available_models(free_only=True)
 
 
-with gr.Blocks(title="OpenRouter Free Models Scanner") as demo:
-    gr.Markdown("# OpenRouter Free Models Scanner")
-    gr.Markdown("Auto-scan all available models from OpenRouter (free priority)")
+def get_scan_status():
+    scan_result = model_tester.scan_result
+    total = scan_result.get("total_available", 0)
+    free = scan_result.get("free_available", 0)
+    timestamp = scan_result.get("timestamp", "Never")
+    return f"Free: {free} | Total: {total} | Last: {timestamp}"
 
-    with gr.Row():
-        gr.Markdown(f"### API Keys: {get_api_keys_status()}")
-        gr.Markdown(f"### {get_interval_info()}")
 
-    gr.Markdown("---")
+def format_model_list(models):
+    return "\n".join(models) if models else "No models available"
 
-    with gr.Row():
-        gr.Markdown("## Status")
-        status_display = gr.Textbox(
-            value=get_scan_status(),
-            label="Scan Status",
-            interactive=False
-        )
-
-    with gr.Row():
-        scan_btn = gr.Button("Trigger Scan Now", variant="primary")
-        scan_result = gr.Textbox(
-            value="Click button to trigger scan",
-            label="Scan Result",
-            interactive=False
-        )
-
-    scan_btn.click(
-        fn=trigger_scan,
-        inputs=[],
-        outputs=[scan_result]
-    )
 
+with gr.Blocks(title="OpenRouter Free API") as demo:
+    gr.Markdown("# OpenRouter Free API")
+    gr.Markdown("Standard OpenAI-compatible API with free model support")
+    
+    gr.Markdown("## API Endpoints")
+    gr.Markdown("""
+    - `GET /v1/models` - List available models  
+    - `POST /v1/chat/completions` - Chat completion
+    """)
+    
     gr.Markdown("---")
-    gr.Markdown("## Available Free Models (Priority)")
-
-    free_models_display = gr.Textbox(
-        value=get_available_free_models(),
-        label=f"Free Models ({len(app_state.scan_result.get('available_free_models', []))})",
-        interactive=False,
-        lines=15
+    gr.Markdown(f"## Status: {get_scan_status()}")
+    
+    gr.Markdown("## Available Free Models")
+    free_display = gr.Textbox(
+        value=format_model_list(get_available_free_models()), 
+        lines=15, 
+        interactive=False
     )
-
+    
     gr.Markdown("## All Available Models")
-
-    models_display = gr.Textbox(
-        value=get_available_models(),
-        label=f"All Models ({len(app_state.scan_result.get('available_models', []))})",
-        interactive=False,
-        lines=15
-    )
-
-    demo.load(
-        fn=lambda: (
-            get_scan_status(),
-            get_available_free_models(),
-            get_available_models()
-        ),
-        outputs=[status_display, free_models_display, models_display]
+    all_display = gr.Textbox(
+        value=format_model_list(get_available_models()), 
+        lines=15, 
+        interactive=False
     )
 
-    app_state.scheduler.start()
 
-    gr.API(documentation=None)(lambda free_only=False: get_models_api(free_only))
-    gr.API(documentation=None)(chat)
+app = gr.mount_gradio_app(fastapi_app, demo, path="/")
 
 
 if __name__ == "__main__":
-    demo.launch()
+    uvicorn.run(app, host="0.0.0.0", port=7860)

requirements.txt

@@ -3,3 +3,6 @@ requests
 python-dotenv
 schedule
 aiohttp
+fastapi
+uvicorn
+sse-starlette

src/model_tester.py

@@ -21,6 +21,13 @@ class ModelTester:
         self._available_free_models: List[str] = []
         self._scan_in_progress = False
         self._last_scan_time: Optional[datetime] = None
+        self.scan_result: Dict[str, Any] = {
+            "available_models": [],
+            "available_free_models": [],
+            "total_available": 0,
+            "free_available": 0,
+            "timestamp": None
+        }
 
     def refresh_model_list(self):
         """Get latest model list from API"""
@@ -106,15 +113,17 @@ class ModelTester:
         self._last_scan_time = datetime.now()
         self._scan_in_progress = False
         
-        print(f"Scan complete: {len(self._available_free_models)} free, {len(self._available_models)} total available")
-        
-        return {
+        self.scan_result = {
             "available_models": self._available_models,
             "available_free_models": self._available_free_models,
             "total_available": len(self._available_models),
             "free_available": len(self._available_free_models),
             "timestamp": self._last_scan_time.isoformat() if self._last_scan_time else None
         }
+        
+        print(f"Scan complete: {len(self._available_free_models)} free, {len(self._available_models)} total available")
+        
+        return self.scan_result
 
     def scan_all_models(self):
         """Sync wrapper for scan"""