You Tube 2

import os import glob import pickle import pandas as pd import numpy as np from datetime import datetime, timedelta import nltk from nltk.sentiment.vader import SentimentIntensityAnalyzer from googleapiclient.discovery import build from googleapiclient.errors import HttpError from sklearn.ensemble import IsolationForest from sklearn.cluster import KMeans from sklearn.preprocessing import StandardScaler # Скачиваем лексикон для анализа тональности (работает для английского и базового транслита/эмодзи) # Для глубокого русского языка лучше использовать глубокие модели (библиотека dostoevsky), # но VADER — идеальный легковесный вариант, не требующий терабайтов памяти. try: nltk.data.find('sentiment/vader_lexicon.zip') except LookupError: nltk.download('vader_lexicon', quiet=True) class YouTubeAIAnalyticSystem: def __init__(self, api_key=None, credentials_path='client_secrets.json', db_path='yt_analytics_db.csv'): self.api_key = api_key self.db_path = db_path self.sia = SentimentIntensityAnalyzer() self.scaler = StandardScaler() self.anomaly_model = None self.cluster_model = None # Инициализация YouTube API (публичный доступ или OAuth) if api_key: self.youtube_data = build('youtube', 'v3', developerKey=api_key) self.youtube_analytics = None print("🤖 Система запущена в режиме мониторинга конкурентов (Public API Key).") else: # Если ключа нет, предполагается авторизация через OAuth для своего канала from google_auth_oauthlib.flow import InstalledAppFlow scopes = ['https://googleapis.com', 'https://googleapis.com'] if os.path.exists(credentials_path): flow = InstalledAppFlow.from_client_secrets_file(credentials_path, scopes) credentials = flow.run_local_server(port=0, open_browser=False) self.youtube_data = build('youtube', 'v3', credentials=credentials) self.youtube_analytics = build('youtubeAnalytics', 'v2', credentials=credentials) print("🔒 Система запущена в режиме полной аналитики своего канала (OAuth).") else: raise FileNotFoundError(f"Не найден файл авторизации {credentials_path} или API_KEY!") def fetch_video_comments(self, video_id, max_comments=30): """Парсинг комментариев и анализ их тональности (Sentiment Analysis)""" comments_data = [] try: response = self.youtube_data.commentThreads().list( part="snippet", videoId=video_id, textFormat="plainText", maxResults=max_comments ).execute() for item in response.get('items', []): comment = item['snippet']['topLevelComment']['snippet']['textDisplay'] # Вычисляем полярность текста (-1 негатив, 0 нейтрально, 1 позитив) score = self.sia.polarity_scores(comment)['compound'] comments_data.append(score) except HttpError: # Комментарии могут быть отключены у видео pass if not comments_data: return 0.0 # Нейтрально, если комментов нет return float(np.mean(comments_data)) def get_channel_videos(self, channel_id, limit=50): """Парсит метаданные видео любого канала (своего или конкурента)""" print(f"📥 Сбор данных с канала: {channel_id}...") try: # Получаем ID плейлиста "Загрузки" (Uploads) этого канала ch_resp = self.youtube_data.channels().list(part="contentDetails", id=channel_id).execute() if not ch_resp.get('items'): print("❌ Канал не найден.") return pd.DataFrame() uploads_playlist_id = ch_resp['items'][0]['contentDetails']['relatedPlaylists']['uploads'] video_list = [] next_page_token = None while len(video_list) < limit: playlist_resp = self.youtube_data.playlistItems().list( part="snippet,contentDetails", playlistId=uploads_playlist_id, maxResults=min(50, limit - len(video_list)), pageToken=next_page_token ).execute() for item in playlist_resp.get('items', []): video_list.append({ 'video_id': item['contentDetails']['videoId'], 'title': item['snippet']['title'], 'published_at': item['snippet']['publishedAt'] }) next_page_token = playlist_resp.get('nextPageToken') if not next_page_token: break return self._enrich_video_data(video_list, channel_id) except Exception as e: print(f"⚠️ Ошибка при парсинге данных канала: {e}") return pd.DataFrame() def _enrich_video_data(self, video_list, channel_id): """Обогащает базовый список видео глубокой аналитикой""" enriched_data = [] for v in video_list: v_id = v['video_id'] try: # 1. Публичная статистика видео v_resp = self.youtube_data.videos().list(part="statistics,contentDetails", id=v_id).execute() if not v_resp.get('items'): continue stats = v_resp['items'][0]['statistics'] details = v_resp['items'][0]['contentDetails'] views = int(stats.get('viewCount', 0)) comments_count = int(stats.get('commentCount', 0)) likes = int(stats.get('likeCount', 0)) # Парсинг длительности ISO 8601 (упрощенный перевод в секунды) duration_str = details.get('duration', '') duration_sec = self._parse_duration(duration_str) # 2. Тональность комментариев sentiment_score = self.fetch_video_comments(v_id) # 3. Закрытая статистика (Смотрибельность / Удержание) — доступна ТОЛЬКО для своего канала по OAuth retention_rate = 0.0 if self.youtube_analytics: end_date = datetime.today().strftime('%Y-%m-%d') start_date = (datetime.today() - timedelta(days=90)).strftime('%Y-%m-%d') analytics_resp = self.youtube_analytics.reports().query( ids=f'channel=={channel_id}', startDate=start_date, endDate=end_date, metrics='averageViewDuration', dimensions='video', filters=f'video=={v_id}' ).execute() if 'rows' in analytics_resp and analytics_resp['rows']: avg_view_sec = analytics_resp['rows'][0][1] if duration_sec > 0: retention_rate = (avg_view_sec / duration_sec) * 100 # Расчет относительных коэффициентов вовлеченности engagement_rate = ((comments_count + likes) / views * 100) if views > 0 else 0 enriched_data.append({ 'video_id': v_id, 'title': v['title'], 'published_at': v['published_at'], 'views': views, 'comments_count': comments_count, 'likes': likes, 'duration_sec': duration_sec, 'sentiment_score': sentiment_score, 'retention_rate': retention_rate if retention_rate > 0 else (engagement_rate * 1.5), # Прокси-метрика для чужих каналов 'engagement_rate': engagement_rate, 'timestamp_analyzed': datetime.now().isoformat() }) except Exception as e: print(f"Пропущено видео {v_id} из-за ошибки: {e}") continue return pd.DataFrame(enriched_data) def _parse_duration(self, duration_str): """Переводит формат YouTube длительности (например, PT1M45S) в секунды""" import re match = re.match(r'PT(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?', duration_str) if not match: return 0 hours = int(match.group(1) or 0) minutes = int(match.group(2) or 0) seconds = int(match.group(3) or 0) return hours * 3600 + minutes * 60 + seconds def update_database(self, new_df): """Интеграция новых данных в историческую БД с защитой от дубликатов""" if new_df.empty: return if os.path.exists(self.db_path): old_df = pd.read_csv(self.db_path) # Удаляем старые записи тех же видео, оставляя обновленные данные combined_df = pd.concat([new_df, old_df]).drop_duplicates(subset=['video_id'], keep='first') else: combined_df = new_df combined_df.to_csv(self.db_path, index=False) return combined_df def train_models_and_predict(self): """Самообучение моделей ИИ на накопленных исторических данных""" if not os.path.exists(self.db_path): print("❌ База данных пуста. Сначала соберите данные каналов.") return None df = pd.read_csv(self.db_path) if len(df) < 5: print("📉 Слишком мало данных в базе для машинного обучения. Требуется минимум 5 видео.") return df