publishes posts on forums

import os import json import asyncio import aiohttp from openai import OpenAI from pathlib import Path from bs4 import BeautifulSoup # --- КРИТИЧЕСКАЯ КОНФИГУРАЦИЯ --- DISCOURSE_URL = os.getenv("DISCOURSE_FORUM_URL", "https://discourse.org") DISCOURSE_API_KEY = os.getenv("DISCOURSE_API_KEY", "mock_discourse_key") DISCOURSE_USER = os.getenv("DISCOURSE_API_USERNAME", "ai_agent_bot") OPENAI_KEY = os.getenv("OPENAI_API_KEY", "mock_openai_key") openai_client = OpenAI(api_key=OPENAI_KEY) MEMORY_FILE = Path("forum_agent_memory.json") class SelfLearningForumAgent: def __init__(self, product_description: str, value_proposition: str): self.product_description = product_description self.value_proposition = value_proposition self.headers = { "Api-Key": DISCOURSE_API_KEY, "Api-Username": DISCOURSE_USER, "Content-Type": "application/json" } self.memory = self._load_memory() # 1. МОДУЛЬ ПАМЯТИ: Загрузка исторического опыта для самообучения def _load_memory(self) -> dict: if MEMORY_FILE.exists(): try: return json.loads(MEMORY_FILE.read_text(encoding="utf-8")) except Exception: return {"posted_topics": [], "performance_records": []} return {"posted_topics": [], "performance_records": []} def _record_experience(self, topic_id: int, status: str, reply_text: str, reason: str = "") -> None: """Сохраняет опыт (успех/ошибка) в локальную базу данных для Few-Shot обучения""" self.memory["posted_topics"].append(topic_id) record = { "topic_id": topic_id, "status": status, # "SUCCESS" или "REJECTED" "reply_length": len(reply_text), "reply_snippet": reply_text[:150] + "...", "reason": reason } self.memory["performance_records"].append(record) # Ограничиваем память последними 50 записями для оптимизации контекстного окна LLM if len(self.memory["performance_records"]) > 50: self.memory["performance_records"].pop(0) MEMORY_FILE.write_text(json.dumps(self.memory, indent=2, ensure_ascii=False), encoding="utf-8") print(f"[🧠 Memory Logged] Outcome: {status} | Topic ID: {topic_id}") def _get_learning_context(self) -> str: """Формирует инструкции на основе анализа прошлых ошибок и успехов""" records = self.memory.get("performance_records", []) if not records: return "No prior experience available. Rely on standard native positioning guidelines." successes = [r for r in records if r["status"] == "SUCCESS"] failures = [r for r in records if r["status"] == "REJECTED"] learning_prompt = "\n=== SELF-LEARNING INSIGHTS FROM PAST OUTCOMES ===\n" learning_prompt += f"Statistical History: {len(successes)} approved posts, {len(failures)} banned/rejected posts.\n" if failures: learning_prompt += "CRITICAL WARNINGS (Avoid these patterns, they triggered automated spam filters):\n" for f in failures[-3:]: learning_prompt += f"- Failed post snippet: \"{f['reply_snippet']}\". Reason: {f['reason']}\n" if successes: learning_prompt += "SUCCESS PATTERNS (Emulate these structures, they successfully passed human moderation):\n" for s in successes[-3:]: learning_prompt += f"- Approved post snippet: \"{s['reply_snippet']}\"\n" return learning_prompt # 2. МОНИТОРИНГ: Асинхронный сбор тем async def fetch_latest_topics(self, session: aiohttp.ClientSession, category_id: int) -> list: url = f"{DISCOURSE_URL}/c/{category_id}.json" async with session.get(url, headers=self.headers) as response: if response.status == 200: data = await response.json() return data.get("topic_list", {}).get("topics", []) print(f"[❌ Error] Failed to scan category: {response.status}") return [] # 3. АНАЛИЗ КОНТЕНТА: Извлечение смыслового ядра обсуждения async def fetch_topic_context(self, session: aiohttp.ClientSession, topic_id: int) -> str: url = f"{DISCOURSE_URL}/t/{topic_id}.json" async with session.get(url, headers=self.headers) as response: if response.status == 200: data = await response.json() post_stream = data.get("post_stream", {}).get("posts", []) context_text = "" for post in post_stream[:3]: username = post.get("username", "User") raw_text = BeautifulSoup(post.get("cooked", ""), "lxml").get_text() context_text += f"{username}: {raw_text}\n---\n" return context_text return "" # 4. КРЕАТИВНЫЙ ИИ-МОДУЛЬ С ПОДДЕРЖКОЙ ОБРАТНОЙ СВЯЗИ (Few-Shot Calibration) def generate_adaptive_reply(self, forum_context: str) -> str: print("[🤖 AI Agent] Dynamic optimization of response strategy...") # Динамическое извлечение уроков из памяти агента experience_guidelines = self._get_learning_context() system_prompt = ( "You are a highly cognitive, adaptive community expert on digital platforms. Your primary goal is to answer forum questions natively. " "Rule 1: Provide high-quality technical or business value first based on the thread context. " "Rule 2: Integrate a subtle, organic mention of a relevant tool ONLY if it naturally solves the user's explicit problem. " "Rule 3: Adhere strictly to the attached Self-Learning insights to fix bugs and errors from your past outreach attempts." ) user_prompt = f""" Our Product Context: {self.product_description} Value Proposition: {self.value_proposition} Current Forum Thread Context: {forum_context} {experience_guidelines} Generate the final responsive message. Be natural, professional, and peer-to-peer. No robotic introductions. """ response = openai_client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt} ], temperature=0.3 # Пониженный шум для предсказуемого и точного выполнения ) return response.choices.message.content # 5. ИСПРАВЛЕННЫЙ МОДУЛЬ ПУБЛИКАЦИИ С ОБРАБОТКОЙ ОШИБОК async def publish_reply(self, session: aiohttp.ClientSession, topic_id: int, reply_text: str) -> bool: url = f"{DISCOURSE_URL}/posts.json" payload = {"topic_id": topic_id, "raw": reply_text} async with session.post(url, headers=self.headers, json=payload) as response: # ИСПРАВЛЕНО: Синтаксическая ошибка решена, введен массив успешных статусов API Discourse if response.status in: print(f"[🚀 Success] Post verified on forum. Released to public loop.") self._record_experience(topic_id, "SUCCESS", reply_text) return True else: error_body = await response.text() # Агент запоминает ошибку, чтобы скорректировать промпт в следующем раунде self._record_experience( topic_id=topic_id, status="REJECTED", reply_text=reply_text, reason=f"HTTP {response.status} - {error_body[:100]}" ) return False # --- ОРКЕСТРАТОР АВТОНОМНОГО ВОРКФЛОУ --- async def start_outreach_flow(self, target_category_id: int): async with aiohttp.ClientSession() as session: print(f"[🔄 Operational] Initializing Node Core on {DISCOURSE_URL}...") topics = await self.fetch_latest_topics(session, target_category_id) posted_topics = set(self.memory.get("posted_topics", [])) for topic in topics: topic_id = topic["id"] # Защита от дублирования сообщений if topic_id in posted_topics: continue print(f"\n[🧵 Thread Active] Processing ID: {topic_id} | Title: {topic['title']}") context = await self.fetch_topic_context(session, topic_id) if not context: continue # ИИ обучается на логах и генерирует выверенный ответ ai_reply = self.generate_adaptive_reply(context) # Публикация и логирование результатов в память await self.publish_reply(session, topic_id, ai_reply) # Интервал защиты от спам-фильтров Cloudflare/Discourse await asyncio.sleep(20) if __name__ == "__main__": agent = SelfLearningForumAgent( product_description="Thompson-Sampling Multi-Armed Bandit Bidding Optimizer Node", value_proposition="Automates programmatic Web3 ad spending to secure target 50% margins." ) # Симуляция обработки маркетинговой категории asyncio.run(agent.start_outreach_flow(target_category_id=5))