chatmemberbot.pyΒΆ

  1#!/usr/bin/env python
  2# pylint: disable=unused-argument
  3# This program is dedicated to the public domain under the CC0 license.
  4
  5"""
  6Simple Bot to handle '(my_)chat_member' updates.
  7Greets new users & keeps track of which chats the bot is in.
  8
  9Usage:
 10Press Ctrl-C on the command line or send a signal to the process to stop the
 11bot.
 12"""
 13
 14import logging
 15from typing import Optional
 16
 17from telegram import Chat, ChatMember, ChatMemberUpdated, Update
 18from telegram.constants import ParseMode
 19from telegram.ext import (
 20    Application,
 21    ChatMemberHandler,
 22    CommandHandler,
 23    ContextTypes,
 24    MessageHandler,
 25    filters,
 26)
 27
 28# Enable logging
 29
 30logging.basicConfig(
 31    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
 32)
 33
 34# set higher logging level for httpx to avoid all GET and POST requests being logged
 35logging.getLogger("httpx").setLevel(logging.WARNING)
 36
 37logger = logging.getLogger(__name__)
 38
 39
 40def extract_status_change(chat_member_update: ChatMemberUpdated) -> Optional[tuple[bool, bool]]:
 41    """Takes a ChatMemberUpdated instance and extracts whether the 'old_chat_member' was a member
 42    of the chat and whether the 'new_chat_member' is a member of the chat. Returns None, if
 43    the status didn't change.
 44    """
 45    status_change = chat_member_update.difference().get("status")
 46    old_is_member, new_is_member = chat_member_update.difference().get("is_member", (None, None))
 47
 48    if status_change is None:
 49        return None
 50
 51    old_status, new_status = status_change
 52    was_member = old_status in [
 53        ChatMember.MEMBER,
 54        ChatMember.OWNER,
 55        ChatMember.ADMINISTRATOR,
 56    ] or (old_status == ChatMember.RESTRICTED and old_is_member is True)
 57    is_member = new_status in [
 58        ChatMember.MEMBER,
 59        ChatMember.OWNER,
 60        ChatMember.ADMINISTRATOR,
 61    ] or (new_status == ChatMember.RESTRICTED and new_is_member is True)
 62
 63    return was_member, is_member
 64
 65
 66async def track_chats(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
 67    """Tracks the chats the bot is in."""
 68    result = extract_status_change(update.my_chat_member)
 69    if result is None:
 70        return
 71    was_member, is_member = result
 72
 73    # Let's check who is responsible for the change
 74    cause_name = update.effective_user.full_name
 75
 76    # Handle chat types differently:
 77    chat = update.effective_chat
 78    if chat.type == Chat.PRIVATE:
 79        if not was_member and is_member:
 80            # This may not be really needed in practice because most clients will automatically
 81            # send a /start command after the user unblocks the bot, and start_private_chat()
 82            # will add the user to "user_ids".
 83            # We're including this here for the sake of the example.
 84            logger.info("%s unblocked the bot", cause_name)
 85            context.bot_data.setdefault("user_ids", set()).add(chat.id)
 86        elif was_member and not is_member:
 87            logger.info("%s blocked the bot", cause_name)
 88            context.bot_data.setdefault("user_ids", set()).discard(chat.id)
 89    elif chat.type in [Chat.GROUP, Chat.SUPERGROUP]:
 90        if not was_member and is_member:
 91            logger.info("%s added the bot to the group %s", cause_name, chat.title)
 92            context.bot_data.setdefault("group_ids", set()).add(chat.id)
 93        elif was_member and not is_member:
 94            logger.info("%s removed the bot from the group %s", cause_name, chat.title)
 95            context.bot_data.setdefault("group_ids", set()).discard(chat.id)
 96    elif not was_member and is_member:
 97        logger.info("%s added the bot to the channel %s", cause_name, chat.title)
 98        context.bot_data.setdefault("channel_ids", set()).add(chat.id)
 99    elif was_member and not is_member:
100        logger.info("%s removed the bot from the channel %s", cause_name, chat.title)
101        context.bot_data.setdefault("channel_ids", set()).discard(chat.id)
102
103
104async def show_chats(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
105    """Shows which chats the bot is in"""
106    user_ids = ", ".join(str(uid) for uid in context.bot_data.setdefault("user_ids", set()))
107    group_ids = ", ".join(str(gid) for gid in context.bot_data.setdefault("group_ids", set()))
108    channel_ids = ", ".join(str(cid) for cid in context.bot_data.setdefault("channel_ids", set()))
109    text = (
110        f"@{context.bot.username} is currently in a conversation with the user IDs {user_ids}."
111        f" Moreover it is a member of the groups with IDs {group_ids} "
112        f"and administrator in the channels with IDs {channel_ids}."
113    )
114    await update.effective_message.reply_text(text)
115
116
117async def greet_chat_members(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
118    """Greets new users in chats and announces when someone leaves"""
119    result = extract_status_change(update.chat_member)
120    if result is None:
121        return
122
123    was_member, is_member = result
124    cause_name = update.chat_member.from_user.mention_html()
125    member_name = update.chat_member.new_chat_member.user.mention_html()
126
127    if not was_member and is_member:
128        await update.effective_chat.send_message(
129            f"{member_name} was added by {cause_name}. Welcome!",
130            parse_mode=ParseMode.HTML,
131        )
132    elif was_member and not is_member:
133        await update.effective_chat.send_message(
134            f"{member_name} is no longer with us. Thanks a lot, {cause_name} ...",
135            parse_mode=ParseMode.HTML,
136        )
137
138
139async def start_private_chat(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
140    """Greets the user and records that they started a chat with the bot if it's a private chat.
141    Since no `my_chat_member` update is issued when a user starts a private chat with the bot
142    for the first time, we have to track it explicitly here.
143    """
144    user_name = update.effective_user.full_name
145    chat = update.effective_chat
146    if chat.type != Chat.PRIVATE or chat.id in context.bot_data.get("user_ids", set()):
147        return
148
149    logger.info("%s started a private chat with the bot", user_name)
150    context.bot_data.setdefault("user_ids", set()).add(chat.id)
151
152    await update.effective_message.reply_text(
153        f"Welcome {user_name}. Use /show_chats to see what chats I'm in."
154    )
155
156
157def main() -> None:
158    """Start the bot."""
159    # Create the Application and pass it your bot's token.
160    application = Application.builder().token("TOKEN").build()
161
162    # Keep track of which chats the bot is in
163    application.add_handler(ChatMemberHandler(track_chats, ChatMemberHandler.MY_CHAT_MEMBER))
164    application.add_handler(CommandHandler("show_chats", show_chats))
165
166    # Handle members joining/leaving chats.
167    application.add_handler(ChatMemberHandler(greet_chat_members, ChatMemberHandler.CHAT_MEMBER))
168
169    # Interpret any other command or text message as a start of a private chat.
170    # This will record the user as being in a private chat with bot.
171    application.add_handler(MessageHandler(filters.ALL, start_private_chat))
172
173    # Run the bot until the user presses Ctrl-C
174    # We pass 'allowed_updates' handle *all* updates including `chat_member` updates
175    # To reset this, simply pass `allowed_updates=[]`
176    application.run_polling(allowed_updates=Update.ALL_TYPES)
177
178
179if __name__ == "__main__":
180    main()