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", "types-PyYAML"]
 
 # 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,42 @@ 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.*",
+    "yaml",
+    "pytz",
+    "pytest",
+]
+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", "union-attr", "operator", "index"]
+
+[[tool.mypy.overrides]]
+module = ["tests.*"]
+disable_error_code = ["arg-type", "union-attr", "misc", "return-value", "assignment", "typeddict-item", "has-type", "attr-defined", "dict-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,call-arg]
             else:
-                query_func(query)  # type: ignore
+                query_func(query)  # type: ignore[operator,call-arg]
         except sqlite3.Error as ex:
             logger.error("DbManager.%s(): %s", error_str, ex)
 
@@ -181,7 +181,7 @@ def count_from(cls, table_name: str, select: str = "*", where: str = "", where_a
         conn.commit()
         cur.close()
         conn.close()
-        return query_result[0]["number"] if len(query_result) > 0 else None
+        return query_result[0]["number"] if len(query_result) > 0 else 0
 
     @classmethod
     def insert_into(cls, table_name: str, values: tuple, columns: tuple | str = "", multiple_rows: bool = False):

src/spotted/data/pending_post.py

@@ -43,6 +43,7 @@ def create(
         Returns:
             instance of the class
         """
+        assert user_message.from_user is not None
         user_id = user_message.from_user.id
         u_message_id = user_message.message_id
         date = datetime.now(tz=timezone.utc)
@@ -141,13 +142,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 +201,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/debug/log_manager.py

@@ -25,7 +25,9 @@ async def error_handler(update: Update, context: CallbackContext):
     """
     logger.error(msg="Exception while handling an update:", exc_info=context.error)
 
-    traceback_list = traceback.format_exception(None, context.error, context.error.__traceback__)
+    traceback_list = traceback.format_exception(
+        None, context.error, context.error.__traceback__ if context.error else None
+    )
     traceback_str = "".join(traceback_list)
 
     min_traceback_list = [line for line in traceback_list if "modules" in line]
@@ -90,9 +92,9 @@ async def log_message(update: Update, _: CallbackContext):
                 message = (
                     f"\n___ID MESSAGE:  {str(update.message.message_id)} ____\n"
                     "___INFO USER___\n"
-                    f"user_id:  {str(user.id)}\n"
-                    f"user_name:  {str(user.username)}\n"
-                    f"user_first_lastname: {str(user.first_name)} {str(user.last_name)}\n"
+                    f"user_id:  {str(user.id if user else 'N/A')}\n"
+                    f"user_name:  {str(user.username if user else 'N/A')}\n"
+                    f"user_first_lastname: {str(user.first_name if user else 'N/A')} {str(user.last_name if user else 'N/A')}\n"
                     "___INFO CHAT___\n"
                     f"chat_id:  {str(chat.id)}\n"
                     f"chat_type:  {str(chat.type)}\n"

src/spotted/handlers/__init__.py

@@ -170,6 +170,7 @@ def add_jobs(app: Application):
     Args:
         app: supplied application
     """
+    assert app.job_queue is not None
     app.job_queue.run_daily(clean_pending_job, time=time(hour=5, tzinfo=utc))  # run each day at 05:00 utc
     app.job_queue.run_daily(db_backup_job, time=time(hour=5, tzinfo=utc))  # run each day at 05:00 utc
     app.job_queue.run_daily(clean_muted_users, time=time(hour=5, tzinfo=utc))  # run each day at 05:00 utc

src/spotted/handlers/purge.py

@@ -35,7 +35,7 @@ async def purge_cmd(update: Update, context: CallbackContext):
                     message_id=published_post["c_message_id"],
                     disable_notification=True,
                 )
-                message.delete()
+                await message.delete()
             except Exception:  # pylint: disable=broad-except
                 lost_posts += 1
                 sleep(10)

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

@@ -71,6 +71,16 @@ async def report_cmd(update: Update, context: CallbackContext) -> int:
     return ConversationState.REPORTING_USER.value
 
 
+async def _report_user_invalid_username(info: EventInfo) -> int:
+    """Sends the invalid username error message and returns the current state."""
+    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
+
+
 async def report_user_msg(update: Update, context: CallbackContext) -> int:
     """Handles the reply to the /report command after sent the @username.
     Checks the the user wants to report, and goes to ask the reason
@@ -83,14 +93,11 @@ 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:
+        return await _report_user_invalid_username(info)
     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(
-            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
+        return await _report_user_invalid_username(info)
 
     if context.user_data is None:
         return ConversationState.END.value

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