autodist/telegram/openmamba_bot.py

206 lines
7.2 KiB
Python
Raw Normal View History

2020-11-19 12:34:53 +01:00
#!/usr/bin/env python3
2017-12-30 23:14:48 +01:00
# -*- coding: utf-8 -*-
#
# openmamba bot for Telegram
#
# Copyright (C) 2016-2023 by Silvan Calarco <silvan.calarco@mambasoft.it>
2017-12-30 23:14:48 +01:00
#
# GPL v3 license
from telegram import (ParseMode, Update)
2017-12-30 23:14:48 +01:00
from telegram.ext import (Updater, CommandHandler, MessageHandler, Filters, RegexHandler,
ConversationHandler, Job, CallbackContext)
2017-12-30 23:14:48 +01:00
import logging
import subprocess
import sqlite3
import os
# Enable logging
logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
level=logging.INFO)
logger = logging.getLogger(__name__)
timers = dict()
social_log_last_ids = dict()
reps = [ 'base', 'devel-makedist', 'devel-autodist', 'devel-kernel', 'devel-misc', 'devel-future', 'devel-past' ]
2017-12-30 23:14:48 +01:00
def start(update: Update, context: CallbackContext):
2017-12-30 23:14:48 +01:00
user = update.message.from_user
logger.info("User %s started the conversation." % user.first_name)
update.message.reply_text(
'Hi! This is the openmamba GNU/Linux Bot.\n'
'Send any text to search for packages.\n'
'Send /details [name] to see details of specified source package.\n'
2017-12-30 23:14:48 +01:00
'Send /set [seconds] to enable notifications.\n'
'Send /unset to disable notifications.\n\n')
2017-12-30 23:14:48 +01:00
return
def help(update: Update, context: CallbackContext):
update.message.reply_text(
'Send any text to search for packages.\n'
'Send /details [name] to see details of specified source package.\n'
'Send /set [seconds] to enable notifications.\n'
'Send /unset to disable notifications.\n\n')
return
def query(update: Update, context: CallbackContext):
2017-12-30 23:14:48 +01:00
user = update.message.from_user
logger.info("Query of %s: %s" % (user.first_name, update.message.text))
response = ""
for rep in reps:
2017-12-30 23:14:48 +01:00
conn = sqlite3.connect('/var/webbuild/db/%s-sources.db' % rep, check_same_thread=False)
cursor = conn.cursor()
print(rep)
2017-12-30 23:14:48 +01:00
cursor.execute(
'SELECT id,name,version,release,summary,url FROM sources where name like "%%%s%%" or summary like "%%%s%%" '
'ORDER BY name="%s" DESC, name like "%s%%" DESC, name like "%%%s%%" DESC limit 10'
% (update.message.text, update.message.text, update.message.text, update.message.text, update.message.text))
for row in cursor:
response += "<b>%s</b> %s-%s (%s)\n%s\n%s\n\n" % (row[1], row[2], row[3], rep, row[4], row[5])
if response != "":
update.message.reply_text(response, parse_mode=ParseMode.HTML)
else:
update.message.reply_text('No results found.')
return
def details(update: Update, context: CallbackContext):
2017-12-30 23:14:48 +01:00
user = update.message.from_user
logger.info("Details of %s: %s" % (user.first_name, update.message.text))
response = ""
for rep in reps:
2017-12-30 23:14:48 +01:00
conn = sqlite3.connect('/var/webbuild/db/%s-sources.db' % rep, check_same_thread=False)
cursor = conn.cursor()
cursor.execute(
'SELECT id,name,version,release,summary,url,description FROM sources where name="%s"' % (context.args[0]))
2017-12-30 23:14:48 +01:00
for row in cursor:
response += "<b>%s</b> %s-%s (devel)\n%s\n%s\n\n<i>%s</i>\n\n" % (row[1], row[2], row[3], row[4], row[5], row[6])
for arch in [ 'x86_64', 'aarch64', 'i586', 'arm' ]:
2017-12-30 23:14:48 +01:00
conn1 = sqlite3.connect('/var/webbuild/db/' + rep + '-' + arch + '.db', check_same_thread=False)
cursor1 = conn1.cursor()
cursor1.execute('SELECT id,name FROM packages where id_source=%s' % (row[0]))
for row1 in cursor1:
response += "<b>%s</b>(%s) " % (row1[1], arch)
response += "\n\n\n"
if response != "":
update.message.reply_text(response[0:4096], parse_mode=ParseMode.HTML)
2017-12-30 23:14:48 +01:00
else:
update.message.reply_text('No results found.')
return
def alarm(context):
2017-12-30 23:14:48 +01:00
social_log_conn = sqlite3.connect('/var/webbuild/webbuild.db')
social_log_cursor = social_log_conn.cursor()
social_log_cursor.execute('SELECT id,user,text,datetime(time,\'localtime\'),type FROM social_log where id>%s' % social_log_last_ids[context.job])
2017-12-30 23:14:48 +01:00
response = ""
for row in social_log_cursor:
if row[4] == "job":
response += "Job run by <i>%s</i> %s (%s)\n" % (row[1], row[2], row[3])
else:
response += "<i>%s</i> %s (%s)\n" % (row[1], row[2], row[3])
if response != "":
bot.sendMessage(context.job, response, parse_mode=ParseMode.HTML)
social_log_last_ids[context.job] = row[0]
2017-12-30 23:14:48 +01:00
def set(update: Update, context: CallbackContext):
2017-12-30 23:14:48 +01:00
chat_id = update.message.chat_id
try:
due = int(context.args[0])
2017-12-30 23:14:48 +01:00
if due < 0:
update.message.reply_text('Sorry we can not go back to future!')
return
#job = Job(alarm, context, next_t=due, repeat=True, context=chat_id)
#timers[chat_id] = job
context.job_queue.run_repeating(alarm, due)
#context.jobqueue.put(job)
2017-12-30 23:14:48 +01:00
social_log_conn = sqlite3.connect('/var/webbuild/webbuild.db')
social_log_cursor = social_log_conn.cursor()
social_log_cursor.execute('''SELECT MAX(id) FROM social_log''')
social_log_last_ids[chat_id] = social_log_cursor.fetchone()[0] - 5
update.message.reply_text('Notifications enabled!')
except (IndexError, ValueError):
update.message.reply_text('Usage: /set <seconds>')
def unset(update: Update, context: CallbackContext):
2017-12-30 23:14:48 +01:00
chat_id = update.message.chat_id
if chat_id not in timers:
update.message.reply_text('Notification were not enabled')
return
job = timers[chat_id]
job.schedule_removal()
del timers[chat_id]
update.message.reply_text('Notifications disabled')
def cancel(update: Update, context: CallbackContext):
2017-12-30 23:14:48 +01:00
user = update.message.from_user
logger.info("User %s canceled the conversation." % user.first_name)
update.message.reply_text('Bye!')
return ConversationHandler.END
def error(update: Update, context: CallbackContext):
logger.warning('Update "%s" caused error "%s"' % (update, context.error))
2017-12-30 23:14:48 +01:00
def main():
# Load bot token from external configuration file
bot_token = ""
with open("/etc/autodist/secrets") as myfile:
for line in myfile:
name, var = line.rstrip().split("=")
if name == "TELEGRAM_BOT_TOKEN":
bot_token = var
2017-12-30 23:14:48 +01:00
# Create the EventHandler and pass it your bot's token.
updater = Updater(bot_token)
# Get the dispatcher to register handlers
dp = updater.dispatcher
dp.add_handler(CommandHandler('start', start))
dp.add_handler(CommandHandler('help', help))
2017-12-30 23:14:48 +01:00
dp.add_handler(CommandHandler('details', details, pass_args=True))
dp.add_handler(CommandHandler('set', set, pass_args=True, pass_job_queue=True))
dp.add_handler(CommandHandler('unset', unset))
dp.add_handler(MessageHandler(Filters.text, query))
2017-12-30 23:14:48 +01:00
# log all errors
dp.add_error_handler(error)
# Start the Bot
updater.start_polling()
# Run the bot until the you presses Ctrl-C or the process receives SIGINT,
# SIGTERM or SIGABRT. This should be used most of the time, since
# start_polling() is non-blocking and will stop the bot gracefully.
updater.idle()
if __name__ == '__main__':
main()