add fetch_transcript + download_video update

Author: mithmith

app/integrations/ytdlp.py

@@ -1,7 +1,11 @@
 import asyncio
+import glob
 import json
 import locale
+import os
+import re
 import subprocess
+import tempfile
 from pathlib import Path
 from typing import Optional
 
@@ -86,25 +90,32 @@ def _extract_video_list(self) -> tuple[list[VideoSchema], str]:
         return video_list, channel_id
 
     @staticmethod
-    async def download_video(video_info: VideoDownloadSchema, format: str = "best[ext=mp4]") -> None:
-        if Path(video_info.video_file_download_path).exists():
-            logger.info(f"Видео уже скачано: {video_info.video_file_download_path}")
+    async def download_video(
+        video_info: VideoDownloadSchema,
+        format: str = "bestvideo[ext=mp4]+bestaudio[ext=m4a]/best[ext=mp4]/best",
+        ensure_mp4: bool = True,
+    ) -> None:
+        out_path = Path(video_info.video_file_download_path)
+        if out_path.exists():
+            logger.info(f"Видео уже скачано: {out_path}")
             return
 
-        command = f'yt-dlp -f "{format}" -o "{video_info.video_file_download_path}" {video_info.video_url}'
-        logger.debug("Downloading video: {}".format(video_info.video_url))
-        logger.debug("Executing command: {}".format(command))
-        process = await asyncio.create_subprocess_shell(
+        postproc_flag = "--recode-video mp4" if ensure_mp4 else "--merge-output-format mp4"
+        command = f'yt-dlp -f "{format}" {postproc_flag} -o "{out_path}" {video_info.video_url}'
+
+        logger.debug(f"Downloading video: {video_info.video_url}")
+        logger.debug(f"Command: {command}")
+
+        proc = await asyncio.create_subprocess_shell(
             command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
         )
+        _, stderr = await proc.communicate()
 
-        _, stderr = await process.communicate()
-
-        if process.returncode == 0:
-            logger.info(f"Видео скачано: {video_info.video_file_download_path}")
+        if proc.returncode == 0:
+            logger.info(f"Видео скачано: {out_path}")
         else:
-            stderr_text = stderr.decode(locale.getpreferredencoding(False), errors="replace").strip()
-            logger.error(f"Ошибка скачивания видео: {stderr_text}")
+            err = stderr.decode(locale.getpreferredencoding(False), "replace").strip()
+            logger.error(f"Ошибка скачивания видео: {err}")
 
     def download_thumbnail(self, video_id: str) -> None:
         video: Video = self._repository.get_video(video_id)
@@ -127,7 +138,8 @@ def update_video_formats(self) -> None:
                 self._repository.add_video_format(format_data, v_id)
             logger.debug(f"[{i+1}/{len(video_ids)}] Added video formats for v_id: {v_id}")
 
-    def get_video_formats(self, video_id: str) -> list[YTFormatSchema]:
+    @staticmethod
+    def get_video_formats(video_id: str) -> list[YTFormatSchema]:
         result = subprocess.run(
             [
                 "yt-dlp",
@@ -159,6 +171,120 @@ def get_video_formats(self, video_id: str) -> list[YTFormatSchema]:
             logger.error(f"Не удалось декодировать JSON: {e}")
             return []
 
+    @staticmethod
+    def fetch_transcript(
+        video_id: str,
+        preferred_langs: tuple[str, ...] = ("ru",),
+    ) -> Optional[str]:
+        """
+        Возвращает расшифровку речи YouTube-видео (чистый текст без тайм-кодов).
+
+        Порядок поиска языка:
+        1. Перебор preferred_langs в automatic_captions
+        2. Первый язык из automatic_captions
+        3. Перебор preferred_langs в subtitles
+        4. Первый язык из subtitles
+        5. None, если субтитров нет
+
+        Parameters
+        ----------
+        video_id : str
+            Идентификатор ролика (например, '7NSBuTHngK0').
+        preferred_langs : tuple[str, ...], optional
+            Коды языков, которые пробуем в первую очередь.
+
+        Returns
+        -------
+        str | None
+            Сплошной текст субтитров или None, если ничего не найдено.
+        """
+        video_url = f"https://www.youtube.com/watch?v={video_id}"
+
+        # 1. Получаем метаданные ролика
+        try:
+            proc = subprocess.run(
+                ["yt-dlp", "-j", video_url],
+                check=True,
+                capture_output=True,
+                text=True,
+            )
+            info = json.loads(proc.stdout)
+        except (subprocess.CalledProcessError, json.JSONDecodeError) as e:
+            logger.error(f"Не удалось получить JSON-метаданные: {e}")
+            return None
+
+        auto_caps: dict[str, list] = info.get("automatic_captions", {}) or {}
+        man_caps: dict[str, list] = info.get("subtitles", {}) or {}
+
+        # 2. Выбираем язык и тип субтитров
+        lang, use_auto = None, None
+
+        for pl in preferred_langs:
+            if pl in auto_caps:
+                lang, use_auto = pl, True
+                break
+        if lang is None and auto_caps:
+            lang, use_auto = next(iter(auto_caps)), True
+
+        if lang is None:
+            for pl in preferred_langs:
+                if pl in man_caps:
+                    lang, use_auto = pl, False
+                    break
+        if lang is None and man_caps:
+            lang, use_auto = next(iter(man_caps)), False
+
+        if lang is None:
+            logger.warning(f"У ролика {video_id} нет субтитров")
+            return None
+
+        # 3. Скачиваем json3 и читаем
+        with tempfile.TemporaryDirectory() as tmpdir:
+            outtmpl = os.path.join(tmpdir, "subs")  # yt-dlp → subs.<lang>.json3
+            cmd = [
+                "yt-dlp",
+                "--skip-download",
+                "--sub-lang",
+                lang,
+                "--sub-format",
+                "json3",
+                "-o",
+                outtmpl,
+                video_url,
+            ]
+            cmd.insert(1, "--write-auto-subs" if use_auto else "--write-subs")
+
+            try:
+                subprocess.run(cmd, check=True, capture_output=True, text=True)
+            except subprocess.CalledProcessError as e:
+                logger.error(f"Ошибка скачивания субтитров: {e.stderr.strip()}")
+                return None
+
+            pattern = os.path.join(tmpdir, f"subs.{lang}*.json3")
+            files = glob.glob(pattern)
+            if not files:
+                logger.error(f"Файл субтитров не найден ({pattern})")
+                return None
+
+            try:
+                with open(files[0], encoding="utf-8") as fp:
+                    data = json.load(fp)
+            except (OSError, json.JSONDecodeError) as e:
+                logger.error(f"Не удалось прочитать JSON3 файл: {e}")
+                return None
+
+        # 4. Превращаем JSON3 в сплошной текст
+        if "events" not in data:
+            logger.error("JSON3 не содержит ключ 'events'")
+            return None
+
+        text_chunks = [seg["utf8"] for ev in data["events"] if "segs" in ev for seg in ev["segs"] if "utf8" in seg]
+        # Убираем лишние пробелы/переводы строк подряд
+        transcript = re.sub(r"\s+", " ", " ".join(text_chunks)).strip()
+
+        logger.info(f"Расшифровка {video_id} ({lang}) успешно получена ({len(transcript)} символов)")
+        return transcript
+
     def channel_exist(self, channel_id: str) -> bool:
         return bool(self._repository.get_channel_by_id(channel_id))
 

app/service/yt_monitor.py

@@ -334,8 +334,8 @@ def _process_old_videos(self, old_videos: list[VideoSchema]) -> None:
 
     def _generate_shorts_download_path(self, channel_name: str, video_id: str, format: str = "mp4") -> Path:
         video_file_name = f"{channel_name}_{video_id}.{format}"
-        return (self._short_download_path / video_file_name)
+        return self._short_download_path / video_file_name
 
     def _generate_videos_download_path(self, channel_name: str, video_id: str, format: str = "mp4") -> Path:
         video_file_name = f"{channel_name}_{video_id}.{format}"
-        return (self._video_download_path / video_file_name)
+        return self._video_download_path / video_file_name