import asyncio import re import random import json import os from datetime import datetime from typing import List, Set import logging from concurrent.futures import ThreadPoolExecutor from telethon import TelegramClient from telethon.tl.types import Message # تنظیمات لاگ logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('proxy_bot.log', encoding='utf-8'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) class ProxyCache: """کلاس مدیریت کش فایل‌محور""" def __init__(self, cache_file: str = None): if cache_file is None: # Use environment variable or default to data directory cache_file = os.getenv('CACHE_FILE', '/app/data/proxy_cache.json') self.cache_file = cache_file # Create directory if it doesn't exist os.makedirs(os.path.dirname(self.cache_file), exist_ok=True) self.cache = self._load_cache() def _load_cache(self) -> dict: """بارگذاری کش از فایل""" try: if os.path.exists(self.cache_file): with open(self.cache_file, 'r', encoding='utf-8') as f: return json.load(f) except Exception as e: logger.error(f"Error loading cache: {e}") return {"sent_proxies": [], "collected_proxies": [], "last_update": None} def _save_cache(self): """ذخیره کش در فایل""" try: with open(self.cache_file, 'w', encoding='utf-8') as f: json.dump(self.cache, f, ensure_ascii=False, indent=2) except Exception as e: logger.error(f"Error saving cache: {e}") def add_sent_proxy(self, proxy: str): """اضافه کردن پروکسی ارسال شده""" if proxy not in self.cache["sent_proxies"]: self.cache["sent_proxies"].append(proxy) # حفظ فقط آخرین 1000 پروکسی ارسال شده if len(self.cache["sent_proxies"]) > 1000: self.cache["sent_proxies"] = self.cache["sent_proxies"][-1000:] self._save_cache() def is_sent(self, proxy: str) -> bool: """بررسی ارسال شدن پروکسی""" return proxy in self.cache["sent_proxies"] def update_collected_proxies(self, proxies: List[str]): """بروزرسانی پروکسی‌های جمع‌آوری شده""" self.cache["collected_proxies"] = proxies self.cache["last_update"] = datetime.now().isoformat() self._save_cache() def get_unsent_proxies(self) -> List[str]: """دریافت پروکسی‌های ارسال نشده""" return [p for p in self.cache["collected_proxies"] if not self.is_sent(p)] class ProxyBot: def __init__(self, api_id: int, api_hash: str, phone_number: str, backup_channel: str): self.api_id = api_id self.api_hash = api_hash self.phone_number = phone_number self.backup_channel = backup_channel self.client = TelegramClient('proxy_bot', api_id, api_hash) self.cache = ProxyCache() async def start(self): """شروع کلاینت""" try: await self.client.start(phone=self.phone_number) logger.info("✅ Bot client started successfully!") except Exception as e: logger.error(f"❌ Error starting client: {str(e)}") raise def extract_proxies(self, text: str) -> List[str]: """استخراج ساده‌تر پروکسی‌ها از متن""" if not text or not text.strip(): return [] proxies = [] # الگوهای پروکسی patterns = [ # vmess:// r'vmess://[A-Za-z0-9+/=]+', # vless:// r'vless://[a-fA-F0-9-]+@[^:\s]+:[0-9]+[^\s]*', # trojan:// r'trojan://[^@\s]+@[^:\s]+:[0-9]+[^\s]*', # ss:// r'ss://[A-Za-z0-9+/=]+@[^:\s]+:[0-9]+[^\s]*', # ssr:// r'ssr://[A-Za-z0-9+/=]+', # MTProto - لینک کامل r'(?:tg://proxy\?|https://t\.me/proxy\?)server=[^&\s]+&port=[0-9]+&secret=[a-fA-F0-9]+', ] # استخراج پروکسی‌های معمولی for pattern in patterns: try: matches = re.findall(pattern, text, re.IGNORECASE) proxies.extend(matches) except Exception as e: logger.warning(f"Error with pattern: {str(e)}") # استخراج MTProto ساده (حتی با secret ناقص) mtproto_proxies = self.extract_simple_mtproto(text) proxies.extend(mtproto_proxies) return list(set(proxies)) # حذف تکراری def extract_simple_mtproto(self, text: str) -> List[str]: """استخراج MTProto با secret ناقص""" mtproto_proxies = [] # فرمت‌های ساده MTProto patterns = [ # IP:PORT:SECRET (حتی اگر secret کوتاه باشد) r'([0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}):([0-9]+):([a-fA-F0-9]+)', # Server Port Secret در خطوط جداگانه r'(?i)server[:\s]*([0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3})[,\s\n]*port[:\s]*([0-9]+)[,\s\n]*secret[:\s]*([a-fA-F0-9]+)', ] for pattern in patterns: try: matches = re.findall(pattern, text, re.MULTILINE) for match in matches: if len(match) == 3: ip, port, secret = match # بررسی ساده IP و Port if self.is_valid_ip(ip) and self.is_valid_port(port) and len(secret) >= 1: # ایجاد لینک MTProto حتی با secret ناقص mtproto_link = f"https://t.me/proxy?server={ip}&port={port}&secret={secret}" mtproto_proxies.append(mtproto_link) except Exception as e: logger.warning(f"Error extracting MTProto: {str(e)}") return mtproto_proxies def is_valid_ip(self, ip: str) -> bool: """بررسی ساده IP""" try: parts = ip.split('.') return len(parts) == 4 and all(0 <= int(part) <= 255 for part in parts) except: return False def is_valid_port(self, port: str) -> bool: """بررسی ساده Port""" try: return 1 <= int(port) <= 65535 except: return False async def collect_from_channel(self, channel: str) -> List[str]: """جمع‌آوری پروکسی از یک کانال""" proxies = [] try: async for message in self.client.iter_messages(channel, limit=50): if message and message.text: found_proxies = self.extract_proxies(message.text) proxies.extend(found_proxies) logger.info(f"✅ {channel}: {len(proxies)} proxies") return proxies except Exception as e: logger.error(f"❌ Error in {channel}: {str(e)}") return [] async def collect_all_proxies_parallel(self, channels: List[str]) -> List[str]: """جمع‌آوری موازی از همه کانال‌ها""" logger.info(f"🔄 Collecting from {len(channels)} channels in parallel...") tasks = [self.collect_from_channel(channel) for channel in channels] results = await asyncio.gather(*tasks, return_exceptions=True) all_proxies = [] for i, result in enumerate(results): if isinstance(result, list): all_proxies.extend(result) else: logger.error(f"❌ Channel {channels[i]} failed: {result}") unique_proxies = list(set(all_proxies)) logger.info(f"📊 Total: {len(all_proxies)} | Unique: {len(unique_proxies)}") return unique_proxies def create_simple_message(self, proxy: str) -> str: """ایجاد پیام ساده برای یک پروکسی""" timestamp = datetime.now().strftime("%H:%M") # تشخیص نوع پروکسی if proxy.startswith('vmess://'): proxy_type = "VMess" emoji = "🔵" elif proxy.startswith('vless://'): proxy_type = "VLESS" emoji = "🟢" elif proxy.startswith('trojan://'): proxy_type = "Trojan" emoji = "🔴" elif proxy.startswith('ss://'): proxy_type = "Shadowsocks" emoji = "🟡" elif 'proxy?' in proxy: proxy_type = "MTProto" emoji = "🟦" else: proxy_type = "Proxy" emoji = "⚪" message = f"🌟 ═══════════════════════════════════ 🌟\n" message += f"🚀 **PREMIUM PROXY** 🚀\n" message += f"🌟 ═══════════════════════════════════ 🌟\n\n" message += f"📅 **{timestamp}** | {emoji} **{proxy_type}**\n\n" message += f"```{proxy}```\n\n" message += f"🌟 ═══════════════════════════════════ 🌟\n" message += f"💎 **Follow @fastify_proxy for more!** 💎\n" message += f"🌟 ═══════════════════════════════════ 🌟" return message async def send_single_proxy(self, proxy: str): """ارسال یک پروکسی""" try: message = self.create_simple_message(proxy) await self.client.send_message(self.backup_channel, message, parse_mode='markdown') self.cache.add_sent_proxy(proxy) logger.info(f"✅ Sent proxy: {proxy[:50]}...") except Exception as e: logger.error(f"❌ Error sending proxy: {str(e)}") async def load_channels(self, file_path: str) -> List[str]: """خواندن کانال‌ها از فایل""" try: with open(file_path, 'r', encoding='utf-8') as f: channels = [line.strip() for line in f if line.strip() and not line.strip().startswith('#')] return channels except Exception as e: logger.error(f"Error loading channels: {str(e)}") return [] async def collection_task(self, channels_file: str): """وظیفه جمع‌آوری پروکسی‌ها (هر 30 دقیقه)""" while True: try: logger.info("🔄 Starting collection cycle...") channels = await self.load_channels(channels_file) if channels: proxies = await self.collect_all_proxies_parallel(channels) self.cache.update_collected_proxies(proxies) logger.info(f"✅ Updated cache with {len(proxies)} proxies") else: logger.warning("⚠️ No channels loaded!") # انتظار 30 دقیقه await asyncio.sleep(30 * 60) except Exception as e: logger.error(f"❌ Collection error: {str(e)}") await asyncio.sleep(60) # انتظار 1 دقیقه در صورت خطا async def sending_task(self): """وظیفه ارسال پروکسی‌ها (هر دقیقه)""" while True: try: unsent_proxies = self.cache.get_unsent_proxies() if unsent_proxies: # انتخاب تصادفی یک پروکسی proxy = random.choice(unsent_proxies) await self.send_single_proxy(proxy) else: logger.info("💤 No unsent proxies available") # انتظار 1 دقیقه await asyncio.sleep(60) except Exception as e: logger.error(f"❌ Sending error: {str(e)}") await asyncio.sleep(30) # انتظار 30 ثانیه در صورت خطا async def run_forever(self, channels_file: str): """اجرای مداوم بات""" logger.info("🚀 Starting bot in continuous mode...") # شروع هر دو وظیفه به صورت موازی await asyncio.gather( self.collection_task(channels_file), self.sending_task() ) async def disconnect(self): """قطع اتصال""" try: await self.client.disconnect() logger.info("✅ Client disconnected") except Exception as e: logger.error(f"❌ Disconnect error: {str(e)}") async def main(): # تنظیمات از متغیرهای محیطی API_ID = int(os.getenv('API_ID', '2413260')) API_HASH = os.getenv('API_HASH', 'bf18c75d154b31b294857a28079163e3') PHONE_NUMBER = os.getenv('PHONE_NUMBER', '+989991219273') BACKUP_CHANNEL = os.getenv('BACKUP_CHANNEL', '@fastify_proxy') CHANNELS_FILE = os.getenv('CHANNELS_FILE', 'channels.txt') bot = ProxyBot(API_ID, API_HASH, PHONE_NUMBER, BACKUP_CHANNEL) try: await bot.start() await bot.run_forever(CHANNELS_FILE) except KeyboardInterrupt: logger.info("🛑 Bot stopped by user") except Exception as e: logger.error(f"❌ Bot error: {str(e)}") # Don't exit on error, restart after delay await asyncio.sleep(60) logger.info("🔄 Restarting bot...") await main() # Restart the bot finally: await bot.disconnect() if __name__ == "__main__": asyncio.run(main())