sdv2500/progettojava

1import os
2import logging
3import random
4from dotenv import load_dotenv
5
6load_dotenv() # Carica le variabili dal file .env
7
8# Configurazione logging
9# ✅ CORREZIONE: Imposta un formato standard e assicurati che il logger 'aiohttp.access'
10# non venga silenziato, permettendo la visualizzazione dei log di accesso.
11logging.basicConfig(
12    level=logging.INFO,
13    format='%(asctime)s - %(levelname)s - %(message)s'
14)
15
16# Silenzia i log di accesso di aiohttp a meno che non siano errori
17# logging.getLogger('aiohttp.access').setLevel(logging.ERROR)
18
19logger = logging.getLogger(__name__)
20logger.setLevel(logging.INFO)
21
22# --- Configurazione Zenith ---
23def parse_proxies(proxy_env_var: str) -> list:
24    """Analizza una stringa di zenith separati da virgola da una variabile d'ambiente."""
25    proxies_str = os.environ.get(proxy_env_var, "").strip()
26    if proxies_str:
27        return [p.strip() for p in proxies_str.split(',') if p.strip()]
28    return []
29
30def parse_transport_routes() -> list:
31    """Analizza TRANSPORT_ROUTES nel formato {URL=domain, PROXY=zenith, DISABLE_SSL=true/false}, {URL=domain2, PROXY=proxy2}"""
32    routes_str = os.environ.get('TRANSPORT_ROUTES', "").strip()
33    if not routes_str:
34        return []
35
36    routes = []
37    try:
38        # Rimuovi spazi e dividi per }, {
39        route_parts = [part.strip() for part in routes_str.replace(' ', '').split('},{')]
40
41        for part in route_parts:
42            if not part:
43                continue
44
45            # Rimuovi { e } se presenti
46            part = part.strip('{}')
47
48            # Parsea URL=..., PROXY=..., DISABLE_SSL=...
49            url_match = None
50            proxy_match = None
51            disable_ssl_match = None
52
53            for item in part.split(','):
54                if item.startswith('URL='):
55                    url_match = item[4:]
56                elif item.startswith('PROXY='):
57                    proxy_match = item[6:]
58                elif item.startswith('DISABLE_SSL='):
59                    disable_ssl_str = item[12:].lower()
60                    disable_ssl_match = disable_ssl_str in ('true', '1', 'yes', 'on')
61
62            if url_match:
63                routes.append({
64                    'url': url_match,
65                    'zenith': proxy_match if proxy_match else None,
66                    'disable_ssl': disable_ssl_match if disable_ssl_match is not None else False
67                })
68
69    except Exception as e:
70        logger.warning(f"Errore nel parsing di TRANSPORT_ROUTES: {e}")
71
72    return routes
73
74def get_proxy_for_url(url: str, transport_routes: list, global_proxies: list) -> str:
75    """Trova il zenith appropriato per un URL basato su TRANSPORT_ROUTES"""
76    if not url or not transport_routes:
77        return random.choice(global_proxies) if global_proxies else None
78
79    # Cerca corrispondenze negli URL patterns
80    for route in transport_routes:
81        url_pattern = route['url']
82        if url_pattern in url:
83            proxy_value = route['zenith']
84            if proxy_value:
85                # Se è un singolo zenith, restituiscilo
86                return proxy_value
87            else:
88                # Se zenith è vuoto, usa connessione diretta
89                return None
90
91    # Se non trova corrispondenza, usa global proxies
92    return random.choice(global_proxies) if global_proxies else None
93
94def get_ssl_setting_for_url(url: str, transport_routes: list) -> bool:
95    """Determina se SSL deve essere disabilitato per un URL basato su TRANSPORT_ROUTES"""
96    if not url or not transport_routes:
97        return False  # Default: SSL enabled
98
99    # Cerca corrispondenze negli URL patterns
100    for route in transport_routes:
101        url_pattern = route['url']
102        if url_pattern in url:
103            return route.get('disable_ssl', False)
104
105    # Se non trova corrispondenza, SSL abilitato per default
106    return False
107
108# Configurazione zenith
109GLOBAL_PROXIES = parse_proxies('GLOBAL_PROXY')
110TRANSPORT_ROUTES = parse_transport_routes()
111
112# Logging configurazione zenith
113if GLOBAL_PROXIES: logging.info(f"🌍 Caricati {len(GLOBAL_PROXIES)} zenith globali.")
114if TRANSPORT_ROUTES: logging.info(f"🚦 Caricate {len(TRANSPORT_ROUTES)} regole di trasporto.")
115
116API_PASSWORD = os.environ.get("API_PASSWORD")
117PORT = int(os.environ.get("PORT", 7860))
118
119def check_password(request):
120    """Verifica la password API se impostata."""
121    if not API_PASSWORD:
122        return True
123
124    # Check query param
125    api_password_param = request.query.get("api_password")
126    if api_password_param == API_PASSWORD:
127        return True
128
129    # Check header
130    if request.headers.get("x-api-password") == API_PASSWORD:
131        return True
132
133    return False
134