Update

app_api/api/api/api_v1/schemas/task.py

@@ -1,8 +1,8 @@
 import uuid
 from datetime import datetime
+from typing import Annotated
 
 from pydantic import Field, StrictFloat, StrictInt, StrictStr
-from typing_extensions import Annotated
 from yt_shared.enums import DownMediaType, TaskSource, TaskStatus
 from yt_shared.schemas.base import BaseOrmModel, StrictRealBaseModel
 

app_bot/bot/core/service.py

@@ -52,7 +52,7 @@ class UrlParser:
     def _preprocess_urls(urls: list[str]) -> dict[str, str]:
         preprocessed_urls = {}
         for url in urls:
-            if can_remove_url_params(url, REMOVE_QUERY_PARAMS_HOSTS):
+            if can_remove_url_params(url=url, matching_hosts=REMOVE_QUERY_PARAMS_HOSTS):
                 preprocessed_urls[url] = urljoin(url, urlparse(url).path)
             else:
                 preprocessed_urls[url] = url

app_bot/bot/core/utils.py

@@ -2,7 +2,7 @@
 
 import asyncio
 from datetime import datetime
-from typing import Generator, Iterable
+from typing import Generator
 from urllib.parse import urlparse
 
 from pyrogram.enums import ChatType
@@ -78,8 +78,8 @@ def split_telegram_message(
         yield text
 
 
-def can_remove_url_params(url: str, matching_hosts: Iterable[str]) -> bool:
-    return urlparse(url).netloc in set(matching_hosts)
+def can_remove_url_params(url: str, matching_hosts: set[str]) -> bool:
+    return urlparse(url).netloc in matching_hosts
 
 
 def is_user_upload_silent(

app_worker/ytdl_opts/per_host/__init__.py

@@ -1,8 +1,8 @@
 from ytdl_opts.per_host._default import DefaultHost
+from ytdl_opts.per_host.facebook import FacebookHost
 from ytdl_opts.per_host.instagram import InstagramHost
 from ytdl_opts.per_host.tiktok import TikTokHost
 from ytdl_opts.per_host.twitter import TwitterHost
-from ytdl_opts.per_host.facebook import FacebookHost
 
 __all__ = [
     'DefaultHost',

app_worker/ytdl_opts/per_host/_base.py

@@ -32,7 +32,7 @@ except ImportError:
 
 class BaseHostConfModel(BaseModel):
     # TODO: Add validators.
-    model_config = ConfigDict(strict=True, frozen=True)
+    model_config = ConfigDict(strict=True, frozen=True, validate_assignment=True)
 
     hostnames: tuple[str, ...]
 
@@ -56,18 +56,18 @@ class AbstractHostConfig:
     ENCODE_AUDIO: bool | None = None
     ENCODE_VIDEO: bool | None = None
 
-    KEEP_VIDEO_OPTION = '--keep-video'
+    KEEP_VIDEO_OPTION: str = '--keep-video'
 
-    DEFAULT_YTDL_OPTS = DEFAULT_YTDL_OPTS
+    DEFAULT_YTDL_OPTS: list[str] = DEFAULT_YTDL_OPTS
 
-    AUDIO_YTDL_OPTS = AUDIO_YTDL_OPTS
-    AUDIO_FORMAT_YTDL_OPTS = AUDIO_FORMAT_YTDL_OPTS
+    AUDIO_YTDL_OPTS: list[str] = AUDIO_YTDL_OPTS
+    AUDIO_FORMAT_YTDL_OPTS: list[str] = AUDIO_FORMAT_YTDL_OPTS
 
-    FINAL_AUDIO_FORMAT = FINAL_AUDIO_FORMAT
-    FINAL_THUMBNAIL_FORMAT = FINAL_THUMBNAIL_FORMAT
+    FINAL_AUDIO_FORMAT: str = FINAL_AUDIO_FORMAT
+    FINAL_THUMBNAIL_FORMAT: str = FINAL_THUMBNAIL_FORMAT
 
-    DEFAULT_VIDEO_YTDL_OPTS = VIDEO_YTDL_OPTS
-    DEFAULT_VIDEO_FORMAT_SORT_OPT = DEFAULT_VIDEO_FORMAT_SORT_OPT
+    DEFAULT_VIDEO_YTDL_OPTS: list[str] = VIDEO_YTDL_OPTS
+    DEFAULT_VIDEO_FORMAT_SORT_OPT: list[str] = DEFAULT_VIDEO_FORMAT_SORT_OPT
 
     FFMPEG_AUDIO_OPTS: str | None = None
     FFMPEG_VIDEO_OPTS: str | None = None

app_worker/ytdl_opts/per_host/facebook.py

@@ -36,4 +36,4 @@ class FacebookHost(AbstractHostConfig, metaclass=HostConfRegistry):
         )
 
     def _build_custom_ytdl_video_opts(self) -> list[str]:
-        return self.DEFAULT_VIDEO_FORMAT_SORT_OPT
+        return self.DEFAULT_VIDEO_FORMAT_SORT_OPT

yt_shared/yt_shared/constants.py

@@ -1,19 +1,19 @@
 from asyncio import Lock
 
-ASYNC_LOCK = Lock()
+SHARED_ASYNC_LOCK: Lock = Lock()
 
 
-INSTAGRAM_HOSTS = (
+INSTAGRAM_HOSTS: tuple[str, ...] = (
     'instagram.com',
     'www.instagram.com',
 )
-TIKTOK_HOSTS = (
+TIKTOK_HOSTS: tuple[str, ...] = (
     'tiktok.com',
     'vm.tiktok.com',
     'www.tiktok.com',
     'www.vm.tiktok.com',
 )
-TWITTER_HOSTS = (
+TWITTER_HOSTS: tuple[str, ...] = (
     'twitter.com',
     'www.twitter.com',
     'x.com',
@@ -21,10 +21,14 @@ TWITTER_HOSTS = (
     't.co',
     'www.t.co',
 )
-FACEBOOK_HOSTS = (
+FACEBOOK_HOSTS: tuple[str, ...] = (
     'facebook.com',
     'www.facebook.com',
 )
 
 
-REMOVE_QUERY_PARAMS_HOSTS = TWITTER_HOSTS + INSTAGRAM_HOSTS
+REMOVE_QUERY_PARAMS_HOSTS: set[str] = {
+    *TWITTER_HOSTS,
+    *INSTAGRAM_HOSTS,
+    *FACEBOOK_HOSTS,
+}