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

Compare commits

base: astra:master
Branches Tags
astra:master
astra:astra-topics
..
compare: astra:astra-topics
Branches Tags
astra:master
astra:astra-topics

No commits in common. "master" and "astra-topics" have entirely different histories.
master ... astra-topi

6 changed files with 51 additions and 360 deletions
Download patch file Download diff file Expand all files Collapse all files

11
Dockerfile
View file

@ -1,11 +0,0 @@
FROM python:3.12-alpine
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY supportbot/ ./supportbot/
ENTRYPOINT ["python", "-m", "supportbot"]
CMD ["-c", "/config.yaml"]

86
README.md
View file

@ -1,86 +0,0 @@
# tgsupportbot
A Telegram support bot that forwards messages from users to a group chat, allowing a support team to respond privately.
## How it works
Users message the bot directly. The bot forwards each message to a configured support group (optionally into a specific topic thread). Support staff reply in the group by replying to the forwarded message; the bot delivers the reply back to the user.
If a message goes unanswered, the bot sends a reminder to the group every 6 hours until the conversation is marked finished.
## Setup
### Docker (recommended)
```sh
docker build -t tgsupportbot .
docker run -d \
-v /path/to/config.yaml:/config.yaml \
-v /path/to/data:/data \
tgsupportbot
```
Set `database: "/data/instance.db"` in your config so the database persists outside the container.
### Manual
**Requirements:** Python 3, pip
```sh
pip install -r requirements.txt
```
Copy `config.yaml.example` to `config.yaml` and fill in the values (see [Configuration](#configuration) below).
Create a bot via [@BotFather](https://t.me/BotFather), add it to your support group, and give it permission to read and send messages.
```sh
python -m supportbot -c config.yaml
```
## Configuration
```yaml
# Telegram bot token from @BotFather
bot_token: "BOT_TOKEN_HERE"
# Path to the database file
database: "./instance.db"
# ID of the support group where messages are forwarded
target_group: -12345678
# Optional: topic thread ID to use within the group
message_thread_id: 1
# Message shown to users on /start (HTML)
welcome_text: |
<b>Welcome</b>. Please describe your issue.
# Message shown after a user sends a message (HTML)
reply_text: |
Thanks, we'll get back to you shortly.
# Optional: link template shown in the group alongside user info.
# %d is replaced with the user's Telegram ID.
integration_fmt: "https://example.com/users/%d"
# Optional: how often to send reminders for unanswered messages, in minutes (default: 360)
reminder_interval_minutes: 360
```
## Group commands
All commands are issued in the support group by replying to a forwarded message.
| Command | Description |
| ------------ | ---------------------------------------------------- |
| `/status` | Show the user's info, reply status, and ban state |
| `/ban` | Ban the user permanently |
| `/ban 7d` | Ban the user for a duration (`s`, `m`, `h`, `d`, `w`) |
| `/unban` | Unban the user |
| `/finished` | Mark the conversation as resolved and stop reminders |
## Reminders
When a user sends a message, the bot starts a reminder timer. If `/finished` has not been used before the interval elapses, a reminder is posted to the group. Reminders repeat at the same interval until `/finished` is used. The timer resets if the user sends another message.

12
config.yaml.example
View file

@ -1,8 +1,8 @@
# Telegram bot token
bot_token: "BOT_TOKEN_HERE"
# Database path (JSON file)
database: "./instance.json"
# Database path
database: "./instance.db"
# ID of support bot group where messages will go
target_group: -12345678
@ -16,11 +16,3 @@ welcome_text: |
# Message shown when user writes a message to bot (HTML)
reply_text: |
Thank you for contacting us, we will reply to your message shortly.
# How often to send reminders for unanswered messages, in minutes (default: 360)
# reminder_interval_minutes: 360
# Quiet hours: suppress reminders between these hours (24h clock).
# At the end of quiet hours, one batch message is sent with links to all pending messages.
# quiet_hours_start: 0
# quiet_hours_end: 6

BIN
instance
View file

Binary file not shown.

21
supportbot/__main__.py
View file

@ -1,9 +1,12 @@
#!/usr/bin/env python3
import logging
import yaml
import threading
import sys
import os
import shelve
import getopt
import yaml
from pickle import Unpickler
from . import bot
@ -22,20 +25,30 @@ def usage():
print(" -q Quiet, set log level to WARNING")
print(" -c Location of config file (default: ./config.yaml)")
# well this is just dumb
class RenamingUnpickler(Unpickler):
def find_class(self, module, name):
if module == "src.core":
module = "supportbot.bot"
return super().find_class(module, name)
def main(configpath, loglevel=logging.INFO):
with open(configpath, "r") as f:
config = yaml.safe_load(f)
logging.basicConfig(format="%(levelname)-7s [%(asctime)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S", level=loglevel)
db = bot.JsonDb(config["database"])
shelve.Unpickler = RenamingUnpickler
db = shelve.open(config["database"])
bot.init(config, db)
try:
start_new_thread(bot.reminder_loop)
start_new_thread(bot.run, join=True)
except KeyboardInterrupt:
logging.info("Interrupted, exiting")
db.close()
os._exit(1)
if __name__ == "__main__":
try:
@ -44,6 +57,7 @@ if __name__ == "__main__":
print(str(e))
exit(1)
# Process command line args
def readopt(name):
for e in opts:
if e[0] == name:
@ -58,4 +72,5 @@ if __name__ == "__main__":
if readopt("-c") is not None:
configpath = readopt("-c")
# Run the actual program
main(configpath, loglevel)

281
supportbot/bot.py
View file

@ -2,15 +2,12 @@ import telebot
import logging
import time
import json
import os
import threading
from datetime import datetime, timedelta
from typing import Optional
LONG_LONG_TIME = datetime(2100, 1, 1)
ID_REMIND_DURATION = timedelta(days=2)
BAN_NOTSENT_WARNING = timedelta(minutes=10)
REMINDER_CHECK_INTERVAL = 300 # seconds between checks
ALL_CONTENT_TYPES = ('animation', 'audio', 'contact', 'dice', 'document',
'game', 'location', 'photo', 'sticker', 'story', 'text', 'venue', 'video',
'video_note', 'voice')
@ -18,8 +15,7 @@ ALL_CONTENT_TYPES = ('animation', 'audio', 'contact', 'dice', 'document',
TMessage = telebot.types.Message
bot: telebot.TeleBot = None
db: 'JsonDb' = None
db_lock = threading.Lock()
db = None
bot_self_id: int = None
target_group: Optional[int] = None
@ -27,13 +23,9 @@ message_thread_id: Optional[int] = None
welcome_text: str = None
reply_text: str = None
integration_fmt: Optional[str] = None
reply_reminder_interval: timedelta = timedelta(hours=6)
quiet_hours_start: Optional[int] = None
quiet_hours_end: Optional[int] = None
_in_quiet_hours: bool = False
def init(config: dict, _db):
global bot, db, bot_self_id, target_group, message_thread_id, welcome_text, reply_text, integration_fmt, reply_reminder_interval, quiet_hours_start, quiet_hours_end, _in_quiet_hours
global bot, db, bot_self_id, target_group, message_thread_id, welcome_text, reply_text, integration_fmt
if not config.get("bot_token"):
logging.error("No telegram token specified.")
exit(1)
@ -49,13 +41,6 @@ def init(config: dict, _db):
welcome_text = config["welcome_text"]
reply_text = config["reply_text"]
integration_fmt = config.get("integration_fmt")
if config.get("reminder_interval_minutes"):
reply_reminder_interval = timedelta(minutes=int(config["reminder_interval_minutes"]))
if config.get("quiet_hours_start") is not None:
quiet_hours_start = int(config["quiet_hours_start"])
if config.get("quiet_hours_end") is not None:
quiet_hours_end = int(config["quiet_hours_end"])
_in_quiet_hours = _is_quiet_hour(datetime.now())
set_handler(handle_msg, content_types=ALL_CONTENT_TYPES)
bot_self_id = bot.get_me().id
@ -79,22 +64,6 @@ def run():
logging.warning("%s while polling Telegram, retrying.", type(e).__name__)
time.sleep(1)
def callwrapper_ex(f):
"""Like callwrapper but also returns the Telegram error text on failure."""
while True:
try:
f()
except telebot.apihelper.ApiException as e:
status = check_telegram_exc(e)
if not status:
continue
try:
err = json.loads(e.result.text).get("description", e.result.text)
except Exception:
err = e.result.text
return status, err
return None, None
def callwrapper(f) -> Optional[str]:
while True:
try:
@ -119,50 +88,20 @@ def check_telegram_exc(e):
time.sleep(d)
return False # retry
logging.error("Telegram API error: %s", e.result.text)
logging.exception("API exception")
return "exception"
### db
class JsonDb:
def __init__(self, path: str):
self.path = path
self._data = {"users": {}, "msg_map": {}}
if os.path.exists(path):
with open(path) as f:
self._data = json.load(f)
def _save(self):
with open(self.path, "w") as f:
json.dump(self._data, f, indent=2)
def get_user(self, user_id: int) -> Optional['User']:
d = self._data["users"].get(str(user_id))
return User.from_dict(d) if d is not None else None
def set_user(self, user: 'User'):
self._data["users"][str(user.id)] = user.to_dict()
self._save()
def get_msg_user(self, msg_id: int) -> Optional[int]:
return self._data["msg_map"].get(str(msg_id))
def set_msg_user(self, msg_id: int, user_id: int):
self._data["msg_map"][str(msg_id)] = user_id
self._save()
def all_users(self) -> list:
return [User.from_dict(d) for d in self._data["users"].values()]
class ModificationContext():
def __init__(self, obj):
def __init__(self, key, obj):
self.key = key
self.obj = obj
def __enter__(self) -> 'User':
return self.obj
def __exit__(self, exc_type, *_):
if exc_type is None:
with db_lock:
db.set_user(self.obj)
db[self.key] = self.obj
class User():
id: int
@ -170,18 +109,12 @@ class User():
realname: str
last_messaged: datetime
banned_until: Optional[datetime]
replied_to: bool
last_reminder_sent: Optional[datetime]
last_fwd_msg_id: Optional[int]
def __init__(self):
self.id = None
self.username = None
self.realname = None
self.last_messaged = None
self.banned_until = None
self.replied_to = True
self.last_reminder_sent = None
self.last_fwd_msg_id = None
def __eq__(self, other):
if isinstance(other, User):
return self.id == other.id
@ -190,52 +123,34 @@ class User():
return "<User id=%d>" % self.id
def defaults(self):
self.last_messaged = datetime(1970, 1, 1)
self.replied_to = True
self.last_reminder_sent = None
def to_dict(self) -> dict:
return {
"id": self.id,
"username": self.username,
"realname": self.realname,
"last_messaged": self.last_messaged.isoformat() if self.last_messaged else None,
"banned_until": self.banned_until.isoformat() if self.banned_until else None,
"replied_to": self.replied_to,
"last_reminder_sent": self.last_reminder_sent.isoformat() if self.last_reminder_sent else None,
"last_fwd_msg_id": self.last_fwd_msg_id,
}
@classmethod
def from_dict(cls, d: dict) -> 'User':
u = cls()
u.id = d["id"]
u.username = d.get("username")
u.realname = d.get("realname")
u.last_messaged = datetime.fromisoformat(d["last_messaged"]) if d.get("last_messaged") else datetime(1970, 1, 1)
u.banned_until = datetime.fromisoformat(d["banned_until"]) if d.get("banned_until") else None
u.replied_to = d.get("replied_to", True)
u.last_reminder_sent = datetime.fromisoformat(d["last_reminder_sent"]) if d.get("last_reminder_sent") else None
u.last_fwd_msg_id = d.get("last_fwd_msg_id")
return u
# this is kinda shit
db_last_sync = 0
def db_auto_sync():
global db_last_sync
now = int(time.time())
if now > db_last_sync + 15:
db_last_sync = now
db.sync()
#
def db_get_user(id) -> User:
with db_lock:
user = db.get_user(id)
if user is None:
raise KeyError(id)
return user
return db["u%d" % id]
def db_modify_user(id, allow_new=False) -> ModificationContext:
with db_lock:
obj = db.get_user(id)
key = "u%d" % id
obj = db.get(key)
if obj is None:
if allow_new:
obj = User()
else:
raise KeyError(id)
return ModificationContext(obj)
raise KeyError
return ModificationContext(key, obj)
### Main stuff
def handle_msg(ev: TMessage):
db_auto_sync()
if ev.chat.type in ("group", "supergroup"):
if ev.chat.id == target_group:
return handle_group(ev)
@ -250,8 +165,7 @@ def handle_group(ev: TMessage):
if ev.reply_to_message.from_user.id != bot_self_id:
return
with db_lock:
user_id = db.get_msg_user(ev.reply_to_message.message_id)
user_id = db.get("m%d" % ev.reply_to_message.message_id)
logging.debug("found id = %d mapped to user %s", ev.reply_to_message.message_id, user_id)
if user_id is None:
logging.warning("Couldn't find replied to message in target group")
@ -270,23 +184,13 @@ def handle_group(ev: TMessage):
return callwrapper(lambda: bot.send_message(target_group, message_thread_id=message_thread_id, text=msg))
# deliver message
def copy_and_log():
result = bot.copy_message(user_id, ev.chat.id, ev.message_id)
logging.info("copy_message response: %s", result)
res, err_text = callwrapper_ex(copy_and_log)
if res:
reply_id = ev.message_id
msg = "Failed to deliver: %s" % err_text
callwrapper(lambda: bot.send_message(target_group, message_thread_id=message_thread_id,
text=msg, reply_to_message_id=reply_id))
else:
with db_modify_user(user_id) as u:
u.replied_to = True
res = callwrapper(lambda: bot.copy_message(user_id, ev.chat.id, ev.message_id))
if res == "blocked":
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):
if c == "status":
user = db_get_user(user_id)
msg = format_user_info(user) + "\n" + format_ticket_info(user)
if c == "info":
msg = format_user_info(db_get_user(user_id))
return callwrapper(lambda: bot.send_message(target_group, message_thread_id=message_thread_id, text=msg, parse_mode="HTML"))
elif c == "ban":
delta = parse_timedelta(arg)
@ -308,12 +212,6 @@ def handle_group_command(ev: TMessage, user_id: int, c: str, arg: str):
user.banned_until = None
msg = "User was unbanned."
return callwrapper(lambda: bot.send_message(target_group, message_thread_id=message_thread_id, text=msg))
elif c == "finished":
with db_modify_user(user_id) as user:
user.replied_to = True
user.last_reminder_sent = None
msg = "Marked as replied. Reminders stopped."
return callwrapper(lambda: bot.send_message(target_group, message_thread_id=message_thread_id, text=msg))
def handle_private(ev: TMessage):
if target_group is None:
@ -361,22 +259,15 @@ def handle_private(ev: TMessage):
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):
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)
with db_lock:
db.set_msg_user(ev2.message_id, user_id)
with db_modify_user(user_id) as u:
u.last_fwd_msg_id = ev2.message_id
db["m%d" % ev2.message_id] = user_id
logging.debug("delivered msg from %s -> id = %d", user, ev2.message_id)
res = callwrapper(f)
if res == "blocked":
return
callwrapper(f)
if reply_text:
callwrapper(lambda: bot.send_message(ev.chat.id, reply_text, parse_mode="HTML"))
with db_modify_user(user.id) as user:
user.last_messaged = now
user.replied_to = False
user.last_reminder_sent = now # first reminder fires reply_reminder_interval from now
def handle_private_command(ev: TMessage, user: User, c):
if c == "start":
@ -385,98 +276,6 @@ def handle_private_command(ev: TMessage, user: User, c):
elif c == "stop":
return True
### Reminders
def reminder_loop():
while True:
try:
_send_pending_reminders()
except Exception:
logging.exception("Exception in reminder loop")
time.sleep(REMINDER_CHECK_INTERVAL)
def _is_quiet_hour(now: datetime) -> bool:
if quiet_hours_start is None or quiet_hours_end is None:
return False
h = now.hour
if quiet_hours_start <= quiet_hours_end:
return quiet_hours_start <= h < quiet_hours_end
return h >= quiet_hours_start or h < quiet_hours_end # wraps midnight
def _send_pending_reminders():
global _in_quiet_hours
if target_group is None:
return
now = datetime.now()
if _is_quiet_hour(now):
_in_quiet_hours = True
return
with db_lock:
users = db.all_users()
all_pending = []
due_for_reminder = []
for user in users:
if user.replied_to:
continue
all_pending.append(user)
last_reminder = user.last_reminder_sent
if last_reminder is None or now - last_reminder >= reply_reminder_interval:
due_for_reminder.append(user)
was_quiet = _in_quiet_hours
_in_quiet_hours = False
if was_quiet and all_pending:
_send_batch_reminder(all_pending, now)
else:
for user in due_for_reminder:
try:
_send_reminder(user, now)
except Exception:
logging.exception("Failed to send reminder for user %d", user.id)
def _send_batch_reminder(users: list, now: datetime):
group_str = str(target_group)
channel_id = group_str[4:] if group_str.startswith('-100') else group_str.lstrip('-')
lines = ["\U0001f514 Pending replies:"]
for user in users:
if user.last_fwd_msg_id is None:
continue
lines.append("https://t.me/c/%s/%d" % (channel_id, user.last_fwd_msg_id))
if len(lines) == 1:
return
msg_text = "\n".join(lines)
callwrapper(lambda: bot.send_message(chat_id=target_group, message_thread_id=message_thread_id,
text=msg_text))
for user in users:
if user.last_fwd_msg_id is None:
continue
with db_modify_user(user.id) as u:
u.replied_to = False
u.last_reminder_sent = now
logging.info("Sent batch reminder for %d users", len(users))
def _send_reminder(user: User, now: datetime):
if user.last_fwd_msg_id is None:
return
def f():
bot.send_message(chat_id=target_group, message_thread_id=message_thread_id,
text="\U0001f514 Pending reply", reply_to_message_id=user.last_fwd_msg_id)
status = callwrapper(f)
if status:
logging.warning("Reminder failed for user %d: %s", user.id, status)
return
with db_modify_user(user.id) as u:
u.replied_to = False
u.last_reminder_sent = now
logging.info("Sent reminder for user %d", user.id)
### Helpers
def str_is_printable(s):
@ -503,24 +302,6 @@ def escape_html(s):
def format_datetime(dt):
return dt.strftime("%Y-%m-%d %H:%M:%S")
def format_ticket_info(user: User):
now = datetime.now()
lines = []
if user.replied_to:
lines.append("Status: replied")
else:
lines.append("Status: <b>pending reply</b>")
if user.last_reminder_sent:
lines.append("Last reminder: %s" % format_datetime(user.last_reminder_sent))
if user.banned_until is not None:
if user.banned_until >= LONG_LONG_TIME:
lines.append("Ban: permanent")
elif user.banned_until > now:
lines.append("Ban: until %s" % format_datetime(user.banned_until))
else:
lines.append("Ban: expired")
return "\n".join(lines)
def format_user_info(user: User):
realname = user.realname
if not str_is_printable(realname):
@ -530,6 +311,6 @@ def format_user_info(user: User):
if user.username is not None:
s += " (@%s)" % escape_html(user.username)
if integration_fmt:
s += "\n➤ " + (integration_fmt % user.id)
s += "\n\u27a4 " + (integration_fmt % user.id)
s += "\nID: <code>%d</code>" % user.id
return s