ryg/royalnet
ryg/royalnet
1
Fork 0
mirror of https://github.com/RYGhub/royalnet.git synced 2024-11-27 13:34:28 +00:00
Code Activity

Handle database sessions in the telegram bot differently

Browse source
This commit is contained in:
Steffo 2019-02-09 20:10:52 +01:00
parent 11b119b08e
commit 80a2a7642b
1 changed files with 241 additions and 271 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

512
telegrambot.py
View file

@ -75,11 +75,13 @@ def reply(bot: telegram.Bot, update: telegram.Update, string: str, ignore_escapi
pass pass
def catch_and_report(func: "function"): # noinspection PyUnresolvedReferences
def command(func: "function"):
def new_func(bot: telegram.Bot, update: telegram.Update): def new_func(bot: telegram.Bot, update: telegram.Update):
# noinspection PyBroadException # noinspection PyBroadException
try: try:
return func(bot, update) session = db.Session()
return func(bot, update, session)
except TimedOut: except TimedOut:
logger.warning(f"Telegram timed out in {update}") logger.warning(f"Telegram timed out in {update}")
except Exception: except Exception:
@ -104,16 +106,18 @@ def catch_and_report(func: "function"):
"update": update.to_dict() "update": update.to_dict()
}) })
sentry.captureException() sentry.captureException()
finally:
session.close()
return new_func return new_func
@catch_and_report @command
def cmd_ping(bot: telegram.Bot, update: telegram.Update): def cmd_ping(bot: telegram.Bot, update: telegram.Update, session: db.Session):
reply(bot, update, strings.PONG) reply(bot, update, strings.PONG)
@catch_and_report @command
def cmd_link(bot: telegram.Bot, update: telegram.Update): def cmd_link(bot: telegram.Bot, update: telegram.Update, session: db.Session):
session = db.Session() session = db.Session()
try: try:
try: try:
@ -141,8 +145,8 @@ def cmd_link(bot: telegram.Bot, update: telegram.Update):
session.close() session.close()
@catch_and_report @command
def cmd_cv(bot: telegram.Bot, update: telegram.Update): def cmd_cv(bot: telegram.Bot, update: telegram.Update, session: db.Session):
if discord_connection is None: if discord_connection is None:
reply(bot, update, strings.TELEGRAM.ERRORS.INACTIVE_BRIDGE) reply(bot, update, strings.TELEGRAM.ERRORS.INACTIVE_BRIDGE)
return return
@ -155,8 +159,8 @@ def cmd_cv(bot: telegram.Bot, update: telegram.Update):
bot.send_message(update.message.chat.id, message, disable_web_page_preview=True, parse_mode="HTML") bot.send_message(update.message.chat.id, message, disable_web_page_preview=True, parse_mode="HTML")
@catch_and_report @command
def cmd_cast(bot: telegram.Bot, update: telegram.Update): def cmd_cast(bot: telegram.Bot, update: telegram.Update, session: db.Session):
try: try:
spell: str = update.message.text.split(" ", 1)[1] spell: str = update.message.text.split(" ", 1)[1]
except IndexError: except IndexError:
@ -176,28 +180,28 @@ def cmd_cast(bot: telegram.Bot, update: telegram.Update):
parse_mode="HTML") parse_mode="HTML")
@catch_and_report @command
def cmd_color(bot: telegram.Bot, update: telegram.Update): def cmd_color(bot: telegram.Bot, update: telegram.Update, session: db.Session):
bot.send_message(update.message.chat.id, "I am sorry, unknown error occured during working with your request," bot.send_message(update.message.chat.id, "I am sorry, unknown error occured during working with your request,"
" Admin were notified") " Admin were notified")
@catch_and_report @command
def cmd_smecds(bot: telegram.Bot, update: telegram.Update): def cmd_smecds(bot: telegram.Bot, update: telegram.Update, session: db.Session):
ds = random.sample(stagismo.listona, 1)[0] ds = random.sample(stagismo.listona, 1)[0]
bot.send_message(update.message.chat.id, f"Secondo me, è colpa {ds}.") bot.send_message(update.message.chat.id, f"Secondo me, è colpa {ds}.")
@catch_and_report @command
def cmd_ciaoruozi(bot: telegram.Bot, update: telegram.Update): def cmd_ciaoruozi(bot: telegram.Bot, update: telegram.Update, session: db.Session):
if update.message.from_user.username.lstrip("@") == "MeStakes": if update.message.from_user.username.lstrip("@") == "MeStakes":
bot.send_message(update.message.chat.id, "Ciao me!") bot.send_message(update.message.chat.id, "Ciao me!")
else: else:
bot.send_message(update.message.chat.id, "Ciao Ruozi!") bot.send_message(update.message.chat.id, "Ciao Ruozi!")
@catch_and_report @command
def cmd_ahnonlosoio(bot: telegram.Bot, update: telegram.Update): def cmd_ahnonlosoio(bot: telegram.Bot, update: telegram.Update, session: db.Session):
if update.message.reply_to_message is not None and update.message.reply_to_message.text in [ if update.message.reply_to_message is not None and update.message.reply_to_message.text in [
"/ahnonlosoio", "/ahnonlosoio@royalgamesbot", "Ah, non lo so io!", "Ah, non lo so neppure io!" "/ahnonlosoio", "/ahnonlosoio@royalgamesbot", "Ah, non lo so io!", "Ah, non lo so neppure io!"
]: ]:
@ -206,8 +210,8 @@ def cmd_ahnonlosoio(bot: telegram.Bot, update: telegram.Update):
bot.send_message(update.message.chat.id, "Ah, non lo so io!") bot.send_message(update.message.chat.id, "Ah, non lo so io!")
@catch_and_report @command
def cmd_balurage(bot: telegram.Bot, update: telegram.Update): def cmd_balurage(bot: telegram.Bot, update: telegram.Update, session: db.Session):
session = db.Session() session = db.Session()
try: try:
user = session.query(db.Telegram).filter_by(telegram_id=update.message.from_user.id).one_or_none() user = session.query(db.Telegram).filter_by(telegram_id=update.message.from_user.id).one_or_none()
@ -231,80 +235,70 @@ def cmd_balurage(bot: telegram.Bot, update: telegram.Update):
session.close() session.close()
@catch_and_report @command
def cmd_diario(bot: telegram.Bot, update: telegram.Update): def cmd_diario(bot: telegram.Bot, update: telegram.Update, session: db.Session):
session = db.Session() user = session.query(db.Telegram).filter_by(telegram_id=update.message.from_user.id).one_or_none()
if user is None:
reply(bot, update, strings.TELEGRAM.ERRORS.ROYALNET_NOT_LINKED)
return
try: try:
user = session.query(db.Telegram).filter_by(telegram_id=update.message.from_user.id).one_or_none() text = update.message.text.split(" ", 1)[1]
if user is None: author = session.query(db.Telegram).filter_by(telegram_id=update.message.from_user.id).one_or_none()
reply(bot, update, strings.TELEGRAM.ERRORS.ROYALNET_NOT_LINKED) saver = author
except IndexError:
if update.message.reply_to_message is None:
reply(bot, update, strings.DIARIO.ERRORS.INVALID_SYNTAX)
return return
try: text = update.message.reply_to_message.text
text = update.message.text.split(" ", 1)[1] author = session.query(db.Telegram)\
author = session.query(db.Telegram).filter_by(telegram_id=update.message.from_user.id).one_or_none() .filter_by(telegram_id=update.message.reply_to_message.from_user.id)\
saver = author .one_or_none()
except IndexError: saver = session.query(db.Telegram).filter_by(telegram_id=update.message.from_user.id).one_or_none()
if update.message.reply_to_message is None: if text is None:
reply(bot, update, strings.DIARIO.ERRORS.INVALID_SYNTAX) reply(bot, update, strings.DIARIO.ERRORS.NO_TEXT)
return return
text = update.message.reply_to_message.text diario = db.Diario(timestamp=datetime.datetime.now(),
author = session.query(db.Telegram)\ saver=saver,
.filter_by(telegram_id=update.message.reply_to_message.from_user.id)\ text=text)
.one_or_none() session.add(diario)
saver = session.query(db.Telegram).filter_by(telegram_id=update.message.from_user.id).one_or_none() session.commit()
if text is None: reply(bot, update, strings.DIARIO.SUCCESS, ignore_escaping=True, diario=diario.to_telegram())
reply(bot, update, strings.DIARIO.ERRORS.NO_TEXT)
return
diario = db.Diario(timestamp=datetime.datetime.now(),
saver=saver,
text=text)
session.add(diario)
session.commit()
reply(bot, update, strings.DIARIO.SUCCESS, ignore_escaping=True, diario=diario.to_telegram())
except Exception:
raise
finally:
session.close()
@catch_and_report @command
def cmd_vote(bot: telegram.Bot, update: telegram.Update): def cmd_vote(bot: telegram.Bot, update: telegram.Update, session: db.Session):
session = db.Session() user = session.query(db.Telegram).filter_by(telegram_id=update.message.from_user.id).one_or_none()
if user is None:
bot.send_message(update.message.chat.id,
"⚠ Il tuo account Telegram non è registrato al RYGdb!"
" Registrati con `/register@royalgamesbot <nomeutenteryg>`.", parse_mode="Markdown")
return
try: try:
user = session.query(db.Telegram).filter_by(telegram_id=update.message.from_user.id).one_or_none() _, mode, question = update.message.text.split(" ", 2)
if user is None: except IndexError:
bot.send_message(update.message.chat.id, bot.send_message(update.message.chat.id,
"⚠ Il tuo account Telegram non è registrato al RYGdb!" "⚠ Non hai specificato tutti i parametri necessari!"
" Registrati con `/register@royalgamesbot <nomeutenteryg>`.", parse_mode="Markdown") "Sintassi: `/vote@royalgamesbot <public|secret> <domanda>`", parse_mode="Markdown")
return return
try: if mode == "public":
_, mode, question = update.message.text.split(" ", 2) vote = db.VoteQuestion(question=question, anonymous=False)
except IndexError: elif mode == "secret":
bot.send_message(update.message.chat.id, vote = db.VoteQuestion(question=question, anonymous=True)
"⚠ Non hai specificato tutti i parametri necessari!" else:
"Sintassi: `/vote@royalgamesbot <public|secret> <domanda>`", parse_mode="Markdown") bot.send_message(update.message.chat.id,
return "⚠ Non hai specificato una modalità valida!"
if mode == "public": "Sintassi: `/vote@royalgamesbot <public|secret> <domanda>`", parse_mode="Markdown")
vote = db.VoteQuestion(question=question, anonymous=False) return
elif mode == "secret": session.add(vote)
vote = db.VoteQuestion(question=question, anonymous=True) session.flush()
else: inline_keyboard = IKMarkup([[IKButton("🔵 Sì", callback_data="vote_yes")],
bot.send_message(update.message.chat.id, [IKButton("🔴 No", callback_data="vote_no")],
"⚠ Non hai specificato una modalità valida!" [IKButton("⚫️ Astieniti", callback_data="vote_abstain")]])
"Sintassi: `/vote@royalgamesbot <public|secret> <domanda>`", parse_mode="Markdown") message = bot.send_message(update.message.chat.id, vote.generate_text(session=session),
return reply_markup=inline_keyboard,
session.add(vote) parse_mode="HTML")
session.flush() vote.message_id = message.message_id
inline_keyboard = IKMarkup([[IKButton("🔵 Sì", callback_data="vote_yes")], session.commit()
[IKButton("🔴 No", callback_data="vote_no")],
[IKButton("⚫️ Astieniti", callback_data="vote_abstain")]])
message = bot.send_message(update.message.chat.id, vote.generate_text(session=session),
reply_markup=inline_keyboard,
parse_mode="HTML")
vote.message_id = message.message_id
session.commit()
finally:
session.close()
def generate_search_message(term, entries): def generate_search_message(term, entries):
@ -324,80 +318,68 @@ def generate_search_message(term, entries):
return msg return msg
@catch_and_report @command
def cmd_search(bot: telegram.Bot, update: telegram.Update): def cmd_search(bot: telegram.Bot, update: telegram.Update, session: db.Session):
session = db.Session()
try: try:
try: query = update.message.text.split(" ", 1)[1]
query = update.message.text.split(" ", 1)[1] except IndexError:
except IndexError: reply(bot, update, strings.DIARIOSEARCH.ERRORS.INVALID_SYNTAX, command="search")
reply(bot, update, strings.DIARIOSEARCH.ERRORS.INVALID_SYNTAX, command="search") return
return query = query.replace('%', '\\%').replace('_', '\\_')
query = query.replace('%', '\\%').replace('_', '\\_') entries = session.query(db.Diario)\
entries = session.query(db.Diario)\ .filter(db.Diario.text.op("~*")(r"(?:[\s\.,:;!?\"'<{([]+|^)"
.filter(db.Diario.text.op("~*")(r"(?:[\s\.,:;!?\"'<{([]+|^)" + query +
+ query + r"(?:[\s\.,:;!?\"'>\})\]]+|$)"))\
r"(?:[\s\.,:;!?\"'>\})\]]+|$)"))\ .order_by(db.Diario.id.desc())\
.order_by(db.Diario.id.desc())\ .all()
.all() bot.send_message(update.message.chat.id, generate_search_message(f"<b>{query}</b>", entries), parse_mode="HTML")
bot.send_message(update.message.chat.id, generate_search_message(f"<b>{query}</b>", entries), parse_mode="HTML")
finally:
session.close()
@catch_and_report @command
def cmd_regex(bot: telegram.Bot, update: telegram.Update): def cmd_regex(bot: telegram.Bot, update: telegram.Update, session: db.Session):
session = db.Session()
try: try:
try: query = update.message.text.split(" ", 1)[1]
query = update.message.text.split(" ", 1)[1] except IndexError:
except IndexError: reply(bot, update, strings.DIARIOSEARCH.ERRORS.INVALID_SYNTAX, command="regex")
reply(bot, update, strings.DIARIOSEARCH.ERRORS.INVALID_SYNTAX, command="regex") return
return query = query.replace('%', '\\%').replace('_', '\\_')
query = query.replace('%', '\\%').replace('_', '\\_') entries = session.query(db.Diario).filter(db.Diario.text.op("~*")(query)).order_by(db.Diario.id.desc()).all()
entries = session.query(db.Diario).filter(db.Diario.text.op("~*")(query)).order_by(db.Diario.id.desc()).all()
try:
bot.send_message(update.message.chat.id, generate_search_message(f"<code>{query}</code>", entries), parse_mode="HTML")
except (BadRequest, TelegramError):
reply(bot, update, strings.DIARIOSEARCH.ERRORS.RESULTS_TOO_LONG)
finally:
session.close()
@catch_and_report
def cmd_mm(bot: telegram.Bot, update: telegram.Update):
session = db.Session()
try: try:
user = session.query(db.Telegram).filter_by(telegram_id=update.message.from_user.id).one_or_none() bot.send_message(update.message.chat.id, generate_search_message(f"<code>{query}</code>", entries), parse_mode="HTML")
if user is None: except (BadRequest, TelegramError):
reply(bot, update, strings.TELEGRAM.ERRORS.ROYALNET_NOT_LINKED) reply(bot, update, strings.DIARIOSEARCH.ERRORS.RESULTS_TOO_LONG)
return
match = re.match(r"/(?:mm|matchmaking)(?:@royalgamesbot)?(?: (?:([0-9]+)-)?([0-9]+))? (?:per )?([A-Za-z0-9!\-_\. ]+)(?:.*\n(.+))?",
update.message.text)
if match is None:
reply(bot, update, strings.MATCHMAKING.ERRORS.INVALID_SYNTAX)
return
min_players, max_players, match_name, match_desc = match.group(1, 2, 3, 4)
db_match = db.Match(timestamp=datetime.datetime.now(),
match_title=match_name,
match_desc=match_desc,
min_players=min_players,
max_players=max_players,
creator=user)
session.add(db_match)
session.flush()
inline_keyboard = IKMarkup([([IKButton(strings.MATCHMAKING.BUTTONS[key], callback_data=key)]) for key in strings.MATCHMAKING.BUTTONS])
message = bot.send_message(config["Telegram"]["announcement_group"], db_match.generate_text(session=session),
parse_mode="HTML",
reply_markup=inline_keyboard)
db_match.message_id = message.message_id
session.commit()
finally:
session.close()
@catch_and_report @command
def on_callback_query(bot: telegram.Bot, update: telegram.Update): def cmd_mm(bot: telegram.Bot, update: telegram.Update, session: db.Session):
user = session.query(db.Telegram).filter_by(telegram_id=update.message.from_user.id).one_or_none()
if user is None:
reply(bot, update, strings.TELEGRAM.ERRORS.ROYALNET_NOT_LINKED)
return
match = re.match(r"/(?:mm|matchmaking)(?:@royalgamesbot)?(?: (?:([0-9]+)-)?([0-9]+))? (?:per )?([A-Za-z0-9!\-_\. ]+)(?:.*\n(.+))?",
update.message.text)
if match is None:
reply(bot, update, strings.MATCHMAKING.ERRORS.INVALID_SYNTAX)
return
min_players, max_players, match_name, match_desc = match.group(1, 2, 3, 4)
db_match = db.Match(timestamp=datetime.datetime.now(),
match_title=match_name,
match_desc=match_desc,
min_players=min_players,
max_players=max_players,
creator=user)
session.add(db_match)
session.flush()
inline_keyboard = IKMarkup([([IKButton(strings.MATCHMAKING.BUTTONS[key], callback_data=key)]) for key in strings.MATCHMAKING.BUTTONS])
message = bot.send_message(config["Telegram"]["announcement_group"], db_match.generate_text(session=session),
parse_mode="HTML",
reply_markup=inline_keyboard)
db_match.message_id = message.message_id
session.commit()
@command
def on_callback_query(bot: telegram.Bot, update: telegram.Update, session: db.Session):
if update.callback_query.data.startswith("vote_"): if update.callback_query.data.startswith("vote_"):
if update.callback_query.data == "vote_yes": if update.callback_query.data == "vote_yes":
status = db.VoteChoices.YES status = db.VoteChoices.YES
@ -410,110 +392,102 @@ def on_callback_query(bot: telegram.Bot, update: telegram.Update):
emoji = "⚫️" emoji = "⚫️"
else: else:
raise NotImplementedError() raise NotImplementedError()
session = db.Session() user = session.query(db.Telegram).filter_by(telegram_id=update.callback_query.from_user.id).one_or_none()
try: if user is None:
user = session.query(db.Telegram).filter_by(telegram_id=update.callback_query.from_user.id).one_or_none() bot.answer_callback_query(update.callback_query.id, show_alert=True,
if user is None: text=strings.TELEGRAM.ERRORS.ROYALNET_NOT_LINKED,
bot.answer_callback_query(update.callback_query.id, show_alert=True, parse_mode="Markdown")
text=strings.TELEGRAM.ERRORS.ROYALNET_NOT_LINKED, return
parse_mode="Markdown") question = session.query(db.VoteQuestion)\
return .filter_by(message_id=update.callback_query.message.message_id)\
question = session.query(db.VoteQuestion)\ .one()
.filter_by(message_id=update.callback_query.message.message_id)\ answer = session.query(db.VoteAnswer).filter_by(question=question, user=user).one_or_none()
.one() if answer is None:
answer = session.query(db.VoteAnswer).filter_by(question=question, user=user).one_or_none() answer = db.VoteAnswer(question=question, choice=status, user=user)
if answer is None: session.add(answer)
answer = db.VoteAnswer(question=question, choice=status, user=user) bot.answer_callback_query(update.callback_query.id, text=f"Hai votato {emoji}.", cache_time=1)
session.add(answer) elif answer.choice == status:
bot.answer_callback_query(update.callback_query.id, text=f"Hai votato {emoji}.", cache_time=1) session.delete(answer)
elif answer.choice == status: bot.answer_callback_query(update.callback_query.id, text=f"Hai ritratto il tuo voto.", cache_time=1)
session.delete(answer) else:
bot.answer_callback_query(update.callback_query.id, text=f"Hai ritratto il tuo voto.", cache_time=1) answer.choice = status
else: bot.answer_callback_query(update.callback_query.id, text=f"Hai cambiato il tuo voto in {emoji}.",
answer.choice = status cache_time=1)
bot.answer_callback_query(update.callback_query.id, text=f"Hai cambiato il tuo voto in {emoji}.", session.commit()
cache_time=1) inline_keyboard = IKMarkup([[IKButton("🔵 Sì", callback_data="vote_yes")],
session.commit() [IKButton("🔴 No", callback_data="vote_no")],
inline_keyboard = IKMarkup([[IKButton("🔵 Sì", callback_data="vote_yes")], [IKButton("⚫️ Astieniti", callback_data="vote_abstain")]])
[IKButton("🔴 No", callback_data="vote_no")], bot.edit_message_text(message_id=update.callback_query.message.message_id,
[IKButton("⚫️ Astieniti", callback_data="vote_abstain")]]) chat_id=update.callback_query.message.chat.id,
bot.edit_message_text(message_id=update.callback_query.message.message_id, text=question.generate_text(session),
chat_id=update.callback_query.message.chat.id, reply_markup=inline_keyboard,
text=question.generate_text(session), parse_mode="HTML")
reply_markup=inline_keyboard,
parse_mode="HTML")
finally:
session.close()
elif update.callback_query.data.startswith("match_"): elif update.callback_query.data.startswith("match_"):
session = db.Session() user = session.query(db.Telegram).filter_by(telegram_id=update.callback_query.from_user.id).one_or_none()
try: if user is None:
user = session.query(db.Telegram).filter_by(telegram_id=update.callback_query.from_user.id).one_or_none() bot.answer_callback_query(update.callback_query.id,
if user is None: show_alert=True,
text=strings.TELEGRAM.ERRORS.ROYALNET_NOT_LINKED,
parse_mode="Markdown")
return
match = session.query(db.Match).filter_by(message_id=update.callback_query.message.message_id).one()
if update.callback_query.data == "match_close":
if match.creator != user:
bot.answer_callback_query(update.callback_query.id, bot.answer_callback_query(update.callback_query.id,
show_alert=True, show_alert=True,
text=strings.TELEGRAM.ERRORS.ROYALNET_NOT_LINKED, text=strings.MATCHMAKING.ERRORS.NOT_ADMIN)
parse_mode="Markdown")
return return
match = session.query(db.Match).filter_by(message_id=update.callback_query.message.message_id).one() match.closed = True
if update.callback_query.data == "match_close": for player in match.players:
if match.creator != user: if player.status >= 1:
bot.answer_callback_query(update.callback_query.id, reply_msg(bot, player.user.telegram_id, strings.MATCHMAKING.GAME_START[player.status], **match.format_dict())
show_alert=True, elif update.callback_query.data == "match_cancel":
text=strings.MATCHMAKING.ERRORS.NOT_ADMIN) if not (match.creator == user or user.telegram_id == 25167391):
return bot.answer_callback_query(update.callback_query.id,
match.closed = True show_alert=True,
for player in match.players: text=strings.MATCHMAKING.ERRORS.NOT_ADMIN)
if player.status >= 1: return
reply_msg(bot, player.user.telegram_id, strings.MATCHMAKING.GAME_START[player.status], **match.format_dict()) match.closed = True
elif update.callback_query.data == "match_cancel": status = {
if not (match.creator == user or user.telegram_id == 25167391): "match_ready": db.MatchmakingStatus.READY,
bot.answer_callback_query(update.callback_query.id, "match_wait_for_me": db.MatchmakingStatus.WAIT_FOR_ME,
show_alert=True, "match_maybe": db.MatchmakingStatus.MAYBE,
text=strings.MATCHMAKING.ERRORS.NOT_ADMIN) "match_ignore": db.MatchmakingStatus.IGNORED,
return "match_close": None,
match.closed = True "match_cancel": None,
status = { }.get(update.callback_query.data)
"match_ready": db.MatchmakingStatus.READY, if status:
"match_wait_for_me": db.MatchmakingStatus.WAIT_FOR_ME, if match.closed:
"match_maybe": db.MatchmakingStatus.MAYBE, bot.answer_callback_query(update.callback_query.id,
"match_ignore": db.MatchmakingStatus.IGNORED, show_alert=True,
"match_close": None, text=strings.MATCHMAKING.ERRORS.MATCH_CLOSED)
"match_cancel": None, return
}.get(update.callback_query.data) player = session.query(db.MatchPartecipation).filter_by(match=match, user=user).one_or_none()
if status: if player is None:
if match.closed: player = db.MatchPartecipation(match=match, status=status.value, user=user)
bot.answer_callback_query(update.callback_query.id, session.add(player)
show_alert=True,
text=strings.MATCHMAKING.ERRORS.MATCH_CLOSED)
return
player = session.query(db.MatchPartecipation).filter_by(match=match, user=user).one_or_none()
if player is None:
player = db.MatchPartecipation(match=match, status=status.value, user=user)
session.add(player)
else:
player.status = status.value
session.commit()
bot.answer_callback_query(update.callback_query.id,
text=strings.MATCHMAKING.TICKER_TEXT[update.callback_query.data],
cache_time=1)
if not match.closed:
inline_keyboard = IKMarkup([([IKButton(strings.MATCHMAKING.BUTTONS[key], callback_data=key)]) for key in strings.MATCHMAKING.BUTTONS])
else: else:
inline_keyboard = None player.status = status.value
try: session.commit()
bot.edit_message_text(message_id=update.callback_query.message.message_id, bot.answer_callback_query(update.callback_query.id,
chat_id=config["Telegram"]["announcement_group"], text=strings.MATCHMAKING.TICKER_TEXT[update.callback_query.data],
text=match.generate_text(session), cache_time=1)
reply_markup=inline_keyboard, if not match.closed:
parse_mode="HTML") inline_keyboard = IKMarkup([([IKButton(strings.MATCHMAKING.BUTTONS[key], callback_data=key)]) for key in strings.MATCHMAKING.BUTTONS])
except BadRequest: else:
pass inline_keyboard = None
finally: try:
session.close() bot.edit_message_text(message_id=update.callback_query.message.message_id,
chat_id=config["Telegram"]["announcement_group"],
text=match.generate_text(session),
reply_markup=inline_keyboard,
parse_mode="HTML")
except BadRequest:
pass
@catch_and_report @command
def cmd_eat(bot: telegram.Bot, update: telegram.Update): def cmd_eat(bot: telegram.Bot, update: telegram.Update, session: db.Session):
try: try:
food: str = update.message.text.split(" ", 1)[1].capitalize() food: str = update.message.text.split(" ", 1)[1].capitalize()
except IndexError: except IndexError:
@ -525,8 +499,8 @@ def cmd_eat(bot: telegram.Bot, update: telegram.Update):
reply(bot, update, strings.EAT.NORMAL, food=food) reply(bot, update, strings.EAT.NORMAL, food=food)
@catch_and_report @command
def cmd_ship(bot: telegram.Bot, update: telegram.Update): def cmd_ship(bot: telegram.Bot, update: telegram.Update, session: db.Session):
try: try:
_, name_one, name_two = update.message.text.split(" ", 2) _, name_one, name_two = update.message.text.split(" ", 2)
except ValueError: except ValueError:
@ -546,8 +520,8 @@ def cmd_ship(bot: telegram.Bot, update: telegram.Update):
result=mixed.capitalize()) result=mixed.capitalize())
@catch_and_report @command
def cmd_bridge(bot: telegram.Bot, update: telegram.Update): def cmd_bridge(bot: telegram.Bot, update: telegram.Update, session: db.Session):
try: try:
data = update.message.text.split(" ", 1)[1] data = update.message.text.split(" ", 1)[1]
except IndexError: except IndexError:
@ -612,8 +586,8 @@ def parse_timestring(timestring: str) -> typing.Union[datetime.timedelta, dateti
raise ValueError("Nothing was found.") raise ValueError("Nothing was found.")
@catch_and_report @command
def cmd_newevent(bot: telegram.Bot, update: telegram.Update): def cmd_newevent(bot: telegram.Bot, update: telegram.Update, session: db.Session):
try: try:
_, timestring, name_desc = update.message.text.split(" ", 2) _, timestring, name_desc = update.message.text.split(" ", 2)
except ValueError: except ValueError:
@ -645,7 +619,6 @@ def cmd_newevent(bot: telegram.Bot, update: telegram.Update):
"per favore inserisci una data futura.\n", parse_mode="Markdown") "per favore inserisci una data futura.\n", parse_mode="Markdown")
return return
# Create the event # Create the event
session = db.Session()
telegram_user = session.query(db.Telegram)\ telegram_user = session.query(db.Telegram)\
.filter_by(telegram_id=update.message.from_user.id)\ .filter_by(telegram_id=update.message.from_user.id)\
.join(db.Royal)\ .join(db.Royal)\
@ -662,15 +635,12 @@ def cmd_newevent(bot: telegram.Bot, update: telegram.Update):
# Save the event # Save the event
session.add(event) session.add(event)
session.commit() session.commit()
session.close()
bot.send_message(update.message.chat.id, "✅ Evento aggiunto al Calendario Royal Games!") bot.send_message(update.message.chat.id, "✅ Evento aggiunto al Calendario Royal Games!")
@catch_and_report @command
def cmd_calendar(bot: telegram.Bot, update: telegram.Update): def cmd_calendar(bot: telegram.Bot, update: telegram.Update, session: db.Session):
session = db.Session()
next_events = session.query(db.Event).filter(db.Event.time > datetime.datetime.now()).order_by(db.Event.time).all() next_events = session.query(db.Event).filter(db.Event.time > datetime.datetime.now()).order_by(db.Event.time).all()
session.close()
msg = "📆 Prossimi eventi\n" msg = "📆 Prossimi eventi\n"
for event in next_events: for event in next_events:
if event.time_left.days >= 1: if event.time_left.days >= 1:
@ -683,8 +653,8 @@ def cmd_calendar(bot: telegram.Bot, update: telegram.Update):
bot.send_message(update.message.chat.id, msg, parse_mode="HTML", disable_web_page_preview=True) bot.send_message(update.message.chat.id, msg, parse_mode="HTML", disable_web_page_preview=True)
@catch_and_report @command
def cmd_markov(bot: telegram.Bot, update: telegram.Update): def cmd_markov(bot: telegram.Bot, update: telegram.Update, session: db.Session):
if model is None: if model is None:
bot.send_message(update.message.chat.id, strings.MARKOV.ERRORS.NO_MODEL) bot.send_message(update.message.chat.id, strings.MARKOV.ERRORS.NO_MODEL)
return return
@ -710,8 +680,8 @@ def cmd_markov(bot: telegram.Bot, update: telegram.Update):
bot.send_message(update.message.chat.id, sentence) bot.send_message(update.message.chat.id, sentence)
@catch_and_report @command
def cmd_roll(bot: telegram.Bot, update: telegram.Update): def cmd_roll(bot: telegram.Bot, update: telegram.Update, session: db.Session):
dice_string = update.message.text.split(" ", 1)[1] dice_string = update.message.text.split(" ", 1)[1]
try: try:
result = dice.roll(f"{dice_string}t") result = dice.roll(f"{dice_string}t")
@ -721,8 +691,8 @@ def cmd_roll(bot: telegram.Bot, update: telegram.Update):
bot.send_message(update.message.chat.id, f"🎲 {result}") bot.send_message(update.message.chat.id, f"🎲 {result}")
@catch_and_report @command
def cmd_start(bot: telegram.Bot, update: telegram.Update): def cmd_start(bot: telegram.Bot, update: telegram.Update, session: db.Session):
reply(bot, update, strings.TELEGRAM.BOT_STARTED) reply(bot, update, strings.TELEGRAM.BOT_STARTED)