forked from python-telegram-bot/python-telegram-bot
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy path333
139 lines (106 loc) · 5.07 KB
/
333
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
import logging
from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes
# 配置日志
logging.basicConfig(
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
level=logging.INFO
)
logger = logging.getLogger(__name__)
# 定义机器人API Token
TOKEN = "7289484401:AAEap4K7VMwEWdg0X3a3zdCpKq-iwsMvdpw"
# 处理 /start 命令
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
user = update.effective_user
await update.message.reply_html(
rf"Hi {user.mention_html()}! 欢迎使用 365体育管理机器人！"
)
# 处理 /status 命令，返回当前的状态
async def status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
bot_info = await context.bot.get_me()
status_message = f"机器人用户名: {bot_info.username}\n"
status_message += f"是否可以加入群组: {bot_info.can_join_groups}\n"
status_message += f"是否可以阅读所有群组消息: {bot_info.can_read_all_group_messages}\n"
status_message += f"是否支持内联查询: {bot_info.supports_inline_queries}\n"
await update.message.reply_text(status_message)
# 处理管理员命令，踢出成员
async def kick(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
if update.message.chat.type in ['group', 'supergroup']:
if update.message.reply_to_message:
user_to_kick = update.message.reply_to_message.from_user
await context.bot.ban_chat_member(update.message.chat_id, user_to_kick.id)
await update.message.reply_text(f"用户 {user_to_kick.first_name} 已被踢出群组。")
else:
await update.message.reply_text("请回复需要踢出的用户的消息。")
else:
await update.message.reply_text("此命令仅在群组中可用。")
# 处理未知命令
async def unknown(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
await update.message.reply_text("抱歉，我无法理解此命令。")
# 主程序
async def main() -> None:
# 创建应用实例
application = Application.builder().token(TOKEN).build()
# 添加命令处理器
application.add_handler(CommandHandler("start", start))
application.add_handler(CommandHandler("status", status))
application.add_handler(CommandHandler("kick", kick))
application.add_handler(MessageHandler(filters.COMMAND, unknown))
# 启动轮询模式
await application.run_polling()
if __name__ == "__main__":
import asyncio
asyncio.run(main())
import logging
from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes
# 配置日志
logging.basicConfig(
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
level=logging.INFO
)
logger = logging.getLogger(__name__)
# 定义机器人API Token
TOKEN = "7289484401:AAEap4K7VMwEWdg0X3a3zdCpKq-iwsMvdpw"
# 处理 /start 命令
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
user = update.effective_user
await update.message.reply_html(
rf"Hi {user.mention_html()}! 欢迎使用 365体育管理机器人！"
)
# 处理 /status 命令，返回当前的状态
async def status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
bot_info = await context.bot.get_me()
status_message = f"机器人用户名: {bot_info.username}\n"
status_message += f"是否可以加入群组: {bot_info.can_join_groups}\n"
status_message += f"是否可以阅读所有群组消息: {bot_info.can_read_all_group_messages}\n"
status_message += f"是否支持内联查询: {bot_info.supports_inline_queries}\n"
await update.message.reply_text(status_message)
# 处理管理员命令，踢出成员
async def kick(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
if update.message.chat.type in ['group', 'supergroup']:
if update.message.reply_to_message:
user_to_kick = update.message.reply_to_message.from_user
await context.bot.ban_chat_member(update.message.chat_id, user_to_kick.id)
await update.message.reply_text(f"用户 {user_to_kick.first_name} 已被踢出群组。")
else:
await update.message.reply_text("请回复需要踢出的用户的消息。")
else:
await update.message.reply_text("此命令仅在群组中可用。")
# 处理未知命令
async def unknown(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
await update.message.reply_text("抱歉，我无法理解此命令。")
# 主程序
async def main() -> None:
# 创建应用实例
application = Application.builder().token(TOKEN).build()
# 添加命令处理器
application.add_handler(CommandHandler("start", start))
application.add_handler(CommandHandler("status", status))
application.add_handler(CommandHandler("kick", kick))
application.add_handler(MessageHandler(filters.COMMAND, unknown))
# 启动轮询模式
await application.run_polling()
if __name__ == "__main__":
import asyncio
asyncio.run(main())