refactor: fix lint errors with ruff check . --fix

This commit is contained in:
dark_zoul committed 2026-10-19 21:52:30 +03:00
1 parent 47f88bc6a4
commit 5b393c0bcd
28 files changed
+99 -107

No files matched your search

+4 -4
View File
@@ -3,17 +3,17 @@ from __future__ import annotations
import argparse
import asyncio
import logging
import re
from pathlib import Path
from .config.settings import Settings
from .core.database.db import Database
from .core.sync.service import SyncService
from .core.sync.executor import ActionExecutor
from .core.events.event_bus import EventBus
import re
from .core.utils.yt import extract_playlist_id
from .core.sync.executor import ActionExecutor
from .core.sync.service import SyncService
from .core.utils.deps import DependencyError
from .core.utils.logging_setup import configure_logging
from .core.utils.yt import extract_playlist_id
def main(argv: list[str] | None = None) -> int:
+6 -6
View File
@@ -3,7 +3,7 @@ from __future__ import annotations
import json
import os
from pathlib import Path
from typing import Any, Dict, List
from typing import Any
def _default_ffmpeg_path() -> str:
@@ -12,7 +12,7 @@ def _default_ffmpeg_path() -> str:
return "./bin/ffmpeg"
DEFAULT_CONFIG: Dict[str, Any] = {
DEFAULT_CONFIG: dict[str, Any] = {
"playlists": [],
"download_mode": "video",
"max_download_quality": "1080p",
@@ -29,7 +29,7 @@ class Settings:
base_dir = Path("config")
base_dir.mkdir(parents=True, exist_ok=True)
self.path = (base_dir / "yt-playlist-config.json").resolve()
self.data: Dict[str, Any] = dict(DEFAULT_CONFIG)
self.data: dict[str, Any] = dict(DEFAULT_CONFIG)
# Ensure there is always a config file at the default path.
if not self.path.exists():
@@ -46,7 +46,7 @@ class Settings:
def _write_default_config(self, path: Path) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
default_payload: Dict[str, Any] = {
default_payload: dict[str, Any] = {
"playlists": [
{
"url": "https://www.youtube.com/playlist?list=YOUR_PLAYLIST_ID",
@@ -60,7 +60,7 @@ class Settings:
path.write_text(json.dumps(default_payload, indent=2) + "\n", encoding="utf-8")
@property
def playlists(self) -> List[Dict[str, Any]]:
def playlists(self) -> list[dict[str, Any]]:
global_defaults = {
"download_mode": self.data.get("download_mode", DEFAULT_CONFIG["download_mode"]),
"max_download_quality": self.data.get("max_download_quality", DEFAULT_CONFIG["max_download_quality"]),
@@ -71,7 +71,7 @@ class Settings:
"retry_delay_seconds": self.data.get("retry_delay_seconds", DEFAULT_CONFIG["retry_delay_seconds"]),
}
results: List[Dict[str, Any]] = []
results: list[dict[str, Any]] = []
for pl in list(self.data.get("playlists", [])):
if not isinstance(pl, dict):
continue
+4 -4
View File
@@ -1,10 +1,10 @@
"""Core backend modules (scanner, sync, download, db, scheduler, events)."""
__all__ = [
"scanner",
"sync",
"download",
"database",
"scheduler",
"download",
"events",
"scanner",
"scheduler",
"sync",
]
+1 -2
View File
@@ -1,9 +1,8 @@
from __future__ import annotations
import sqlite3
from collections.abc import Iterable
from pathlib import Path
from typing import Iterable
SCHEMA = """
PRAGMA journal_mode=WAL;
+6 -9
View File
@@ -1,7 +1,5 @@
from __future__ import annotations
from typing import Optional
from .queue_manager import DownloadJob, JobState
@@ -11,7 +9,7 @@ class Downloader:
where actual download logic will land (audio/video/both).
"""
def __init__(self, yt_dlp_path: Optional[str] = None, ffmpeg_path: Optional[str] = None) -> None:
def __init__(self, yt_dlp_path: str | None = None, ffmpeg_path: str | None = None) -> None:
self.yt_dlp_path = yt_dlp_path
self.ffmpeg_path = ffmpeg_path
@@ -72,8 +70,9 @@ class Downloader:
import asyncio
def run():
import yt_dlp # type: ignore
from pathlib import Path
import yt_dlp # type: ignore
try:
from yt_dlp.utils import DownloadCancelled # type: ignore
except Exception: # pragma: no cover - optional
@@ -138,15 +137,13 @@ class Downloader:
should_emit = status in {"finished", "error"}
if not should_emit:
if now - last_ts >= 0.25:
should_emit = True
elif pct is not None and (last_pct < 0 or abs(pct - last_pct) >= 0.01):
if now - last_ts >= 0.25 or pct is not None and (last_pct < 0 or abs(pct - last_pct) >= 0.01):
should_emit = True
if should_emit:
setattr(job, "_last_progress_emit_ts", now)
job._last_progress_emit_ts = now
if pct is not None:
setattr(job, "_last_progress_emit_pct", pct)
job._last_progress_emit_pct = pct
progress_cb(payload)
except Exception:
pass
+12 -11
View File
@@ -1,10 +1,11 @@
from __future__ import annotations
import asyncio
from collections.abc import Callable
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
from typing import Any, Callable, Optional
from typing import Any
from ..models import PlaylistItem
@@ -22,17 +23,17 @@ class JobState(str, Enum):
@dataclass
class DownloadJob:
item: PlaylistItem
output_path: Optional[Path] = None
url: Optional[str] = None
output_path: Path | None = None
url: str | None = None
mode: str = "audio" # audio|video
state: JobState = JobState.QUEUED
error: Optional[str] = None
ffmpeg_path: Optional[str] = None
max_download_quality: Optional[str] = None
playlist_id: Optional[str] = None
progress_callback: Optional[Callable[[dict[str, Any]], None]] = None
cancel_check: Optional[Callable[[], bool]] = None
audio_output_path: Optional[Path] = None # when mode=video and we also want mp3
error: str | None = None
ffmpeg_path: str | None = None
max_download_quality: str | None = None
playlist_id: str | None = None
progress_callback: Callable[[dict[str, Any]], None] | None = None
cancel_check: Callable[[], bool] | None = None
audio_output_path: Path | None = None # when mode=video and we also want mp3
keep_video: bool = True
@@ -45,7 +46,7 @@ class QueueManager:
"""
def __init__(self, concurrency: int = 2) -> None:
self._queue: "asyncio.Queue[DownloadJob]" = asyncio.Queue()
self._queue: asyncio.Queue[DownloadJob] = asyncio.Queue()
self._concurrency = max(1, concurrency)
self._workers: list[asyncio.Task[None]] = []
self._stopped = asyncio.Event()
+2 -1
View File
@@ -2,9 +2,10 @@ from __future__ import annotations
import asyncio
import logging
from ..utils.rate_limit import is_youtube_rate_limit_error
from .downloader import Downloader
from .queue_manager import DownloadJob, JobState
from ..utils.rate_limit import is_youtube_rate_limit_error
async def default_worker(job: DownloadJob, *, max_retries: int = 2, delay_seconds: float = 1.5):
+5 -5
View File
@@ -1,21 +1,21 @@
from __future__ import annotations
from collections import defaultdict
from typing import Any, Awaitable, Callable, DefaultDict, Dict, List
from collections.abc import Awaitable, Callable
from typing import Any
EventHandler = Callable[[Dict[str, Any]], Awaitable[None]]
EventHandler = Callable[[dict[str, Any]], Awaitable[None]]
class EventBus:
"""Simple async pub/sub event bus used by backend and (later) GUI."""
def __init__(self) -> None:
self._subs: DefaultDict[str, List[EventHandler]] = defaultdict(list)
self._subs: defaultdict[str, list[EventHandler]] = defaultdict(list)
def subscribe(self, event_name: str, handler: EventHandler) -> None:
self._subs[event_name].append(handler)
async def publish(self, event_name: str, payload: Dict[str, Any]) -> None:
async def publish(self, event_name: str, payload: dict[str, Any]) -> None:
for h in list(self._subs.get(event_name, [])):
await h(payload)
+5 -6
View File
@@ -3,7 +3,6 @@ from __future__ import annotations
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
from typing import Optional
class DownloadMode(str, Enum):
@@ -15,7 +14,7 @@ class DownloadMode(str, Enum):
@dataclass(frozen=True)
class Playlist:
id: str
name: Optional[str]
name: str | None
url: str
path: Path
mode: DownloadMode = DownloadMode.audio
@@ -29,7 +28,7 @@ class PlaylistItem:
video_id: str
title: str
playlist_index: int
local_filename: Optional[str] = None
local_filename: str | None = None
downloaded: bool = False
@@ -45,9 +44,9 @@ class SyncActionType(str, Enum):
@dataclass(frozen=True)
class SyncAction:
type: SyncActionType
item: Optional[PlaylistItem] = None
from_name: Optional[str] = None
to_name: Optional[str] = None
item: PlaylistItem | None = None
from_name: str | None = None
to_name: str | None = None
@dataclass(frozen=True)
+2 -3
View File
@@ -1,7 +1,6 @@
from __future__ import annotations
from pathlib import Path
from typing import List, Optional
from ..models import PlaylistItem
@@ -19,7 +18,7 @@ class PlaylistScanner:
def __init__(self) -> None:
pass
def scan(self, playlist_url: str, playlist_id: str, *, ffmpeg_path: Optional[str] = None) -> List[PlaylistItem]:
def scan(self, playlist_url: str, playlist_id: str, *, ffmpeg_path: str | None = None) -> list[PlaylistItem]:
"""Return the current remote playlist entries as `PlaylistItem` records."""
try:
import yt_dlp # type: ignore
@@ -46,7 +45,7 @@ class PlaylistScanner:
info = ydl.extract_info(playlist_url, download=False)
entries = info.get("entries", []) if isinstance(info, dict) else []
items: List[PlaylistItem] = []
items: list[PlaylistItem] = []
for idx, v in enumerate(entries, start=1):
if not v:
continue
+1 -1
View File
@@ -1,7 +1,7 @@
from __future__ import annotations
from collections.abc import Awaitable, Callable
from datetime import timedelta
from typing import Awaitable, Callable
class Scheduler:
+3 -3
View File
@@ -1,6 +1,6 @@
from __future__ import annotations
from typing import Iterable, List, Mapping, Sequence
from collections.abc import Iterable, Mapping, Sequence
from ..models import FilesystemEntry, PlaylistItem, SyncAction, SyncActionType
@@ -18,8 +18,8 @@ class DiffEngine:
db_index: Mapping[str, PlaylistItem],
fs_entries: Iterable[FilesystemEntry],
extension: str,
) -> List[SyncAction]:
actions: List[SyncAction] = []
) -> list[SyncAction]:
actions: list[SyncAction] = []
desired_names = {
item.video_id: f"{item.playlist_index:04d} - {item.title}{extension}"
+7 -7
View File
@@ -1,20 +1,20 @@
from __future__ import annotations
import asyncio
import time
import shutil
import time
from collections.abc import Iterable
from pathlib import Path
from typing import Iterable, List
from ..database.db import Database
from ..download.queue_manager import DownloadJob, QueueManager
from ..download.workers import default_worker
from ..events.event_bus import EventBus
from ..models import SyncAction, SyncActionType
from ..sync.reorder import safe_multi_rename
from ..database.db import Database
from ..utils.yt import extract_playlist_id
from ..events.event_bus import EventBus
from ..utils.deps import ensure_ffmpeg_available, ensure_yt_dlp_available
from ..utils.rate_limit import is_youtube_rate_limit_error
from ..utils.yt import extract_playlist_id
class ActionExecutor:
@@ -140,7 +140,7 @@ class ActionExecutor:
playlist_id = extract_playlist_id(playlist_cfg.get("url", "")) or playlist_cfg.get("url", "")
audio_renames = []
video_renames = []
applied: List[SyncAction] = []
applied: list[SyncAction] = []
for a in actions:
if a.type != SyncActionType.RENAME or not a.from_name or not a.to_name:
continue
@@ -284,7 +284,7 @@ class ActionExecutor:
await queue.start(worker)
try:
jobs: List[DownloadJob] = []
jobs: list[DownloadJob] = []
# Collapse 'both' into single video download + local audio extraction
# Build per-video desired outputs
+3 -3
View File
@@ -1,14 +1,14 @@
from __future__ import annotations
from collections.abc import Sequence
from pathlib import Path
from typing import List, Sequence
from ..models import FilesystemEntry
def list_files(root: Path, extensions: Sequence[str]) -> List[FilesystemEntry]:
def list_files(root: Path, extensions: Sequence[str]) -> list[FilesystemEntry]:
exts = {e.lower() for e in extensions}
results: List[FilesystemEntry] = []
results: list[FilesystemEntry] = []
if not root.exists():
return results
for p in root.glob("**/*"):
+3 -3
View File
@@ -1,10 +1,10 @@
from __future__ import annotations
from collections.abc import Iterable
from pathlib import Path
from typing import Dict, Iterable, Tuple
def safe_multi_rename(renames: Iterable[Tuple[Path, Path]]) -> None:
def safe_multi_rename(renames: Iterable[tuple[Path, Path]]) -> None:
"""
Apply multiple renames safely using a two-pass strategy to avoid
name collisions. Each item is a tuple (src_path, dst_path).
@@ -14,7 +14,7 @@ def safe_multi_rename(renames: Iterable[Tuple[Path, Path]]) -> None:
existing_dests = {dst for _, dst in planned}
# Pass 1: move all sources that would collide to temporary names
temps: Dict[Path, Path] = {}
temps: dict[Path, Path] = {}
for src, dst in planned:
if not src.exists():
continue
+4 -5
View File
@@ -1,7 +1,6 @@
from __future__ import annotations
from pathlib import Path
from typing import List
from ..database.db import Database
from ..models import PlaylistItem, SyncAction
@@ -34,7 +33,7 @@ class SyncService:
return [".mp3", ".mp4"]
return [".mp4"]
def sync_from_config(self, playlist_cfg: dict) -> List[SyncAction]:
def sync_from_config(self, playlist_cfg: dict) -> list[SyncAction]:
"""Return the sync actions required to bring one playlist in sync.
This method does not apply any changes itself. It normalizes the
@@ -60,7 +59,7 @@ class SyncService:
ffmpeg_cfg = playlist_cfg.get("ffmpeg_path")
items = self.scanner.scan(url, playlist_id, ffmpeg_path=str(ffmpeg_cfg) if ffmpeg_cfg is not None else None)
sanitized: List[PlaylistItem] = []
sanitized: list[PlaylistItem] = []
for it in items:
safe_title = sanitize_title(it.title, it.video_id)
sanitized.append(
@@ -100,7 +99,7 @@ class SyncService:
)
# Augment remote items with DB-known filenames/download flags
augmented: List[PlaylistItem] = []
augmented: list[PlaylistItem] = []
for it in sanitized:
known = db_index.get(it.video_id)
if known is None:
@@ -118,7 +117,7 @@ class SyncService:
)
exts = self._mode_to_extensions(mode)
merged_actions: List[SyncAction] = []
merged_actions: list[SyncAction] = []
# Compute per-extension actions against respective roots
for ext in exts:
+6 -7
View File
@@ -6,7 +6,6 @@ import sys
from dataclasses import dataclass
from pathlib import Path
from shutil import which
from typing import Optional, Tuple
class DependencyError(RuntimeError):
@@ -16,14 +15,14 @@ class DependencyError(RuntimeError):
@dataclass(frozen=True)
class DependencyStatus:
yt_dlp_ok: bool
yt_dlp_version: Optional[str]
yt_dlp_version: str | None
ffmpeg_ok: bool
ffmpeg_path: Optional[str]
ffmpeg_path: str | None
ffprobe_ok: bool
ffprobe_path: Optional[str]
ffprobe_path: str | None
def get_yt_dlp_version() -> Optional[str]:
def get_yt_dlp_version() -> str | None:
try:
import yt_dlp # type: ignore
@@ -44,7 +43,7 @@ def ensure_yt_dlp_available() -> str:
return get_yt_dlp_version() or "unknown"
def _resolve_tool_paths(tool_hint: Optional[str], exe_name: str) -> Tuple[Optional[str], Optional[str]]:
def _resolve_tool_paths(tool_hint: str | None, exe_name: str) -> tuple[str | None, str | None]:
"""
Returns (resolved_exe_path, resolved_dir_hint).
- If tool_hint is a file path, use it.
@@ -70,7 +69,7 @@ def _resolve_tool_paths(tool_hint: Optional[str], exe_name: str) -> Tuple[Option
return None, None
def ensure_ffmpeg_available(ffmpeg_hint: Optional[str]) -> Tuple[str, str]:
def ensure_ffmpeg_available(ffmpeg_hint: str | None) -> tuple[str, str]:
"""
Ensures ffmpeg is runnable. Returns (ffmpeg_path, ffmpeg_path).
+4 -2
View File
@@ -1,5 +1,7 @@
from __future__ import annotations
from typing import Any, Dict
from typing import Any
from PySide6 import QtCore
from ..core.events.event_bus import EventBus
@@ -32,7 +34,7 @@ class BusBridge(QtCore.QObject):
self._bus.subscribe(name, self._make_handler(name))
def _make_handler(self, name: str):
async def handler(payload: Dict[str, Any]) -> None:
async def handler(payload: dict[str, Any]) -> None:
# Ensure delivery on the Qt main thread.
self.event.emit(name, dict(payload))
+4 -4
View File
@@ -3,12 +3,12 @@ from __future__ import annotations
import json
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict
from typing import Any
@dataclass(frozen=True)
class AppConfig:
data: Dict[str, Any]
data: dict[str, Any]
path: Path
@@ -19,13 +19,13 @@ def load_config(path: Path) -> AppConfig:
return AppConfig(data=raw, path=path)
def save_config(path: Path, data: Dict[str, Any]) -> None:
def save_config(path: Path, data: dict[str, Any]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
payload = json.dumps(data, indent=2, ensure_ascii=False) + "\n"
path.write_text(payload, encoding="utf-8")
def normalize_config(data: Dict[str, Any]) -> Dict[str, Any]:
def normalize_config(data: dict[str, Any]) -> dict[str, Any]:
"""
Ensure basic expected shape for config dict.
Keeps unknown keys intact.
+6 -6
View File
@@ -7,15 +7,15 @@ from PySide6 import QtCore, QtGui, QtWidgets
from ..config.settings import Settings
from ..core.events.event_bus import EventBus
from .bus_bridge import BusBridge
from .app_icon import load_app_icon
from .bus_bridge import BusBridge
from .config_store import load_config
from .runner import SyncRequest, SyncRunner
from .pages.about import AboutPage
from .pages.logs import LogsPage
from .pages.playlists import PlaylistManagerPage
from .pages.queue import QueuePage
from .pages.logs import LogsPage
from .pages.settings import SettingsPage
from .pages.about import AboutPage
from .runner import SyncRequest, SyncRunner
class MainWindow(QtWidgets.QMainWindow):
@@ -334,8 +334,8 @@ class MainWindow(QtWidgets.QMainWindow):
pass
# Continue "sync all" chain if active.
if hasattr(self, "_sync_queue") and getattr(self, "_sync_queue"):
nxt = getattr(self, "_sync_queue").pop(0)
if hasattr(self, "_sync_queue") and self._sync_queue:
nxt = self._sync_queue.pop(0)
self._sync_playlist_index(nxt)
@QtCore.Slot()
+2 -2
View File
@@ -1,16 +1,16 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
from pathlib import Path
from typing import Any
from PySide6 import QtCore, QtGui, QtWidgets
from ...config.settings import Settings
from ...core.database.db import Database
from ...core.utils.yt import extract_playlist_id
from ..smooth_scroll import enable_smooth_scrolling
from ..config_store import load_config, normalize_config, save_config
from ..smooth_scroll import enable_smooth_scrolling
@dataclass(frozen=True)
+3 -3
View File
@@ -4,19 +4,19 @@ import asyncio
import threading
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict
from typing import Any
from PySide6 import QtCore
from ..core.database.db import Database
from ..core.events.event_bus import EventBus
from ..core.sync.executor import ActionExecutor
from ..core.sync.service import SyncService
from ..core.events.event_bus import EventBus
@dataclass(frozen=True)
class SyncRequest:
playlist_cfg: Dict[str, Any]
playlist_cfg: dict[str, Any]
apply: bool = True
db_path: Path = Path("db/app.db")
cancel_flag: threading.Event | None = None
+1 -1
View File
@@ -11,7 +11,7 @@ class _SmoothWheelFilter(QtCore.QObject):
self._anim = QtCore.QPropertyAnimation(area.verticalScrollBar(), b"value", self)
self._anim.setEasingCurve(QtCore.QEasingCurve.Type.OutCubic)
def eventFilter(self, obj: QtCore.QObject, event: QtCore.QEvent) -> bool: # noqa: N802
def eventFilter(self, obj: QtCore.QObject, event: QtCore.QEvent) -> bool:
if event.type() != QtCore.QEvent.Type.Wheel:
return super().eventFilter(obj, event)
+2 -2
View File
@@ -12,10 +12,10 @@ from pathlib import Path
from .config.settings import Settings
from .core.database.db import Database
from .core.sync.service import SyncService
from .core.sync.executor import ActionExecutor
from .core.utils.yt import extract_playlist_id
from .core.sync.service import SyncService
from .core.utils.deps import DependencyError
from .core.utils.yt import extract_playlist_id
def bootstrap(db_path: Path | None = None) -> None:
+1 -2
View File
@@ -5,11 +5,10 @@ import os
import pytest
from app.core.database.db import Database
from app.core.sync.executor import ActionExecutor
from app.core.events.event_bus import EventBus
from app.core.sync.executor import ActionExecutor
from app.core.sync.service import SyncService
PLAYLIST_URL = os.getenv("TEST_PLAYLIST_URL")
+1 -2
View File
@@ -5,11 +5,10 @@ import os
import pytest
from app.core.database.db import Database
from app.core.sync.executor import ActionExecutor
from app.core.events.event_bus import EventBus
from app.core.sync.executor import ActionExecutor
from app.core.sync.service import SyncService
PLAYLIST_URL = os.getenv("TEST_PLAYLIST_URL")
+1 -2
View File
@@ -5,11 +5,10 @@ import os
import pytest
from app.core.database.db import Database
from app.core.sync.executor import ActionExecutor
from app.core.events.event_bus import EventBus
from app.core.sync.executor import ActionExecutor
from app.core.sync.service import SyncService
PLAYLIST_URL = os.getenv("TEST_PLAYLIST_URL")
-1
View File
@@ -2,6 +2,5 @@ from __future__ import annotations
from app.gui.main import main
if __name__ == "__main__":
raise SystemExit(main())