Update app.py

app.py

@@ -3,116 +3,93 @@ import requests
 import threading
 import time
 import random
-from datetime import datetime
 
 # ==============================================================================
 # ⚙️ CONFIGURATION
 # ==============================================================================
-TARGET_COUNT = 20         # 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)
+TARGET_POOL_SIZE = 15      # Keep 15 good proxies ready
+TIMEOUT_STRICT = 5         # Reject anything slower than 5 seconds
+CHECK_URL = "https://m.youtube.com" # We check against YouTube Mobile (faster)
 
-# Shared Memory (The "Live File")
 proxy_storage = {
-    "valid_proxies": [],
-    "last_updated": "Not yet started"
+    "pool": [],
+    "last_update": "Initializing..."
 }
 
 # ==============================================================================
-# 🕵️‍♂️ PROXY WORKER (Background Thread)
+# 🕵️‍♂️ REFINERY WORKER
 # ==============================================================================
 def check_proxy(ip):
-    """Returns True if proxy connects to YouTube in < 5 seconds."""
+    """Strictly checks if proxy can load YouTube."""
     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
+        # We use a Session for speed
+        with requests.Session() as s:
+            s.proxies.update(proxies)
+            # HEAD request is faster than GET
+            r = s.head(CHECK_URL, timeout=TIMEOUT_STRICT)
+            if r.status_code == 200:
+                return True
     except:
         return False
     return False
 
-def worker_loop():
+def refinery_loop():
     while True:
-        print(f"\n[{datetime.now().strftime('%H:%M')}] ♻️ Starting Validation Cycle...")
+        print(f"♻️ Refining Pool... Current: {len(proxy_storage['pool'])}")
         
-        # 1. RE-VALIDATE EXISTING (Keep the good ones)
-        current_list = proxy_storage["valid_proxies"]
-        still_good = []
-        for ip in current_list:
+        # 1. Clean existing pool (Re-check)
+        valid_proxies = []
+        for ip in proxy_storage['pool']:
             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 = []
+                valid_proxies.append(ip)
+        proxy_storage['pool'] = valid_proxies
+
+        # 2. Refill if needed
+        if len(proxy_storage['pool']) < TARGET_POOL_SIZE:
+            print("   📉 Low fuel. Fetching fresh proxies...")
+            raw_list = []
             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:
+            for url in sources:
                 try:
-                    r = requests.get(s, timeout=10)
-                    raw_proxies += r.text.strip().split("\n")
+                    r = requests.get(url, timeout=10)
+                    raw_list += 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
+            # Shuffle and Test
+            random.shuffle(raw_list)
+            for ip in raw_list:
+                if len(proxy_storage['pool']) >= TARGET_POOL_SIZE: break
                 
-                # 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)
+                ip = ip.strip()
+                if ip and ip not in proxy_storage['pool']:
+                    if check_proxy(ip):
+                        print(f"   ✅ Verified: {ip}")
+                        proxy_storage['pool'].append(ip)
 
-# Start Background Thread
-threading.Thread(target=worker_loop, daemon=True).start()
+        proxy_storage['last_update'] = time.strftime("%H:%M:%S")
+        time.sleep(120) # Run every 2 minutes for freshness
 
+threading.Thread(target=refinery_loop, daemon=True).start()
 
 # ==============================================================================
-# 🔌 API ENDPOINT (FIXED)
+# 🔌 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']}"
+def serve_proxies():
+    return proxy_storage['pool'], f"Last Refined: {proxy_storage['last_update']}"
 
 with gr.Blocks() as app:
-    gr.Markdown("## 🚦 Proxy Engine (Backend)")
-    
+    gr.Markdown("## 🏭 Proxy Refinery")
     with gr.Row():
-        json_out = gr.JSON(label="Live Proxy Pool")
-        status_out = gr.Textbox(label="Last Update")
-    
-    refresh_btn = gr.Button("Refresh View")
-    
-    # --- THIS IS THE FIX: We added api_name="get_proxies" ---
-    refresh_btn.click(get_proxies_api, outputs=[json_out, status_out], api_name="get_proxies")
+        out_json = gr.JSON(label="Active High-Speed Pool")
+        out_status = gr.Textbox(label="Status")
     
-    # Load immediately on open
-    app.load(get_proxies_api, outputs=[json_out, status_out])
+    btn = gr.Button("Refresh")
+    btn.click(serve_proxies, outputs=[out_json, out_status], api_name="get_proxies")
+    app.load(serve_proxies, outputs=[out_json, out_status])
 
 app.launch()