From c88b0e4f6d4a7f755376e35a8cbbaf2fc7e5a676 Mon Sep 17 00:00:00 2001
From: jze9 <jze9@moon-1.home>
Date: Tue, 14 Jul 2026 15:33:01 +0500
Subject: [PATCH] =?UTF-8?q?=D0=A0=D0=B5=D1=82=D1=80=D0=B0=D0=B8=20=D0=B8?=
 =?UTF-8?q?=20=D0=B4=D0=B5=D0=B4=D1=83=D0=BF=D0=BB=D0=B8=D0=BA=D0=B0=D1=86?=
 =?UTF-8?q?=D0=B8=D1=8F=20=D0=B2=20=D0=B7=D0=B0=D0=B4=D0=B0=D1=87=D0=B0?=
 =?UTF-8?q?=D1=85?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- элемент плейлиста повторяется до 3 раз с паузой 30 с (разовые 403/429 от YouTube)
- yt-dlp: retries/fragment_retries/extractor_retries
- уже обработанные URL пропускаются — повторная постановка плейлиста
  доделывает только упавшее

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 api/db/video.py     |  6 +++++
 api/lib/jobqueue.py | 57 +++++++++++++++++++++++++++++++++------------
 api/lib/youtube.py  |  4 ++++
 3 files changed, 52 insertions(+), 15 deletions(-)

diff --git a/api/db/video.py b/api/db/video.py
index 34bc1a0..f78bf2d 100644
--- a/api/db/video.py
+++ b/api/db/video.py
@@ -40,6 +40,12 @@ async def get_video(video_uuid: str) -> Video | None:
         return res.scalars().first()
 
 
+async def get_video_by_url(url: str) -> Video | None:
+    async with SessionLocal() as session:
+        res = await session.execute(select(Video).where(Video.source_url == url))
+        return res.scalars().first()
+
+
 async def list_videos(
     *,
     q: str | None = None,
diff --git a/api/lib/jobqueue.py b/api/lib/jobqueue.py
index 4c75919..9271bad 100644
--- a/api/lib/jobqueue.py
+++ b/api/lib/jobqueue.py
@@ -10,10 +10,15 @@ import asyncio
 import logging
 
 from api.db.job import get_job, update_job
-from api.lib.processing import process_single_video
+from api.db.video import get_video_by_url
+from api.lib.processing import PipelineError, process_single_video
 from api.lib.youtube import expand_url
 
 
+_ITEM_ATTEMPTS = 3  # попытки на один элемент (YouTube любит разовые 403/429)
+_RETRY_DELAY = 30.0
+
+
 logger = logging.getLogger("jobs")
 
 _queue: asyncio.Queue[str] = asyncio.Queue()
@@ -95,21 +100,43 @@ async def _run_job(job_id: str) -> None:
         def _notify(msg: str, _label: str = label) -> None:
             _live[job_id] = f"{_label} — {msg}"
 
-        try:
-            await process_single_video(
-                entry["url"],
-                section_id=params.get("section_id"),
-                model_path=params.get("model_path"),
-                summary_max_sentences=params.get("summary_max_sentences"),
-                keep_video=params.get("keep_video"),
-                job_id=job_id,
-                progress=_notify,
-            )
+        # дедупликация: повторная постановка плейлиста не переделывает готовое
+        if await get_video_by_url(entry["url"]) is not None:
             done += 1
-        except Exception as e:
-            failed += 1
-            errors.append(f"{entry['title']}: {e}")
-            logger.warning("job %s item %d/%d failed: %s", job_id, i, n, e)
+            _live[job_id] = f"{label} — уже в библиотеке, пропуск"
+            logger.info("job %s item %d/%d already processed, skip", job_id, i, n)
+            await update_job(job_id, done_items=done)
+            continue
+
+        for attempt in range(1, _ITEM_ATTEMPTS + 1):
+            try:
+                await process_single_video(
+                    entry["url"],
+                    section_id=params.get("section_id"),
+                    model_path=params.get("model_path"),
+                    summary_max_sentences=params.get("summary_max_sentences"),
+                    keep_video=params.get("keep_video"),
+                    job_id=job_id,
+                    progress=_notify,
+                )
+                done += 1
+                break
+            except Exception as e:
+                retryable = not (isinstance(e, PipelineError) and e.status == 400)
+                if retryable and attempt < _ITEM_ATTEMPTS and not is_cancelled(job_id):
+                    logger.warning(
+                        "job %s item %d/%d attempt %d failed, retrying: %s",
+                        job_id, i, n, attempt, e,
+                    )
+                    _live[job_id] = (
+                        f"{label} — ошибка, повтор {attempt}/{_ITEM_ATTEMPTS - 1} через {int(_RETRY_DELAY)} с"
+                    )
+                    await asyncio.sleep(_RETRY_DELAY)
+                    continue
+                failed += 1
+                errors.append(f"{entry['title']}: {e}")
+                logger.warning("job %s item %d/%d failed: %s", job_id, i, n, e)
+                break
         await update_job(
             job_id, done_items=done, failed_items=failed, error="\n".join(errors)[:4000]
         )
diff --git a/api/lib/youtube.py b/api/lib/youtube.py
index 07786a9..693b3ea 100644
--- a/api/lib/youtube.py
+++ b/api/lib/youtube.py
@@ -96,6 +96,10 @@ def download_video_with_subs(
         "quiet": True,
         "no_warnings": True,
         "noplaylist": True,  # ссылка с list= в задаче на одно видео не тянет весь плейлист
+        # YouTube периодически отвечает 403/сбросом соединения — повторяем сами
+        "retries": 10,
+        "fragment_retries": 10,
+        "extractor_retries": 3,
     }
     if audio_only:
         base_opts["format"] = "ba/b"