Добавлено описание проекта PVideoDl, включая функциональность, стек технологий, архитектуру, инструкции по запуску и API. Обновлён README.md для лучшего понимания проекта.

app/__init__.py

@@ -0,0 +1 @@
+"""PVideoDl — локальная скачивалка файлов и видео с веб-интерфейсом."""

app/api/deps.py

@@ -0,0 +1,25 @@
+"""Доступ к разделяемым компонентам приложения через app.state.
+
+Компоненты (Storage, DownloadQueue, EventBus, WorkerPool) создаются в lifespan
+в main.py и кладутся в app.state. Роуты достают их отсюда через Depends.
+"""
+
+from __future__ import annotations
+
+from fastapi import Request
+
+from app.core.events import EventBus
+from app.core.queue import DownloadQueue
+from app.core.storage import Storage
+
+
+def get_storage(request: Request) -> Storage:
+    return request.app.state.storage
+
+
+def get_queue(request: Request) -> DownloadQueue:
+    return request.app.state.queue
+
+
+def get_events(request: Request) -> EventBus:
+    return request.app.state.events

app/api/downloads.py

@@ -0,0 +1,53 @@
+"""Роуты управления загрузками: создание, список, удаление."""
+
+from __future__ import annotations
+
+from fastapi import APIRouter, Depends, HTTPException, status
+
+from app.api.deps import get_events, get_queue, get_storage
+from app.core.events import EventBus
+from app.core.queue import DownloadQueue
+from app.core.storage import Storage
+from app.models import CreateDownloads, Download, DownloadEvent
+
+router = APIRouter(prefix="/api/downloads", tags=["downloads"])
+
+
+@router.post("", status_code=status.HTTP_201_CREATED, response_model=list[Download])
+async def create_downloads(
+    payload: CreateDownloads,
+    storage: Storage = Depends(get_storage),
+    queue: DownloadQueue = Depends(get_queue),
+    events: EventBus = Depends(get_events),
+) -> list[Download]:
+    """Принять ссылки (по одной на строку), создать задачи и поставить в очередь."""
+    urls = [u.strip() for u in payload.urls if u.strip()]
+    if not urls:
+        raise HTTPException(status.HTTP_400_BAD_REQUEST, "Список ссылок пуст")
+
+    created: list[Download] = []
+    for url in urls:
+        download = Download(url=url)
+        await storage.create(download)
+        await queue.add(download.id)
+        await events.publish(DownloadEvent(type="created", download=download))
+        created.append(download)
+    return created
+
+
+@router.get("", response_model=list[Download])
+async def list_downloads(storage: Storage = Depends(get_storage)) -> list[Download]:
+    """Список всех загрузок — для гидрации UI при загрузке страницы."""
+    return await storage.list()
+
+
+@router.delete("/{download_id}", status_code=status.HTTP_204_NO_CONTENT)
+async def delete_download(
+    download_id: str,
+    storage: Storage = Depends(get_storage),
+    events: EventBus = Depends(get_events),
+) -> None:
+    deleted = await storage.delete(download_id)
+    if not deleted:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, "Загрузка не найдена")
+    await events.publish(DownloadEvent(type="deleted", id=download_id))

app/api/events.py

@@ -0,0 +1,37 @@
+"""SSE-поток обновлений прогресса: GET /api/events."""
+
+from __future__ import annotations
+
+import asyncio
+
+from fastapi import APIRouter, Depends, Request
+from sse_starlette.sse import EventSourceResponse
+
+from app.api.deps import get_events
+from app.core.events import EventBus
+
+router = APIRouter(tags=["events"])
+
+# Если за это время не было ни одного события — шлём комментарий-пинг,
+# чтобы прокси/браузер не закрыли «висящее» соединение.
+_KEEPALIVE_SECONDS = 15.0
+
+
+@router.get("/api/events")
+async def events_stream(
+    request: Request,
+    events: EventBus = Depends(get_events),
+) -> EventSourceResponse:
+    async def generator():
+        async with events.subscribe() as queue:
+            while True:
+                if await request.is_disconnected():
+                    break
+                try:
+                    event = await asyncio.wait_for(queue.get(), timeout=_KEEPALIVE_SECONDS)
+                except asyncio.TimeoutError:
+                    yield {"event": "ping", "data": "{}"}
+                    continue
+                yield {"event": event.type, "data": event.model_dump_json()}
+
+    return EventSourceResponse(generator())

app/config.py

@@ -0,0 +1,43 @@
+"""Настройки приложения. Значения берутся из переменных окружения с префиксом PVDL_."""
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from pathlib import Path
+
+BASE_DIR = Path(__file__).resolve().parent.parent
+
+
+def _env_path(name: str, default: Path) -> Path:
+    value = os.environ.get(name)
+    return Path(value).expanduser().resolve() if value else default
+
+
+def _env_int(name: str, default: int) -> int:
+    try:
+        return int(os.environ.get(name, default))
+    except (TypeError, ValueError):
+        return default
+
+
+@dataclass(frozen=True)
+class Settings:
+    # Куда складывать скачанные файлы.
+    download_dir: Path = _env_path("PVDL_DOWNLOAD_DIR", BASE_DIR / "downloads")
+    # Файл SQLite.
+    db_path: Path = _env_path("PVDL_DB_PATH", BASE_DIR / "app.db")
+    # Каталог собранной статики фронтенда (SvelteKit build).
+    static_dir: Path = _env_path("PVDL_STATIC_DIR", BASE_DIR / "frontend" / "build")
+    # Сколько воркеров разбирают очередь параллельно.
+    workers: int = _env_int("PVDL_WORKERS", 3)
+    # Размер chunk'а при скачивании через httpx (байт).
+    chunk_size: int = _env_int("PVDL_CHUNK_SIZE", 64 * 1024)
+    # Как часто (в секундах) пушить событие прогресса, чтобы не заспамить шину.
+    progress_interval: float = 0.3
+    # Хост/порт веб-сервера.
+    host: str = os.environ.get("PVDL_HOST", "127.0.0.1")
+    port: int = _env_int("PVDL_PORT", 8000)
+
+
+settings = Settings()

app/core/events.py

@@ -0,0 +1,42 @@
+"""Шина событий для трансляции прогресса в SSE.
+
+Простая in-memory реализация на основе asyncio-очередей подписчиков (fan-out).
+Каждый SSE-клиент подписывается, получает свою очередь и читает из неё события.
+Завтра при необходимости тут окажется Redis pub/sub — интерфейс не изменится.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+from collections.abc import AsyncIterator
+
+from app.models import DownloadEvent
+
+
+class EventBus:
+    def __init__(self, max_queue: int = 1000) -> None:
+        self._subscribers: set[asyncio.Queue[DownloadEvent]] = set()
+        self._max_queue = max_queue
+
+    async def publish(self, event: DownloadEvent) -> None:
+        # Рассылаем всем подписчикам. Если чья-то очередь переполнена
+        # (медленный клиент) — дропаем самое старое событие, не блокируясь.
+        for queue in list(self._subscribers):
+            if queue.full():
+                with contextlib.suppress(asyncio.QueueEmpty):
+                    queue.get_nowait()
+            queue.put_nowait(event)
+
+    @contextlib.asynccontextmanager
+    async def subscribe(self) -> AsyncIterator[asyncio.Queue[DownloadEvent]]:
+        queue: asyncio.Queue[DownloadEvent] = asyncio.Queue(maxsize=self._max_queue)
+        self._subscribers.add(queue)
+        try:
+            yield queue
+        finally:
+            self._subscribers.discard(queue)
+
+    @property
+    def subscriber_count(self) -> int:
+        return len(self._subscribers)

app/core/queue.py

@@ -0,0 +1,31 @@
+"""Абстракция очереди задач.
+
+Сегодня — обёртка над asyncio.Queue. Завтра можно подменить на Redis/RabbitMQ,
+не трогая вызовы в сервисах: интерфейс DownloadQueue остаётся прежним.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+
+class DownloadQueue:
+    """Очередь идентификаторов загрузок, ожидающих обработки воркерами."""
+
+    def __init__(self) -> None:
+        self._queue: asyncio.Queue[str] = asyncio.Queue()
+
+    async def add(self, download_id: str) -> None:
+        await self._queue.put(download_id)
+
+    async def get(self) -> str:
+        return await self._queue.get()
+
+    def task_done(self) -> None:
+        self._queue.task_done()
+
+    async def join(self) -> None:
+        await self._queue.join()
+
+    def qsize(self) -> int:
+        return self._queue.qsize()

app/core/storage.py

@@ -0,0 +1,164 @@
+"""Абстракция хранилища загрузок.
+
+Интерфейс Storage + реализация на aiosqlite. Сегодня SQLite, завтра Postgres —
+сервисы и роуты зовут только методы Storage и не знают о бэкенде.
+"""
+
+from __future__ import annotations
+
+import json
+from abc import ABC, abstractmethod
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+import aiosqlite
+
+from app.models import Download, DownloadStatus
+
+
+class Storage(ABC):
+    @abstractmethod
+    async def init(self) -> None: ...
+
+    @abstractmethod
+    async def close(self) -> None: ...
+
+    @abstractmethod
+    async def create(self, download: Download) -> None: ...
+
+    @abstractmethod
+    async def update(self, download_id: str, **fields: Any) -> Download | None: ...
+
+    @abstractmethod
+    async def get(self, download_id: str) -> Download | None: ...
+
+    @abstractmethod
+    async def list(self) -> list[Download]: ...
+
+    @abstractmethod
+    async def delete(self, download_id: str) -> bool: ...
+
+
+_COLUMNS = (
+    "id",
+    "url",
+    "filename",
+    "status",
+    "progress",
+    "size_bytes",
+    "downloaded_bytes",
+    "speed",
+    "eta",
+    "error",
+    "created_at",
+)
+
+
+def _row_to_download(row: aiosqlite.Row) -> Download:
+    data = dict(row)
+    data["status"] = DownloadStatus(data["status"])
+    data["created_at"] = datetime.fromisoformat(data["created_at"])
+    return Download.model_validate(data)
+
+
+def _to_db_value(key: str, value: Any) -> Any:
+    if isinstance(value, DownloadStatus):
+        return value.value
+    if isinstance(value, datetime):
+        return value.isoformat()
+    return value
+
+
+class SqliteStorage(Storage):
+    def __init__(self, db_path: Path) -> None:
+        self._db_path = db_path
+        self._db: aiosqlite.Connection | None = None
+
+    @property
+    def _conn(self) -> aiosqlite.Connection:
+        if self._db is None:
+            raise RuntimeError("Storage не инициализирован — вызовите init()")
+        return self._db
+
+    async def init(self) -> None:
+        self._db_path.parent.mkdir(parents=True, exist_ok=True)
+        self._db = await aiosqlite.connect(self._db_path)
+        self._db.row_factory = aiosqlite.Row
+        await self._db.execute(
+            """
+            CREATE TABLE IF NOT EXISTS downloads (
+                id               TEXT PRIMARY KEY,
+                url              TEXT NOT NULL,
+                filename         TEXT,
+                status           TEXT NOT NULL,
+                progress         REAL NOT NULL DEFAULT 0,
+                size_bytes       INTEGER,
+                downloaded_bytes INTEGER NOT NULL DEFAULT 0,
+                speed            REAL,
+                eta              REAL,
+                error            TEXT,
+                created_at       TEXT NOT NULL
+            )
+            """
+        )
+        # На старте всё, что осталось "качающимся" после прошлого запуска,
+        # помечаем упавшим — воркеры этого процесса о них не знают.
+        await self._db.execute(
+            "UPDATE downloads SET status = ?, error = ? WHERE status = ?",
+            (
+                DownloadStatus.FAILED.value,
+                "Прервано при перезапуске приложения",
+                DownloadStatus.DOWNLOADING.value,
+            ),
+        )
+        await self._db.commit()
+
+    async def close(self) -> None:
+        if self._db is not None:
+            await self._db.close()
+            self._db = None
+
+    async def create(self, download: Download) -> None:
+        await self._conn.execute(
+            f"INSERT INTO downloads ({', '.join(_COLUMNS)}) "
+            f"VALUES ({', '.join('?' for _ in _COLUMNS)})",
+            tuple(_to_db_value(c, getattr(download, c)) for c in _COLUMNS),
+        )
+        await self._conn.commit()
+
+    async def update(self, download_id: str, **fields: Any) -> Download | None:
+        if not fields:
+            return await self.get(download_id)
+        allowed = {k: v for k, v in fields.items() if k in _COLUMNS and k != "id"}
+        if not allowed:
+            return await self.get(download_id)
+        assignments = ", ".join(f"{k} = ?" for k in allowed)
+        params = [_to_db_value(k, v) for k, v in allowed.items()]
+        params.append(download_id)
+        await self._conn.execute(
+            f"UPDATE downloads SET {assignments} WHERE id = ?", params
+        )
+        await self._conn.commit()
+        return await self.get(download_id)
+
+    async def get(self, download_id: str) -> Download | None:
+        async with self._conn.execute(
+            "SELECT * FROM downloads WHERE id = ?", (download_id,)
+        ) as cursor:
+            row = await cursor.fetchone()
+        return _row_to_download(row) if row else None
+
+    async def list(self) -> list[Download]:
+        async with self._conn.execute(
+            "SELECT * FROM downloads ORDER BY created_at DESC"
+        ) as cursor:
+            rows = await cursor.fetchall()
+        return [_row_to_download(r) for r in rows]
+
+    async def delete(self, download_id: str) -> bool:
+        cursor = await self._conn.execute(
+            "DELETE FROM downloads WHERE id = ?", (download_id,)
+        )
+        await self._conn.commit()
+        return cursor.rowcount > 0

app/main.py

@@ -0,0 +1,119 @@
+"""Точка входа: сборка FastAPI-приложения, lifespan, роутеры, отдача статики."""
+
+from __future__ import annotations
+
+import logging
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import FileResponse
+from fastapi.staticfiles import StaticFiles
+from starlette.requests import Request
+from starlette.responses import JSONResponse, Response
+
+from app.api import downloads, events
+from app.config import settings
+from app.core.events import EventBus
+from app.core.queue import DownloadQueue
+from app.core.storage import SqliteStorage
+from app.models import DownloadStatus
+from app.services.worker import WorkerPool
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
+logger = logging.getLogger("pvideodl")
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    storage = SqliteStorage(settings.db_path)
+    await storage.init()
+
+    queue = DownloadQueue()
+    event_bus = EventBus()
+    pool = WorkerPool(queue, storage, event_bus, settings.workers)
+
+    app.state.storage = storage
+    app.state.queue = queue
+    app.state.events = event_bus
+    app.state.pool = pool
+
+    # Перезаливаем в очередь всё, что осталось в pending с прошлого запуска.
+    for download in await storage.list():
+        if download.status == DownloadStatus.PENDING:
+            await queue.add(download.id)
+
+    await pool.start()
+    settings.download_dir.mkdir(parents=True, exist_ok=True)
+    logger.info("PVideoDl запущен. Файлы → %s", settings.download_dir)
+    try:
+        yield
+    finally:
+        await pool.stop()
+        await storage.close()
+
+
+def create_app() -> FastAPI:
+    app = FastAPI(title="PVideoDl", version="0.1.0", lifespan=lifespan)
+
+    # В деве фронт крутится на Vite (5173) и ходит на /api — пускаем его.
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=["http://localhost:5173", "http://127.0.0.1:5173"],
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )
+
+    app.include_router(downloads.router)
+    app.include_router(events.router)
+
+    @app.get("/api/health")
+    async def health() -> dict:
+        return {"status": "ok"}
+
+    _mount_frontend(app)
+    return app
+
+
+def _mount_frontend(app: FastAPI) -> None:
+    """Отдаём собранную статику SvelteKit, если она есть. SPA-fallback на index.html."""
+    static_dir = settings.static_dir
+    index = static_dir / "index.html"
+    if not index.exists():
+        @app.get("/")
+        async def no_frontend() -> JSONResponse:
+            return JSONResponse(
+                {
+                    "message": "Фронтенд не собран. Соберите его (npm run build в frontend/) "
+                    "или откройте API напрямую.",
+                    "api": "/api/downloads",
+                }
+            )
+        return
+
+    app.mount("/_app", StaticFiles(directory=static_dir / "_app"), name="_app")
+
+    @app.get("/{full_path:path}")
+    async def spa(full_path: str, request: Request) -> Response:
+        # Несуществующий /api путь — это 404, а не SPA-страница.
+        if full_path.startswith("api/"):
+            return JSONResponse({"detail": "Not Found"}, status_code=404)
+        # Реальный файл из статики (favicon, изображения и т.п.) — отдаём как есть.
+        candidate = (static_dir / full_path).resolve()
+        if candidate.is_file() and static_dir.resolve() in candidate.parents:
+            return FileResponse(candidate)
+        return FileResponse(index)
+
+
+app = create_app()
+
+
+def run() -> None:
+    """Запуск одной командой: `pvideodl` или `python -m app.main`."""
+    import uvicorn
+
+    uvicorn.run(app, host=settings.host, port=settings.port)
+
+
+if __name__ == "__main__":
+    run()

app/models.py

@@ -0,0 +1,56 @@
+"""Pydantic-модели и перечисления доменного слоя."""
+
+from __future__ import annotations
+
+import uuid
+from datetime import datetime, timezone
+from enum import Enum
+
+from pydantic import BaseModel, Field
+
+
+class DownloadStatus(str, Enum):
+    PENDING = "pending"
+    DOWNLOADING = "downloading"
+    DONE = "done"
+    FAILED = "failed"
+    PAUSED = "paused"  # задел на будущее
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _new_id() -> str:
+    return uuid.uuid4().hex
+
+
+class Download(BaseModel):
+    id: str = Field(default_factory=_new_id)
+    url: str
+    filename: str | None = None
+    status: DownloadStatus = DownloadStatus.PENDING
+    progress: float = 0.0  # 0.0–100.0
+    size_bytes: int | None = None
+    downloaded_bytes: int = 0
+    speed: float | None = None  # байт/сек
+    eta: float | None = None  # секунд до конца, оценка
+    error: str | None = None
+    created_at: datetime = Field(default_factory=_now)
+
+
+class CreateDownloads(BaseModel):
+    """Тело запроса POST /api/downloads — список ссылок (по одной на строку или массивом)."""
+
+    urls: list[str] = Field(default_factory=list, min_length=1)
+
+
+# --- События шины (то, что улетает в SSE) ---
+
+
+class DownloadEvent(BaseModel):
+    """Обёртка над снимком загрузки, транслируемая в браузер через SSE."""
+
+    type: str  # "created" | "progress" | "done" | "failed" | "deleted"
+    download: Download | None = None
+    id: str | None = None  # для type="deleted" хватает id

app/services/downloader.py

@@ -0,0 +1,209 @@
+"""Логика скачивания.
+
+Две стратегии за общим интерфейсом:
+- HttpxDownloader — прямые ссылки на файлы, прогресс по chunk'ам.
+- YtDlpDownloader — видео/медиа с сайтов (YouTube и сотни других) через yt-dlp.
+
+`pick_downloader()` выбирает стратегию по URL. Прогресс отдаётся через
+async-колбэк on_progress, который дёргается не чаще, чем раз в progress_interval.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import re
+import time
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
+from pathlib import Path
+from urllib.parse import unquote, urlparse
+
+import httpx
+
+from app.config import settings
+
+# Расширения, которые качаем напрямую через httpx, а не через yt-dlp.
+_DIRECT_EXTENSIONS = {
+    ".zip", ".rar", ".7z", ".tar", ".gz", ".tgz", ".bz2", ".xz",
+    ".iso", ".dmg", ".exe", ".msi", ".apk", ".deb", ".rpm", ".appimage",
+    ".pdf", ".epub", ".mobi", ".djvu",
+    ".jpg", ".jpeg", ".png", ".gif", ".webp", ".svg", ".bmp", ".tiff",
+    ".mp3", ".flac", ".wav", ".ogg", ".m4a", ".aac",
+    ".mp4", ".mkv", ".webm", ".mov", ".avi", ".flv", ".m4v",  # прямые ссылки на медиа
+    ".doc", ".docx", ".xls", ".xlsx", ".ppt", ".pptx", ".csv", ".txt", ".json",
+    ".bin", ".img", ".dll", ".so",
+}
+
+OnProgress = Callable[["Progress"], Awaitable[None]]
+
+
+@dataclass
+class Progress:
+    downloaded_bytes: int
+    total_bytes: int | None
+    speed: float | None
+    eta: float | None
+    filename: str | None = None
+
+
+@dataclass
+class DownloadResult:
+    filename: str
+    size_bytes: int | None
+    path: Path
+
+
+def _safe_filename(name: str) -> str:
+    """Чистим имя файла от разделителей пути и опасных символов."""
+    name = unquote(name).strip().replace("\\", "/").split("/")[-1]
+    name = re.sub(r'[<>:"/\\|?*\x00-\x1f]', "_", name)
+    name = name.strip(". ") or "download"
+    return name[:200]
+
+
+def _unique_path(directory: Path, filename: str) -> Path:
+    """Не перезатираем существующие файлы — добавляем (1), (2), ..."""
+    candidate = directory / filename
+    if not candidate.exists():
+        return candidate
+    stem, suffix = candidate.stem, candidate.suffix
+    i = 1
+    while True:
+        candidate = directory / f"{stem} ({i}){suffix}"
+        if not candidate.exists():
+            return candidate
+        i += 1
+
+
+def is_direct_file(url: str) -> bool:
+    path = urlparse(url).path.lower()
+    return any(path.endswith(ext) for ext in _DIRECT_EXTENSIONS)
+
+
+class Downloader:
+    """Базовый интерфейс стратегии скачивания."""
+
+    async def download(self, url: str, on_progress: OnProgress) -> DownloadResult:
+        raise NotImplementedError
+
+
+class HttpxDownloader(Downloader):
+    def __init__(self) -> None:
+        self._dir = settings.download_dir
+        self._chunk = settings.chunk_size
+        self._interval = settings.progress_interval
+
+    def _filename_from_response(self, url: str, resp: httpx.Response) -> str:
+        cd = resp.headers.get("content-disposition", "")
+        match = re.search(r"filename\*=(?:UTF-8'')?([^;]+)|filename=\"?([^\";]+)\"?", cd)
+        if match:
+            raw = match.group(1) or match.group(2)
+            if raw:
+                return _safe_filename(raw)
+        name = urlparse(str(resp.url)).path
+        return _safe_filename(name) if name and name != "/" else "download"
+
+    async def download(self, url: str, on_progress: OnProgress) -> DownloadResult:
+        self._dir.mkdir(parents=True, exist_ok=True)
+        timeout = httpx.Timeout(30.0, read=None)
+        async with httpx.AsyncClient(follow_redirects=True, timeout=timeout) as client:
+            async with client.stream("GET", url) as resp:
+                resp.raise_for_status()
+                filename = self._filename_from_response(url, resp)
+                total = int(resp.headers["content-length"]) if "content-length" in resp.headers else None
+                target = _unique_path(self._dir, filename)
+
+                downloaded = 0
+                start = time.monotonic()
+                last_emit = 0.0
+                try:
+                    with target.open("wb") as fh:
+                        async for chunk in resp.aiter_bytes(self._chunk):
+                            fh.write(chunk)
+                            downloaded += len(chunk)
+                            now = time.monotonic()
+                            if now - last_emit >= self._interval:
+                                elapsed = now - start
+                                speed = downloaded / elapsed if elapsed > 0 else None
+                                eta = (
+                                    (total - downloaded) / speed
+                                    if total and speed and speed > 0
+                                    else None
+                                )
+                                await on_progress(
+                                    Progress(downloaded, total, speed, eta, target.name)
+                                )
+                                last_emit = now
+                except BaseException:
+                    target.unlink(missing_ok=True)
+                    raise
+
+        elapsed = time.monotonic() - start
+        speed = downloaded / elapsed if elapsed > 0 else None
+        await on_progress(Progress(downloaded, total or downloaded, speed, 0, target.name))
+        return DownloadResult(target.name, downloaded, target)
+
+
+class YtDlpDownloader(Downloader):
+    """Скачивание через yt-dlp. yt-dlp синхронный, поэтому крутим его в потоке,
+    а progress-хуки прокидываем обратно в event loop через call_soon_threadsafe."""
+
+    def __init__(self) -> None:
+        self._dir = settings.download_dir
+        self._interval = settings.progress_interval
+
+    async def download(self, url: str, on_progress: OnProgress) -> DownloadResult:
+        self._dir.mkdir(parents=True, exist_ok=True)
+        loop = asyncio.get_running_loop()
+        last_emit = 0.0
+        result_holder: dict[str, object] = {}
+
+        def hook(d: dict) -> None:
+            nonlocal last_emit
+            status = d.get("status")
+            if status == "downloading":
+                now = time.monotonic()
+                if now - last_emit < self._interval:
+                    return
+                last_emit = now
+                downloaded = d.get("downloaded_bytes") or 0
+                total = d.get("total_bytes") or d.get("total_bytes_estimate")
+                progress = Progress(
+                    downloaded_bytes=downloaded,
+                    total_bytes=total,
+                    speed=d.get("speed"),
+                    eta=d.get("eta"),
+                    filename=os.path.basename(d.get("filename") or "") or None,
+                )
+                asyncio.run_coroutine_threadsafe(on_progress(progress), loop)
+            elif status == "finished":
+                result_holder["path"] = d.get("filename")
+
+        def run_blocking() -> DownloadResult:
+            # Импортируем лениво, чтобы httpx-only сценарий не тянул yt-dlp.
+            from yt_dlp import YoutubeDL
+
+            ydl_opts = {
+                "outtmpl": str(self._dir / "%(title)s [%(id)s].%(ext)s"),
+                "progress_hooks": [hook],
+                "noprogress": True,
+                "quiet": True,
+                "no_warnings": True,
+                "noplaylist": True,
+            }
+            with YoutubeDL(ydl_opts) as ydl:
+                info = ydl.extract_info(url, download=True)
+                final_path = result_holder.get("path")
+                if not final_path:
+                    final_path = ydl.prepare_filename(info)
+            path = Path(str(final_path))
+            size = path.stat().st_size if path.exists() else None
+            return DownloadResult(path.name, size, path)
+
+        return await loop.run_in_executor(None, run_blocking)
+
+
+def pick_downloader(url: str) -> Downloader:
+    """Прямые ссылки на файлы — httpx, всё остальное (страницы сайтов) — yt-dlp."""
+    return HttpxDownloader() if is_direct_file(url) else YtDlpDownloader()

app/services/worker.py

@@ -0,0 +1,113 @@
+"""Воркеры, разбирающие очередь загрузок.
+
+Каждый воркер — отдельная asyncio-таска: берёт id из очереди, грузит запись из
+Storage, качает через выбранный downloader, по ходу обновляет Storage и публикует
+события в EventBus. Менеджер WorkerPool поднимает N таких воркеров и гасит их.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+
+from app.core.events import EventBus
+from app.core.queue import DownloadQueue
+from app.core.storage import Storage
+from app.models import Download, DownloadEvent, DownloadStatus
+from app.services.downloader import Progress, pick_downloader
+
+logger = logging.getLogger("pvideodl.worker")
+
+
+class WorkerPool:
+    def __init__(
+        self,
+        queue: DownloadQueue,
+        storage: Storage,
+        events: EventBus,
+        worker_count: int,
+    ) -> None:
+        self._queue = queue
+        self._storage = storage
+        self._events = events
+        self._count = worker_count
+        self._tasks: list[asyncio.Task[None]] = []
+
+    async def start(self) -> None:
+        self._tasks = [
+            asyncio.create_task(self._run(i), name=f"worker-{i}")
+            for i in range(self._count)
+        ]
+        logger.info("Запущено воркеров: %d", self._count)
+
+    async def stop(self) -> None:
+        for task in self._tasks:
+            task.cancel()
+        await asyncio.gather(*self._tasks, return_exceptions=True)
+        self._tasks.clear()
+
+    async def _run(self, index: int) -> None:
+        while True:
+            download_id = await self._queue.get()
+            try:
+                await self._process(download_id)
+            except asyncio.CancelledError:
+                self._queue.task_done()
+                raise
+            except Exception:  # noqa: BLE001 — воркер не должен падать целиком
+                logger.exception("Воркер %d упал на задаче %s", index, download_id)
+            finally:
+                self._queue.task_done()
+
+    async def _process(self, download_id: str) -> None:
+        download = await self._storage.get(download_id)
+        if download is None or download.status != DownloadStatus.PENDING:
+            return
+
+        updated = await self._storage.update(
+            download_id, status=DownloadStatus.DOWNLOADING, error=None
+        )
+        await self._publish("progress", updated)
+
+        async def on_progress(p: Progress) -> None:
+            fields: dict = {
+                "downloaded_bytes": p.downloaded_bytes,
+                "size_bytes": p.total_bytes,
+                "speed": p.speed,
+                "eta": p.eta,
+            }
+            if p.total_bytes:
+                fields["progress"] = min(100.0, p.downloaded_bytes / p.total_bytes * 100)
+            if p.filename:
+                fields["filename"] = p.filename
+            snapshot = await self._storage.update(download_id, **fields)
+            await self._publish("progress", snapshot)
+
+        try:
+            downloader = pick_downloader(download.url)
+            result = await downloader.download(download.url, on_progress)
+        except Exception as exc:  # noqa: BLE001
+            logger.warning("Ошибка скачивания %s: %s", download.url, exc)
+            failed = await self._storage.update(
+                download_id, status=DownloadStatus.FAILED, error=str(exc), speed=None, eta=None
+            )
+            await self._publish("failed", failed)
+            return
+
+        done = await self._storage.update(
+            download_id,
+            status=DownloadStatus.DONE,
+            filename=result.filename,
+            size_bytes=result.size_bytes,
+            downloaded_bytes=result.size_bytes or 0,
+            progress=100.0,
+            speed=None,
+            eta=0,
+            error=None,
+        )
+        await self._publish("done", done)
+
+    async def _publish(self, event_type: str, download: Download | None) -> None:
+        if download is None:
+            return
+        await self._events.publish(DownloadEvent(type=event_type, download=download))