app.py

import os
import json
import random
import re
import requests
import asyncio
from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse, Response
from threading import Lock
import uvicorn
from faker import Faker
from bs4 import BeautifulSoup

app = FastAPI()
faker = Faker()
valid_proxies = {}
invalid_proxies = {}
proxies_lock = Lock()

def create_headers():
    user_agent = faker.user_agent()
    random_ip = faker.ipv4()
    return {
        "User-Agent": user_agent,
        "X-Forwarded-For": random_ip,
        "Client-IP": random_ip,
        "X-Real-IP": random_ip
    }

def is_valid_proxy(proxy: str) -> bool:
    ip_port_pattern = re.compile(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}:\d{2,5}$')
    if ip_port_pattern.match(proxy) is None:
        return False
    ip, port = proxy.split(':')
    if any(char in ip for char in 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ') or \
       any(char in port for char in 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'):
        return False
    return True

async def verify_proxy(proxy: str):
    if not is_valid_proxy(proxy):
        with proxies_lock:
            invalid_proxies[proxy] = True
        return proxy, False

    test_urls = [
        "https://google.com",
        "https://www.youtube.com/",
        "http://httpbin.org/ip",
        "https://instagram.com"
    ]
    proxy_dict = {
        "http": f"http://{proxy}",
        "https": f"http://{proxy}"
    }
    headers = create_headers()
    try:
        for url in test_urls:
            try:
                response = await asyncio.to_thread(requests.get, url, proxies=proxy_dict, headers=headers, timeout=10)
                response.raise_for_status()
            except (requests.RequestException, ValueError):
                with proxies_lock:
                    invalid_proxies[proxy] = True
                return proxy, False
        with proxies_lock:
            valid_proxies[proxy] = True
        return proxy, True
    except Exception:
        with proxies_lock:
            valid_proxies.pop(proxy, None)
            invalid_proxies[proxy] = True
        return proxy, False

async def verify_proxies_in_background():
    while True:
        with proxies_lock:
            proxies_to_verify = list(valid_proxies.keys())
        if proxies_to_verify:
            tasks = [verify_proxy(proxy) for proxy in proxies_to_verify]
            results = await asyncio.gather(*tasks)
            for proxy, is_valid in results:
                if not is_valid:
                    with proxies_lock:
                        valid_proxies.pop(proxy, None)
                        invalid_proxies[proxy] = True
        await asyncio.sleep(0)

async def fetch_proxies_from_sources():
    proxy_urls = [
        os.getenv("PROXY_API_URL", "http://pubproxy.com/api/proxy?format=txt&level=anonymous,elite&type=http,socks4,socks5&last_check=60&speed=25&limit=1&post=true&user_agent=true&cookies=true&referer=true"),
        'https://raw.githubusercontent.com/clarketm/proxy-list/master/proxy-list-raw.txt',
        'https://raw.githubusercontent.com/TheSpeedX/PROXY-List/master/http.txt',
        'https://raw.githubusercontent.com/TheSpeedX/SOCKS-List/master/socks4.txt',
        'https://raw.githubusercontent.com/TheSpeedX/SOCKS-List/master/socks5.txt',
        'https://raw.githubusercontent.com/ShiftyTR/Proxy-List/master/proxy.txt',
        'https://raw.githubusercontent.com/sunny9577/proxy-scraper/master/proxies.txt',
        "https://storage.googleapis.com/river-treat-249913.appspot.com/p.txt",
        "https://storage.googleapis.com/river-treat-249913.appspot.com/proxy.txt",
        "https://storage.googleapis.com/river-treat-249913.appspot.com/ultimate.txt",
        'https://raw.githubusercontent.com/proxylist/proxylist/master/proxy.txt',
        'https://raw.githubusercontent.com/scrapfly/proxy-list/main/proxies.txt',
        'https://raw.githubusercontent.com/roosterkid/openproxylist/main/HTTP.txt',
        'https://raw.githubusercontent.com/roosterkid/openproxylist/main/SOCKS4.txt',
        'https://raw.githubusercontent.com/roosterkid/openproxylist/main/SOCKS5.txt',
        'https://raw.githubusercontent.com/roosterkid/openproxylist/main/HTTPS.txt',
        'https://raw.githubusercontent.com/roosterkid/openproxylist/main/ALL.txt',
        'https://raw.githubusercontent.com/proxylist/proxylist/master/https.txt',
        'https://raw.githubusercontent.com/proxylist/proxylist/master/socks4.txt',
        'https://raw.githubusercontent.com/proxylist/proxylist/master/socks5.txt',
        'https://raw.githubusercontent.com/proxylist/proxylist/master/http.txt',
        'https://raw.githubusercontent.com/proxylist/proxylist/master/all.txt',
        'https://raw.githubusercontent.com/jetlore/proxies/master/proxy-list.txt',
        'https://raw.githubusercontent.com/hookzof/proxy-list/main/proxy.txt',
        'https://raw.githubusercontent.com/zzlol123/proxy-list/main/proxies.txt',
        'https://raw.githubusercontent.com/sqSfg/Proxy-List/master/http.txt',
        'https://raw.githubusercontent.com/sqSfg/Proxy-List/master/https.txt',
        'https://raw.githubusercontent.com/sqSfg/Proxy-List/master/socks4.txt',
        'https://raw.githubusercontent.com/sqSfg/Proxy-List/master/socks5.txt',
        'https://raw.githubusercontent.com/sqSfg/Proxy-List/master/all.txt',
        'https://www.proxy-list.download/api/v1/get?type=https',
        'https://www.proxy-list.download/api/v1/get?type=http',
        'https://www.proxy-list.download/api/v1/get?type=socks4',
        'https://www.proxy-list.download/api/v1/get?type=socks5',
        'https://www.proxy-list.download/api/v1/get?type=all',
        'https://www.sslproxies.org/',
        'https://www.us-proxy.org/',
        'https://free-proxy-list.net/',
        'https://www.proxy-list.download/',
        'https://www.proxy-list.org/eng/proxylist.txt',
        "https://api.proxyscrape.com/v2/?request=displayproxies&protocol=http&timeout=10000&country=all&ssl=all&anonymity=all"
    ]
    
    for i in range(2, 52): 
        proxy_urls.append(f'https://free-proxy-list.net/uk-proxy.html?page={i}')

    proxies = set()

    for url in proxy_urls:
        response_text = fetch_response(url)
        if response_text:
            if response_text.startswith('{') and response_text.endswith('}'):
                try:
                    data = json.loads(response_text)
                    if isinstance(data, dict) and 'data' in data:
                        new_proxies = {f"{proxy_data['ip']}:{proxy_data['port']}" for proxy_data in data['data'] if 'ip' in proxy_data and 'port' in proxy_data}
                        proxies.update(new_proxies)
                except ValueError:
                    pass
            else:
                if 'free-proxy-list.net' in url:
                    soup = BeautifulSoup(response_text, 'html.parser')
                    table = soup.find('table', {'id': 'proxylisttable'})
                    if table:  # Verifica si se encontró la tabla
                        for row in table.find_all('tr')[1:]:
                            try:
                                columns = row.find_all('td')
                                ip = columns[0].text.strip()
                                port = columns[1].text.strip()
                                proxy = f"{ip}:{port}"
                                proxies.add(proxy)
                            except:
                                pass
                else:
                    lines = response_text.splitlines()
                    new_proxies = {line.strip() for line in lines if line.strip()}
                    proxies.update(new_proxies)

    with proxies_lock:
        for proxy in proxies:
            if is_valid_proxy(proxy):
                valid_proxies[proxy] = True
            else:
                invalid_proxies[proxy] = True
    return valid_proxies, invalid_proxies

@app.on_event("startup")
async def on_startup():
    global valid_proxies, invalid_proxies
    valid_proxies, invalid_proxies = await fetch_proxies_from_sources()

def fetch_response(url):
    headers = create_headers()
    try:
        response = requests.get(url, headers=headers)
        response.raise_for_status()
        return response.text
    except requests.RequestException:
        return None


@app.get("/")
async def root():
    html_content = '''
    <html>
        <body>
            <h1>Proxy List</h1>
            <p><a href="/valid">Valid Proxies</a></p>
            <p><a href="/invalid">Invalid Proxies</a></p>
        </body>
    </html>
    '''
    return HTMLResponse(content=html_content, status_code=200)

@app.get("/valid")
async def valid_proxies_endpoint():
    with proxies_lock:
        valid_proxies_list = list(valid_proxies.keys())
        if len(valid_proxies_list) > 10:
            valid_proxies_list = random.sample(valid_proxies_list, 10)
    response_content = "\n".join(valid_proxies_list)
    return Response(content=response_content, media_type='text/plain')

@app.get("/invalid")
async def invalid_proxies_endpoint():
    with proxies_lock:
        invalid_proxies_list = list(invalid_proxies.keys())
        if len(invalid_proxies_list) > 10:
            invalid_proxies_list = random.sample(invalid_proxies_list, 10)
    response_content = "\n".join(invalid_proxies_list)
    return Response(content=response_content, media_type='text/plain')

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        with proxies_lock:
            proxies_list = list(valid_proxies.keys())
        await websocket.send_text("\n".join(proxies_list))
        await asyncio.sleep(0)

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=7860)