ci: enforce static checks mypy, flake8

.flake8

@@ -0,0 +1,17 @@
+[flake8]
+exclude =
+    .git,
+    __pycache__,
+    .venv,
+    venv,
+    env,
+    .env,
+    build,
+    dist,
+    *.egg-info,
+    .tox,
+    .mypy_cache,
+    .pytest_cache,
+    htmlcov
+max-line-length = 127
+max-complexity = 10

.github/workflows/lint.yml

@@ -40,9 +40,15 @@ jobs:
       - name: Lint with pylint
         run: pylint src tests
 
-      # Removed because coroutines :(
-      # - name: Lint with mypy
-      #   run: mypy src tests
+      - name: Lint with flake8
+        run: |
+          # stop the build if there are Python syntax errors or undefined names
+          flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
+          # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
+          flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
+
+      - name: Lint with mypy
+        run: mypy src tests
 
       - name: Lint with isort
         run: isort --check-only --diff src tests

pyproject.toml

@@ -50,7 +50,7 @@ dynamic = ['version']
 # e.g. `pip install telegram-spotted-dmi-bot[test]`
 [project.optional-dependencies]
 test = ["pytest", "pytest-asyncio", "pytest-cov", "pytest-mock"]
-lint = ["pylint", "black", "isort"]
+lint = ["pylint", "black", "isort", "flake8", "mypy"]
 
 # URLs of the project
 [project.urls]
@@ -107,6 +107,7 @@ disable = [
     "missing-module-docstring",
     "too-few-public-methods",
     "too-many-positional-arguments",
+    "line-too-long",
 ]
 
 [tool.pylint.format]
@@ -117,3 +118,39 @@ max-args = 7
 
 [tool.pylint.similarities]
 min-similarity-lines = 6
+
+# Mypy configuration (static type checker)
+[tool.mypy]
+python_version = "3.10"
+warn_return_any = false
+warn_unused_configs = true
+disallow_untyped_defs = false
+check_untyped_defs = true
+no_implicit_optional = true
+warn_redundant_casts = true
+warn_unused_ignores = false
+warn_no_return = true
+strict_equality = true
+explicit_package_bases = true
+mypy_path = "src"
+namespace_packages = true
+
+[[tool.mypy.overrides]]
+module = [
+    "telegram.*",
+    "telegram_api",
+    "telegram_simulator",
+    "pyzipper.*",
+    "pytest_asyncio.*",
+    "APScheduler.*",
+]
+ignore_missing_imports = true
+
+# Handlers can assume certain values are not None in their context
+[[tool.mypy.overrides]]
+module = ["spotted.handlers.*"]
+disable_error_code = ["arg-type"]
+
+[[tool.mypy.overrides]]
+module = ["tests.*"]
+disable_error_code = ["arg-type", "union-attr", "misc", "return-value", "assignment", "typeddict-item"]

src/spotted/__main__.py

@@ -1,7 +1,7 @@
 """Main module"""
 
 import argparse
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, cast
 
 from spotted import run_bot
 from spotted.data import Config
@@ -30,7 +30,7 @@ def parse_args() -> "SpottedArgs":
     parser.add_argument("--version", action="version", version=__version__)
     parser.add_argument("-s", "--settings", help="Path to the settings file", default="settings.yaml")
     parser.add_argument("-a", "--autoreplies", help="Path to the autoreplies file", default="autoreplies.yaml")
-    return parser.parse_args()
+    return cast("SpottedArgs", parser.parse_args())
 
 
 def main():

src/spotted/data/__init__.py

@@ -11,6 +11,19 @@
 from .report import Report
 from .user import User
 
+__all__ = [
+    "Application",
+    "Config",
+    "get_abs_path",
+    "read_md",
+    "DbManager",
+    "PendingPost",
+    "PostData",
+    "PublishedPost",
+    "Report",
+    "User",
+]
+
 
 def init_db():
     """Initialize the database.

src/spotted/data/config.py

@@ -36,6 +36,8 @@
     "max_n_warns",
     "warn_expiration_days",
     "mute_default_duration_days",
+    "autoreplies_per_page",
+    "reject_after_autoreply",
 ]
 SettingsKeysType = Literal[SettingsKeys, SettingsPostKeys, SettingsDebugKeys]
 AutorepliesKeysType = Literal["autoreplies"]
@@ -46,8 +48,8 @@
 class Config:
     """Configurations"""
 
-    DEFAULT_SETTINGS_PATH = os.path.join(resources.files("spotted"), "config", "yaml", "settings.yaml")
-    DEFAULT_AUTOREPLIES_PATH = os.path.join(resources.files("spotted"), "config", "yaml", "autoreplies.yaml")
+    DEFAULT_SETTINGS_PATH = str(resources.files("spotted") / "config" / "yaml" / "settings.yaml")
+    DEFAULT_AUTOREPLIES_PATH = str(resources.files("spotted") / "config" / "yaml" / "autoreplies.yaml")
     __instance: "Config | None" = None
 
     SETTINGS_PATH = "settings.yaml"
@@ -219,7 +221,7 @@ def __load_configuration(cls, path: str) -> dict:
         Returns:
             configuration dictionary
         """
-        conf = {}
+        conf: dict[str, Any] = {}
         if os.path.exists(path):
             with open(path, "r", encoding="utf-8") as conf_file:
                 conf = cls.__merge_settings(conf, yaml.load(conf_file, Loader=yaml.SafeLoader))
@@ -242,7 +244,9 @@ def __read_env_settings(self):
                         new_vars[match.group(1).lower()] = match.group(2)
 
         for key in os.environ:
-            new_vars[key.lower()] = os.getenv(key)
+            env_val = os.getenv(key)
+            if env_val is not None:
+                new_vars[key.lower()] = env_val
 
         for key, value in new_vars.items():
             if key.startswith("post_"):

src/spotted/data/data_reader.py

@@ -1,6 +1,5 @@
 """Read data from files"""
 
-import os
 from importlib import resources
 
 from telegram.helpers import escape_markdown
@@ -17,7 +16,11 @@ def get_abs_path(*root_file_path: str) -> str:
     Returns:
         corresponding abs path
     """
-    return os.path.join(resources.files("spotted"), *root_file_path)
+    base_path = resources.files("spotted")
+    result_path = base_path
+    for part in root_file_path:
+        result_path = result_path / part
+    return str(result_path)
 
 
 def read_file(*root_file_path: str) -> str:

src/spotted/data/db_manager.py

@@ -53,9 +53,9 @@ def __query_execute(
 
         try:
             if args:
-                query_func(query, args)
+                query_func(query, args)  # type: ignore[operator]
             else:
-                query_func(query)  # type: ignore
+                query_func(query)  # type: ignore[operator]
         except sqlite3.Error as ex:
             logger.error("DbManager.%s(): %s", error_str, ex)
 

src/spotted/data/pending_post.py

@@ -141,13 +141,21 @@ def get_all(admin_group_id: int, before: datetime | None = None) -> list["Pendin
         pending_posts = []
         for post in pending_posts_id:
             g_message_id = int(post["g_message_id"])
-            pending_posts.append(PendingPost.from_group(admin_group_id=admin_group_id, g_message_id=g_message_id))
+            pending_post = PendingPost.from_group(admin_group_id=admin_group_id, g_message_id=g_message_id)
+            if pending_post is not None:
+                pending_posts.append(pending_post)
         return pending_posts
 
     def save_post(self) -> "PendingPost":
         """Saves the pending_post in the database"""
-        columns = ("user_id", "u_message_id", "g_message_id", "admin_group_id", "message_date")
-        values = (self.user_id, self.u_message_id, self.g_message_id, self.admin_group_id, self.date)
+        columns: tuple[str, ...] = ("user_id", "u_message_id", "g_message_id", "admin_group_id", "message_date")
+        values: tuple[int | datetime | str, ...] = (
+            self.user_id,
+            self.u_message_id,
+            self.g_message_id,
+            self.admin_group_id,
+            self.date,
+        )
         if self.credit_username is not None:
             columns += ("credit_username",)
             values += (self.credit_username,)
@@ -192,7 +200,7 @@ def get_list_admin_votes(self, vote: "bool | None" = None) -> "list[int] | list[
         """
 
         where = "g_message_id = %s and admin_group_id = %s"
-        where_args = (self.g_message_id, self.admin_group_id)
+        where_args: tuple[int | bool, ...] = (self.g_message_id, self.admin_group_id)
 
         if vote is not None:
             where += " and is_upvote = %s"

src/spotted/data/report.py

@@ -25,10 +25,10 @@ class Report:
     user_id: int
     admin_group_id: int
     g_message_id: int
-    channel_id: int = None
-    c_message_id: int = None
-    target_username: str = None
-    date: datetime = None
+    channel_id: int | None = None
+    c_message_id: int | None = None
+    target_username: str | None = None
+    date: datetime | None = None
 
     @property
     def minutes_passed(self) -> float:

src/spotted/debug/__init__.py

@@ -1,3 +1,5 @@
 """Modules used during debug"""
 
 from .log_manager import error_handler, log_message, logger
+
+__all__ = ["error_handler", "log_message", "logger"]

src/spotted/handlers/reply.py

@@ -24,7 +24,7 @@ async def reply_cmd(update: Update, context: CallbackContext):
             "seguito da ciò che gli si vuole dire",
         )
         return
-    ### build the reply text from the args
+    # Build the reply text from the args
     reply_text = " ".join(info.args)
     g_message_id = update.message.reply_to_message.message_id
     if (pending_post := PendingPost.from_group(admin_group_id=info.chat_id, g_message_id=g_message_id)) is not None:

src/spotted/handlers/report_spot.py

@@ -103,7 +103,8 @@ async def report_spot_msg(update: Update, context: CallbackContext) -> int:
     channel_id, target_message_id = context.user_data["current_post_reported"].split(",")
 
     await info.bot.forward_message(chat_id=chat_id, from_chat_id=channel_id, message_id=target_message_id)
-    admin_message = await info.bot.sendMessage(chat_id=chat_id, text="🚨🚨 SEGNALAZIONE 🚨🚨\n\n" + info.text)
+    report_text = info.text if info.text is not None else ""
+    admin_message = await info.bot.sendMessage(chat_id=chat_id, text="🚨🚨 SEGNALAZIONE 🚨🚨\n\n" + report_text)
     await info.bot.send_message(
         chat_id=info.chat_id, text="Gli admins verificheranno quanto accaduto. Grazie per la collaborazione!"
     )

src/spotted/handlers/report_user.py

@@ -83,6 +83,13 @@ async def report_user_msg(update: Update, context: CallbackContext) -> int:
         next state of the conversation
     """
     info = EventInfo.from_message(update, context)
+    if info.text is None:
+        await info.bot.send_message(
+            chat_id=info.chat_id,
+            text="Questo tipo di messaggio non è supportato\n"
+            "È consentito solo username telegram. Puoi annullare il processo con /cancel",
+        )
+        return ConversationState.REPORTING_USER.value
     reported_user = info.text.strip()
     if not info.is_valid_message_type or not reported_user.startswith("@") or " " in reported_user:
         await info.bot.send_message(

src/spotted/handlers/sban.py

@@ -19,13 +19,11 @@ async def sban_cmd(update: Update, context: CallbackContext):
     info = EventInfo.from_message(update, context)
     sban_fail = []
     if context.args is None or len(context.args) == 0:  # if no args have been passed
-        banned_users = "\n".join(
+        banned_users_str = "\n".join(
             f"#{idx} ({user.ban_date:%d/%m/%Y %H:%M})" for idx, user in enumerate(User.banned_users())
         )
-        banned_users = "Nessuno" if len(banned_users) == 0 else f"{banned_users}"
-        text = (
-            f"[uso]: /sban <user_id1|#idx> [...(user_id2|#idx)]\nGli utenti attualmente bannati sono:\n{banned_users}"
-        )
+        banned_users_str = "Nessuno" if len(banned_users_str) == 0 else f"{banned_users_str}"
+        text = f"[uso]: /sban <user_id1|#idx> [...(user_id2|#idx)]\nGli utenti attualmente bannati sono:\n{banned_users_str}"
         await info.bot.send_message(chat_id=info.chat_id, text=text)
         return
 

src/spotted/handlers/spot.py

@@ -129,6 +129,8 @@ async def spot_preview_query(update: Update, context: CallbackContext) -> int:
         next state of the conversation
     """
     info = EventInfo.from_callback(update, context)
+    if info.query_data is None:
+        return ConversationState.END.value
     arg = info.query_data.split(",")[1]
     info.user_data["show_preview"] = arg == "accept"
     await info.bot.edit_message_text(
@@ -155,6 +157,8 @@ async def spot_confirm_query(update: Update, context: CallbackContext) -> int:
         next state of the conversation
     """
     info = EventInfo.from_callback(update, context)
+    if info.query_data is None:
+        return ConversationState.END.value
     arg = info.query_data.split(",")[1]
     text = "Qualcosa è andato storto!"
     if arg == "submit":  # if the the user wants to publish the post

src/spotted/handlers/unmute.py

@@ -20,15 +20,13 @@ async def unmute_cmd(update: Update, context: CallbackContext):
     failed_unmute = []
     if context.args is None or len(context.args) == 0:  # if no args have been passed
         if len(User.muted_users()) == 0:
-            muted_users = "Nessuno"
+            muted_users_str = "Nessuno"
         else:
-            muted_users = "\n".join(
+            muted_users_str = "\n".join(
                 f"#{i} (Mute: {user.mute_date:%d/%m/%Y %H:%M} - Exp: {user.mute_expire_date:%d/%m/%Y %H:%M} )"
                 for i, user in enumerate(User.muted_users())
             )
-        text = (
-            f"[uso]: /unmute <user_id1|#idx> [...(user_id2|#idx)]\nGli utenti attualmente mutati sono:\n{muted_users}"
-        )
+        text = f"[uso]: /unmute <user_id1|#idx> [...(user_id2|#idx)]\nGli utenti attualmente mutati sono:\n{muted_users_str}"
         await info.bot.send_message(chat_id=info.chat_id, text=text)
         return
 

src/spotted/scripts/f_crypto.py

@@ -1,5 +1,5 @@
 import argparse
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, cast
 
 from cryptography.fernet import Fernet
 
@@ -41,7 +41,7 @@ def parse_args() -> "DecryptArgs":
     # generate_key
     subparsers.add_parser("generate_key")
 
-    return parser.parse_args()
+    return cast("DecryptArgs", parser.parse_args())
 
 
 def main():

src/spotted/scripts/run_sql.py

@@ -1,5 +1,5 @@
 import argparse
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, cast
 
 from spotted.data import Config, DbManager
 
@@ -25,7 +25,7 @@ def parse_args() -> "RunSQLArgs":
         help="Path to the SQL file. Multiple queries must be separated by a ';'",
     )
     parser.add_argument("db_file", type=str, help="Path to the database file")
-    return parser.parse_args()
+    return cast("RunSQLArgs", parser.parse_args())
 
 
 def main():

src/spotted/utils/__init__.py

@@ -11,3 +11,16 @@
     get_settings_kb,
 )
 from .utils import get_user_by_id_or_index
+
+__all__ = [
+    "conv_cancel",
+    "conv_fail",
+    "EventInfo",
+    "get_approve_kb",
+    "get_confirm_kb",
+    "get_paused_kb",
+    "get_preview_kb",
+    "get_published_post_kb",
+    "get_settings_kb",
+    "get_user_by_id_or_index",
+]