mirror of
https://github.com/nikdoof/smsbot.git
synced 2025-12-13 18:12:15 +00:00
Initial rework into classes
This commit is contained in:
249
smsbot.py
Normal file → Executable file
249
smsbot.py
Normal file → Executable file
@@ -1,147 +1,160 @@
|
||||
try:
|
||||
import os
|
||||
import logging
|
||||
from functools import wraps
|
||||
from flask import Flask, request, abort
|
||||
from waitress import serve
|
||||
from twilio.request_validator import RequestValidator
|
||||
from telegram import ParseMode
|
||||
from telegram.ext import Updater, CommandHandler
|
||||
except ImportError as err:
|
||||
print(f"Failed to import required modules: {err}")
|
||||
#!/usr/bin/env python
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
from functools import wraps
|
||||
|
||||
# FB - Enable logging
|
||||
logging.basicConfig(
|
||||
format='%(asctime)s - %(levelname)s - %(message)s', level=logging.INFO)
|
||||
logger = logging.getLogger(__name__)
|
||||
from flask import Flask, abort, request, current_app
|
||||
from telegram import ParseMode
|
||||
from telegram.ext import CommandHandler, Updater
|
||||
from twilio.request_validator import RequestValidator
|
||||
from waitress import serve
|
||||
|
||||
webhook_listener = Flask(__name__)
|
||||
__version__ = '0.0.1'
|
||||
|
||||
|
||||
class TelegramSmsBot(object):
|
||||
|
||||
owner_id = None
|
||||
subscriber_ids = []
|
||||
|
||||
def __init__(self, token):
|
||||
self.logger = logging.getLogger(self.__class__.__name__)
|
||||
self.bot_token = token
|
||||
|
||||
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(__version__))
|
||||
|
||||
def subscribe_handler(self, update, context):
|
||||
self.logger.info('/subscribe command received')
|
||||
if not update.message.chat['id'] in self.subscriber_ids:
|
||||
self.logger.info('{0} subscribed'.format(update.message.chat['username']))
|
||||
self.subscriber_ids.append(update.message.chat['id'])
|
||||
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'])
|
||||
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)
|
||||
|
||||
def send_message(self, message, chat_id):
|
||||
self.bot.sendMessage(text=message, chat_id=chat_id, parse_mode=ParseMode.MARKDOWN_V2)
|
||||
|
||||
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(f):
|
||||
"""Validates that incoming requests genuinely originated from Twilio"""
|
||||
@wraps(f)
|
||||
def decorated_function(*args, **kwargs):
|
||||
# FB - Create an instance of the RequestValidator class
|
||||
validator = RequestValidator(os.environ.get('TWILIO_AUTH_TOKEN'))
|
||||
# Create an instance of the RequestValidator class
|
||||
twilio_token = os.environ.get('SMSBOT_TWILIO_AUTH_TOKEN')
|
||||
|
||||
# FB - Validate the request using its URL, POST data, and X-TWILIO-SIGNATURE header
|
||||
if not twilio_token:
|
||||
current_app.logger.warning('Twilio request validation skipped due to SMSBOT_TWILIO_AUTH_TOKEN missing')
|
||||
return f(*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', ''))
|
||||
|
||||
# FB - Continue processing the request if it's valid, return a 403 error if it's not
|
||||
if request_valid:
|
||||
# Continue processing the request if it's valid, return a 403 error if
|
||||
# it's not
|
||||
if request_valid or current_app.debug:
|
||||
return f(*args, **kwargs)
|
||||
else:
|
||||
logger.error('Invalid twilio request, aborting')
|
||||
return abort(403)
|
||||
return decorated_function
|
||||
|
||||
|
||||
def tg_help_handler(update, context):
|
||||
"""Send a message when the command /help is issued."""
|
||||
logger.info('/help command received in chat: %s', update.message.chat)
|
||||
update.message.reply_markdown(
|
||||
'Find out more on [Github](https://github.com/FiveBoroughs/Twilio2Telegram)')
|
||||
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('/message', 'message', self.message, methods=['POST'])
|
||||
self.app.add_url_rule('/call', 'call', self.call, methods=['POST'])
|
||||
|
||||
def tg_error_handler(update, context):
|
||||
"""Log Errors caused by Updates."""
|
||||
logger.warning('Update "%s" caused error "%s"', update, context.error)
|
||||
def set_bot(self, bot):
|
||||
self.bot = bot
|
||||
|
||||
def index(self):
|
||||
return 'Smsbot v{0}'.format(__version__)
|
||||
|
||||
def tg_bot_start():
|
||||
"""Start the telegram bot."""
|
||||
# FB - Create the Updater
|
||||
updater = Updater(os.environ.get('TELEGRAM_BOT_TOKEN'), use_context=True)
|
||||
@validate_twilio_request
|
||||
def message(self):
|
||||
return '<response></response>'
|
||||
|
||||
# FB - Get the dispatcher to register handlers
|
||||
dispatcher = updater.dispatcher
|
||||
@validate_twilio_request
|
||||
def call(self):
|
||||
# Always reject calls
|
||||
return '<Response><Reject/></Response>'
|
||||
|
||||
# FB - on /help command
|
||||
dispatcher.add_handler(CommandHandler("help", tg_help_handler))
|
||||
|
||||
# FB - log all errors
|
||||
dispatcher.add_error_handler(tg_error_handler)
|
||||
|
||||
# FB - Start the Bot
|
||||
updater.start_polling()
|
||||
|
||||
return updater.bot
|
||||
|
||||
|
||||
def tg_send_owner_message(message):
|
||||
"""Send telegram message to owner."""
|
||||
telegram_bot.sendMessage(text=message, chat_id=os.environ.get('TELEGRAM_OWNER'), parse_mode=ParseMode.MARKDOWN)
|
||||
|
||||
|
||||
def tg_send_subscribers_message(message):
|
||||
"""Send telegram messages to subscribers."""
|
||||
for telegram_destination in os.environ.get('TELEGRAM_SUBSCRIBERS').split(','):
|
||||
telegram_bot.sendMessage(
|
||||
text=message, chat_id=telegram_destination, parse_mode=ParseMode.MARKDOWN)
|
||||
|
||||
|
||||
@webhook_listener.route('/', methods=['GET'])
|
||||
def index():
|
||||
"""Upon call of homepage."""
|
||||
logger.info('"/" reached, IP: %s', request.remote_addr)
|
||||
|
||||
return webhook_listener.send_static_file('Index.html')
|
||||
|
||||
|
||||
@webhook_listener.route('/message', methods=['POST'])
|
||||
@validate_twilio_request
|
||||
def recv_message():
|
||||
"""Upon reception of a SMS."""
|
||||
logger.info(' "/message" reached, IP: %s', request.remote_addr)
|
||||
# FB - Format telegram Message
|
||||
telegram_message = 'Text from `{From}` ({Country}, {State}) :``` {Body}```'.format(
|
||||
From=request.values.get('From', 'unknown'),
|
||||
Country=request.values.get('FromCountry', 'unknown'),
|
||||
State=request.values.get('FromState', 'unknown'),
|
||||
Body=request.values.get('Body', 'unknown')
|
||||
)
|
||||
|
||||
logger.info(telegram_message)
|
||||
# FB - Send telegram alerts
|
||||
tg_send_owner_message('Twilio ID : `{Id}`\n'.format(
|
||||
Id=request.values.get('MessageSid', 'unknown')) + telegram_message)
|
||||
tg_send_subscribers_message(telegram_message)
|
||||
|
||||
# FB - return empty response to avoid further Twilio fees
|
||||
return '<response></response>'
|
||||
|
||||
|
||||
|
||||
@webhook_listener.route('/call', methods=['POST'])
|
||||
@validate_twilio_request
|
||||
def recv_call():
|
||||
"""Upon reception of a call."""
|
||||
logger.info(' "/call" reached, IP: %s', request.remote_addr)
|
||||
# FB - Format telegram Message
|
||||
telegram_message = 'Call from `{From}` ({Country}, {State}) :``` {Status}```'.format(
|
||||
From=request.values.get('From', 'unknown'),
|
||||
Country=request.values.get('FromCountry', 'unknown'),
|
||||
State=request.values.get('FromState', 'unknown'),
|
||||
Status=request.values.get('CallStatus', 'unknown')
|
||||
)
|
||||
|
||||
logger.info(telegram_message)
|
||||
# FB - Send telegram alerts
|
||||
tg_send_owner_message('Twilio ID : `{Id}`\n'.format(
|
||||
Id=request.values.get('CallSid', 'unknown')) + telegram_message)
|
||||
tg_send_subscribers_message(telegram_message)
|
||||
|
||||
# FB - reject the call without being billed
|
||||
return '<Response><Reject/></Response>'
|
||||
def serve(self, host='0.0.0.0', port=80, debug=False):
|
||||
serve(self.app, host=host, port=port)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
logger.info('Starting bot')
|
||||
# FB - Start the telegram bot
|
||||
telegram_bot = tg_bot_start()
|
||||
# FB - Start the website
|
||||
serve(webhook_listener, host='0.0.0.0', port=80)
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
|
||||
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.exception('Telegram Bot token missing')
|
||||
sys.exit(1)
|
||||
|
||||
# Start bot
|
||||
telegram_bot = TelegramSmsBot(token)
|
||||
telegram_bot.start()
|
||||
|
||||
# Start webhooks
|
||||
webhooks = TwilioWebhookHandler()
|
||||
webhooks.set_bot(telegram_bot)
|
||||
webhooks.serve()
|
||||
|
||||
Reference in New Issue
Block a user