refactor: Move to pre-commit repos, lint with ruff

Signed-off-by: yshalsager <[email protected]>

.pre-commit-config.yaml

@@ -66,3 +66,8 @@ repos:
     rev: 'v1.3.0'  # Use the sha / tag you want to point at
     hooks:
       - id: mypy
+        args:
+          - --install-types
+        additional_dependencies:
+          - SQLAlchemy==2.0.15
+

droos_bot/__init__.py

@@ -1,11 +1,10 @@
-""" Bot initialization """
+"""Bot initialization."""
 import json
 import logging
 from functools import partial
 from logging.handlers import TimedRotatingFileHandler
 from pathlib import Path
 from sys import stderr, stdout
-from typing import Dict
 
 from telegram.constants import ParseMode
 from telegram.ext import ApplicationBuilder, Defaults, PicklePersistence
@@ -21,7 +20,7 @@
 CONFIG = json.loads((PARENT_DIR / "config.json").read_text(encoding="utf-8"))
 BOT_TOKEN = CONFIG["tg_bot_token"]
 TG_BOT_ADMINS = CONFIG["tg_bot_admins"]
-DATA_COLUMNS: Dict[str, str] = CONFIG["data_columns"]
+DATA_COLUMNS: dict[str, str] = CONFIG["data_columns"]
 
 # Logging
 LOG_FILE = PARENT_DIR / "last_run.log"
@@ -62,3 +61,5 @@
     CONFIG["sheet_name"],
     DATA_COLUMNS,
 )
+
+logging.getLogger("httpx").setLevel(logging.WARNING)

droos_bot/__main__.py

@@ -1,4 +1,4 @@
-""" Bot Entry Point """
+"""Bot Entry Point."""
 from droos_bot.bot import main
 
 if __name__ == "__main__":

droos_bot/bot.py

@@ -1,6 +1,4 @@
-"""
-Telegram Bot
-"""
+"""Telegram Bot."""
 from droos_bot import application
 from droos_bot.modules import ALL_MODULES
 from droos_bot.utils.modules_loader import load_modules

droos_bot/db/__init__.py

@@ -1,3 +1,7 @@
-from .models.chat import Chat
-from .models.lecture import Lecture
-from .models.series import Series
+"""Database package entry."""
+
+from droos_bot.db.models.chat import Chat
+from droos_bot.db.models.lecture import Lecture
+from droos_bot.db.models.series import Series
+
+__all__ = ["Chat", "Lecture", "Series"]

droos_bot/db/curd.py

@@ -1,5 +1,3 @@
-from typing import List, Optional, Tuple
-
 from pandas import DataFrame
 from sqlalchemy import Row
 from sqlalchemy.sql.functions import sum
@@ -11,7 +9,7 @@
 
 
 def increment_series_requests(series_info: DataFrame) -> None:
-    series: Optional[Series] = (
+    series: Series | None = (
         session.query(Series)
         .filter(Series.id == series_info.series_slug.item())
         .first()
@@ -24,7 +22,7 @@ def increment_series_requests(series_info: DataFrame) -> None:
 
 
 def increment_lecture_requests(lecture_info: DataFrame) -> None:
-    lecture: Optional[Lecture] = (
+    lecture: Lecture | None = (
         session.query(Lecture).filter(Lecture.id == lecture_info.id.item()).first()
     )
     if lecture:
@@ -48,24 +46,24 @@ def increment_usage(user_id: int) -> None:
     session.commit()
 
 
-def get_chats_count() -> Tuple[int, int]:
+def get_chats_count() -> tuple[int, int]:
     all_chats = session.query(Chat).count()
     active_chats = session.query(Chat).filter(Chat.usage_times > 0).count()
     return all_chats, active_chats
 
 
-def get_usage_count() -> Tuple[int, int, int]:
-    usage_times_row: Optional[Row] = session.query(
+def get_usage_count() -> tuple[int, int, int]:
+    usage_times_row: Row | None = session.query(
         sum(Chat.usage_times).label("usage_times")
     ).first()
     usage_times: int = usage_times_row.usage_times if usage_times_row else 0
-    series_requests_row: Optional[Row] = session.query(
+    series_requests_row: Row | None = session.query(
         sum(Series.requests).label("series_requests")
     ).first()
     series_requests: int = (
         series_requests_row.series_requests if series_requests_row else 0
     )
-    lecture_requests_row: Optional[Row] = session.query(
+    lecture_requests_row: Row | None = session.query(
         sum(Lecture.requests).label("lecture_requests")
     ).first()
     lecture_requests: int = (
@@ -74,13 +72,13 @@ def get_usage_count() -> Tuple[int, int, int]:
     return usage_times, series_requests, lecture_requests
 
 
-def get_top_series() -> List[Series]:
+def get_top_series() -> list[Series]:
     return session.query(Series).order_by(Series.requests.desc()).limit(5).all()
 
 
-def get_top_lectures() -> List[Lecture]:
+def get_top_lectures() -> list[Lecture]:
     return session.query(Lecture).order_by(Lecture.requests.desc()).limit(5).all()
 
 
-def get_all_chats() -> List[Chat]:
+def get_all_chats() -> list[Chat]:
     return session.query(Chat).all()

droos_bot/db/models/chat.py

@@ -5,10 +5,12 @@
 
 class Chat(Base):
     __tablename__ = "chats"
-    id: int = Column(INT(), primary_key=True, autoincrement=True, nullable=False)
+    id: int = Column(  # noqa: A003
+        INT(), primary_key=True, autoincrement=True, nullable=False
+    )
     user_id: int = Column(BIGINT(), unique=True, nullable=False)
     user_name: str = Column(VARCHAR(), nullable=False)
-    type: int = Column(INT(), nullable=False)  # 0=user, 1=group, 2=channel
+    type: int = Column(INT(), nullable=False)  # 0=user, 1=group, 2=channel # noqa: A003
     usage_times: int = Column(INT(), nullable=False, default=0)
 
     def __repr__(self) -> str:

droos_bot/db/models/lecture.py

@@ -5,7 +5,7 @@
 
 class Lecture(Base):
     __tablename__ = "lectures"
-    id: str = Column(VARCHAR, primary_key=True, nullable=False)
+    id: str = Column(VARCHAR, primary_key=True, nullable=False)  # noqa: A003
     requests: int = Column(Integer, nullable=False, default=0)
 
     def __repr__(self) -> str:

droos_bot/db/models/series.py

@@ -5,7 +5,7 @@
 
 class Series(Base):
     __tablename__ = "series"
-    id: str = Column(VARCHAR, primary_key=True, nullable=False)
+    id: str = Column(VARCHAR, primary_key=True, nullable=False)  # noqa: A003
     requests: int = Column(Integer, nullable=False, default=0)
 
     def __repr__(self) -> str:

droos_bot/db/session.py

@@ -1,4 +1,4 @@
-"""Database initialization"""
+"""Database initialization."""
 from sqlalchemy import create_engine
 from sqlalchemy.orm import scoped_session, sessionmaker
 

droos_bot/gsheet/spreadsheet.py

@@ -1,6 +1,5 @@
 import logging
 from pathlib import Path
-from typing import Dict
 
 from gspread_pandas import Client, Spread, conf
 from pandas import DataFrame
@@ -14,16 +13,16 @@ def __init__(
         service_account: str,
         sheet_id: str,
         sheet_name: str,
-        data_columns: Dict[str, str],
+        data_columns: dict[str, str],
     ) -> None:
         self._client: Client = Client(
             config=conf.get_config(
                 conf_dir=str(Path(service_account).parent), file_name=service_account
             )
         )
         self.worksheet: Spread = Spread(sheet_id, sheet=sheet_name, client=self._client)
-        # self.table_headers: List[str] = self._sheet.sheet.row_values(1)
-        # self.items: List[Dict[str, str]] = self._sheet.sheet.get_all_records()[1:]
+        # self.table_headers: list[str] = self._sheet.sheet.row_values(1)
+        # self.items: list[Dict[str, str]] = self._sheet.sheet.get_all_records()[1:]
         # self.df = self.worksheet.sheet_to_df()
         self.df: DataFrame = self.worksheet.sheet_to_df().iloc[1:]
         # self.series = self.df.series.unique().tolist()

droos_bot/modules/__init__.py

@@ -1,4 +1,4 @@
-"""Modules loader"""
+"""Modules loader."""
 from pathlib import Path
 
 from droos_bot.utils.modules_loader import get_modules

droos_bot/modules/admin_files.py

@@ -1,6 +1,5 @@
-"""
-Admin files handler module.
-"""
+"""Admin files handler module."""
+from typing import cast
 
 from telegram import Audio, Document, Update, Video, Voice
 from telegram.ext import (
@@ -23,15 +22,15 @@ async def start_receiving(_: Update, __: CallbackContext) -> int:
 
 
 async def stop_receiving(_: Update, __: CallbackContext) -> int:
-    return ConversationHandler.END
+    return cast(int, ConversationHandler.END)
 
 
 async def files_receiver(update: Update, _: ContextTypes.DEFAULT_TYPE) -> None:
-    """Sends explanation on how to use the bot."""
+    """Send explanation on how to use the bot."""
     assert update.effective_message is not None
     if not isinstance(
         update.effective_message.effective_attachment,
-        (Document, Audio, Video, Voice, list),
+        Document | Audio | Video | Voice | list,
     ):
         message = f"Ͱ`{update.effective_message.text_html_urled}`"
     else:

droos_bot/modules/broadcast.py

@@ -1,6 +1,4 @@
-"""
-broadcast module.
-"""
+"""broadcast module."""
 import logging
 from time import sleep
 

droos_bot/modules/droos.py

@@ -1,9 +1,6 @@
-"""
-Droos handler module.
-"""
+"""Droos handler module."""
 from functools import partial
 from math import ceil
-from typing import Optional, Tuple, Union
 
 from pandas import DataFrame, Series
 from telegram import CallbackQuery, InlineKeyboardButton, InlineKeyboardMarkup, Update
@@ -32,7 +29,7 @@
 }
 
 
-def get_lecture_message_text(item: Union[Series, DataFrame]) -> str:
+def get_lecture_message_text(item: Series | DataFrame) -> str:
     if isinstance(item.series, str):
         series_text, lecture = item.series, item.lecture
     else:
@@ -42,7 +39,7 @@ def get_lecture_message_text(item: Union[Series, DataFrame]) -> str:
 
 def get_data(
     data: Series, data_column_id: str, page: int = 1
-) -> Tuple[str, InlineKeyboardMarkup]:
+) -> tuple[str, InlineKeyboardMarkup]:
     text = "*اختر ما تريد من القائمة:*"
     paginator = InlineKeyboardPaginator(
         ceil(len(data) / page_size),
@@ -89,7 +86,7 @@ async def data_callback_handler(update: Update, _: ContextTypes.DEFAULT_TYPE) ->
 
 @tg_exceptions_handler
 async def droos_handler(update: Update, _: ContextTypes.DEFAULT_TYPE) -> None:
-    """Sends Droos list."""
+    """Send Droos list."""
     query: CallbackQuery = update.callback_query
     await query.answer()
     # get series query
@@ -117,7 +114,7 @@ async def droos_handler(update: Update, _: ContextTypes.DEFAULT_TYPE) -> None:
         return
     if data_column_id != "series":
         # Handle author/category
-        series_data: Series = data.groupby(f"series_slug")["series"].unique()
+        series_data: Series = data.groupby("series_slug")["series"].unique()
         text, reply_markup = get_data(series_data, "series", page=page_idx)
         await query.edit_message_text(
             text=text,
@@ -152,7 +149,7 @@ async def droos_handler(update: Update, _: ContextTypes.DEFAULT_TYPE) -> None:
 @analysis
 async def get_lecture_callback_handler(
     update: Update, _: CallbackContext
-) -> Optional[Series]:
+) -> Series | None:
     query = update.callback_query
     await query.answer()
     __, data, lecture_id = query.data.split("|")

droos_bot/modules/feedback_and_files.py

@@ -1,6 +1,4 @@
-"""
-Feedback and files handler module.
-"""
+"""Feedback and files handler module."""
 from telegram import Update
 from telegram.constants import MessageLimit, ParseMode
 from telegram.ext import (

droos_bot/modules/restart.py

@@ -1,5 +1,5 @@
-""" Bot restart module"""
-import pickle
+"""Bot restart module."""
+import json
 from os import execl
 from pathlib import Path
 from sys import executable
@@ -12,13 +12,13 @@
 
 
 async def restart(update: Update, _: ContextTypes.DEFAULT_TYPE) -> None:
-    """restarts the bot."""
+    """Restarts the bot."""
     restart_message = await update.message.reply_text(
         "`Restarting, please wait...`",
     )
     chat_info = {"chat": restart_message.chat_id, "message": restart_message.message_id}
-    Path(f"{PARENT_DIR}/restart.pickle").write_bytes(pickle.dumps(chat_info))
-    execl(executable, executable, "-m", __package__.split(".")[0])
+    Path(f"{PARENT_DIR}/restart.json").write_text(json.dumps(chat_info))
+    execl(executable, executable, "-m", __package__.split(".")[0])  # noqa: S606
 
 
 filter_bot_admin = FilterBotAdmin()

droos_bot/modules/search.py

@@ -1,7 +1,5 @@
-"""
-Data search module
-"""
-from typing import Optional
+"""Data search module."""
+from typing import cast
 
 from telegram import Update
 from telegram.ext import (
@@ -32,9 +30,7 @@ async def search_handler(update: Update, _: ContextTypes.DEFAULT_TYPE) -> int:
 
 
 @tg_exceptions_handler
-async def search_for_text(
-    update: Update, _: ContextTypes.DEFAULT_TYPE
-) -> Optional[int]:
+async def search_for_text(update: Update, _: ContextTypes.DEFAULT_TYPE) -> int | None:
     assert update.effective_message is not None
     search_text = update.effective_message.text.strip()
     match = (
@@ -55,15 +51,15 @@ async def search_for_text(
         reply_to_message_id=update.effective_message.message_id,
     )
     await cancel_search_handler(update, _)
-    return ConversationHandler.END
+    return cast(int, ConversationHandler.END)
 
 
 async def cancel_search_handler(update: Update, _: ContextTypes.DEFAULT_TYPE) -> int:
     await update.message.reply_text(
         "يمكنك متابعة استخدام البوت من خلال الأزرار الظاهرة بالأسفل",
         reply_markup=main_keyboard,
     )
-    return ConversationHandler.END
+    return cast(int, ConversationHandler.END)
 
 
 search_conversation_handler = ConversationHandler(

droos_bot/modules/stats.py

@@ -1,5 +1,4 @@
-# """ Bot stats module"""
-from typing import List
+"""Bot stats module."""
 
 from pandas import DataFrame
 from telegram import Update
@@ -21,15 +20,15 @@ async def stats(update: Update, _: ContextTypes.DEFAULT_TYPE) -> None:
     stats_message = await update.message.reply_text("جاري تحضير الإحصائيات…")
     all_chats, active_chats = get_chats_count()
     usage_times, series_requests, lecture_requests = get_usage_count()
-    top_series: List[Series] = get_top_series()
-    top_lectures: List[Lecture] = get_top_lectures()
+    top_series: list[Series] = get_top_series()
+    top_lectures: list[Lecture] = get_top_lectures()
 
     message = (
-        f"**المستخدمون الحاليون**: {str(active_chats)}\n"
-        f"**كل المستخدمون**: {str(all_chats)}\n"
-        f"**إجمالي مرات الاستخدام**: {str(usage_times)}\n"
-        f"**إجمالي الملفات المرسلة**: {str(series_requests)}\n"
-        f"**إجمالي الدروس المطلوبة**: {str(lecture_requests)}\n\n"
+        f"**المستخدمون الحاليون**: {active_chats!s}\n"
+        f"**كل المستخدمون**: {all_chats!s}\n"
+        f"**إجمالي مرات الاستخدام**: {usage_times!s}\n"
+        f"**إجمالي الملفات المرسلة**: {series_requests!s}\n"
+        f"**إجمالي الدروس المطلوبة**: {lecture_requests!s}\n\n"
         f"**أكثر السلاسل طلبًا**:\n"
         f"$top_series\n"
         f"**أكثر الدروس طلبًا**:\n"