import atexit from logging.config import dictConfig from threading import Thread import random from flask import Flask, request import requests from rollbot import Rollbot, RollbotMessage, RollbotPlugin import plugins import db from config import BOTS_LOOKUP, get_config, get_secret GLOBAL_ADMINS = get_secret("auths.global") GROUP_ADMINS = get_secret("auths.group") dictConfig({ "version": 1, "formatters": {"default": { "format": "[%(asctime)s] %(levelname)s in %(module)s: %(message)s", }}, "handlers": {"wsgi": { "class": "logging.StreamHandler", "stream": "ext://flask.logging.wsgi_errors_stream", "formatter": "default" }}, "root": { "level": "INFO", "handlers": ["wsgi"] } }) def post_groupme_message(msg, group_id): bot_id = BOTS_LOOKUP[group_id] requests.post( "https://api.groupme.com/v3/bots/post", json={ "bot_id": bot_id, "text": msg }, timeout=10 ) sponsor_count = -1 sponsors = [ ("RAID: Shadow Legends - the hot new mobile game with over a billion players", "RAIDBOT"), ("Raycon Earbuds - the best true wireless earbuds that sponsor Rollbot", "ROLLBUDS"), ("Audible dot com - sign up today for a free license to view an eBook", "READBOT"), ("Casper Mattresses - the beds are haunted, actually haunted! How is that allowed?", "ROLLBED"), ("MeUndies - think about rollbot but on your junk", "ROLLBOXERS I guess"), ("PepsiCo Products - the refreshing taste of PepsiCo products", "Sinaltrainal v. Coca-Cola Co."), ("RAIDCON Shadow Buds - nightmare nightmare nightmare nightmare", "nightmarenightmarenightmare"), ] def post_sponsored(msg, group_id): global sponsor_count if random.randint(1, 3) == 3: sponsor_count += 1 (sponsor, code) = sponsors[sponsor_count] if sponsor_count < len(sponsors) else random.choice(sponsors) post_groupme_message(f"This action was sponsored by {sponsor}! Use code {code} for {random.choice([10, 15, 25])}% off!", group_id) post_groupme_message(msg, group_id) app = Flask(__name__) app.config["PROPAGATE_EXCEPTIONS"] = True rollbot = Rollbot( logger=app.logger, plugin_classes=RollbotPlugin.find_all_plugins(), aliases=get_config("aliases"), responses=get_config("responses"), sleep_time=float(get_config("sleep_time")), callback=post_sponsored, session_factory=db.Session ) app.logger.info("Initializing database tables") db.init_db() app.logger.info("Finished initializing database") rollbot.start_plugins() atexit.register(rollbot.shutdown_plugins) @app.route("/", methods=["POST"]) def execute(): json = request.get_json() msg = RollbotMessage.from_groupme(json, global_admins=GLOBAL_ADMINS, group_admins=GROUP_ADMINS) if msg.group_id not in BOTS_LOOKUP: app.logger.warning(f"Received message from unknown group ID {msg.group_id}") return "", 400 t = Thread(target=lambda: rollbot.handle_command(msg)) t.start() return "", 204 @app.route("/health") def health(): return "Rollbot healthy!", 200 if __name__ == "__main__": app.run(host="0.0.0.0", port=6070)