Ретраи и дедупликация в задачах
- элемент плейлиста повторяется до 3 раз с паузой 30 с (разовые 403/429 от YouTube) - yt-dlp: retries/fragment_retries/extractor_retries - уже обработанные URL пропускаются — повторная постановка плейлиста доделывает только упавшее Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -40,6 +40,12 @@ async def get_video(video_uuid: str) -> Video | None:
|
|||||||
return res.scalars().first()
|
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(
|
async def list_videos(
|
||||||
*,
|
*,
|
||||||
q: str | None = None,
|
q: str | None = None,
|
||||||
|
|||||||
@@ -10,10 +10,15 @@ import asyncio
|
|||||||
import logging
|
import logging
|
||||||
|
|
||||||
from api.db.job import get_job, update_job
|
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
|
from api.lib.youtube import expand_url
|
||||||
|
|
||||||
|
|
||||||
|
_ITEM_ATTEMPTS = 3 # попытки на один элемент (YouTube любит разовые 403/429)
|
||||||
|
_RETRY_DELAY = 30.0
|
||||||
|
|
||||||
|
|
||||||
logger = logging.getLogger("jobs")
|
logger = logging.getLogger("jobs")
|
||||||
|
|
||||||
_queue: asyncio.Queue[str] = asyncio.Queue()
|
_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:
|
def _notify(msg: str, _label: str = label) -> None:
|
||||||
_live[job_id] = f"{_label} — {msg}"
|
_live[job_id] = f"{_label} — {msg}"
|
||||||
|
|
||||||
try:
|
# дедупликация: повторная постановка плейлиста не переделывает готовое
|
||||||
await process_single_video(
|
if await get_video_by_url(entry["url"]) is not None:
|
||||||
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
|
done += 1
|
||||||
except Exception as e:
|
_live[job_id] = f"{label} — уже в библиотеке, пропуск"
|
||||||
failed += 1
|
logger.info("job %s item %d/%d already processed, skip", job_id, i, n)
|
||||||
errors.append(f"{entry['title']}: {e}")
|
await update_job(job_id, done_items=done)
|
||||||
logger.warning("job %s item %d/%d failed: %s", job_id, i, n, e)
|
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(
|
await update_job(
|
||||||
job_id, done_items=done, failed_items=failed, error="\n".join(errors)[:4000]
|
job_id, done_items=done, failed_items=failed, error="\n".join(errors)[:4000]
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -96,6 +96,10 @@ def download_video_with_subs(
|
|||||||
"quiet": True,
|
"quiet": True,
|
||||||
"no_warnings": True,
|
"no_warnings": True,
|
||||||
"noplaylist": True, # ссылка с list= в задаче на одно видео не тянет весь плейлист
|
"noplaylist": True, # ссылка с list= в задаче на одно видео не тянет весь плейлист
|
||||||
|
# YouTube периодически отвечает 403/сбросом соединения — повторяем сами
|
||||||
|
"retries": 10,
|
||||||
|
"fragment_retries": 10,
|
||||||
|
"extractor_retries": 3,
|
||||||
}
|
}
|
||||||
if audio_only:
|
if audio_only:
|
||||||
base_opts["format"] = "ba/b"
|
base_opts["format"] = "ba/b"
|
||||||
|
|||||||
Reference in New Issue
Block a user