autodist/telegram/openmamba_bot.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# openmamba bot for Telegram
#
# Copyright (C) 2016-2023 by Silvan Calarco <silvan.calarco@mambasoft.it>
#
# GPL v3 license

from telegram import (ParseMode, Update)

from telegram.ext import (Updater, CommandHandler, MessageHandler, Filters, RegexHandler,
                          ConversationHandler, Job, CallbackContext)

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' ]

def start(update: Update, context: CallbackContext):

    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'
        'Send /set [seconds] to enable notifications.\n'
        'Send /unset to disable notifications.\n\n')

    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):
    user = update.message.from_user
    logger.info("Query of %s: %s" % (user.first_name, update.message.text))

    response = ""
    for rep in reps:
        conn = sqlite3.connect('/var/webbuild/db/%s-sources.db' % rep, check_same_thread=False)
        cursor = conn.cursor()
        print(rep)
        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):
    user = update.message.from_user
    logger.info("Details of %s: %s" % (user.first_name, update.message.text))

    response = ""
    for rep in reps:
        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]))
        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' ]:
                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)
    else:
        update.message.reply_text('No results found.')
    return


def alarm(context):
    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])
    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]


def set(update: Update, context: CallbackContext):
    chat_id = update.message.chat_id
    try:
        due = int(context.args[0])
        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)

        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):
    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):
    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))


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

    # 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))
    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))

    # 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()