astra
/
tgsupportbot
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

astra-topics #1

Merged
astra merged 2 commits from astra-topics into master 2024-12-05 01:13:28 +01:00
Conversation 0 Commits 2 Files Changed 2
2 changed files with 12 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
config.yaml.example
View File

@ -6,6 +6,7 @@ database: "./instance.db"
# ID of support bot group where messages will go # ID of support bot group where messages will go
target_group: -12345678 target_group: -12345678
message_thread_id: 1
# Message shown to users when they /start the bot (HTML) # Message shown to users when they /start the bot (HTML)
welcome_text: | welcome_text: |

19
supportbot/bot.py
View File

@ -19,12 +19,13 @@ db = None
bot_self_id: int = None bot_self_id: int = None
target_group: Optional[int] = None target_group: Optional[int] = None
message_thread_id: Optional[int] = None
welcome_text: str = None welcome_text: str = None
reply_text: str = None reply_text: str = None
integration_fmt: Optional[str] = None integration_fmt: Optional[str] = None
def init(config: dict, _db): def init(config: dict, _db):
global bot, db, bot_self_id, target_group, welcome_text, reply_text, integration_fmt global bot, db, bot_self_id, target_group, message_thread_id, welcome_text, reply_text, integration_fmt
if not config.get("bot_token"): if not config.get("bot_token"):
logging.error("No telegram token specified.") logging.error("No telegram token specified.")
exit(1) exit(1)
@ -35,6 +36,8 @@ def init(config: dict, _db):
db = _db db = _db
if config.get("target_group"): if config.get("target_group"):
target_group = int(config["target_group"]) target_group = int(config["target_group"])
if config.get("message_thread_id"):
message_thread_id = int(config["message_thread_id"])
welcome_text = config["welcome_text"] welcome_text = config["welcome_text"]
reply_text = config["reply_text"] reply_text = config["reply_text"]
integration_fmt = config.get("integration_fmt") integration_fmt = config.get("integration_fmt")
@ -178,17 +181,17 @@ def handle_group(ev: TMessage):
if user.banned_until is not None and (user.banned_until >= now and if user.banned_until is not None and (user.banned_until >= now and
now - user.last_messaged >= BAN_NOTSENT_WARNING): now - user.last_messaged >= BAN_NOTSENT_WARNING):
msg = "Message was not delivered, unban recipient first." msg = "Message was not delivered, unban recipient first."
return callwrapper(lambda: bot.send_message(target_group, msg)) return callwrapper(lambda: bot.send_message(target_group, message_thread_id=message_thread_id, text=msg))
# deliver message # deliver message
res = callwrapper(lambda: bot.copy_message(user_id, ev.chat.id, ev.message_id)) res = callwrapper(lambda: bot.copy_message(user_id, ev.chat.id, ev.message_id))
if res == "blocked": if res == "blocked":
callwrapper(lambda: bot.send_message(target_group, "Bot was blocked by user.")) callwrapper(lambda: bot.send_message(target_group, message_thread_id=message_thread_id, text="Bot was blocked by user."))
def handle_group_command(ev: TMessage, user_id: int, c: str, arg: str): def handle_group_command(ev: TMessage, user_id: int, c: str, arg: str):
if c == "info": if c == "info":
msg = format_user_info(db_get_user(user_id)) msg = format_user_info(db_get_user(user_id))
return callwrapper(lambda: bot.send_message(target_group, msg, parse_mode="HTML")) return callwrapper(lambda: bot.send_message(target_group, message_thread_id=message_thread_id, text=msg, parse_mode="HTML"))
elif c == "ban": elif c == "ban":
delta = parse_timedelta(arg) delta = parse_timedelta(arg)
if not delta: if not delta:
@ -199,7 +202,7 @@ def handle_group_command(ev: TMessage, user_id: int, c: str, arg: str):
msg = "User banned until %s." % format_datetime(until) msg = "User banned until %s." % format_datetime(until)
with db_modify_user(user_id) as user: with db_modify_user(user_id) as user:
user.banned_until = until user.banned_until = until
return callwrapper(lambda: bot.send_message(target_group, msg)) return callwrapper(lambda: bot.send_message(target_group, message_thread_id=message_thread_id, text=msg))
elif c == "unban": elif c == "unban":
msg = None msg = None
with db_modify_user(user_id) as user: with db_modify_user(user_id) as user:
@ -208,7 +211,7 @@ def handle_group_command(ev: TMessage, user_id: int, c: str, arg: str):
else: else:
user.banned_until = None user.banned_until = None
msg = "User was unbanned." msg = "User was unbanned."
return callwrapper(lambda: bot.send_message(target_group, msg)) return callwrapper(lambda: bot.send_message(target_group, message_thread_id=message_thread_id, text=msg))
def handle_private(ev: TMessage): def handle_private(ev: TMessage):
if target_group is None: if target_group is None:
@ -253,9 +256,9 @@ def handle_private(ev: TMessage):
if now - user.last_messaged >= ID_REMIND_DURATION: if now - user.last_messaged >= ID_REMIND_DURATION:
msg = "---------------------------------------\n" msg = "---------------------------------------\n"
msg += format_user_info(user) msg += format_user_info(user)
callwrapper(lambda: bot.send_message(target_group, msg, parse_mode="HTML")) callwrapper(lambda: bot.send_message(chat_id=target_group, message_thread_id=message_thread_id, text=msg, parse_mode="HTML"))
def f(user_id=user.id): def f(user_id=user.id):
ev2 = bot.forward_message(target_group, ev.chat.id, ev.message_id) ev2 = bot.forward_message(chat_id=target_group, from_chat_id=ev.chat.id, message_id=ev.message_id, message_thread_id=message_thread_id)
db["m%d" % ev2.message_id] = user_id db["m%d" % ev2.message_id] = user_id
logging.debug("delivered msg from %s -> id = %d", user, ev2.message_id) logging.debug("delivered msg from %s -> id = %d", user, ev2.message_id)
callwrapper(f) callwrapper(f)