Fixes and stubs for type correctness

This commit is contained in:
Christoph Stahl 2023-11-01 18:02:38 +01:00
parent 31c45e3fe4
commit 5679dfab67
5 changed files with 71 additions and 7 deletions

4
pyqrcodeng.pyi Normal file
View file

@ -0,0 +1,4 @@
class QRCode:
def terminal(self, quiet_zone: int) -> str: ...
def create(data: str) -> QRCode: ...

60
pytube.pyi Normal file
View file

@ -0,0 +1,60 @@
from __future__ import annotations
from collections.abc import Iterable
from typing import Any, Callable, Iterator, Optional
class exceptions:
class PytubeError(Exception): ...
class Channel:
channel_id: str
def __init__(self, url: str) -> None:
pass
class innertube:
class InnerTube:
base_url: str
base_data: dict[str, str]
base_params: dict[str, str]
def _call_api(
self, endpoint: str, params: dict[str, str], data: dict[str, str]
) -> dict[str, Any]: ...
def __init__(self, client: str) -> None: ...
class Stream:
resolution: str
is_progressive: bool
is_adaptive: bool
abr: str
def download(
self,
output_path: Optional[str] = None,
filename_prefix: Optional[str] = None,
) -> str: ...
class StreamQuery(Iterable[Stream]):
resolution: str
def filter(
self,
type: Optional[str] = None,
custom_filter_functions: Optional[
list[Callable[[StreamQuery], bool]]
] = None,
only_audio: bool = False,
) -> StreamQuery: ...
def __iter__(self) -> Iterator[Stream]: ...
class YouTube:
def __init__(self, url: str) -> None: ...
length: int
title: str
author: str
watch_url: str
streams: StreamQuery
class Search:
results: Optional[list[YouTube]]
def __init__(self, query: str) -> None: ...

View file

@ -62,7 +62,7 @@ sources: dict[str, Source] = {}
currentLock: asyncio.Semaphore = asyncio.Semaphore(0) currentLock: asyncio.Semaphore = asyncio.Semaphore(0)
def default_config(): def default_config() -> dict[str, Optional[int | str]]:
return { return {
"preview_duration": 3, "preview_duration": 3,
"last_song": None, "last_song": None,

View file

@ -148,16 +148,16 @@ class S3Source(Source):
if os.path.splitext(entry.ident)[1] == ".cdg": if os.path.splitext(entry.ident)[1] == ".cdg":
cdg_filename: str = os.path.basename(entry.ident) cdg_filename: str = os.path.basename(entry.ident)
path_to_file: str = os.path.dirname(entry.ident) path_to_files: str = os.path.dirname(entry.ident)
cdg_path: str = os.path.join(path_to_file, cdg_filename) cdg_path: str = os.path.join(path_to_files, cdg_filename)
target_file_cdg: str = os.path.join(self.tmp_dir, cdg_path) target_file_cdg: str = os.path.join(self.tmp_dir, cdg_path)
ident_mp3: str = entry.ident[:-3] + "mp3" ident_mp3: str = entry.ident[:-3] + "mp3"
target_file_mp3: str = target_file_cdg[:-3] + "mp3" target_file_mp3: str = target_file_cdg[:-3] + "mp3"
os.makedirs(os.path.dirname(target_file_cdg), exist_ok=True) os.makedirs(os.path.dirname(target_file_cdg), exist_ok=True)
video_task: asyncio.Task[Any] = asyncio.create_task( cdg_task: asyncio.Task[Any] = asyncio.create_task(
asyncio.to_thread( asyncio.to_thread(
self.minio.fget_object, self.minio.fget_object,
self.bucket, self.bucket,
@ -174,7 +174,7 @@ class S3Source(Source):
) )
) )
await video_task await cdg_task
await audio_task await audio_task
return target_file_cdg, target_file_mp3 return target_file_cdg, target_file_mp3
video_filename: str = os.path.basename(entry.ident) video_filename: str = os.path.basename(entry.ident)

View file

@ -19,7 +19,7 @@ from pytube import Search
from pytube import Stream from pytube import Stream
from pytube import StreamQuery from pytube import StreamQuery
from pytube import YouTube from pytube import YouTube
from pytube.exceptions import PytubeError from pytube import exceptions
try: try:
from yt_dlp import YoutubeDL from yt_dlp import YoutubeDL
@ -152,7 +152,7 @@ class YoutubeSource(Source):
artist=yt_song.author, artist=yt_song.author,
performer=performer, performer=performer,
) )
except PytubeError: except exceptions.PytubeError:
return None return None
return await asyncio.to_thread(_get_entry, performer, ident) return await asyncio.to_thread(_get_entry, performer, ident)