Type hint (#2)

Preprocessing/BOT_EXTRACTION_AUDIO.py

@@ -5,7 +5,7 @@
 from typing import Optional
 
 
-def get_best_audio_stream(video_file) -> Optional[int]:
+def get_best_audio_stream(video_file: str) -> Optional[int]:
     """Zwraca indeks najlepszej ścieżki audio na podstawie bitrate."""
     try:
         cmd = f'ffprobe -v quiet -print_format json -show_streams -select_streams a "{video_file}"'
@@ -20,7 +20,7 @@ def get_best_audio_stream(video_file) -> Optional[int]:
         return None
 
 
-def convert_and_normalize_audio(video_file, audio_index, output_audio_file) -> None:
+def convert_and_normalize_audio(video_file: str, audio_index: int, output_audio_file: str) -> None:
     """Konwertuje wybraną ścieżkę audio do formatu WAV, mono, z normalizacją głośności."""
     try:
         # Konwersja do formatu WAV, mono
@@ -42,7 +42,7 @@ def convert_and_normalize_audio(video_file, audio_index, output_audio_file) -> N
         print(f"Błąd podczas konwersji audio: {e}")
 
 
-def process_folder(input_folder, output_folder) -> None:
+def process_folder(input_folder: str, output_folder: str) -> None:
     """Przetwarza wszystkie pliki wideo w podanym folderze, zachowując strukturę folderów."""
     for root, _, files in os.walk(input_folder):
         for file in files:

Preprocessing/BOT_MAKE_TRANSCRIPTION.py

@@ -3,7 +3,7 @@
 import subprocess
 
 
-def przetworz_folder(folder_wejsciowy, folder_wyjsciowy) -> None:
+def przetworz_folder(folder_wejsciowy: str, folder_wyjsciowy: str) -> None:
     model = "large-v3"
     jezyk = "Polish"
     urzadzenie = "cuda"

Preprocessing/DataExtractionFromWiki/ModifyJson.py

@@ -2,13 +2,13 @@
 import os
 
 
-def load_episode_info():
+def load_episode_info() -> json:
     with open('EpisodeInfo.json', 'r', encoding='utf-8') as file:
         episode_info = json.load(file)
     return episode_info
 
 
-def add_episode_info_to_transcriptions(base_path="RANCZO-TRANSKRYPCJE"):
+def add_episode_info_to_transcriptions(base_path: str = "RANCZO-TRANSKRYPCJE") -> None:
     episode_info = load_episode_info()
 
     for season in range(1, 11):  # Dla 10 sezonów

Preprocessing/PreprocessingVideo.py

@@ -3,7 +3,7 @@
 import subprocess
 
 
-def convert_videos(input_dir, output_dir) -> None:
+def convert_videos(input_dir: str, output_dir: str) -> None:
     for root, _, files in os.walk(input_dir):
         for file in files:
             if file.endswith(".mp4"):

bot/handlers/admin_tools.py

@@ -1,13 +1,18 @@
+from datetime import date
 import logging
+from typing import (
+    List,
+    Optional,
+)
 
 from aiogram import (
     Dispatcher,
     Router,
     types,
 )
 from aiogram.filters import Command
+import asyncpg
 from tabulate import tabulate
-from typing import Optional
 
 from bot.middlewares.auth_middleware import AuthorizationMiddleware
 from bot.middlewares.error_middleware import ErrorHandlerMiddleware
@@ -22,43 +27,51 @@
 # Definicja UserManager dla łatwiejszego dostępu do funkcji zarządzania użytkownikami
 class UserManager:
     @staticmethod
-    async def add_user(username, is_admin=False, is_moderator=False, full_name=None, email=None, phone=None, subscription_days=None) -> None:
+    async def add_user(
+        username: str, is_admin: Optional[bool] = False, is_moderator: Optional[bool] = False,
+        full_name: Optional[str] = None, email: Optional[str] = None, phone: Optional[str] = None,
+        subscription_days: Optional[int] = None,
+    ) -> None:
         await DatabaseManager.add_user(username, is_admin, is_moderator, full_name, email, phone, subscription_days)
 
     @staticmethod
-    async def remove_user(username) -> None:
+    async def remove_user(username: str) -> None:
         await DatabaseManager.remove_user(username)
 
     @staticmethod
-    async def update_user(username, is_admin=None, is_moderator=None, full_name=None, email=None, phone=None, subscription_end=None) -> None:
+    async def update_user(
+        username: str, is_admin: Optional[bool] = False, is_moderator: Optional[bool] = False,
+        full_name: Optional[str] = None, email: Optional[str] = None, phone: Optional[str] = None,
+        subscription_end: Optional[int] = None,
+    ) -> None:
         await DatabaseManager.update_user(username, is_admin, is_moderator, full_name, email, phone, subscription_end)
 
     @staticmethod
-    async def get_all_users() -> Optional[list]: # TO DO: Change return type
+    async def get_all_users() -> Optional[List[asyncpg.Record]]:
         return await DatabaseManager.get_all_users()
 
     @staticmethod
-    async def get_admin_users() -> Optional[list]: # TO DO: Change return type
+    async def get_admin_users() -> Optional[List[asyncpg.Record]]:
         return await DatabaseManager.get_admin_users()
 
     @staticmethod
-    async def get_moderator_users() -> Optional[list]: # TO DO: Change return type
+    async def get_moderator_users() -> Optional[List[asyncpg.Record]]:
         return await DatabaseManager.get_moderator_users()
 
     @staticmethod
-    async def add_subscription(username, days) -> Optional[str]: # TO DO: Change return type
+    async def add_subscription(username: str, days: int) -> Optional[date]:
         return await DatabaseManager.add_subscription(username, days)
 
     @staticmethod
-    async def remove_subscription(username) -> None:
+    async def remove_subscription(username: str) -> None:
         await DatabaseManager.remove_subscription(username)
 
     @staticmethod
-    async def is_user_admin(username) -> Optional[bool]: # TO DO: Change return type
+    async def is_user_admin(username: str) -> Optional[bool]:
         return await DatabaseManager.is_user_admin(username)
 
     @staticmethod
-    async def is_user_moderator(username) -> Optional[bool]: # TO DO: Change return type
+    async def is_user_moderator(username: str) -> Optional[bool]:
         return await DatabaseManager.is_user_moderator(username)
 
 

bot/handlers/clip_search.py

@@ -1,4 +1,9 @@
+import json
 import logging
+from typing import (
+    Dict,
+    List,
+)
 
 from aiogram import (
     Bot,
@@ -17,8 +22,8 @@
 router = Router()
 dis = Dispatcher()
 
-last_search_quotes = {}
-last_search_terms = {}  # Store search terms
+last_search_quotes: Dict[int, List[json]] = {}
+last_search_terms: Dict[int, str] = {}
 
 
 @router.message(Command(commands=['szukaj', 'search', 'sz']))

bot/handlers/compile_selected.py

@@ -1,6 +1,8 @@
+import json
 import logging
 import os
 import tempfile
+from typing import Dict
 
 from aiogram import (
     Bot,
@@ -19,8 +21,7 @@
 logger = logging.getLogger(__name__)
 router = Router()
 
-# Definicja last_compiled_clip
-last_compiled_clip = {}
+last_compiled_clip: Dict[int, json] = {}
 
 
 @router.message(Command(commands=['polaczklipy', 'concatclips', 'pk']))

bot/handlers/episode_list.py

@@ -1,4 +1,9 @@
 import logging
+from typing import (
+    List,
+    Optional,
+    Tuple,
+)
 
 from aiogram import (
     Bot,
@@ -7,7 +12,6 @@
     types,
 )
 from aiogram.filters import Command
-from typing import Optional
 
 from bot.middlewares.auth_middleware import AuthorizationMiddleware
 from bot.middlewares.error_middleware import ErrorHandlerMiddleware
@@ -18,14 +22,15 @@
 router = Router()
 dis = Dispatcher()
 
-def adjust_episode_number(absolute_episode) -> Optional[tuple[int, int]]:
+
+def adjust_episode_number(absolute_episode: int) -> Optional[Tuple[int, int]]:
     """ Adjust the absolute episode number to season and episode format """
     season = (absolute_episode - 1) // 13 + 1
     episode = (absolute_episode - 1) % 13 + 1
     return season, episode
 
 
-def split_message(message, max_length=4096) -> Optional[list[str]]:
+def split_message(message: types.Message, max_length: int = 4096) -> Optional[List[str]]:
     """ Splits a message into chunks to fit within the Telegram message length limit """
     parts = []
     while len(message) > max_length:

bot/handlers/handle_clip.py

@@ -1,4 +1,6 @@
+import json
 import logging
+from typing import Dict
 
 from aiogram import (
     Bot,
@@ -22,8 +24,7 @@
 router = Router()
 dis = Dispatcher()
 
-# Definicja last_selected_segment
-last_selected_segment = {}
+last_selected_segment: Dict[int, json] = {}
 
 
 @router.message(Command(commands=['klip', 'clip', 'k']))

bot/handlers/list_clips.py

@@ -1,4 +1,5 @@
 import logging
+from typing import Dict
 
 from aiogram import (
     Bot,
@@ -15,7 +16,7 @@
 logger = logging.getLogger(__name__)
 router = Router()
 
-number_to_emoji = {
+number_to_emoji: Dict[str, str] = {
     '0': '0️⃣',
     '1': '1️⃣',
     '2': '2️⃣',
@@ -29,7 +30,7 @@
 }
 
 
-def convert_number_to_emoji(number) -> str:
+def convert_number_to_emoji(number: int) -> str:
     return ''.join(number_to_emoji.get(digit, digit) for digit in str(number))
 
 

bot/handlers/manual_clip.py

@@ -1,5 +1,11 @@
+import json
 import logging
 import os
+from typing import (
+    Dict,
+    Optional,
+    Tuple,
+)
 
 from aiogram import (
     Bot,
@@ -8,7 +14,6 @@
     types,
 )
 from aiogram.filters import Command
-from typing import Optional
 
 from bot.middlewares.auth_middleware import AuthorizationMiddleware
 from bot.middlewares.error_middleware import ErrorHandlerMiddleware
@@ -20,10 +25,10 @@
 router = Router()
 dis = Dispatcher()
 
-last_manual_clip = {}  # Dictionary to store the last manual clip per chat ID
+last_manual_clip: Dict[int, json] = {}
 
 
-def minutes_str_to_seconds(time_str) -> Optional[float]:
+def minutes_str_to_seconds(time_str: str) -> Optional[float]:
     """ Convert time string in the format MM:SS.ms to seconds """
     try:
         minutes, seconds = time_str.split(':')
@@ -34,7 +39,7 @@ def minutes_str_to_seconds(time_str) -> Optional[float]:
         return None
 
 
-def adjust_episode_number(absolute_episode) -> Optional[tuple[int, int]]:
+def adjust_episode_number(absolute_episode: int) -> Optional[Tuple[int, int]]:
     """ Adjust the absolute episode number to season and episode format """
     season = (absolute_episode - 1) // 13 + 1
     episode = (absolute_episode - 1) % 13 + 1

bot/handlers/subscription_status.py

@@ -1,5 +1,9 @@
 from datetime import date
 import logging
+from typing import (
+    Optional,
+    Tuple,
+)
 
 from aiogram import (
     Bot,
@@ -8,7 +12,6 @@
     types,
 )
 from aiogram.filters import Command
-from typing import Optional
 
 from bot.middlewares.auth_middleware import AuthorizationMiddleware
 from bot.middlewares.error_middleware import ErrorHandlerMiddleware
@@ -20,7 +23,7 @@
 
 class UserManager:
     @staticmethod
-    async def get_subscription_status(username: str) -> Optional[tuple[date, int]]: # TO DO: Change return type
+    async def get_subscription_status(username: str) -> Optional[Tuple[date, int]]:
         subscription_end = await DatabaseManager.get_user_subscription(username)
         if subscription_end is None:
             return None

bot/main.py

@@ -1,6 +1,9 @@
 import asyncio
+from asyncio.selector_events import BaseSelectorEventLoop  # pylint: disable=unused-import
 import logging
+from logging import LogRecord
 import os
+from typing import Optional
 
 from aiogram import (
     Bot,
@@ -9,26 +12,26 @@
 from aiogram.fsm.storage.memory import MemoryStorage
 
 from bot.handlers import register_handlers
-from bot.middlewares.auth_middleware import AuthorizationMiddleware  # Import AuthorizationMiddleware
-from bot.middlewares.error_middleware import ErrorHandlerMiddleware  # Import ErrorHandlerMiddleware
+from bot.middlewares.auth_middleware import AuthorizationMiddleware
+from bot.middlewares.error_middleware import ErrorHandlerMiddleware
 from bot.settings import settings
 from bot.utils.database import DatabaseManager
 
-# Configure logging
+# TODO loglevel from env
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
 
 class DBLogHandler(logging.Handler):
     def __init__(self) -> None:
         super().__init__()
-        self.loop = None
+        self.loop: Optional[asyncio.selector_events.BaseSelectorEventLoop] = None
 
-    def emit(self, record) -> None:
+    def emit(self, record: LogRecord) -> None:
         if self.loop is not None:
             self.loop.create_task(self.log_to_db(record))
 
-    async def log_to_db(self, record) -> None:
+    async def log_to_db(self, record: LogRecord) -> None:
         log_message = self.format(record)
         await DatabaseManager.log_system_message(record.levelname, log_message)
 
@@ -38,8 +41,8 @@ async def log_to_db(self, record) -> None:
 dp = Dispatcher(storage=MemoryStorage())
 
 # Add middlewares
-dp.update.middleware(AuthorizationMiddleware())  # Register AuthorizationMiddleware
-dp.update.middleware(ErrorHandlerMiddleware())  # Register ErrorHandlerMiddleware
+dp.update.middleware(AuthorizationMiddleware())
+dp.update.middleware(ErrorHandlerMiddleware())
 
 
 async def on_startup() -> None:
@@ -69,8 +72,7 @@ async def main() -> None:
 
 
 if __name__ == "__main__":
-    loop = asyncio.get_event_loop()
     db_log_handler = DBLogHandler()
-    db_log_handler.loop = loop
+    db_log_handler.loop = asyncio.get_event_loop()
     logging.getLogger().addHandler(db_log_handler)
     asyncio.run(main())