you tybe bot

import os import json import asyncio import aiohttp from openai import OpenAI from pathlib import Path # --- КОНФИГУРАЦИЯ --- # Обязательно активируйте "YouTube Data API v3" в Google Cloud Console YOUTUBE_API_KEY = os.getenv("YOUTUBE_API_KEY", "YOUR_YOUTUBE_API_KEY_HERE") OPENAI_KEY = os.getenv("OPENAI_API_KEY", "YOUR_OPENAI_API_KEY_HERE") # API Эндпоинты Google SEARCH_URL = "https://www.googleapis.com/youtube/v3/search" VIDEOS_URL = "https://googleapis.com" COMMENTS_URL = "https://googleapis.com" openai_client = OpenAI(api_key=OPENAI_KEY) BRAIN_FILE = Path("youtube_agent_brain.json") class YouTubeLearningAgent: def __init__(self, niche_keyword: str): self.niche_keyword = niche_keyword self.brain_data = self._load_brain() def _load_brain(self) -> dict: """Загружает долговременную память агента""" if BRAIN_FILE.exists(): try: return json.loads(BRAIN_FILE.read_text(encoding="utf-8")) except Exception: return {"analyzed_videos": [], "learned_insights": []} return {"analyzed_videos": [], "learned_insights": []} def _save_brain(self, video_id: str, title: str, insights: dict): """Интегрирует новый опыт в базу знаний ИИ""" self.brain_data["analyzed_videos"].append(video_id) experience_entry = { "video_id": video_id, "title": title, "audience_pains": insights.get("audience_pains", []), "recommended_next_topics": insights.get("recommended_next_topics", []), "sentiment_score": insights.get("sentiment_score", "Neutral") } self.brain_data["learned_insights"].append(experience_entry) # Запись на диск BRAIN_FILE.write_text(json.dumps(self.brain_data, indent=2, ensure_ascii=False), encoding="utf-8") print(f"[🧠 Memory Updated] AI successfully learned from video: '{title}'") # 1. СКАУТИНГ ВИДЕО: Оптимизированный поиск трендов async def fetch_trending_videos(self, session: aiohttp.ClientSession, max_results: int = 2) -> list: """Ищет видео, отсортированные по числу просмотров""" params = { "part": "snippet", "q": self.niche_keyword, "maxResults": max_results, "type": "video", "order": "viewCount", # Сортировка по просмотрам "key": YOUTUBE_API_KEY } async with session.get(SEARCH_URL, params=params) as response: if response.status != 200: print(f"[❌ API Error] Search failed. Status: {response.status}") return [] data = await response.json() return data.get("items", []) # 2. ПАРСИНГ КОММЕНТАРИЕВ: Глубокая выгрузка текстов async def fetch_video_comments(self, session: aiohttp.ClientSession, video_id: str, max_comments: int = 15) -> list: """Выгружает текстовые треды комментариев под конкретным видео""" params = { "part": "snippet", "videoId": video_id, "maxResults": max_comments, "textFormat": "plainText", "key": YOUTUBE_API_KEY } async with session.get(COMMENTS_URL, params=params) as response: if response.status != 200: # Некоторые каналы отключают комментарии print(f"[⚠️ Warning] Comments are likely disabled for video {video_id}.") return [] data = await response.json() comments = [] for item in data.get("items", []): # Извлекаем чистый текст верхнеуровневого комментария top_comment = item["snippet"]["topLevelComment"]["snippet"]["textDisplay"] comments.append(top_comment) return comments # 3. МОДУЛЬ САМООБУЧЕНИЯ (LLM Cognitive Loop) def analyze_and_learn_from_feedback(self, video_title: str, comments: list) -> dict: """Изучает пул комментариев, вычленяя инсайты и паттерны поведения аудитории""" print(f"[🤖 AI Brain] Analyzing {len(comments)} audience comments for cognitive insights...") if not comments: return {"audience_pains": ["None"], "recommended_next_topics": ["General intro"], "sentiment_score": "Unknown"} # Проектируем строгий JSON-формат для вывода структурированных инсайтов json_schema = { "name": "youtube_audience_insights", "strict": True, "schema": { "type": "object", "properties": { "audience_pains": { "type": "array", "items": {"type": "string"}, "description": "What problems, bugs, or difficulties are users complaining about in comments?" }, "recommended_next_topics": { "type": "array", "items": {"type": "string"}, "description": "Based on what users ask or want to see, what should our next high-traffic video topic be?" }, "sentiment_score": { "type": "string", "enum": ["Positive", "Neutral", "Negative", "Highly Skeptical"], "description": "General emotional response of the audience to this video content." } }, "required": ["audience_pains", "recommended_next_topics", "sentiment_score"], "additionalProperties": False } } system_prompt = ( "You are an Elite YouTube Growth Strategist Node. Your goal is to reverse-engineer viral videos. " "Analyze the comments to find out exactly what the audience needs, what they are struggling with, " "and what content gaps you can exploit to build a better, more profitable script." ) user_prompt = f""" Video Title: {video_title} Collected Audience Comments: {json.dumps(comments, ensure_ascii=False)} """ response = openai_client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt} ], response_format={"type": "json_schema", "json_schema": json_schema} ) return json.loads(response.choices.message.content) # --- ГЛАВНЫЙ ИСПОЛНИТЕЛЬНЫЙ ЦИКЛ (ОРКЕСТРАТОР) --- async def execute_node_lifecycle(self): async with aiohttp.ClientSession() as session: print(f"[🚀 Initialization] Starting YouTube Agent for niche: '{self.niche_keyword}'...") # Шаг 1: Сканируем трендовые видео в нише videos = await self.fetch_trending_videos(session, max_results=2) analyzed_set = set(self.brain_data.get("analyzed_videos", [])) for video in videos: video_id = video["id"]["videoId"] video_title = video["snippet"]["title"] # Защита от дублирования: не анализируем одно видео дважды if video_id in analyzed_set: print(f"[⏭️ Skip] Video '{video_title}' has already been processed.") continue print(f"\n🎬 [Processing Video] Title: {video_title} | ID: {video_id}") # Шаг 2: Глубокий асинхронный парсинг комментариев comments = await self.fetch_video_comments(session, video_id, max_comments=10) print(f"[📥 Parser] Extracted {len(comments)} comments successfully.") # Шаг 3: Самообучение и вычленение бизнес-инсайтов insights = self.analyze_and_learn_from_feedback(video_title, comments) # Шаг 4: Запись опыта в долговременную память (JSON) self._save_brain(video_id, video_title, insights) # Задержка для защиты лимитов await asyncio.sleep(2) if __name__ == "__main__": # Настраиваем агента на парсинг и самообучение в трендовой нише ИИ-ассистентов # Для реального запуска убедитесь, что в переменных среды прописаны ваши API-ключи agent = YouTubeLearningAgent(niche_keyword="AI agents code tutorial") # Запуск автономного воркфлоу asyncio.run(agent.execute_node_lifecycle())