Upload 9 files

.env.example

@@ -0,0 +1,7 @@
+OPENROUTER_API_KEYS=your_key1,your_key2,your_key3
+SCAN_INTERVAL_HOURS=1
+MAX_RETRIES=3
+RETRY_DELAY=5
+REQUEST_TIMEOUT=30
+MAX_CONCURRENCY=5
+TEST_PROMPT=hi

app.py

@@ -0,0 +1,114 @@
+import gradio as gr
+from datetime import datetime
+
+from src.config import get_api_keys, get_scan_interval_hours
+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": [],
+            "total_tested": 0,
+            "available_count": 0,
+            "timestamp": None
+        }
+        self.last_update_time = None
+
+    def run_scan(self):
+        result = self.model_tester.test_all_models()
+        self.scan_result = result
+        self.last_update_time = format_timestamp()
+
+
+app_state = AppState()
+
+
+def get_available_models():
+    models = app_state.scan_result.get("available_models", [])
+    return "\n".join(models) if models else "No models available yet"
+
+
+def get_scan_status():
+    count = app_state.scan_result.get("available_count", 0)
+    tested = app_state.scan_result.get("total_tested", 0)
+    timestamp = app_state.last_update_time or "Never"
+    return f"Available: {count}/{tested} | Last scan: {timestamp}"
+
+
+def trigger_scan():
+    app_state.scheduler.run_now()
+    return "Scan triggered! Results will update shortly..."
+
+
+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_interval_info():
+    return f"Scan interval: every {get_scan_interval_hours()} hour(s)"
+
+
+with gr.Blocks(title="OpenRouter Free Models Scanner") as demo:
+    gr.Markdown("# OpenRouter Free Models Scanner")
+    gr.Markdown("Auto-scan available free models from OpenRouter")
+
+    with gr.Row():
+        gr.Markdown(f"### API Keys: {get_api_keys_status()}")
+        gr.Markdown(f"### {get_interval_info()}")
+
+    gr.Markdown("---")
+
+    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]
+    )
+
+    gr.Markdown("---")
+    gr.Markdown("## Available Models")
+
+    models_display = gr.Textbox(
+        value=get_available_models(),
+        label=f"Available Models ({len(app_state.scan_result.get('available_models', []))})",
+        interactive=False,
+        lines=20
+    )
+
+    demo.load(
+        fn=lambda: (
+            get_scan_status(),
+            get_available_models()
+        ),
+        outputs=[status_display, models_display]
+    )
+
+    app_state.scheduler.start()
+
+
+if __name__ == "__main__":
+    demo.launch()

requirements.txt

@@ -0,0 +1,4 @@
+gradio
+requests
+python-dotenv
+schedule

src/config.py

@@ -0,0 +1,39 @@
+import os
+from typing import List
+from dotenv import load_dotenv
+
+load_dotenv()
+
+
+def get_api_keys() -> List[str]:
+    keys_str = os.getenv("OPENROUTER_API_KEYS", "")
+    if not keys_str:
+        raise ValueError("OPENROUTER_API_KEYS environment variable is not set")
+    keys = [k.strip() for k in keys_str.split(",") if k.strip()]
+    if not keys:
+        raise ValueError("No valid API keys found in OPENROUTER_API_KEYS")
+    return keys
+
+
+def get_scan_interval_hours() -> float:
+    return float(os.getenv("SCAN_INTERVAL_HOURS", "1"))
+
+
+def get_max_retries() -> int:
+    return int(os.getenv("MAX_RETRIES", "3"))
+
+
+def get_retry_delay() -> int:
+    return int(os.getenv("RETRY_DELAY", "5"))
+
+
+def get_request_timeout() -> int:
+    return int(os.getenv("REQUEST_TIMEOUT", "30"))
+
+
+def get_max_concurrency() -> int:
+    return int(os.getenv("MAX_CONCURRENCY", "5"))
+
+
+def get_test_prompt() -> str:
+    return os.getenv("TEST_PROMPT", "hi")

src/model_tester.py

@@ -0,0 +1,72 @@
+import concurrent.futures
+from typing import List, Dict, Any, Set
+from datetime import datetime
+
+from .openrouter_client import OpenRouterClient
+from . import config
+from .utils import clean_model_name
+
+
+class ModelTester:
+    def __init__(self):
+        self.client = OpenRouterClient()
+        self.max_concurrency = config.get_max_concurrency()
+        self.test_prompt = config.get_test_prompt()
+
+    def get_all_free_models(self) -> List[str]:
+        models = self.client.get_models()
+        free_models = []
+
+        for model in models:
+            model_id = model.get("id", "")
+            if ":free" in model_id:
+                free_models.append(model_id)
+
+        return free_models
+
+    def test_single_model(self, model_id: str) -> tuple[str, bool]:
+        is_available = self.client.test_model(model_id, self.test_prompt)
+        cleaned_name = clean_model_name(model_id)
+        return cleaned_name, is_available
+
+    def test_all_models(self) -> Dict[str, Any]:
+        print(f"[{datetime.now()}] Starting model scan...")
+
+        free_models = self.get_all_free_models()
+        print(f"Found {len(free_models)} free models to test")
+
+        available_models: Set[str] = set()
+        tested_count = 0
+        success_count = 0
+
+        with concurrent.futures.ThreadPoolExecutor(
+            max_workers=self.max_concurrency
+        ) as executor:
+            future_to_model = {
+                executor.submit(self.test_single_model, model_id): model_id
+                for model_id in free_models
+            }
+
+            for future in concurrent.futures.as_completed(future_to_model):
+                tested_count += 1
+                try:
+                    cleaned_name, is_available = future.result()
+                    if is_available:
+                        available_models.add(cleaned_name)
+                        success_count += 1
+                        print(f"[{tested_count}/{len(free_models)}] ✓ {cleaned_name}")
+                    else:
+                        print(f"[{tested_count}/{len(free_models)}] ✗ {cleaned_name}")
+                except Exception as e:
+                    model_id = future_to_model[future]
+                    print(f"[{tested_count}/{len(free_models)}] ! {model_id}: {e}")
+
+        result = {
+            "available_models": sorted(list(available_models)),
+            "total_tested": tested_count,
+            "available_count": success_count,
+            "timestamp": datetime.now().isoformat()
+        }
+
+        print(f"Scan complete: {success_count}/{tested_count} models available")
+        return result

src/openrouter_client.py

@@ -0,0 +1,100 @@
+import random
+import time
+import requests
+from typing import Optional, Dict, Any, List
+
+from . import config
+
+
+class OpenRouterClient:
+    def __init__(self):
+        self.api_keys = config.get_api_keys()
+        self.max_retries = config.get_max_retries()
+        self.retry_delay = config.get_retry_delay()
+        self.timeout = config.get_request_timeout()
+        self.current_key_index = 0
+
+    def get_random_key(self) -> str:
+        return random.choice(self.api_keys)
+
+    def get_next_key(self) -> str:
+        key = self.api_keys[self.current_key_index]
+        self.current_key_index = (self.current_key_index + 1) % len(self.api_keys)
+        return key
+
+    def _make_request(
+        self,
+        method: str,
+        url: str,
+        api_key: str,
+        **kwargs
+    ) -> requests.Response:
+        headers = kwargs.pop("headers", {})
+        headers["Authorization"] = f"Bearer {api_key}"
+        headers["Content-Type"] = "application/json"
+
+        return requests.request(
+            method=method,
+            url=url,
+            headers=headers,
+            timeout=self.timeout,
+            **kwargs
+        )
+
+    def request_with_retry(
+        self,
+        method: str,
+        url: str,
+        **kwargs
+    ) -> Optional[requests.Response]:
+        last_error = None
+
+        for attempt in range(self.max_retries):
+            api_key = self.get_random_key()
+
+            try:
+                response = self._make_request(method, url, api_key, **kwargs)
+
+                if response.status_code == 200:
+                    return response
+                elif response.status_code == 429:
+                    last_error = f"Rate limited (429)"
+                    time.sleep(self.retry_delay * (attempt + 1))
+                elif response.status_code == 401:
+                    last_error = f"Unauthorized (401)"
+                else:
+                    last_error = f"HTTP {response.status_code}: {response.text[:100]}"
+                    time.sleep(self.retry_delay)
+
+            except requests.exceptions.Timeout:
+                last_error = "Request timeout"
+                time.sleep(self.retry_delay)
+            except requests.exceptions.RequestException as e:
+                last_error = f"Request error: {str(e)}"
+                time.sleep(self.retry_delay)
+
+        print(f"Request failed after {self.max_retries} attempts: {last_error}")
+        return None
+
+    def get_models(self) -> List[Dict[str, Any]]:
+        url = "https://openrouter.ai/api/v1/models"
+        response = self.request_with_retry("GET", url)
+
+        if response and response.status_code == 200:
+            data = response.json()
+            return data.get("data", [])
+        return []
+
+    def test_model(self, model_id: str, prompt: str = "hi") -> bool:
+        url = "https://openrouter.ai/api/v1/chat/completions"
+        payload = {
+            "model": model_id,
+            "messages": [{"role": "user", "content": prompt}],
+            "max_tokens": 10
+        }
+
+        response = self.request_with_retry("POST", url, json=payload)
+
+        if response and response.status_code == 200:
+            return True
+        return False

src/scheduler.py

@@ -0,0 +1,50 @@
+import threading
+import time
+from datetime import datetime
+from typing import Callable, Optional
+
+import schedule
+
+from . import config
+
+
+class Scheduler:
+    def __init__(self, task_callback: Callable):
+        self.task_callback = task_callback
+        self.interval_hours = config.get_scan_interval_hours()
+        self._stop_event = threading.Event()
+        self._thread: Optional[threading.Thread] = None
+        self._running = False
+
+    def start(self):
+        if self._running:
+            return
+
+        schedule.every(self.interval_hours).hours.do(self._run_task)
+        self._run_task()
+
+        self._running = True
+        self._stop_event.clear()
+        self._thread = threading.Thread(target=self._scheduler_loop, daemon=True)
+        self._thread.start()
+        print(f"Scheduler started, will run every {self.interval_hours} hour(s)")
+
+    def stop(self):
+        self._running = False
+        self._stop_event.set()
+        schedule.clear()
+        print("Scheduler stopped")
+
+    def run_now(self):
+        self._run_task()
+
+    def _run_task(self):
+        try:
+            self.task_callback()
+        except Exception as e:
+            print(f"Error in scheduled task: {e}")
+
+    def _scheduler_loop(self):
+        while not self._stop_event.is_set():
+            schedule.run_pending()
+            time.sleep(1)

src/utils.py

@@ -0,0 +1,16 @@
+import re
+from datetime import datetime
+
+
+def clean_model_name(model_id: str) -> str:
+    if "/" in model_id:
+        model_id = model_id.split("/")[-1]
+    if ":free" in model_id:
+        model_id = model_id.replace(":free", "")
+    return model_id.strip()
+
+
+def format_timestamp(dt: datetime = None) -> str:
+    if dt is None:
+        dt = datetime.now()
+    return dt.strftime("%Y-%m-%d %H:%M:%S")