minetest-telegram-bridge/bot/server.py

65 lines
1.6 KiB
Python

#!/usr/bin/env python3
from flask import Flask, request
import sqlite3
import config
import telegram
import re
from emoji import emojize
def escape_html(message):
return re.sub("<", "&lt;",
re.sub(">", "&rt;",
re.sub("&", "&amp;", message)))
def get(db):
cursor = db.cursor()
query = "SELECT * FROM messages ORDER BY ROWID"
cursor.execute(query)
text = ""
for row in cursor:
text += row[0] + "\n"
cursor.execute("DELETE FROM messages")
db.commit()
return text
def post(db):
message = request.form['message']
cursor = db.cursor()
query = 'INSERT INTO messages values(?)'
cursor.execute(query, (message,))
db.commit()
return ""
def send(bot, code):
message = request.form["message"]
if code:
message = escape_html(message)
message = "<code>" + message + "</code>"
for i in config.groups:
bot.send_message(chat_id=i, text=emojize(message),
parse_mode="HTML")
else:
for i in config.groups:
bot.send_message(chat_id=i, text=emojize(message))
return ""
app = Flask(__name__)
db = sqlite3.connect(":memory:", check_same_thread=False)
bot = telegram.Bot(token=config.api_key)
# Set up tables
cursor = db.cursor()
with open('schema.sql') as fp:
cursor.executescript(fp.read())
app.add_url_rule('/get', 'get', lambda: get(db))
app.add_url_rule('/post', 'post', lambda: post(db), methods=["POST"])
app.add_url_rule('/send', 'send', lambda: send(bot, False), methods=["POST"])
app.add_url_rule('/sendCode', 'sc', lambda: send(bot, True), methods=["POST"])