neo/research-agent-plugin
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
d4731bee201ab0feba0edee0cc9cfcb991656b37
research-agent-plugin/scripts/youtube_analyzer.py
T

480 lines
18 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""
YouTube Analyzer — анализ видео как Perplexity Deep Research.
Три режима:
1. По URL: youtube_analyzer.py --url "https://youtube.com/watch?v=..."
2. Поиск: youtube_analyzer.py --search "LLM агенты 2026"
3. Пакет: youtube_analyzer.py --search "AI" --limit 3 --compare (сравнение нескольких)
Алгоритм:
- Получение субтитров через youtube-transcript-api
- Если субтитров нет — pytube + whisper
- LLM-синтез → structured report с таймкодами и цитатами
Использование:
python3 scripts/youtube_analyzer.py --url "https://youtube.com/watch?v=..."
python3 scripts/youtube_analyzer.py --search "термоядерный синтез 2026" --lang ru
python3 scripts/youtube_analyzer.py --search "LLM agents" --limit 3 --compare --lang en
python3 scripts/youtube_analyzer.py --url "..." --transcript-only
Опции:
--url URL видео
--search Поисковый запрос для поиска видео
--limit 1-5 (default: 1, для --search и --compare)
--compare Сравнить несколько видео по теме
--lang Язык субтитров: ru (default) | en
--transcript-only Только текст без LLM-анализа
--output Сохранить в файл
Зависимости: youtube-transcript-api, pytube, openai-whisper (установлены)
"""
import argparse
import html
import json
import os
import os
import re
import sys
import time
import urllib.parse
from datetime import datetime, timedelta
import requests
# Utils для LLM — ищем в scripts/ плагина или в ~/.hermes/scripts/
import importlib.util as _importlib_util
import sys as _sys
_utils_paths = [
os.path.join(os.path.dirname(os.path.abspath(__file__)), "research_llm_utils.py"),
"/home/neo/.hermes/scripts/research_llm_utils.py",
]
_utils_path = None
for _up in _utils_paths:
if os.path.exists(_up):
_utils_path = _up
break
if _utils_path is None:
raise ImportError("research_llm_utils.py не найден. Установите плагин research-agent полностью.")
_spec = _importlib_util.spec_from_file_location("research_llm_utils", _utils_path)
_utils_mod = _importlib_util.module_from_spec(_spec)
_spec.loader.exec_module(_utils_mod)
llm_call_safe = _utils_mod.llm_call_safe
validate_and_warn = _utils_mod.validate_and_warn
REASON_FIRST_PREFIX = _utils_mod.REASON_FIRST_PREFIX
log = _utils_mod.log
apply_cli_overrides = _utils_mod.apply_cli_overrides
SEARXNG_URL = os.environ.get("SEARXNG_URL", "http://localhost:8888")
LLM_URL = os.environ.get("LLM_URL", "http://192.168.1.152:80/v1/chat/completions")
LLM_MODEL = os.environ.get("LLM_MODEL", "Qwen3.6-35B-A3B-UD-IQ4_NL.gguf")
LLM_TEMPERATURE = float(os.environ.get("LLM_TEMPERATURE", "0.3"))
USER_AGENT = "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36"
def log(msg):
print(f"[youtube-analyzer] {msg}", file=sys.stderr)
def llm_call(system, user, max_tokens=2048, temperature=None):
if temperature is None:
temperature = LLM_TEMPERATURE
payload = {
"model": LLM_MODEL,
"messages": [
{"role": "system", "content": system},
{"role": "user", "content": user},
],
"max_tokens": max_tokens,
"temperature": temperature,
"stream": False,
}
try:
resp = requests.post(LLM_URL, json=payload, timeout=180)
resp.raise_for_status()
data = resp.json()
content = data["choices"][0]["message"]["content"].strip()
if not content:
log("LLM error: empty response")
return None
return content
except Exception as e:
log(f"LLM error: {e}")
return None
def extract_video_id(url):
"""Извлечь video ID из URL YouTube."""
patterns = [
r"(?:v=|/v/|youtu\.be/|/embed/|/shorts/)([a-zA-Z0-9_-]{11})",
r"youtube\.com/watch\?.*v=([a-zA-Z0-9_-]{11})",
]
for p in patterns:
m = re.search(p, url)
if m:
return m.group(1)
return None
def get_video_title(video_id):
"""Получить название видео через оython youtube-транскрипт-api или прямой запрос."""
# Пробуем через страницу
try:
resp = requests.get(
f"https://www.youtube.com/watch?v={video_id}",
headers={"User-Agent": USER_AGENT},
timeout=10,
)
m = re.search(r"<title>([^<]+)</title>", resp.text)
if m:
title = html.unescape(m.group(1))
title = re.sub(r"\s*-\s*YouTube\s*$", "", title)
return title.strip()
except Exception:
pass
return f"Video {video_id}"
def get_transcript(video_id, lang="ru"):
"""Получить транскрипт видео (API youtube-transcript-api v2.x)."""
try:
from youtube_transcript_api import YouTubeTranscriptApi
lang_list = [lang, "en", "a.en", "ru", "es", "de", "fr"]
api = YouTubeTranscriptApi()
transcript = api.fetch(video_id, languages=lang_list)
raw = transcript.to_raw_data()
segments = [
{
"text": seg.get("text", ""),
"start": seg.get("start", 0),
"duration": seg.get("duration", 0),
}
for seg in raw
]
return segments, None
except ImportError:
return None, "youtube-transcript-api не установлен"
except Exception as e:
return None, str(e)
def format_timestamp(seconds):
"""MM:SS или HH:MM:SS."""
s = int(seconds)
h, r = divmod(s, 3600)
m, s = divmod(r, 60)
if h:
return f"{h}:{m:02d}:{s:02d}"
return f"{m}:{s:02d}"
def transcript_to_text(segments, with_timestamps=True):
"""Транскрипт в читаемый текст с/без таймкодов."""
lines = []
for seg in segments:
text = seg["text"].strip()
if not text:
continue
if with_timestamps:
ts = format_timestamp(seg["start"])
lines.append(f"[{ts}] {text}")
else:
lines.append(text)
return "\n".join(lines)
def search_youtube_videos(query, lang="ru", limit=5):
"""Поиск видео через SearXNG."""
params = {
"q": f"{query} site:youtube.com",
"format": "json",
"categories": "general",
"language": lang,
"pageno": 1,
}
try:
resp = requests.get(
SEARXNG_URL, params=params, headers={"User-Agent": USER_AGENT}, timeout=20
)
resp.raise_for_status()
data = resp.json()
results = data.get("results", [])
# Фильтр: только YouTube ссылки
videos = []
for r in results:
url = r.get("url", "")
vid = extract_video_id(url)
if vid:
videos.append({
"title": r.get("title", "?"),
"url": url,
"video_id": vid,
"snippet": r.get("content", ""),
})
if len(videos) >= limit:
break
return videos
except Exception as e:
log(f"Search error: {e}")
return []
def transcribe_with_whisper(video_id):
"""Скачать аудио + whisper транскрибация (fallback)."""
try:
from pytube import YouTube
log("Скачивание аудио через pytube...")
yt = YouTube(f"https://www.youtube.com/watch?v={video_id}")
stream = yt.streams.filter(only_audio=True).first()
if not stream:
return None, "Нет аудиопотока"
tmp_path = f"/tmp/yta_{video_id}.mp4"
stream.download(output_path="/tmp", filename=f"yta_{video_id}.mp4")
if not os.path.exists(tmp_path):
return None, "Не удалось скачать аудио"
log("Транскрибация через whisper...")
import whisper
model = whisper.load_model("small") # small — баланс скорость/качество
result = model.transcribe(tmp_path, language="ru")
os.remove(tmp_path)
segments = []
for seg in result.get("segments", []):
segments.append({
"text": seg.get("text", "").strip(),
"start": seg.get("start", 0),
"duration": seg.get("end", 0) - seg.get("start", 0),
})
return segments, None
except ImportError as e:
return None, f"pytube/whisper не установлен: {e}"
except Exception as e:
return None, str(e)
def analyze_single_video(video_id, title, lang="ru", transcript_only=False):
"""Анализ одного видео — полный цикл."""
result = {
"video_id": video_id,
"title": title,
"url": f"https://youtube.com/watch?v={video_id}",
"segments": None,
"error": None,
"transcript_text": "",
"report": "",
}
# Фаза 1: получение транскрипта
log(f"Получение транскрипта ({lang})...")
segments, error = get_transcript(video_id, lang)
used_whisper = False
if segments is None:
log(f"Транскрипт не найден: {error}")
log("Fallback: whisper транскрибация...")
segments, error = transcribe_with_whisper(video_id)
used_whisper = True
if segments is None:
result["error"] = f"Не удалось получить транскрипт: {error}"
return result
result["segments"] = segments
result["transcript_text"] = transcript_to_text(segments, with_timestamps=True)
result["used_whisper"] = used_whisper
log(f"Транскрипт: {len(segments)} сегментов, "
f"{len(result['transcript_text'])} символов")
if transcript_only:
return result
# Фаза 2: LLM-синтез
log("Синтез отчёта...")
text_for_llm = transcript_to_text(segments, with_timestamps=True)
# Ограничение для LLM
if len(text_for_llm) > 12000:
text_for_llm = text_for_llm[:12000] + "\n[...truncated]"
lang_system = "ru" if lang == "ru" else "en"
system = (
"Ты — аналитик контента. Проанализируй транскрипт видео и составь "
"структурированный отчёт.\n\n"
"Формат отчёта СТРОГО:\n\n"
"## [Название видео]\n\n"
"**Ключевой вывод:** [1-2 предложения]\n\n"
"### Краткое содержание\n[2-3 абзаца с ключевыми идеями и таймкодами [MM:SS]]\n\n"
"### Ключевые тезисы\n- **Тезис 1** (таймкод [MM:SS]): [формулировка]\n"
"- **Тезис 2** (таймкод [MM:SS]): [формулировка]\n\n"
"### Яркие цитаты\n> \"[цитата]\" — [MM:SS]\n\n"
"### Оценка\n[Кому будет полезно, качество аргументации, факты/мнение]\n"
)
report = llm_call(system, text_for_llm, max_tokens=2048, temperature=0.2)
if report:
result["report"] = report
else:
result["report"] = "Ошибка синтеза отчёта."
return result
def synthesize_compare(video_analyses, topic):
"""Сравнительный анализ нескольких видео."""
ctx_parts = []
for va in video_analyses:
if va["report"]:
ctx_parts.append(
f"## {va['title']}\n{va['report'][:2000]}"
)
elif va["error"]:
ctx_parts.append(f"## {va['title']}\nОшибка: {va['error']}")
context = "\n\n---\n\n".join(ctx_parts)
system = (
"Ты — аналитик. Составь сравнительный обзор нескольких видео на одну тему.\n\n"
"Формат:\n\n"
"## Сравнительный анализ: [тема]\n\n"
"**Ключевой вывод:** [общий вывод по всем видео]\n\n"
"### Обзор видео\n"
"- [Видео 1]: кратко (1-2 предложения)\n"
"- [Видео 2]: кратко\n\n"
"### Сходства\n[что общего в подаче / фактах / аргументах]\n\n"
"### Различия\n[чем отличаются: угол подачи, глубина, качество]\n\n"
"### Рейтинг (от лучшего к худшему)\n"
"1. [Видео] — почему\n"
"2. [Видео] — почему\n\n"
"### Итог\n[какое видео лучше всего раскрывает тему]\n"
)
prompt = (
f"Тема: {topic}\n\n"
f"Анализы видео:\n\n{context}\n\n"
"Составь сравнительный отчёт."
)
report = llm_call(system, prompt, max_tokens=2048, temperature=0.2)
return report or "Ошибка синтеза."
# ─── Main ──────────────────────────────────────────────────────────────────────
def main():
parser = argparse.ArgumentParser(
description="YouTube Analyzer — анализ видео как Perplexity"
)
parser.add_argument("--url", help="URL видео YouTube")
parser.add_argument("--search", help="Поисковый запрос для поиска видео")
parser.add_argument("--limit", type=int, default=1, choices=range(1, 6),
help="Количество видео для поиска/сравнения")
parser.add_argument("--compare", action="store_true",
help="Сравнить несколько видео")
parser.add_argument("--lang", default="ru", choices=["ru", "en"])
parser.add_argument("--model", help="Имя LLM модели (env: LLM_MODEL)")
parser.add_argument("--temperature", type=float, help="Temperature (env: LLM_TEMPERATURE)")
parser.add_argument("--transcript-only", action="store_true",
help="Только текст транскрипта")
parser.add_argument("--output", "-o", help="Сохранить в файл")
args = parser.parse_args()
if args.model:
LLM_MODEL = args.model
if args.temperature is not None:
LLM_TEMPERATURE = args.temperature
if not args.url and not args.search:
parser.print_help()
sys.exit(1)
start_time = time.time()
# ─── Режим: поиск видео ─────────────────────────────────────────
videos = []
if args.search:
log(f"Поиск видео: {args.search}")
videos = search_youtube_videos(args.search, args.lang, args.limit)
if not videos:
print("Видео не найдены.")
sys.exit(1)
log(f"Найдено видео: {len(videos)}")
for v in videos:
log(f" {v['title'][:80]}")
if not v.get("title"):
v["title"] = get_video_title(v["video_id"])
elif args.url:
vid = extract_video_id(args.url)
if not vid:
print("Не удалось извлечь ID видео из URL")
sys.exit(1)
title = get_video_title(vid)
videos = [{"video_id": vid, "title": title, "url": args.url}]
log(f"Видео: {title}")
# ─── Анализ видео ───────────────────────────────────────────────
analyses = []
for i, v in enumerate(videos):
log(f"\nАнализ [{i+1}/{len(videos)}]: {v['title'][:60]}")
result = analyze_single_video(
v["video_id"],
v.get("title", v.get("title", "?")),
args.lang,
args.transcript_only,
)
analyses.append(result)
# ─── Вывод результатов ──────────────────────────────────────────
output_parts = []
elapsed = time.time() - start_time
if args.compare and len(analyses) > 1:
# Сравнительный отчёт
log("Синтез сравнительного отчёта...")
topic = args.search or "сравнение видео"
compare_report = synthesize_compare(analyses, topic)
output_parts.append(f"# Сравнительный анализ видео\n\n{compare_report}")
output_parts.append(f"\n\n---\n*Проанализировано {len(analyses)} видео за {elapsed:.0f} сек*")
else:
for a in analyses:
title_display = a["title"]
if a.get("error"):
output_parts.append(f"## {title_display}\n\n**Ошибка:** {a['error']}\n\n---")
continue
if args.transcript_only:
# Только текст транскрипта
header = f"## {title_display}\nURL: {a['url']}\n\n"
output_parts.append(header + a["transcript_text"])
else:
output_parts.append(a["report"])
if a.get("used_whisper"):
output_parts.append("\n\n*Транскрипция через whisper (субтитры отсутствовали)*")
if not args.transcript_only:
output_parts.append(f"\n\n---\n*Проанализировано {len(analyses)} видео за {elapsed:.0f} сек*")
output = "\n\n".join(output_parts)
if args.output:
with open(args.output, "w", encoding="utf-8") as f:
f.write(output)
log(f"Отчёт сохранён: {args.output}")
else:
print(output)
if __name__ == "__main__":
main()
Reference in new issue View Git Blame Copy Permalink