File size: 2,499 Bytes
050a103 | 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 | import asyncio
import logging
import urllib.parse
import aiohttp
from proxy_manager import ProxyManager, Proxy
logger = logging.getLogger(__name__)
class ProxyUpdater:
def __init__(self, proxy_manager: ProxyManager):
self.pm = proxy_manager
self.update_url_template = "https://sasasas.zeabur.app/api/set/proxy/{}"
self.running = False
async def update_remote(self, proxy: Proxy):
"""Update the remote instance with the new proxy."""
# URL encode the proxy URL
encoded_proxy = urllib.parse.quote(proxy.url, safe='')
target_url = self.update_url_template.format(encoded_proxy)
logger.info(f"Updating remote instance with proxy: {proxy.url}")
logger.info(f"Target URL: {target_url}")
try:
async with aiohttp.ClientSession() as session:
async with session.get(target_url) as response:
text = await response.text()
if response.status == 200:
logger.info(f"Successfully updated remote proxy. Response: {text}")
else:
logger.error(f"Failed to update remote proxy. Status: {response.status}, Response: {text}")
except Exception as e:
logger.error(f"Exception during remote update: {e}")
async def run_loop(self):
"""Main loop ensuring proxy is updated every 5 minutes."""
self.running = True
while self.running:
logger.info("Starting scheduled proxy update...")
# 1. Refresh and find best proxy
proxies = await self.pm.refresh_proxies()
if self.pm.best_proxy:
# 2. Test one last time before updating (as requested)
logger.info(f"Verifying best proxy {self.pm.best_proxy.url} before update...")
await self.pm.check_proxy(self.pm.best_proxy)
if self.pm.best_proxy.latency != float('inf'):
# 3. Update remote
await self.update_remote(self.pm.best_proxy)
else:
logger.warning("Best proxy failed verification. Retrying in next cycle.")
else:
logger.warning("No valid proxies found to update.")
logger.info("Sleeping for 5 minutes...")
await asyncio.sleep(300) # 5 minutes
def stop(self):
self.running = False
|