felixm
/
TagTimeBot
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Implement ability to respond to all pending pings at once.

main
felixm 2024-01-09 18:42:26 -05:00
parent e6473513f0
commit cbe4db15e8
3 changed files with 96 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
.gitignore vendored
View File

@ -1,4 +1,5 @@
token.txt
__pycache__
deploy.sh
persistence.pkl
tagtimebot.log
__pycache__
token*.txt

2
README.md
View File

@ -5,7 +5,7 @@ Place Telegram bot API key into `token.txt`.
```
pipenv update
pipenv shell
python main.py
python main.py token.txt
```
Currently deployed as `@tagtime_bot`.

127
main.py
View File

@ -1,5 +1,6 @@
import logging
import timer
import sys
from collections import OrderedDict
from typing import Optional
from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update
@ -25,6 +26,9 @@ class UserData:
if not "pending_pings" in user_data:
user_data["pending_pings"] = {}
if not "reply_all" in user_data:
user_data["reply_all"] = None
self.user_data = user_data
@property
@ -43,6 +47,14 @@ class UserData:
def pings(self):
return self.user_data["pings"]
@property
def reply_all(self) -> Optional[tuple[int, list[str]]]:
return self.user_data["reply_all"]
@reply_all.setter
def reply_all(self, value: Optional[tuple[int, list[str]]]):
self.user_data["reply_all"] = value
def add_tag(self, tag):
self.user_data["tags"].add(tag)
@ -66,6 +78,15 @@ class UserData:
del self.user_data["pending_pings"][(chat_id, message_id)]
self.user_data["pings"][ping_id] = ping_list
def submit_pending_pings(self, tags: list[str]) -> list[tuple[int, int]]:
res = []
for (chat_id, message_id), (ping_id, _) in self.user_data["pending_pings"].items():
self.user_data["pings"][ping_id] = tags
res.append((chat_id, message_id))
for (chat_id, message_id) in res:
del self.user_data["pending_pings"][(chat_id, message_id)]
return res
def get_pending_pings(self):
return self.user_data["pending_pings"].values()
@ -143,48 +164,81 @@ async def list_pending_pings_command(update: Update, context: ContextTypes.DEFAU
await context.bot.send_message(chat_id=update.message.chat.id, text=text)
async def reply_all_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
assert update.message is not None
assert context.user_data is not None
user_data = UserData(context.user_data)
user_data.reply_all = None
pings = user_data.get_pending_pings()
if user_data.reply_all is not None:
text = "It seems like there is a pending reply all command."
reply_markup = None
elif len(pings) == 0:
text = "No pending pings to reply to."
reply_markup = None
else:
text = f"Reply to {len(pings)} pending pings."
reply_markup=user_data.get_ping_reply_markup()
message = await context.bot.send_message(chat_id=update.message.chat.id, text=text, reply_markup=reply_markup)
if reply_markup is not None and len(pings) > 0:
user_data.reply_all = (message.id, [])
async def callback_button(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Parses the CallbackQuery and updates the message text."""
assert update.callback_query is not None
assert context.chat_data is not None
assert update.callback_query.message is not None
query = update.callback_query
assert query.message is not None
assert isinstance(query.data, str)
# We have to always to do that because the PTB docs say so.
await query.answer()
user_data = UserData(context.user_data)
tags = user_data.get_ping_list(query.message.chat_id, query.message.id)
tag = update.callback_query.data
text = None
tags, text, reply_all_id = None, None, None
reply_markup = user_data.get_ping_reply_markup()
if user_data.reply_all is not None and query.message.id == user_data.reply_all[0]:
reply_all_id, tags = user_data.reply_all
else:
tags = user_data.get_ping_list(query.message.chat_id, query.message.id)
if tags is None:
text = "Invalid ping. Removed."
reply_markup = None
elif tag == "submit" and len(tags) > 0:
user_data.submit_ping(query.message.chat_id, query.message.id)
tags_text = "', '".join(tags)
text = f"☑️ Submitted '{tags_text}'."
reply_markup = None
elif tag == "submit":
text = "Select at least one tag before submitting."
elif tag == "clear":
tags.clear()
elif tag == "reload":
pass
else:
if tag not in tags:
tags.append(tag)
else:
tags.remove(tag)
match query.data:
case "submit" if len(tags) > 0:
text = f"☑️ Submitted {', '.join(tags)}."
reply_markup = None
if reply_all_id is None:
user_data.submit_ping(query.message.chat_id, query.message.id)
else:
for (chat_id, message_id) in user_data.submit_pending_pings(tags):
await context.bot.edit_message_text(chat_id=chat_id, message_id=message_id, text=text)
case "submit":
text = "Select at least one tag before submitting."
case "clear":
tags.clear()
case "reload":
pass
case tag if tag not in tags:
tags.append(tag)
case tag if tag in tags:
tags.remove(tag)
if text is None:
if tags is not None:
if text is None:
selected = ", ".join(tags)
if selected:
text = f"Selected: {selected}."
text = f"Selected {selected}."
else:
text = "Select tags."
else:
text = "No text or tags."
if text is None:
text = "Something went wrong. Sorry!"
if query.message.text != text or query.message.reply_markup != reply_markup:
await query.edit_message_text(text=text, reply_markup=reply_markup)
@ -223,14 +277,16 @@ async def help_command(update: Update, _: ContextTypes.DEFAULT_TYPE) -> None:
text = ("Commands:\n"
" /start to start receiving pings\n"
"\n"
" /list_tags to see see currently defined tags\n"
" /add_tags to add tags (provide tags separated by space)\n"
" /delete_tags to delete tags (will not affect past pings)\n"
" /listtags to see see currently defined tags\n"
" /addtags to add tags (provide tags separated by space)\n"
" /deletetags to delete tags (will not affect past pings)\n"
"\n"
" /ping to receive a demo ping\n"
" /list_pings to list recently answered pings\n"
" /pending_pings to list pending pings\n"
"")
" /listpings to list recently answered pings\n"
" /pending to list pending pings\n"
" /replyall to tag all pending pings\n"
"\n"
"🫂")
await update.message.reply_text(text)
@ -296,7 +352,7 @@ async def delete_tags_command(update: Update, context: ContextTypes.DEFAULT_TYPE
def main() -> None:
"""Run the bot."""
token = open("token.txt", "r").read().strip()
token = open(sys.argv[1], "r").read().strip()
persistence = PicklePersistence(filepath='persistence.pkl')
application = Application.builder().token(token).persistence(persistence).build()
assert application.job_queue is not None
@ -304,13 +360,14 @@ def main() -> None:
application.add_handler(CommandHandler("start", start_command))
application.add_handler(CommandHandler("help", help_command))
application.add_handler(CommandHandler("ping", ping_command))
application.add_handler(CommandHandler("list_pings", list_pings_command))
application.add_handler(CommandHandler("pending_pings", list_pending_pings_command))
application.add_handler(CommandHandler("listpings", list_pings_command))
application.add_handler(CommandHandler("pending", list_pending_pings_command))
application.add_handler(CommandHandler("replyall", reply_all_command))
application.add_handler(CallbackQueryHandler(callback_button))
application.add_handler(CommandHandler("add_tags", add_tags_command))
application.add_handler(CommandHandler("list_tags", list_tags_command))
application.add_handler(CommandHandler("delete_tags", delete_tags_command))
application.add_handler(CommandHandler("addtags", add_tags_command))
application.add_handler(CommandHandler("listtags", list_tags_command))
application.add_handler(CommandHandler("deletetags", delete_tags_command))
application.job_queue.run_repeating(callback_minute, interval=60, first=5)
application.run_polling(allowed_updates=Update.ALL_TYPES)