Create app.py

app.py

@@ -0,0 +1,117 @@
+import gradio as gr
+import requests
+import threading
+import time
+import random
+from datetime import datetime
+
+# ==============================================================================
+# ⚙️ CONFIGURATION
+# ==============================================================================
+TARGET_COUNT = 10         # We want exactly 10 working proxies
+TIMEOUT_SEC = 5           # Reject if slower than 5 seconds
+CHECK_INTERVAL = 300      # Re-check every 5 minutes (300 seconds)
+
+# Shared Memory (The "Live File")
+proxy_storage = {
+    "valid_proxies": [],
+    "last_updated": "Not yet started"
+}
+
+# ==============================================================================
+# 🕵️‍♂️ PROXY WORKER (Background Thread)
+# ==============================================================================
+def check_proxy(ip):
+    """Returns True if proxy connects to YouTube in < 5 seconds."""
+    proxies = {"http": f"http://{ip}", "https": f"http://{ip}"}
+    try:
+        # We test against YouTube specifically because that's our target
+        r = requests.get("https://www.youtube.com", proxies=proxies, timeout=TIMEOUT_SEC)
+        if r.status_code == 200:
+            return True
+    except:
+        return False
+    return False
+
+def worker_loop():
+    while True:
+        print(f"\n[{datetime.now().strftime('%H:%M')}] ♻️ Starting Validation Cycle...")
+        
+        # 1. RE-VALIDATE EXISTING (Keep the good ones)
+        current_list = proxy_storage["valid_proxies"]
+        still_good = []
+        for ip in current_list:
+            if check_proxy(ip):
+                still_good.append(ip)
+            else:
+                print(f"   ❌ Dropped dead proxy: {ip}")
+        
+        proxy_storage["valid_proxies"] = still_good
+        
+        # 2. FILL THE POOL (If we have less than 10)
+        if len(proxy_storage["valid_proxies"]) < TARGET_COUNT:
+            needed = TARGET_COUNT - len(proxy_storage["valid_proxies"])
+            print(f"   📉 Pool low. Need {needed} more. Fetching fresh list...")
+            
+            # Fetch huge raw list
+            raw_proxies = []
+            sources = [
+                "https://raw.githubusercontent.com/TheSpeedX/SOCKS-List/master/http.txt",
+                "https://raw.githubusercontent.com/monosans/proxy-list/main/proxies/http.txt",
+                "https://api.proxyscrape.com/v2/?request=getproxies&protocol=http&timeout=5000&country=all&ssl=all&anonymity=all"
+            ]
+            for s in sources:
+                try:
+                    r = requests.get(s, timeout=10)
+                    raw_proxies += r.text.strip().split("\n")
+                except: pass
+            
+            # Shuffle to get random ones
+            random.shuffle(raw_proxies)
+            
+            # Test until we hit 10
+            for ip in raw_proxies:
+                ip = ip.strip()
+                if not ip or ip in proxy_storage["valid_proxies"]: continue
+                
+                # Stop if we hit our target
+                if len(proxy_storage["valid_proxies"]) >= TARGET_COUNT:
+                    break
+                    
+                print(f"   Testing {ip}...", end="\r")
+                if check_proxy(ip):
+                    print(f"   ✅ FOUND NEW: {ip}")
+                    proxy_storage["valid_proxies"].append(ip)
+
+        proxy_storage["last_updated"] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+        print(f"   💤 Sleeping for {CHECK_INTERVAL/60} mins. Current Pool: {len(proxy_storage['valid_proxies'])}")
+        
+        # 3. SLEEP
+        time.sleep(CHECK_INTERVAL)
+
+# Start Background Thread
+threading.Thread(target=worker_loop, daemon=True).start()
+
+# ==============================================================================
+# 🔌 API ENDPOINT
+# ==============================================================================
+def get_proxies_api():
+    """Main App calls this to get the list"""
+    return proxy_storage["valid_proxies"], f"Updated: {proxy_storage['last_updated']}"
+
+with gr.Blocks() as app:
+    gr.Markdown("## 🚦 Proxy Engine (Backend)")
+    gr.Markdown(f"Maintains a pool of {TARGET_COUNT} fast proxies.")
+    
+    with gr.Row():
+        json_out = gr.JSON(label="Live Proxy Pool")
+        status_out = gr.Textbox(label="Last Update")
+    
+    # Hidden button to trigger refresh for UI viewers
+    refresh_btn = gr.Button("Refresh View")
+    refresh_btn.click(get_proxies_api, outputs=[json_out, status_out])
+    
+    # Load immediately on open
+    app.load(get_proxies_api, outputs=[json_out, status_out])
+
+app.launch()