Initial plan

Co-authored-by: renovate[bot] <29139614+renovate[bot]@users.noreply.github.com>

.github/renovate.json

@@ -1,5 +1,6 @@
 {
     "extends": [
-        "config:base"
+        "config:base",
+        "github>nikdoof/renovate-config:python"
     ]
 }

.github/workflows/lint.yaml

@@ -16,6 +16,10 @@ jobs:
         python-version: ["3.9"]
     steps:
       - uses: actions/checkout@v4
+      - name: Install Task
+        uses: arduino/setup-task@v2
+        with:
+          version: 3.x
       - name: Setup Python
         uses: actions/setup-python@v5
         with:
@@ -27,4 +31,4 @@ jobs:
           virtualenvs-in-project: true
       - name: Lint with ruff
         run: |
-          make lint
+          task python:lint

.github/workflows/release-chart.yaml

@@ -26,7 +26,7 @@ jobs:
           version: v3.6.3
 
       - name: Run chart-releaser
-        uses: helm/chart-releaser-action@v1.6.0
+        uses: helm/chart-releaser-action@v1.7.0
         env:
           CR_TOKEN: "${{ secrets.GITHUB_TOKEN }}"
           CR_RELEASE_NAME_TEMPLATE: "smsbot-helm-chart-{{ .Version }}"

.github/workflows/release.yaml

@@ -13,6 +13,8 @@ jobs:
 
       - name: Setup Python
         uses: actions/setup-python@v5
+        with:
+          python-version: "3.9"
 
       - name: Install Poetry
         uses: snok/install-poetry@v1

Dockerfile

@@ -17,5 +17,5 @@ FROM base AS runtime
 
 COPY --from=builder /runtime /runtime
 ENV PATH=/runtime/bin:$PATH
-EXPOSE 8000/tcp
+EXPOSE 80/tcp
 CMD ["smsbot"]

Makefile

@@ -1,7 +0,0 @@
-.venv:
-	python3 -m pip install poetry
-	python3 -m poetry install --with github
-
-lint: .venv
-	python3 -m poetry run ruff check --output-format=github --select=E9,F63,F7,F82 --target-version=py37 .
-	python3 -m poetry run ruff check --output-format=github --target-version=py37 .

Taskfile.yaml

@@ -0,0 +1,17 @@
+version: 3
+tasks:
+  python:env:
+    cmds:
+      - poetry install --with=dev,github
+  python:lint:
+    desc: Lint Python files
+    deps:
+      - python:env
+    cmds:
+      - poetry run ruff check --output-format=github --select=E9,F63,F7,F82 --target-version=py39 .
+      - poetry run ruff check --output-format=github --target-version=py39 .
+
+  docker:build:
+    desc: Build the container using Docker
+    cmds:
+      - docker build . -t smsbot:latest

pyproject.toml

@@ -27,7 +27,7 @@ build-backend = "poetry.core.masonry.api"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^8.0.0"
-ruff = "^0.9.0"
+ruff = "^0.12.0"
 pytest-mock = "^3.12.0"
 pytest-flask = "^1.3.0"
 

smsbot/__main__.py

@@ -1,4 +1,4 @@
 from smsbot.cli import main
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     main()

smsbot/cli.py

@@ -3,23 +3,32 @@ import logging
 import os
 
 from smsbot.telegram import TelegramSmsBot
-from smsbot.webhook_handler import TwilioWebhookHandler
 from smsbot.utils import get_smsbot_version
+from smsbot.webhook_handler import TwilioWebhookHandler
 
 
 def main():
-    parser = argparse.ArgumentParser('smsbot')
-    parser.add_argument('--listen-host', default=os.environ.get('SMSBOT_LISTEN_HOST') or '0.0.0.0')
-    parser.add_argument('--listen-port', default=os.environ.get('SMSBOT_LISTEN_PORT') or '80')
-    parser.add_argument('--telegram-bot-token', default=os.environ.get('SMSBOT_TELEGRAM_BOT_TOKEN'))
-    parser.add_argument('--owner-id', default=os.environ.get('SMSBOT_OWNER_ID'))
-    parser.add_argument('--default-subscribers', default=os.environ.get('SMSBOT_DEFAULT_SUBSCRIBERS'))
-    parser.add_argument('--log-level', default='INFO')
+    parser = argparse.ArgumentParser("smsbot")
+    parser.add_argument(
+        "--listen-host", default=os.environ.get("SMSBOT_LISTEN_HOST") or "0.0.0.0"
+    )
+    parser.add_argument(
+        "--listen-port", default=os.environ.get("SMSBOT_LISTEN_PORT") or "80"
+    )
+    parser.add_argument(
+        "--telegram-bot-token", default=os.environ.get("SMSBOT_TELEGRAM_BOT_TOKEN")
+    )
+    parser.add_argument("--owner-id", default=os.environ.get("SMSBOT_OWNER_ID"))
+    parser.add_argument(
+        "--default-subscribers", default=os.environ.get("SMSBOT_DEFAULT_SUBSCRIBERS")
+    )
+    parser.add_argument("--log-level", default="INFO")
     args = parser.parse_args()
 
+    # TODO: Replace for Py >=3.11
     logging.basicConfig(level=logging.getLevelName(args.log_level))
-    logging.info('smsbot v%s', get_smsbot_version())
-    logging.debug('Arguments: %s', args)
+    logging.info("smsbot v%s", get_smsbot_version())
+    logging.debug("Arguments: %s", args)
 
     # Start bot
     telegram_bot = TelegramSmsBot(args.telegram_bot_token)
@@ -28,11 +37,11 @@ def main():
     if args.owner_id:
         telegram_bot.set_owner(args.owner_id)
     else:
-        logging.warning('No Owner ID is set, which is not a good idea...')
+        logging.warning("No Owner ID is set, which is not a good idea...")
 
     # Add default subscribers
     if args.default_subscribers:
-        for chat_id in args.default_subscribers.split(','):
+        for chat_id in args.default_subscribers.split(","):
             telegram_bot.add_subscriber(chat_id)
 
     telegram_bot.start()

smsbot/telegram.py

@@ -5,33 +5,40 @@ from telegram.ext import CommandHandler, Updater
 
 from smsbot.utils import get_smsbot_version
 
-REQUEST_TIME = Summary('telegram_request_processing_seconds', 'Time spent processing request')
-COMMAND_COUNT = Counter('telegram_command_count', 'Total number of commands processed')
+REQUEST_TIME = Summary(
+    "telegram_request_processing_seconds", "Time spent processing request"
+)
+COMMAND_COUNT = Counter("telegram_command_count", "Total number of commands processed")
 
 
 class TelegramSmsBot(object):
-
-    def __init__(self, telegram_token, allow_subscribing=False, owner=None, subscribers=None):
+    def __init__(
+        self, telegram_token, allow_subscribing=False, owner=None, subscribers=None
+    ):
         self.logger = logging.getLogger(self.__class__.__name__)
         self.bot_token = telegram_token
         self.subscriber_ids = subscribers or []
         self.set_owner(owner)
 
         self.updater = Updater(self.bot_token, use_context=True)
-        self.updater.dispatcher.add_handler(CommandHandler('help', self.help_handler))
-        self.updater.dispatcher.add_handler(CommandHandler('start', self.help_handler))
+        self.updater.dispatcher.add_handler(CommandHandler("help", self.help_handler))
+        self.updater.dispatcher.add_handler(CommandHandler("start", self.help_handler))
 
         if allow_subscribing:
-            self.updater.dispatcher.add_handler(CommandHandler('subscribe', self.subscribe_handler))
-            self.updater.dispatcher.add_handler(CommandHandler('unsubscribe', self.unsubscribe_handler))
+            self.updater.dispatcher.add_handler(
+                CommandHandler("subscribe", self.subscribe_handler)
+            )
+            self.updater.dispatcher.add_handler(
+                CommandHandler("unsubscribe", self.unsubscribe_handler)
+            )
 
         self.updater.dispatcher.add_error_handler(self.error_handler)
 
     def start(self):
-        self.logger.info('Starting bot...')
+        self.logger.info("Starting bot...")
         self.updater.start_polling()
         self.bot = self.updater.bot
-        self.logger.info('Bot Ready')
+        self.logger.info("Bot Ready")
 
     def stop(self):
         self.updater.stop()
@@ -39,43 +46,57 @@ class TelegramSmsBot(object):
     @REQUEST_TIME.time()
     def help_handler(self, update, context):
         """Send a message when the command /help is issued."""
-        self.logger.info('/help command received in chat: %s', update.message.chat)
+        self.logger.info("/help command received in chat: %s", update.message.chat)
 
         commands = []
         for command in self.updater.dispatcher.handlers[0]:
-            commands.extend(['/{0}'.format(cmd) for cmd in command.command])
+            commands.extend(["/{0}".format(cmd) for cmd in command.command])
 
-        update.message.reply_markdown('Smsbot v{0}\n\n{1}'.format(get_smsbot_version(), '\n'.join(commands)))
+        update.message.reply_markdown(
+            "Smsbot v{0}\n\n{1}".format(get_smsbot_version(), "\n".join(commands))
+        )
         COMMAND_COUNT.inc()
 
     @REQUEST_TIME.time()
     def subscribe_handler(self, update, context):
-        self.logger.info('/subscribe command received')
-        if update.message.chat['id'] not in self.subscriber_ids:
-            self.logger.info('{0} subscribed'.format(update.message.chat['username']))
-            self.subscriber_ids.append(update.message.chat['id'])
-            self.send_owner('{0} has subscribed'.format(update.message.chat['username']))
-            update.message.reply_markdown('You have been subscribed to SMS notifications')
+        self.logger.info("/subscribe command received")
+        if update.message.chat["id"] not in self.subscriber_ids:
+            self.logger.info("{0} subscribed".format(update.message.chat["username"]))
+            self.subscriber_ids.append(update.message.chat["id"])
+            self.send_owner(
+                "{0} has subscribed".format(update.message.chat["username"])
+            )
+            update.message.reply_markdown(
+                "You have been subscribed to SMS notifications"
+            )
         else:
-            update.message.reply_markdown('You are already subscribed to SMS notifications')
+            update.message.reply_markdown(
+                "You are already subscribed to SMS notifications"
+            )
         COMMAND_COUNT.inc()
 
     @REQUEST_TIME.time()
     def unsubscribe_handler(self, update, context):
-        self.logger.info('/unsubscribe command received')
-        if update.message.chat['id'] in self.subscriber_ids:
-            self.logger.info('{0} unsubscribed'.format(update.message.chat['username']))
-            self.subscriber_ids.remove(update.message.chat['id'])
-            self.send_owner('{0} has unsubscribed'.format(update.message.chat['username']))
-            update.message.reply_markdown('You have been unsubscribed to SMS notifications')
+        self.logger.info("/unsubscribe command received")
+        if update.message.chat["id"] in self.subscriber_ids:
+            self.logger.info("{0} unsubscribed".format(update.message.chat["username"]))
+            self.subscriber_ids.remove(update.message.chat["id"])
+            self.send_owner(
+                "{0} has unsubscribed".format(update.message.chat["username"])
+            )
+            update.message.reply_markdown(
+                "You have been unsubscribed to SMS notifications"
+            )
         else:
-            update.message.reply_markdown('You are not subscribed to SMS notifications')
+            update.message.reply_markdown("You are not subscribed to SMS notifications")
         COMMAND_COUNT.inc()
 
     def error_handler(self, update, context):
         """Log Errors caused by Updates."""
         self.logger.warning('Update "%s" caused error "%s"', update, context.error)
-        self.send_owner('Update "%{0}" caused error "{1}"'.format(update, context.error))
+        self.send_owner(
+            'Update "%{0}" caused error "{1}"'.format(update, context.error)
+        )
 
     def send_message(self, message, chat_id):
         self.bot.sendMessage(text=message, chat_id=chat_id)

smsbot/webhook_handler.py

@@ -1,4 +1,3 @@
-
 import os
 from functools import wraps
 
@@ -10,20 +9,23 @@ from werkzeug.middleware.dispatcher import DispatcherMiddleware
 
 from smsbot.utils import get_smsbot_version
 
-REQUEST_TIME = Summary('webhook_request_processing_seconds', 'Time spent processing request')
-MESSAGE_COUNT = Counter('webhook_message_count', 'Total number of messages processed')
-CALL_COUNT = Counter('webhook_call_count', 'Total number of calls processed')
+REQUEST_TIME = Summary("webhook_request_processing_seconds", "Time spent processing request")
+MESSAGE_COUNT = Counter("webhook_message_count", "Total number of messages processed")
+CALL_COUNT = Counter("webhook_call_count", "Total number of calls processed")
 
 
 def validate_twilio_request(func):
     """Validates that incoming requests genuinely originated from Twilio"""
+
     @wraps(func)
     def decorated_function(*args, **kwargs):  # noqa: WPS430
         # Create an instance of the RequestValidator class
-        twilio_token = os.environ.get('SMSBOT_TWILIO_AUTH_TOKEN')
+        twilio_token = os.environ.get("SMSBOT_TWILIO_AUTH_TOKEN")
 
         if not twilio_token:
-            current_app.logger.warning('Twilio request validation skipped due to SMSBOT_TWILIO_AUTH_TOKEN missing')
+            current_app.logger.warning(
+                "Twilio request validation skipped due to SMSBOT_TWILIO_AUTH_TOKEN missing"
+            )
             return func(*args, **kwargs)
 
         validator = RequestValidator(twilio_token)
@@ -33,7 +35,7 @@ def validate_twilio_request(func):
         request_valid = validator.validate(
             request.url,
             request.form,
-            request.headers.get('X-TWILIO-SIGNATURE', ''),
+            request.headers.get("X-TWILIO-SIGNATURE", ""),
         )
 
         # Continue processing the request if it's valid, return a 403 error if
@@ -41,58 +43,67 @@ def validate_twilio_request(func):
         if request_valid or current_app.debug:
             return func(*args, **kwargs)
         return abort(403)
+
     return decorated_function
 
 
 class TwilioWebhookHandler(object):
-
     def __init__(self):
         self.app = Flask(self.__class__.__name__)
-        self.app.add_url_rule('/', 'index', self.index, methods=['GET'])
-        self.app.add_url_rule('/health', 'health', self.health, methods=['GET'])
-        self.app.add_url_rule('/message', 'message', self.message, methods=['POST'])
-        self.app.add_url_rule('/call', 'call', self.call, methods=['POST'])
+        self.app.add_url_rule("/", "index", self.index, methods=["GET"])
+        self.app.add_url_rule("/health", "health", self.health, methods=["GET"])
+        self.app.add_url_rule("/message", "message", self.message, methods=["POST"])
+        self.app.add_url_rule("/call", "call", self.call, methods=["POST"])
 
         # Add prometheus wsgi middleware to route /metrics requests
-        self.app.wsgi_app = DispatcherMiddleware(self.app.wsgi_app, {
-            '/metrics': make_wsgi_app(),
-        })
+        self.app.wsgi_app = DispatcherMiddleware(
+            self.app.wsgi_app,
+            {
+                "/metrics": make_wsgi_app(),
+            },
+        )
 
     def set_bot(self, bot):  # noqa: WPS615
         self.bot = bot
 
     def index(self):
-        return ''
+        return ""
 
     @REQUEST_TIME.time()
     def health(self):
         return {
-            'version': get_smsbot_version(),
-            'owner': self.bot.owner_id,
-            'subscribers': self.bot.subscriber_ids,
+            "version": get_smsbot_version(),
+            "owner": self.bot.owner_id,
+            "subscribers": self.bot.subscriber_ids,
         }
 
     @REQUEST_TIME.time()
     @validate_twilio_request
     def message(self):
-        current_app.logger.info('Received SMS from {From}: {Body}'.format(**request.values.to_dict()))
+        current_app.logger.info(
+            "Received SMS from {From}: {Body}".format(**request.values.to_dict())
+        )
 
-        message = 'From: {From}\n\n{Body}'.format(**request.values.to_dict())
+        message = "From: {From}\n\n{Body}".format(**request.values.to_dict())
         self.bot.send_subscribers(message)
 
         # Return a blank response
         MESSAGE_COUNT.inc()
-        return '<response></response>'
+        return "<response></response>"
 
     @REQUEST_TIME.time()
     @validate_twilio_request
     def call(self):
-        current_app.logger.info('Received Call from {From}'.format(**request.values.to_dict()))
-        self.bot.send_subscribers('Received Call from {From}, rejecting.'.format(**request.values.to_dict()))
+        current_app.logger.info(
+            "Received Call from {From}".format(**request.values.to_dict())
+        )
+        self.bot.send_subscribers(
+            "Received Call from {From}, rejecting.".format(**request.values.to_dict())
+        )
 
         # Always reject calls
         CALL_COUNT.inc()
-        return '<Response><Reject/></Response>'
+        return "<Response><Reject/></Response>"
 
-    def serve(self, host='0.0.0.0', port=80, debug=False):
+    def serve(self, host="0.0.0.0", port=80, debug=False):
         serve(self.app, host=host, port=port)