Hugging Face's logo

Spaces:
lexicalspace
/
emergency
Paused

App Files Community
emergency
File size: 4,699 Bytes
ad0eb37
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
import gradio as gr
import requests
import threading
import time
import random
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor, as_completed

# ==============================================================================
# ⚙️ CONFIGURATION (SAFE MODE)
# ==============================================================================
TARGET_COUNT = 200
TIMEOUT_SEC = 5
CHECK_INTERVAL = 30     
MAX_THREADS = 40        # ⚠️ REDUCED to 40. Start small. If it works, try 80.

# Shared Memory
proxy_storage = {
    "valid_proxies": [],
    "last_updated": "Waiting for startup..."
}

# ==============================================================================
# 🕵️‍♂️ PROXY WORKER
# ==============================================================================
def check_proxy(ip):
    """Returns IP if valid and FAST."""
    proxies = {"http": f"http://{ip}", "https": f"http://{ip}"}
    try:
        r = requests.get("https://www.youtube.com", proxies=proxies, timeout=TIMEOUT_SEC)
        if r.status_code == 200:
            return ip
    except:
        pass
    return None

def worker_loop():
    # 🛑 CRITICAL FIX: Sleep 15s to let Gradio start first!
    print("⏳ Worker waiting 15s for Gradio to boot...")
    time.sleep(15) 
    
    while True:
        print(f"\n[{datetime.now().strftime('%H:%M:%S')}] ☢️ Starting Scan...")
        
        # 1. FETCH SOURCES (Fixed Syntax Errors)
        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",
            "https://raw.githubusercontent.com/shiftytr/proxy-list/master/proxy.txt",
            "https://raw.githubusercontent.com/hookzof/socks5_list/master/proxy.txt",
            "https://raw.githubusercontent.com/clarketm/proxy-list/master/proxy-list-raw.txt",
            "https://raw.githubusercontent.com/sunny9577/proxy-scraper/master/proxies.txt",
            "https://raw.githubusercontent.com/jetkai/proxy-list/main/online-proxies/txt/proxies-http.txt"
        ]
        
        print("   📥 Downloading proxy lists...")
        for s in sources:
            try:
                r = requests.get(s, timeout=5)
                if r.status_code == 200:
                    raw_proxies += r.text.strip().split("\n")
            except: pass
            
        raw_proxies = list(set(raw_proxies)) 
        random.shuffle(raw_proxies)
        
        # 2. VALIDATE
        check_list = proxy_storage["valid_proxies"] + raw_proxies
        check_list = check_list[:2000] # Hard limit to save RAM

        new_valid_pool = []
        
        print(f"   🚀 Scanning {len(check_list)} proxies with {MAX_THREADS} threads...")
        
        with ThreadPoolExecutor(max_workers=MAX_THREADS) as executor:
            future_to_ip = {executor.submit(check_proxy, ip): ip for ip in check_list}
            
            for future in as_completed(future_to_ip):
                result = future.result()
                if result:
                    if result not in new_valid_pool:
                        new_valid_pool.append(result)
                
                if len(new_valid_pool) >= TARGET_COUNT:
                    executor.shutdown(wait=False, cancel_futures=True)
                    break
        
        proxy_storage["valid_proxies"] = new_valid_pool
        proxy_storage["last_updated"] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        
        print(f"   ✅ Update Complete. Found {len(new_valid_pool)} proxies.")
        time.sleep(CHECK_INTERVAL)

# Start Background Thread
threading.Thread(target=worker_loop, daemon=True).start()

# ==============================================================================
# 🔌 API ENDPOINT
# ==============================================================================
def get_proxies_api():
    return proxy_storage["valid_proxies"], f"Updated: {proxy_storage['last_updated']}"

with gr.Blocks() as app:
    gr.Markdown("## 🚦 Proxy Engine (Safe Mode)")
    with gr.Row():
        json_out = gr.JSON(label="Live Proxy Pool")
        status_out = gr.Textbox(label="Last Update")
    
    refresh_btn = gr.Button("Refresh View")
    refresh_btn.click(get_proxies_api, outputs=[json_out, status_out], api_name="get_proxies")
    app.load(get_proxies_api, outputs=[json_out, status_out])

# Explicitly set server settings to ensure HF can find the app
app.launch(server_name="0.0.0.0", server_port=7860)