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

Breakout into sub modules

Browse Source
This commit is contained in:
Andrew Williams
2022-06-09 09:52:26 +01:00
parent 0c260c9a08
commit 5c89056a9e
7 changed files with 204 additions and 186 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
setup.cfg
View File

@@ -23,7 +23,7 @@ install_requires =
[options.entry_points]
console_scripts =
smsbot = smsbot:main
smsbot = smsbot.cli:main
[flake8]
format = wemake

184
smsbot/__init__.py
View File

@@ -1,184 +0,0 @@
import logging
import os
import sys
from functools import wraps
import pkg_resources
from flask import Flask, abort, current_app, request
from telegram.ext import CommandHandler, Updater
from twilio.request_validator import RequestValidator
from waitress import serve
pkg_version = pkg_resources.require('smsbot')[0].version
class TelegramSmsBot(object):
def __init__(self, token, owner=None, subscribers=None):
self.logger = logging.getLogger(self.__class__.__name__)
self.bot_token = token
self.owner_id = owner
self.subscriber_ids = subscribers or []
def start(self):
self.logger.info('Starting bot...')
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('subscribe', self.subscribe_handler))
self.updater.dispatcher.add_handler(CommandHandler('unsubscribe', self.unsubscribe_handler))
self.updater.dispatcher.add_error_handler(self.error_handler)
self.updater.start_polling()
self.bot = self.updater.bot
self.logger.info('Bot Ready')
def stop(self):
self.updater.stop()
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)
update.message.reply_markdown('Smsbot v{0}\n\n/help\n/subscribe\n/unsubscribe'.format(pkg_version))
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')
else:
update.message.reply_markdown('You are already subscribed to SMS notifications')
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')
else:
update.message.reply_markdown('You are not subscribed to SMS notifications')
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))
def send_message(self, message, chat_id):
self.bot.sendMessage(text=message, chat_id=chat_id)
def send_owner(self, message):
if self.owner_id:
self.send_message(message, self.owner_id)
def send_subscribers(self, message):
for chat_id in self.subscriber_ids:
self.send_message(message, chat_id)
def set_owner(self, chat_id):
self.owner_id = chat_id
def add_subscriber(self, chat_id):
self.subscriber_ids.append(chat_id)
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')
if not twilio_token:
current_app.logger.warning('Twilio request validation skipped due to SMSBOT_TWILIO_AUTH_TOKEN missing')
return func(*args, **kwargs)
validator = RequestValidator(twilio_token)
# Validate the request using its URL, POST data,
# and X-TWILIO-SIGNATURE header
request_valid = validator.validate(
request.url,
request.form,
request.headers.get('X-TWILIO-SIGNATURE', ''),
)
# Continue processing the request if it's valid, return a 403 error if
# it's not
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'])
def set_bot(self, bot): # noqa: WPS615
self.bot = bot
def index(self):
return ''
def health(self):
return '<h1>Smsbot v{0}</h1><p><b>Owner</b>: {1}</p><p><b>Subscribers</b>: {2}</p>'.format(pkg_version, self.bot.owner_id, self.bot.subscriber_ids)
@validate_twilio_request
def message(self):
message = 'From: {From}\n\n{Body}'.format(**request.values.to_dict())
current_app.logger.info('Received SMS from {From}: {Body}'.format(**request.values.to_dict()))
self.bot.send_subscribers(message)
return '<response></response>'
@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()))
# Always reject calls
return '<Response><Reject/></Response>'
def serve(self, host='0.0.0.0', port=80, debug=False):
serve(self.app, host=host, port=port)
def main():
logging.basicConfig(level=logging.INFO)
logging.info('smsbot v%s', pkg_version)
listen_host = os.environ.get('SMSBOT_LISTEN_HOST') or '0.0.0.0'
listen_port = int(os.environ.get('SMSBOT_LISTEN_PORT') or '80')
token = os.environ.get('SMSBOT_TELEGRAM_BOT_TOKEN')
if not token:
logging.error('Telegram Bot token missing')
sys.exit(1)
# Start bot
telegram_bot = TelegramSmsBot(token)
# Set the owner ID if configured
if 'SMSBOT_OWNER_ID' in os.environ:
telegram_bot.set_owner(os.environ.get('SMSBOT_OWNER_ID'))
# Add default subscribers
if 'SMSBOT_DEFAULT_SUBSCRIBERS' in os.environ:
for chat_id in os.environ.get('SMSBOT_DEFAULT_SUBSCRIBERS').split(','):
telegram_bot.add_subscriber(chat_id)
telegram_bot.start()
# Start webhooks
webhooks = TwilioWebhookHandler()
webhooks.set_bot(telegram_bot)
webhooks.serve(host=listen_host, port=listen_port)

2
smsbot/__main__.py
View File

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

42
smsbot/cli.py Normal file
View File

@@ -0,0 +1,42 @@
import logging
import os
import sys
import pkg_resources
from smsbot.telegram import TelegramSmsBot
from smsbot.webhook_handler import TwilioWebhookHandler
pkg_version = pkg_resources.require('smsbot')[0].version
def main():
logging.basicConfig(level=logging.INFO)
logging.info('smsbot v%s', pkg_version)
listen_host = os.environ.get('SMSBOT_LISTEN_HOST') or '0.0.0.0'
listen_port = int(os.environ.get('SMSBOT_LISTEN_PORT') or '80')
token = os.environ.get('SMSBOT_TELEGRAM_BOT_TOKEN')
if not token:
logging.error('Telegram Bot token missing')
sys.exit(1)
# Start bot
telegram_bot = TelegramSmsBot(token)
# Set the owner ID if configured
if 'SMSBOT_OWNER_ID' in os.environ:
telegram_bot.set_owner(os.environ.get('SMSBOT_OWNER_ID'))
# Add default subscribers
if 'SMSBOT_DEFAULT_SUBSCRIBERS' in os.environ:
for chat_id in os.environ.get('SMSBOT_DEFAULT_SUBSCRIBERS').split(','):
telegram_bot.add_subscriber(chat_id)
telegram_bot.start()
# Start webhooks
webhooks = TwilioWebhookHandler()
webhooks.set_bot(telegram_bot)
webhooks.serve(host=listen_host, port=listen_port)

77
smsbot/telegram.py Normal file
View File

@@ -0,0 +1,77 @@
import logging
from telegram.ext import CommandHandler, Updater
from smsbot.utils import get_smsbot_version
class TelegramSmsBot(object):
def __init__(self, token, owner=None, subscribers=None):
self.logger = logging.getLogger(self.__class__.__name__)
self.bot_token = token
self.owner_id = owner
self.subscriber_ids = subscribers or []
def start(self):
self.logger.info('Starting bot...')
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('subscribe', self.subscribe_handler))
self.updater.dispatcher.add_handler(CommandHandler('unsubscribe', self.unsubscribe_handler))
self.updater.dispatcher.add_error_handler(self.error_handler)
self.updater.start_polling()
self.bot = self.updater.bot
self.logger.info('Bot Ready')
def stop(self):
self.updater.stop()
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)
update.message.reply_markdown('Smsbot v{0}\n\n/help\n/subscribe\n/unsubscribe'.format(get_smsbot_version()))
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')
else:
update.message.reply_markdown('You are already subscribed to SMS notifications')
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')
else:
update.message.reply_markdown('You are not subscribed to SMS notifications')
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))
def send_message(self, message, chat_id):
self.bot.sendMessage(text=message, chat_id=chat_id)
def send_owner(self, message):
if self.owner_id:
self.send_message(message, self.owner_id)
def send_subscribers(self, message):
for chat_id in self.subscriber_ids:
self.send_message(message, chat_id)
def set_owner(self, chat_id):
self.owner_id = chat_id
def add_subscriber(self, chat_id):
self.subscriber_ids.append(chat_id)

5
smsbot/utils.py Normal file
View File

@@ -0,0 +1,5 @@
import pkg_resources
def get_smsbot_version():
return pkg_resources.require('smsbot')[0].version

78
smsbot/webhook_handler.py Normal file
View File

@@ -0,0 +1,78 @@
import os
from functools import wraps
from flask import Flask, abort, current_app, request
from twilio.request_validator import RequestValidator
from waitress import serve
from smsbot.utils import get_smsbot_version
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')
if not twilio_token:
current_app.logger.warning('Twilio request validation skipped due to SMSBOT_TWILIO_AUTH_TOKEN missing')
return func(*args, **kwargs)
validator = RequestValidator(twilio_token)
# Validate the request using its URL, POST data,
# and X-TWILIO-SIGNATURE header
request_valid = validator.validate(
request.url,
request.form,
request.headers.get('X-TWILIO-SIGNATURE', ''),
)
# Continue processing the request if it's valid, return a 403 error if
# it's not
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'])
def set_bot(self, bot): # noqa: WPS615
self.bot = bot
def index(self):
return ''
def health(self):
return '<h1>Smsbot v{0}</h1><p><b>Owner</b>: {1}</p><p><b>Subscribers</b>: {2}</p>'.format(get_smsbot_version(), self.bot.owner_id, self.bot.subscriber_ids)
@validate_twilio_request
def message(self):
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())
self.bot.send_subscribers(message)
# Return a blank response
return '<response></response>'
@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()))
# Always reject calls
return '<Response><Reject/></Response>'
def serve(self, host='0.0.0.0', port=80, debug=False):
serve(self.app, host=host, port=port)