new bot in the making, slight improvements to old
This commit is contained in:
Remy Moll
5
bot2/commands/__init__.py
Normal file
5
bot2/commands/__init__.py
Normal file
@@ -0,0 +1,5 @@
|
||||
# Placeholder
|
||||
from . import clock
|
||||
from . import help
|
||||
from . import weather
|
||||
from . import status
|
||||
0
bot2/commands/clock.py
Normal file
0
bot2/commands/clock.py
Normal file
87
bot2/commands/help.py
Normal file
87
bot2/commands/help.py
Normal file
@@ -0,0 +1,87 @@
|
||||
from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update
|
||||
from telegram.ext import Updater, CommandHandler, CallbackQueryHandler, CallbackContext
|
||||
from telegram.ext import (
|
||||
Updater,
|
||||
CommandHandler,
|
||||
CallbackQueryHandler,
|
||||
ConversationHandler,
|
||||
CallbackContext,
|
||||
)
|
||||
|
||||
FIRST = 1
|
||||
|
||||
class Help():
|
||||
"""Shows the functions and their usage"""
|
||||
|
||||
def __init__(self):
|
||||
self.available_commands = {}
|
||||
|
||||
|
||||
def create_handler(self):
|
||||
conv_handler = ConversationHandler(
|
||||
entry_points=[CommandHandler('help', self.entry_point)],
|
||||
states={
|
||||
FIRST: [
|
||||
CallbackQueryHandler(self.print_all, pattern="^all$"),
|
||||
CallbackQueryHandler(self.choose_specific, pattern="^specific$"),
|
||||
CallbackQueryHandler(self.print_one, pattern='func-'),
|
||||
]
|
||||
},
|
||||
fallbacks=[CommandHandler('help', self.entry_point)],
|
||||
)
|
||||
return conv_handler
|
||||
|
||||
def add_commands(self, commands):
|
||||
# commands is a dict {"name": class}
|
||||
for k in commands:
|
||||
self.available_commands[k] = commands[k].__doc__
|
||||
|
||||
def entry_point(self, update: Update, context: CallbackContext) -> None:
|
||||
keyboard = [
|
||||
[
|
||||
InlineKeyboardButton("All commands", callback_data="all"),
|
||||
InlineKeyboardButton("Just one", callback_data="specific"),
|
||||
]
|
||||
]
|
||||
reply_markup = InlineKeyboardMarkup(keyboard)
|
||||
update.message.reply_text("What exactly do you want?", reply_markup=reply_markup)
|
||||
return FIRST
|
||||
|
||||
|
||||
def print_all(self, update: Update, context: CallbackContext) -> None:
|
||||
query = update.callback_query
|
||||
query.answer()
|
||||
|
||||
all_cmd = ""
|
||||
for k in self.available_commands:
|
||||
all_cmd += k + " - " + self.available_commands[k] +"\n"
|
||||
|
||||
query.edit_message_text(text="List of all commands:\n" + all_cmd)
|
||||
return ConversationHandler.END
|
||||
|
||||
|
||||
def choose_specific(self, update: Update, context: CallbackContext) -> None:
|
||||
query = update.callback_query
|
||||
query.answer()
|
||||
|
||||
|
||||
keyboard = [[InlineKeyboardButton(k, callback_data="func-" + k)] for k in self.available_commands]
|
||||
|
||||
reply_markup = InlineKeyboardMarkup(keyboard)
|
||||
query.edit_message_text(
|
||||
text="What command should be printed?", reply_markup=reply_markup
|
||||
)
|
||||
return FIRST
|
||||
|
||||
|
||||
def print_one(self, update: Update, context: CallbackContext) -> None:
|
||||
"""Show new choice of buttons"""
|
||||
query = update.callback_query
|
||||
data = query.data.replace("func-", "")
|
||||
|
||||
query.answer()
|
||||
message = self.available_commands[data]
|
||||
query.edit_message_text(
|
||||
text= message
|
||||
)
|
||||
return ConversationHandler.END
|
||||
92
bot2/commands/status.py
Normal file
92
bot2/commands/status.py
Normal file
@@ -0,0 +1,92 @@
|
||||
from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update
|
||||
from telegram.ext import Updater, CommandHandler, CallbackQueryHandler, CallbackContext
|
||||
from telegram.ext import (
|
||||
Updater,
|
||||
CommandHandler,
|
||||
CallbackQueryHandler,
|
||||
ConversationHandler,
|
||||
CallbackContext,
|
||||
)
|
||||
|
||||
import datetime
|
||||
import requests
|
||||
import socket
|
||||
import numpy as np
|
||||
FIRST = 1
|
||||
|
||||
class Status():
|
||||
"""Shows a short status of the program."""
|
||||
|
||||
def __init__(self, name, version, prst, logger):
|
||||
self.start_time = datetime.datetime.now()
|
||||
self.name = name
|
||||
self.version = version
|
||||
self.persistence = prst
|
||||
self.logger = logger
|
||||
|
||||
def create_handler(self):
|
||||
conv_handler = ConversationHandler(
|
||||
entry_points=[CommandHandler('status', self.entry_point)],
|
||||
states={
|
||||
FIRST: [
|
||||
CallbackQueryHandler(self.print_status, pattern="^status-"),
|
||||
]
|
||||
},
|
||||
fallbacks=[CommandHandler('status', self.entry_point)],
|
||||
)
|
||||
return conv_handler
|
||||
|
||||
|
||||
def entry_point(self, update: Update, context: CallbackContext) -> None:
|
||||
user = update.message.from_user
|
||||
keyboard = [
|
||||
[
|
||||
InlineKeyboardButton("Status", callback_data="status-simple"),
|
||||
InlineKeyboardButton("With log", callback_data="status-full"),
|
||||
]
|
||||
]
|
||||
reply_markup = InlineKeyboardMarkup(keyboard)
|
||||
update.message.reply_text("What exactly do you want?", reply_markup=reply_markup)
|
||||
return FIRST
|
||||
|
||||
|
||||
def print_status(self, update: Update, context: CallbackContext) -> None:
|
||||
query = update.callback_query
|
||||
wanted = query.data.replace("status-","")
|
||||
query.answer()
|
||||
|
||||
delta = str(datetime.datetime.now() - self.start_time)
|
||||
message = "BeebBop, this is " + self.name + " (V." + self.version + ")\n"
|
||||
|
||||
try:
|
||||
ip = requests.get('https://api.ipify.org').text
|
||||
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
|
||||
s.connect(('8.8.8.8', 80))
|
||||
(addr, port) = s.getsockname()
|
||||
local_ips = addr
|
||||
except:
|
||||
ip = "not fetchable"
|
||||
local_ips = "not fetchable"
|
||||
|
||||
message += "<pre>Status: Running 🟢\n"
|
||||
message += "Uptime: " + delta[:delta.rfind(".")] + "\n"
|
||||
message += "Reboots: " + str(self.persistence["global"]["reboots"]) + "\n"
|
||||
message += "IP (public): " + ip + "\n"
|
||||
message += "IP (private): " + str(local_ips) + "\n"
|
||||
tot_r = np.array(self.persistence["bot"]["receive_activity"]["count"]).sum()
|
||||
message += "Total messages read: " + str(tot_r) + "\n"
|
||||
|
||||
tot_s = np.array(self.persistence["bot"]["send_activity"]["count"]).sum()
|
||||
message += "Total messages sent: " + str(tot_s) + "\n"
|
||||
|
||||
tot_e = np.array(self.persistence["bot"]["execute_activity"]["count"]).sum()
|
||||
message += "Commands executed " + str(tot_e) + "</pre>\n"
|
||||
|
||||
if wanted == "full":
|
||||
message += str(dir(self.logger))
|
||||
|
||||
query.edit_message_text(
|
||||
text= message
|
||||
)
|
||||
|
||||
return ConversationHandler.END
|
||||
114
bot2/commands/weather.py
Normal file
114
bot2/commands/weather.py
Normal file
@@ -0,0 +1,114 @@
|
||||
from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update
|
||||
from telegram.ext import Updater, CommandHandler, CallbackQueryHandler, CallbackContext
|
||||
from telegram.ext import (
|
||||
Updater,
|
||||
CommandHandler,
|
||||
CallbackQueryHandler,
|
||||
ConversationHandler,
|
||||
CallbackContext,
|
||||
)
|
||||
|
||||
import datetime
|
||||
FIRST = 1
|
||||
|
||||
class Weather():
|
||||
"""Shows a weatherforecast for a given location"""
|
||||
def __init__(self, api):
|
||||
"""initialize api and persistence"""
|
||||
self.api = api
|
||||
self.city = ""
|
||||
|
||||
def create_handler(self):
|
||||
"""returns the handlers with button-logic"""
|
||||
conv_handler = ConversationHandler(
|
||||
entry_points=[CommandHandler('weather', self.entry_point)],
|
||||
states={
|
||||
FIRST: [
|
||||
CallbackQueryHandler(self.choose_city, pattern="^city-"),
|
||||
CallbackQueryHandler(self.choose_time, pattern="^time-"),
|
||||
]
|
||||
},
|
||||
fallbacks=[CommandHandler('weather', self.entry_point)],
|
||||
)
|
||||
|
||||
return conv_handler
|
||||
|
||||
|
||||
def entry_point(self, update: Update, context: CallbackContext) -> None:
|
||||
"""Reacts the call of the command. Prints the first buttons"""
|
||||
keyboard = [
|
||||
[
|
||||
InlineKeyboardButton("Zürich", callback_data="city-zurich"),
|
||||
InlineKeyboardButton("Freiburg", callback_data="city-freiburg"),
|
||||
InlineKeyboardButton("Mulhouse", callback_data="city-mulhouse"),
|
||||
]
|
||||
]
|
||||
reply_markup = InlineKeyboardMarkup(keyboard)
|
||||
update.message.reply_text("Which city?", reply_markup=reply_markup)
|
||||
return FIRST
|
||||
|
||||
|
||||
def choose_city(self, update: Update, context: CallbackContext) -> None:
|
||||
"""Prompt same text & keyboard as `start` does but not as new message"""
|
||||
# Get CallbackQuery from Update
|
||||
query = update.callback_query
|
||||
data = query.data
|
||||
self.city = data.replace("city-","")
|
||||
query.answer()
|
||||
keyboard = [
|
||||
[
|
||||
InlineKeyboardButton("Now", callback_data="time-now"),
|
||||
InlineKeyboardButton("Tomorrow", callback_data="time-tomorrow"),
|
||||
InlineKeyboardButton("7 days", callback_data="time-7"),
|
||||
]
|
||||
]
|
||||
reply_markup = InlineKeyboardMarkup(keyboard)
|
||||
query.edit_message_text(
|
||||
text = "Which time?", reply_markup=reply_markup
|
||||
)
|
||||
return FIRST
|
||||
|
||||
|
||||
def choose_time(self, update: Update, context: CallbackContext) -> None:
|
||||
"""Show new choice of buttons"""
|
||||
query = update.callback_query
|
||||
query.answer()
|
||||
forecast_time = query.data.replace("time-","")
|
||||
weather = self.get_weather(self.city, forecast_time)
|
||||
query.edit_message_text(
|
||||
text = "Weather: \n" + weather
|
||||
)
|
||||
return ConversationHandler.END
|
||||
|
||||
|
||||
def get_weather(self, city, forecast_time) -> None:
|
||||
"""get the weather that matches the given params"""
|
||||
locations = {"freiburg": [47.9990, 7.8421], "zurich": [47.3769, 8.5417], "mulhouse": [47.7508, 7.3359]}
|
||||
|
||||
city = locations[city]
|
||||
|
||||
categories = {"Clouds": "☁", "Rain": "🌧", "Thunderstorm": "🌩", "Drizzle": ":droplet:", "Snow": "❄", "Clear": "☀", "Mist": "🌫", "Smoke": "Smoke", "Haze": "Haze", "Dust": "Dust", "Fog": "Fog", "Sand": "Sand", "Dust": "Dust", "Ash": "Ash", "Squall": "Squall", "Tornado": "Tornado",}
|
||||
days = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
|
||||
today = datetime.datetime.today().weekday()
|
||||
weather = self.api.show_weather(city)
|
||||
message = ""
|
||||
if forecast_time == "now" or forecast_time == "7":
|
||||
now = weather.pop(0)
|
||||
message += "<b>Now:</b> " + categories[now["short"]] + "\n"
|
||||
message += "🌡" + str(now["temps"][0]) + "°\n\n"
|
||||
tod = weather.pop(0)
|
||||
message += "<b>" + "Today" + ":</b> " + categories[tod["short"]] + "\n"
|
||||
message += "🌡 ❄ " + str(tod["temps"][0]) + "° , 🌡 🔥 " + str(tod["temps"][1]) + "°\n\n"
|
||||
|
||||
if forecast_time == "tomorrow" or forecast_time == "7":
|
||||
tom = weather.pop(0)
|
||||
print(tom)
|
||||
message += "<b>" + "Tomorrow" + ":</b> " + categories[tom["short"]] + "\n"
|
||||
message += "🌡 ❄ " + str(tom["temps"][0]) + "° , 🌡 🔥 " + str(tom["temps"][1]) + "°\n\n"
|
||||
|
||||
if forecast_time == "7":
|
||||
for i, day in enumerate(weather):
|
||||
message += "<b>" + days[(today + i + 2) % 7] + ":</b> " + categories[day["short"]] + "\n"
|
||||
message += "🌡 ❄ " + str(day["temps"][0]) + "° , 🌡 🔥 " + str(day["temps"][1]) + "°\n\n"
|
||||
|
||||
return message
|
||||
Reference in New Issue
Block a user