nikdoof/smsbot
1
0
Fork 0
mirror of https://github.com/nikdoof/smsbot.git synced 2025-12-13 10:02:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore(lint): Linting cleanup

Browse Source
This commit is contained in:
Andrew Williams
2025-05-04 12:47:30 +01:00
parent 5980bc67dd
commit 5cacd8e23a
4 changed files with 107 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
smsbot/__main__.py
View File

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

32
smsbot/cli.py
View File

@@ -3,23 +3,31 @@ 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()
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 +36,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()

77
smsbot/telegram.py
View File

@@ -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)

63
smsbot/webhook_handler.py
View File

@@ -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)