...
 
Commits (4)
......@@ -20,8 +20,10 @@ class BaseModel(Model):
return items
@classmethod
@handle_errors
def get_all(cls):
return handle_errors(cls._get_all)
"""Returns a list with all table entries."""
return cls._get_all()
@classmethod
def _get_entry(cls):
......@@ -30,45 +32,58 @@ class BaseModel(Model):
return cls._create()
@classmethod
@handle_errors
def get_entry(cls):
return handle_errors(cls._get_entry)
"""Returns the first entry in the table."""
return cls._get_entry()
@classmethod
def _update_or_create(cls, id, **kwargs):
def _update_or_create(cls, pk_name, **kwargs):
# update or create für chat admin list sollte diese Methode nutzen:
# http://docs.peewee-orm.com/en/latest/peewee/api.html#Model.get_or_create
try:
item = cls._get(id)
item.__data__ = {"id": id, **kwargs}
item = cls._get(kwargs[pk_name])
item.__data__ = {**kwargs}
except DoesNotExist:
item = cls._create(id=id, **kwargs)
item = cls._create(**kwargs)
item.save()
return item
@classmethod
def update_or_create(cls, id, **kwargs):
return handle_errors(cls._update_or_create, id, **kwargs)
@handle_errors
def update_or_create(cls, pk_name, **kwargs):
"""Updates an entry or creates a new one.
Attributes:
pk_name: Name of the primary key. (example: "id")
**kwargs: Dict with all members of the entry.
(example: {"id": 7, "name": "Bernd"})
"""
return cls._update_or_create(pk_name, **kwargs)
@classmethod
def _create(cls, *args, **kwargs):
return super().create(*args, **kwargs)
def _create(cls, **query):
return super().create(**query)
@classmethod
def create(cls, *args, **kwargs):
return handle_errors(cls._create, *args, **kwargs)
@handle_errors
def create(cls, **query):
return cls._create(**query)
@classmethod
def _select(cls, *args, **kwargs):
return super().select(*args, **kwargs)
def _select(cls, *fields):
return super().select(*fields)
@classmethod
def select(cls, *args, **kwargs):
return handle_errors(cls._select, *args, **kwargs)
@handle_errors
def select(cls, *fields):
return cls._select(*fields)
@classmethod
def _get(cls, *args, **kwargs):
return super().get(*args, **kwargs)
def _get(cls, *query, **filters):
return super().get(*query, **filters)
@classmethod
def get(cls, *args, **kwargs):
return handle_errors(cls._get, *args, **kwargs)
@handle_errors
def get(cls, *query, **filters):
return cls._get(*query, **filters)
......@@ -46,31 +46,35 @@ def __get_peewee_error_handlers(
(OperationalError, __operational_error_handler)]
def __error_callback(e: Exception) -> None:
def __error_callback(error: Exception) -> bool:
"""Returns true if error is fixed."""
print("An error occurred: {}".format(e))
print("An error occurred: {}".format(error))
peewee_err = __get_peewee_error_handlers()
for ex, exfunc in peewee_err:
if isinstance(e, ex):
if isinstance(error, ex):
print("Trying to fix the error.")
exfunc(e)
return
exfunc(error)
return True
print("No error handler found.")
return False
def handle_errors(callback: Callable, *args, **kwargs):
try:
ret = callback(*args, **kwargs)
return ret
except Exception as error:
__error_callback(error)
# after all tries, try again with killing main
try:
ret = callback(*args, **kwargs)
print("Error fix succeeded.")
return ret
except Exception:
print("Unable to fix error. Exiting main!")
print(traceback.format_exc())
interrupt_main()
raise Exception()
def handle_errors(callback: Callable):
def deco(*args, **kwargs):
try:
ret = callback(*args, **kwargs)
return ret
except Exception as error:
fixed = __error_callback(error)
if not fixed:
raise error
# after all tries, try again with killing main
try:
ret = callback(*args, **kwargs)
print("Error fix succeeded.")
return ret
except Exception as error:
print("Unable to fix error. Exiting main!")
interrupt_main()
raise error
return deco
......@@ -76,25 +76,26 @@ def create_w_chat_from_bot(chat_id: Union[int, str], bot: Bot, chat_prio: int,
except ValueError:
continue
picture = None
admin = W_User.update_or_create(
id=admin_raw['user']['id'],
username=admin_raw['user']['username'],
first_name=admin_raw['user']['first_name'],
picture=picture,
is_bot=admin_raw['user']['is_bot'])
user_dict = {"id": admin_raw['user']['id'],
"username": admin_raw['user']['username'],
"first_name": admin_raw['user']['first_name'],
"picture": picture,
"is_bot": admin_raw['user']['is_bot']}
admin = W_User.update_or_create("id", **user_dict)
admins.append(admin)
chat = W_Chat.update_or_create(id=res['id'],
channel=channel,
title=res['title'],
username=res['username'],
name=name,
fullname=fullname,
photo=photo,
pinned_message=pinned_message,
description=res['description'],
user_amount=amount_users,
invite_link=invite_link,
prio=chat_prio)
chat_dict = {"id": res['id'],
"channel": channel,
"title": res['title'],
"username": res['username'],
"name": name,
"fullname": fullname,
"photo": photo,
"pinned_message": pinned_message,
"description": res['description'],
"user_amount": amount_users,
"invite_link": invite_link,
"prio": chat_prio}
chat = W_Chat.update_or_create("id", **chat_dict)
for admin in admins:
W_ChatAdmin.update_or_create_chat_admin(admin=admin, chat=chat)
return chat
......@@ -58,11 +58,12 @@ class W_ChatAdmin(BaseModel):
# update or create für chat admin list sollte diese Methode nutzen:
# http://docs.peewee-orm.com/en/latest/peewee/api.html#Model.get_or_create
try:
item = cls._get(admin=admin, chat=chat)
item = super()._get(admin=admin, chat=chat)
except DoesNotExist:
item = cls._create(admin=admin, chat=chat)
item = super()._create(admin=admin, chat=chat)
return item
@classmethod
@handle_errors
def update_or_create_chat_admin(cls, admin, chat):
return handle_errors(cls._update_or_create_chat_admin, admin, chat)
return cls._update_or_create_chat_admin(admin, chat)
from __future__ import annotations
from telegram import Bot, ReplyKeyboardRemove, Update
from telegram.ext import CommandHandler, Handler
from gmb.pollingbot import PollBotExt
from typing import List
from gmb.telegram_util import send_custom_reply
from .infofetchbot_config import InfoFetchConfig
class InfoFetchConfigExt(PollBotExt):
def get_handlers(self: InfoFetchConfigExt) -> List[Handler]:
return [
CommandHandler('set_update_time', self.set_update_time),
CommandHandler('reset_update_time', self.reset_update_time),
CommandHandler('set_prio_standard', self.set_prio_standard),
CommandHandler('reset_prio_standard', self.reset_prio_standard)
]
def set_update_time(self: InfoFetchConfigExt, bot: Bot, update: Update) -> None:
# TODO remove command from message?
new_time = update.message.text
try:
new_time = int(new_time)
except ValueError:
msg = "Not updated! New update time is not a number."
send_custom_reply(bot, update, msg)
entry = InfoFetchConfig.get_entry()
entry.update_time = new_time
entry.save()
msg = "Update time updated."
send_custom_reply(bot, update, msg)
def reset_update_time(self: InfoFetchConfigExt,
bot: Bot, update: Update) -> None:
update_time = InfoFetchConfig.update_time.default
entry = InfoFetchConfig.get_entry()
entry.update_time = update_time
entry.save()
msg = "Update time resetted."
send_custom_reply(bot, update, msg)
def set_prio_standard(self: InfoFetchConfigExt,
bot: Bot, update: Update) -> None:
# TODO remove command from message?
new_prio = update.message.text
entry = InfoFetchConfig.get_entry()
entry.prio_standard = new_prio
entry.save()
msg = "Prio standard updated."
send_custom_reply(bot, update, msg)
def reset_prio_standard(self: InfoFetchConfigExt,
bot: Bot, update: Update) -> None:
prio_standard = InfoFetchConfig.prio_standard.default
entry = InfoFetchConfig.get_entry()
entry.prio_standard = prio_standard
entry.save()
msg = "Prio standard resetted."
send_custom_reply(bot, update, msg)
......@@ -6,21 +6,16 @@ import time
from typing import List
import telegram
from peewee import IntegerField, TextField
from telegram import Bot
from gmb.config.config import StaticConfig
from gmb.config.db_config import ChatsConfig
from gmb.database.basemodel import BaseModel
from .infofetchbot_config import InfoFetchConfig
from gmb.datastructs.factories import create_w_chat_from_bot
from gmb.defaults import STATUS_INFOFETCH_FETCHING_FILENAME
from gmb.status import DestructionStatus
class InfoFetchConfig(BaseModel):
update_time = IntegerField(null=False, default=600)
prio_standard = TextField(null=False, default=0)
class InfoFetch:
"""Telegram Bot connection, that gets some information about chats."""
......
from gmb.database.basemodel import BaseModel
from peewee import IntegerField, TextField
class InfoFetchConfig(BaseModel):
update_time = IntegerField(null=False, default=600)
prio_standard = TextField(null=False, default=0)
......@@ -80,11 +80,12 @@ def migrate_chatsconfig(config: JsonConfigV224):
prio = int(chat["prio"])
except ValueError:
prio = infofetch_config.prio_standard
ChatsConfig.update_or_create(id=id, name=name, prio=prio)
cc_dict = {"id": id, "name": name, "prio": prio}
ChatsConfig.update_or_create(pk_name="id", **cc_dict)
class JsonConfigV224:
def __init__(self: Config, filename: str):
def __init__(self: JsonConfigV224, filename: str):
self.filename = filename # not saved
self.token = ""
self.chats: List[Dict[str, Union[str, int]]] = []
......@@ -103,7 +104,7 @@ class JsonConfigV224:
self.license_message = "Welcome!\nThis bot is a program which is available under the <b>GNU AGPLv3</b> license at https://git.thevillage.chat/thevillage/groupmanagementbot."
self.dsgvo_message = "Dieser Bot speichert <b>keine persönlichen Daten</b> über normale User wie dich. Jedoch speichert er einige <b>Informationen über unsere Admins</b> ab. Diese wären der <b>Username</b>, der <b>Vorname</b> und ob ein Admin ein <b>Bot ist</b>. Dies wird benötigt um die Admins auf unserer Webseite zu verlinken."
def load(self: Config) -> None:
def load(self: JsonConfigV224) -> None:
try:
file = open(self.filename, "r")
conf_raw = file.read()
......
......@@ -5,7 +5,7 @@ from typing import List
import telegram
from peewee import BooleanField, CharField, TextField
from telegram import (Bot, InlineKeyboardButton, InlineKeyboardMarkup, Message,
Update)
Update, CommandHandler)
from telegram.ext import BaseFilter, Handler, MessageHandler
from gmb.config.db_config import ChatsConfig
......@@ -19,8 +19,7 @@ class LeaveChatConfig(BaseModel):
leave_unknown_chats = BooleanField(null=False, default=True)
leave_message = TextField(
null=False,
default=
'Diese Gruppe gehört nicht zum <a href="https://t.me/TheVillageChats">TheVillage Chats</a> Netzwerk.\n\n<code>{chat_name} ({chat_id})</code>'
default='Diese Gruppe gehört nicht zum <a href="https://t.me/TheVillageChats">TheVillage Chats</a> Netzwerk.\n\n<code>{chat_name} ({chat_id})</code>'
)
leave_button_1_text = CharField(null=False, default="Chat")
leave_button_1_link = CharField(null=False,
......@@ -30,6 +29,141 @@ class LeaveChatConfig(BaseModel):
default="https://thevillage.chat")
class LeaveChatConfigExt(PollBotExt):
def get_handlers(self: LeaveChatConfigExt) -> List[Handler]:
return [
CommandHandler('toggle_leave_chats', self.toggle_leave_chats),
CommandHandler('reset_leave_chats', self.reset_leave_chats),
CommandHandler('set_leave_message', self.set_leave_message),
CommandHandler('reset_leave_message', self.reset_leave_message),
CommandHandler('set_leave_button_1_text', self.set_leave_button_1_text),
CommandHandler('reset_leave_button_1_text', self.reset_leave_button_1_text),
CommandHandler('set_leave_button_1_link', self.set_leave_button_1_link),
CommandHandler('reset_leave_button_1_link', self.reset_leave_button_1_link),
CommandHandler('set_leave_button_2_text', self.set_leave_button_2_text),
CommandHandler('reset_leave_button_2_text', self.reset_leave_button_2_text),
CommandHandler('set_leave_button_2_link', self.set_leave_button_2_link),
CommandHandler('reset_leave_button_2_link', self.reset_leave_button_2_link)
]
def toggle_leave_chats(self: LeaveChatConfigExt, bot: Bot, update: Update) -> None:
# TODO remove command from message?
entry = LeaveChatConfig.get_entry()
entry.leave_unknown_chats = not entry.leave_unknown_chats
entry.save()
if entry.leave_unknown_chats:
msg = "Leave chats is enabled."
else:
msg = "Leave chats is disabled."
send_custom_reply(bot, update, msg)
def reset_leave_chats(self: LeaveChatConfigExt, bot: Bot, update: Update) -> None:
# TODO remove command from message?
entry = LeaveChatConfig.get_entry()
entry.leave_unknown_chats = LeaveChatConfig.leave_unknown_chats.default
entry.save()
if entry.leave_unknown_chats:
msg = "Leave chats is enabled."
else:
msg = "Leave chats is disabled."
send_custom_reply(bot, update, msg)
def set_leave_message(self: LeaveChatConfigExt,
bot: Bot, update: Update) -> None:
# TODO remove command from message?
new_msg = update.message.text
entry = LeaveChatConfig.get_entry()
entry.leave_message = new_msg
entry.save()
msg = "Leave message updated."
send_custom_reply(bot, update, msg)
def reset_leave_message(self: LeaveChatConfigExt,
bot: Bot, update: Update) -> None:
new_msg = LeaveChatConfig.leave_message.default
entry = LeaveChatConfig.get_entry()
entry.leave_message = new_msg
entry.save()
msg = "Start message resetted."
send_custom_reply(bot, update, msg)
def set_leave_button_1_text(self: LeaveChatConfigExt,
bot: Bot, update: Update) -> None:
# TODO remove command from message?
new_msg = update.message.text
entry = LeaveChatConfig.get_entry()
entry.leave_button_1_text = new_msg
entry.save()
msg = "Leave button 1 text updated."
send_custom_reply(bot, update, msg)
def reset_leave_button_1_text(self: LeaveChatConfigExt,
bot: Bot, update: Update) -> None:
new_msg = LeaveChatConfig.leave_button_1_text.default
entry = LeaveChatConfig.get_entry()
entry.leave_button_1_text = new_msg
entry.save()
msg = "Leave button 1 text resetted."
send_custom_reply(bot, update, msg)
def set_leave_button_1_link(self: LeaveChatConfigExt,
bot: Bot, update: Update) -> None:
# TODO remove command from message?
new_msg = update.message.text
entry = LeaveChatConfig.get_entry()
entry.leave_button_1_link = new_msg
entry.save()
msg = "Leave button 1 link updated."
send_custom_reply(bot, update, msg)
def reset_leave_button_1_link(self: LeaveChatConfigExt,
bot: Bot, update: Update) -> None:
new_msg = LeaveChatConfig.leave_button_1_link.default
entry = LeaveChatConfig.get_entry()
entry.leave_button_1_link = new_msg
entry.save()
msg = "Leave button 1 link resetted."
send_custom_reply(bot, update, msg)
def set_leave_button_2_text(self: LeaveChatConfigExt,
bot: Bot, update: Update) -> None:
# TODO remove command from message?
new_msg = update.message.text
entry = LeaveChatConfig.get_entry()
entry.leave_button_2_text = new_msg
entry.save()
msg = "Leave button 2 text updated."
send_custom_reply(bot, update, msg)
def reset_leave_button_2_text(self: LeaveChatConfigExt,
bot: Bot, update: Update) -> None:
new_msg = LeaveChatConfig.leave_button_2_text.default
entry = LeaveChatConfig.get_entry()
entry.leave_button_2_text = new_msg
entry.save()
msg = "Leave button 2 text resetted."
send_custom_reply(bot, update, msg)
def set_leave_button_2_link(self: LeaveChatConfigExt,
bot: Bot, update: Update) -> None:
# TODO remove command from message?
new_msg = update.message.text
entry = LeaveChatConfig.get_entry()
entry.leave_button_2_link = new_msg
entry.save()
msg = "Leave button 2 link updated."
send_custom_reply(bot, update, msg)
def reset_leave_button_2_link(self: LeaveChatConfigExt,
bot: Bot, update: Update) -> None:
new_msg = LeaveChatConfig.leave_button_2_link.default
entry = LeaveChatConfig.get_entry()
entry.leave_button_2_link = new_msg
entry.save()
msg = "Leave button 2 link resetted."
send_custom_reply(bot, update, msg)
class LeaveChatFilter(BaseFilter):
"""Filter for leaving a chat.
......
......@@ -16,21 +16,105 @@ class SmallCommandConfig(BaseModel):
start_message = TextField(null=False, default="Hallo :)")
license_message = TextField(
null=False,
default=
"Welcome!\nThis bot is a program which is available under the <b>GNU AGPLv3</b> license at https://git.thevillage.chat/thevillage/groupmanagementbot."
default="Welcome!\nThis bot is a program which is available under the <b>GNU AGPLv3</b> license at https://git.thevillage.chat/thevillage/groupmanagementbot."
)
help_message = TextField(
null=False,
default=
'Dieser Bot ist Teil des <a href="t.me/thevillagechats">TheVillage Chat</a> Netzwerkes. Er ist nur in TheVillage Gruppen und sammelt Informationen für unsere <a href="https://thevillage.chat">Webseite</a>\n\nZum einfachereren Verlinken der Chats schickt der Bot die Invite Links der Gruppen, wenn man den Hashtag derselbigen schreibt (z.B.: #dorfzentrum)\n\nBefehle:\n/help - <i>Zeigt diese Hilfenachricht an</i>\n/license - <i>Zeigt die Lizenz an</i>\n/dsgvo - <i>Unsere Datenschutzerklärung</i>\n/groups - <i>Zeigt eine Liste mit allen TheVillage Chats an</i>'
default='Dieser Bot ist Teil des <a href="t.me/thevillagechats">TheVillage Chat</a> Netzwerkes. Er ist nur in TheVillage Gruppen und sammelt Informationen für unsere <a href="https://thevillage.chat">Webseite</a>\n\nZum einfachereren Verlinken der Chats schickt der Bot die Invite Links der Gruppen, wenn man den Hashtag derselbigen schreibt (z.B.: #dorfzentrum)\n\nBefehle:\n/help - <i>Zeigt diese Hilfenachricht an</i>\n/license - <i>Zeigt die Lizenz an</i>\n/dsgvo - <i>Unsere Datenschutzerklärung</i>\n/groups - <i>Zeigt eine Liste mit allen TheVillage Chats an</i>'
)
dsgvo_message = TextField(
null=False,
default=
"Dieser Bot speichert <b>keine persönlichen Daten</b> über normale User wie dich. Jedoch speichert er einige <b>Informationen über unsere Admins</b> ab. Diese wären der <b>Username</b>, der <b>Vorname</b> und ob ein Admin ein <b>Bot ist</b>. Dies wird benötigt um die Admins auf unserer Webseite zu verlinken."
default="Dieser Bot speichert <b>keine persönlichen Daten</b> über normale User wie dich. Jedoch speichert er einige <b>Informationen über unsere Admins</b> ab. Diese wären der <b>Username</b>, der <b>Vorname</b> und ob ein Admin ein <b>Bot ist</b>. Dies wird benötigt um die Admins auf unserer Webseite zu verlinken."
)
class SmallCommandConfigExt(PollBotExt):
def get_handlers(self: SmallCommandConfigExt) -> List[Handler]:
return [
CommandHandler('set_start_message', self.set_start_message),
CommandHandler('set_license_message', self.set_license_message),
CommandHandler('set_help_message', self.set_help_message),
CommandHandler('set_dsgvo_message', self.reset_dsgvo_message),
CommandHandler('reset_start_message', self.reset_start_message),
CommandHandler('reset_license_message', self.reset_license_message),
CommandHandler('reset_help_message', self.reset_help_message),
CommandHandler('reset_dsgvo_message', self.reset_dsgvo_message)
]
def set_start_message(self: SmallCommandConfigExt,
bot: Bot, update: Update) -> None:
# TODO remove command from message?
new_msg = update.message.text
entry = SmallCommandConfig.get_entry()
entry.start_message = new_msg
entry.save()
msg = "Start message updated."
send_custom_reply(bot, update, msg)
def reset_start_message(self: SmallCommandConfigExt,
bot: Bot, update: Update) -> None:
start_message = SmallCommandConfig.start_message.default
entry = SmallCommandConfig.get_entry()
entry.start_message = start_message
entry.save()
msg = "Start message resetted."
send_custom_reply(bot, update, msg)
def set_license_message(self: SmallCommandConfigExt,
bot: Bot, update: Update) -> None:
new_msg = update.message.text
entry = SmallCommandConfig.get_entry()
entry.license_message = new_msg
entry.save()
msg = "License message updated."
send_custom_reply(bot, update, msg)
def reset_license_message(self: SmallCommandConfigExt,
bot: Bot, update: Update) -> None:
license_message = SmallCommandConfig.license_message.default
entry = SmallCommandConfig.get_entry()
entry.license_message = license_message
entry.save()
msg = "License message resetted."
send_custom_reply(bot, update, msg)
def set_help_message(self: SmallCommandConfigExt,
bot: Bot, update: Update) -> None:
new_msg = update.message.text
entry = SmallCommandConfig.get_entry()
entry.help_message = new_msg
entry.save()
msg = "Help message updated."
send_custom_reply(bot, update, msg)
def reset_help_message(self: SmallCommandConfigExt,
bot: Bot, update: Update) -> None:
help_message = SmallCommandConfig.help_message.default
entry = SmallCommandConfig.get_entry()
entry.help_message = help_message
entry.save()
msg = "Help message resetted."
send_custom_reply(bot, update, msg)
def set_dsgvo_message(self: SmallCommandConfigExt,
bot: Bot, update: Update) -> None:
new_msg = update.message.text
entry = SmallCommandConfig.get_entry()
entry.dsgvo_message = new_msg
entry.save()
msg = "Dsgvo message updated."
send_custom_reply(bot, update, msg)
def reset_dsgvo_message(self: SmallCommandConfigExt,
bot: Bot, update: Update) -> None:
dsgvo_message = SmallCommandConfig.dsgvo_message.default
entry = SmallCommandConfig.get_entry()
entry.dsgvo_message = dsgvo_message
entry.save()
msg = "Dsgvo message resetted."
send_custom_reply(bot, update, msg)
class SmallCommandExt(PollBotExt):
"""Extension for small commands that just sends a single message."""
......
......@@ -7,6 +7,8 @@ from gmb.pollingbot.pollingbot import PollingBotExtensions
def register_pollingbot_extensions():
PollingBotExtensions.add_extension(SmallCommandExt)
PollingBotExtensions.add_extension(SmallCommandConfigExt)
PollingBotExtensions.add_extension(SendInviteLinkExt)
PollingBotExtensions.add_extension(LeaveChatExt)
PollingBotExtensions.add_extension(LeaveChatConfigExt)
PollingBotExtensions.add_extension(GroupsListExt)