import logging import time import click import sentry_sdk from tabulate import tabulate from slack_sdk import WebClient as SlackWebClient from logging.config import dictConfig from community_qa_bot.settings import DEBUG, SENTRY_DSN, LOG_LEVEL dictConfig({ 'version': 1, 'formatters': { 'default': { 'format': '%(asctime)s - %(name)s:%(lineno)d - %(levelname)s - %(message)s', } }, 'handlers': { 'file': { 'class': 'logging.FileHandler', 'filename': 'app.log', 'formatter': 'default' }, 'console': { 'class': 'logging.StreamHandler', 'formatter': 'default' } }, 'root': { 'level': LOG_LEVEL, 'handlers': ['file', 'console'] } }) def update_log_file_handler(new_file_name): for handler in logging.getLogger().handlers: if isinstance(handler, logging.FileHandler): handler.close() handler.baseFilename = new_file_name handler.stream = open(new_file_name, 'a') if not DEBUG: sentry_sdk.init( dsn=SENTRY_DSN, # Set traces_sample_rate to 1.0 to capture 100% # of transactions for performance monitoring. # We recommend adjusting this value in production. traces_sample_rate=1.0 ) @click.group(context_settings={'max_content_width': 150}) def cli(): pass @cli.group() @click.option('--logfile', '-lf', default='web.log', help='log file name, default is web.log') def web(logfile): update_log_file_handler(logfile) @web.command(name='runserver') @click.option('--host', '-h', default='127.0.0.1', help='bind host, default is 0.0.0.0') @click.option('--port', '-p', default=8080, type=int, help='bind port, default is 8080') @click.option( '--workers', '-w', default=4, type=int, help='number of workers, default is 4, only works in production mode') def run_web_server(host, port, workers): update_log_file_handler('web.log') from community_qa_bot.web.app import app if DEBUG: app.run(debug=DEBUG, host=host, port=port) else: from gunicorn.app.base import BaseApplication class StandaloneApplication(BaseApplication): def __init__(self, app, options=None): self.options = options or {} self.application = app super().__init__() def load_config(self): config = {key: value for key, value in self.options.items() if key in self.cfg.settings and value is not None} for key, value in config.items(): self.cfg.set(key.lower(), value) def load(self): return self.application StandaloneApplication(app, options={'bind': f'{host}:{port}', 'workers': workers}).run() @cli.group() @click.option('--logfile', '-lf', default='slack.log', help='log file name, default is slack.log') def slack(logfile): update_log_file_handler(logfile) @slack.command(name='runserver') def run_slack_server(): from community_qa_bot.slack.app import run_app run_app() @slack.command() @click.option('--include-token', is_flag=True, help='include token in the output') def list_installed_bots(include_token): from community_qa_bot.slack.bot import installation_store data = [] for bot in installation_store.get_all_bots(): bot_info = { 'app_id': bot.app_id, 'team_id': bot.team_id, 'team_name': bot.team_name, 'bot_id': bot.bot_id, 'bot_user_id': bot.bot_user_id, } if include_token: bot_info['token'] = bot.bot_token data.append(bot_info) print(tabulate(data, headers='keys', tablefmt='grid')) @slack.command() @click.option('--bot_token', prompt='slack bot token (xoxb-******)', type=str) def install_bot(bot_token): from community_qa_bot.slack.bot import installation_store, Bot client = SlackWebClient(token=bot_token) bot_auth_info = client.auth_test().data team_id = bot_auth_info['team_id'] team_name = bot_auth_info['team'] bot_id = bot_auth_info['bot_id'] bot_user_id = bot_auth_info['user_id'] bot_info = client.bots_info(bot=bot_id, team_id=team_id).data app_id = bot_info['bot']['app_id'] bot = Bot( app_id=app_id, team_id=team_id, team_name=team_name, bot_id=bot_id, bot_user_id=bot_user_id, bot_token=bot_token, # installed_at is not relevant in our case; using the current time is fine. installed_at=time.time(), ) installation_store.save_bot(bot) print("Bot installed successfully!") @cli.group() @click.option('--logfile', '-lf', default='discord.log', help='log file name, default is discord.log') def discord(logfile): update_log_file_handler(logfile) @discord.command(name='runserver') def run_discord_server(): from community_qa_bot.discord.app import run run() if __name__ == '__main__': cli()