2334

import os import asyncio from telethon import TelegramClient, events from openai import OpenAI from web3 import Web3 # --- КОНФИГУРАЦИЯ И ИНИЦИАЛИЗАЦИЯ --- API_ID = int(os.getenv("TG_API_ID", 123456)) API_HASH = os.getenv("TG_API_HASH", "your_tg_api_hash") OPENAI_KEY = os.getenv("OPENAI_API_KEY", "your_openai_key") WEB3_PROVIDER = os.getenv("RPC_PROVIDER_URL", "https://polygon-rpc.com") USDC_CONTRACT_ADDRESS = "0x2791Bca1f2de4661ED88A30C99A7a9449Aa84174" # USDC на Polygon tg_client = TelegramClient('ai_buyer_session', API_ID, API_HASH) openai_client = OpenAI(api_key=OPENAI_KEY) w3 = Web3(Web3.HTTPProvider(WEB3_PROVIDER)) # Минимальный ABI для проверки баланса ERC-20 токена (USDC/USDT) MIN_ERC20_ABI = [ {"constant": True, "inputs": [{"name": "_owner", "type": "address"}], "name": "balanceOf", "outputs": [{"name": "balance", "type": "uint256"}], "type": "function"}, {"constant": True, "inputs": [], "name": "decimals", "outputs": [{"name": "remaining", "type": "uint8"}], "type": "function"} ] class AITrafficManager: def __init__(self, wallet_address, product_description, target_audience): self.wallet_address = wallet_address self.product_description = product_description self.target_audience = target_audience self.channels_to_scout = [] self.active_campaigns = {} # 1. МОДУЛЬ ФИНАНСОВОГО КОНТРОЛЯ (Web3) def check_ad_budget(self) -> float: """Проверяет баланс USDC на криптокошельке агента перед закупкой""" contract = w3.eth.contract(address=w3.to_checksum_address(USDC_CONTRACT_ADDRESS), abi=MIN_ERC20_ABI) raw_balance = contract.functions.balanceOf(w3.to_checksum_address(self.wallet_address)).call() decimals = contract.functions.decimals().call() balance = raw_balance / (10 ** decimals) print(f"[🌐 Wallet] Current AI budget: {balance} USDC") return balance # 2. КРЕАТИВНЫЙ МОДУЛЬ (OpenAI GPT + DALL-E) def generate_ad_bundle(self) -> dict: """Автономно генерирует рекламный пост и промт для баннера""" print("[🤖 Creative] Generating global ad copy and banner prompt...") # Генерация текста text_prompt = f"Create a high-converting Telegram ad post for this product: '{self.product_description}'. Target audience: {self.target_audience}. Include strong CTA, emojis, but keep it concise." completion = openai_client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": text_prompt}] ) ad_text = completion.choices[0].message.content # Генерация баннера image_prompt = f"Modern, high-quality promo banner for digital product. Description: {self.product_description}. Minimalistic design, cyberpunk or tech aesthetic, no text on image." image_response = openai_client.images.generate( model="dall-e-3", prompt=image_prompt, n=1, size="1024x1024" ) banner_url = image_response.data[0].url return {"text": ad_text, "banner_url": banner_url} # 3. СКАУТ-МОДУЛЬ (Поиск каналов через глобальный парсинг) async def scout_target_channels(self, keyword: str): """Ищет глобальные Telegram-каналы по ключевым словам через глобальный поиск API""" print(f"[🔍 Scout] Searching channels globally for keyword: '{keyword}'...") # Базовый поиск сущностей через Telethon API напрямую в Telegram result = await tg_client(functions.contacts.SearchRequest(q=keyword, limit=10)) for chat in result.chats: if hasattr(chat, 'username') and chat.username: # Фильтруем только каналы if getattr(chat, 'broadcast', False): self.channels_to_scout.append(chat.username) print(f"[🔍 Scout] Found {len(self.channels_to_scout)} potential channels.") # 4. БАЙЕР-МОДУЛЬ (Автономные переговоры через UserBot) async def negotiate_with_admin(self, channel_username: str, ad_text: str): """Автоматически находит контакт админа и предлагает сделку""" try: entity = await tg_client.get_entity(channel_username) # Пытаемся вытащить описание канала, чтобы найти юзернейм админа (обычно указан там) full_chat = await tg_client(functions.channels.GetFullChannelRequest(channel=entity)) about_text = full_chat.full_chat.about print(f"[💬 Buyer] Analyzing channel '{channel_username}' description for admin contact...") # Базовый поиск контакта админа в описании (ищем '@') # В продакшене здесь подключается LLM для точного извлечения контакта из текста words = about_text.split() admin_contact = None for word in words: if word.startswith('@') and word.lower() != f"@{channel_username}".lower(): admin_contact = word break if admin_contact: proposal = f"Hello! I am an AI Media Buyer. I want to buy an ad placement in @{channel_username}. What are your current rates for 1/24 or 2/48 formats? Payment in USDT/USDC available." await tg_client.send_message(admin_contact, proposal) print(f"[🚀 Buyer] Proposal sent successfully to admin {admin_contact}") self.active_campaigns[channel_username] = {"status": "negotiating", "text": ad_text} else: print(f"[⚠️ Buyer] Could not automatically find admin contact for @{channel_username}") except Exception as e: print(f"[❌ Error] Failed to process channel {channel_username}: {e}") # --- ОРКЕСТРАТОР (ЗАПУСК СИСТЕМЫ) --- async def main(): await tg_client.start() # Инициализируем агента (на входе: Web3-адрес, инфо о продукте, ЦА) ai_agent = AITrafficManager( wallet_address="0x71C7656EC7ab88b098defB751B7401B5f6d1476B", product_description="SaaS platform for automated crypto bookkeeping", target_audience="Web3 founders, crypto traders, indie hackers" ) # Шаг 1: Проверка бюджета (финансовый чек) budget = ai_agent.check_ad_budget() if budget < 50: # Условный минимум в USDC print("[🛑 Stop] Insufficient crypto budget to start campaign.") return # Шаг 2: Скаутинг площадок по ключевому слову await ai_agent.scout_target_channels("crypto") if ai_agent.channels_to_scout: # Шаг 3: Генерация креативного бандла bundle = ai_agent.generate_ad_bundle() # Шаг 4: Пилотный заход на первый найденный канал target_channel = ai_agent.channels_to_scout[0] await ai_agent.negotiate_with_admin(target_channel, bundle["text"]) if __name__ == "__main__": from telethon import functions asyncio.run(main())