1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
|
#!/usr/bin/env python3
import argparse
import configparser
import logging
import signal
import sys
import threading
import os
from project import botmanager
from project import uptimeservice
class BotLauncher(object):
CONFIG_FILE = "bot.config"
@staticmethod
def sigint_handler(signal, frame):
"""Exit gracefully when receiving an interrupt signal"""
logging.info("Stopping uptime services (if any)...")
uptimeservice.UptimeService.stop_services()
print("Exiting radiodiodibot...")
sys.exit(0)
def __init__(self):
signal.signal(signal.SIGINT, BotLauncher.sigint_handler)
try: # Python 2.7+
from logging import NullHandler
except ImportError:
class NullHandler(logging.Handler):
def emit(self, record):
pass
# Read configs, unsuccessful reads are
# silently ignored
self.config = configparser.ConfigParser()
self.config.read(BotLauncher.CONFIG_FILE)
# Get the bot token as an argument from the user
parser = argparse.ArgumentParser(description="Telegram bot for the Radiodiodi Student Radio broadcast.")
parser.add_argument("-t", "--token", help="Telegram Bot API token")
parser.add_argument("-U", "--shoutbox-api-url", help="Shoutbox API URL")
parser.add_argument("-C", "--telegram-chat-id", help="Telegram Chat ID")
parser.add_argument("-v", "--verbose", help="Verbose output", action="store_true")
parser.add_argument("-i", "--interval", help="Delay between API update calls in seconds", type=int)
self.args = parser.parse_args()
# Show verbose output with the -v option
if self.args.verbose:
logging_level = logging.INFO
else:
logging_level = logging.WARNING
formatter = logging.Formatter("%(asctime)s %(levelname)s %(message)s")
logging.getLogger().addHandler(NullHandler())
fileLogger = logging.FileHandler("output.log")
fileLogger.setFormatter(formatter)
consoleLogger = logging.StreamHandler()
consoleLogger.setFormatter(formatter)
logging.getLogger().addHandler(fileLogger)
logging.getLogger().addHandler(consoleLogger)
logging.getLogger().setLevel(logging_level)
# Try to import telepot
# If not installed, inform user and fail gracefully
try:
import telepot
except:
print("src needs the telepot module to communicate with Telegram.")
print("Please install telepot before using src.")
sys.exit(1)
def run_uptime(self, port):
logging.info("Setting up uptime service...")
self.uptime_server = uptimeservice.UptimeService(int(port))
self.uptime_server.start()
# Entry point
def main(self):
print("=== Radiodiodibot ===")
print("Press CTRL-C to exit.")
# Get default parameter values from the config file
telegram_bot_token = self.config["GENERAL"]["TelegramBotToken"]
shoutbox_api_url = self.config["GENERAL"]["ShoutboxApiUrl"]
telegram_chat_id = self.config["GENERAL"]["TelegramChatID"]
api_call_interval = int(self.config["GENERAL"]["ApiCallInterval"])
api_auth_token = self.config["GENERAL"]["ApiAuthToken"]
uptime_port = self.config["UPTIME"]["Port"]
# If the user has specified the parameters as command line
# arguments, use them to override the config values
if self.args.token is not None:
telegram_bot_token = self.args.token
if self.args.shoutbox_api_url is not None:
shoutbox_api_url = self.args.shoutbox_api_url
if self.args.telegram_chat_id is not None:
telegram_chat_id = self.args.telegram_chat_id
if self.args.interval is not None:
api_call_interval = self.args.interval
# Store the final values in a manager instance
manager = botmanager.BotManager(telegram_bot_token)
manager.set_parameters(shoutbox_api_url, telegram_chat_id, api_call_interval, api_auth_token)
logging.info("Using Shoutbox API URL: {}".format(shoutbox_api_url))
logging.info("Using Telegram Chat ID: {}".format(telegram_chat_id))
logging.info("Using API call interval of {} seconds.".format(api_call_interval))
# Start uptime server if user has provided a port to bind
if uptime_port is not None:
self.run_uptime(uptime_port)
# Start listening
manager.start()
def launch(self):
self.main()
def main():
l = BotLauncher()
l.launch()
if __name__ == "__main__":
main()
|
