mirror of
https://github.com/woodchen-ink/Q58Bot.git
synced 2025-07-18 13:52:07 +08:00
refactor(bot_commands): 简化命令处理并重构对话管理
重构了命令注册和处理函数，简化了与用户之间的对话管理。通过将关键词和白名单命令处理分离到独立的异步函数中，提高了代码可读性和可维护性。此外，优化了参数处理，允许更直接的命令交互，并对错误情况给出了更清晰的反馈。
This commit is contained in:
parent
3746b4743b
commit
7b73013749
@ -11,6 +11,8 @@
|
||||
|
||||
1. TeleGuard：一个 Telegram 机器人，用于管理群组中的关键词并自动删除包含这些关键词的消息。
|
||||
2. 币安价格更新器：定期获取并发送指定加密货币的价格信息。
|
||||
3. 链接拦截：拦截并撤回非白名单域名链接的第二次发送。
|
||||
|
||||
|
||||
这些功能被整合到一个 Docker 容器中，可以同时运行。
|
||||
|
||||
@ -26,6 +28,10 @@
|
||||
- 发送详细的价格更新，包括当前价格、24小时变化、高低点等
|
||||
- 可自定义更新频率和货币对
|
||||
|
||||
### 链接拦截
|
||||
- 非白名单域名链接, 在发送第二次会被拦截撤回
|
||||
|
||||
|
||||
## 安装与配置
|
||||
|
||||
1. 克隆此仓库到本地
|
||||
|
||||
@ -1,14 +1,6 @@
|
||||
import os
|
||||
from telethon import events
|
||||
from telethon.tl.types import InputPeerUser
|
||||
from telethon.tl.functions.bots import SetBotCommandsRequest
|
||||
from telethon.tl.types import BotCommand
|
||||
import logging
|
||||
import json
|
||||
|
||||
__all__ = ['register_commands', 'handle_command', 'get_keywords', 'get_whitelist']
|
||||
|
||||
|
||||
KEYWORDS_FILE = '/app/data/keywords.json'
|
||||
WHITELIST_FILE = '/app/data/whitelist.json'
|
||||
ADMIN_ID = int(os.environ.get('ADMIN_ID'))
|
||||
@ -24,128 +16,69 @@ def save_json(file_path, data):
|
||||
with open(file_path, 'w') as f:
|
||||
json.dump(data, f)
|
||||
|
||||
async def register_commands(client, admin_id):
|
||||
commands = [
|
||||
BotCommand('add', '添加新的关键词'),
|
||||
BotCommand('delete', '删除现有的关键词'),
|
||||
BotCommand('list', '列出所有当前的关键词'),
|
||||
BotCommand('addwhite', '添加域名到白名单'),
|
||||
BotCommand('delwhite', '从白名单移除域名'),
|
||||
BotCommand('listwhite', '列出白名单域名'),
|
||||
]
|
||||
|
||||
try:
|
||||
await client(SetBotCommandsRequest(
|
||||
commands=commands,
|
||||
scope=InputPeerUser(admin_id, 0),
|
||||
lang_code=''
|
||||
))
|
||||
logging.info("Bot commands registered successfully.")
|
||||
except Exception as e:
|
||||
logging.error(f"Failed to register bot commands: {str(e)}")
|
||||
|
||||
async def handle_keyword_command(event, client):
|
||||
async def handle_command(event, client):
|
||||
sender = await event.get_sender()
|
||||
if sender.id != ADMIN_ID:
|
||||
return
|
||||
|
||||
command = event.message.text.split(maxsplit=1)
|
||||
command, *args = event.message.text.split()
|
||||
command = command.lower()
|
||||
|
||||
if command[0].lower() == '/list':
|
||||
await execute_keyword_command(event, '/list', '')
|
||||
elif len(command) > 1:
|
||||
await execute_keyword_command(event, command[0], command[1])
|
||||
else:
|
||||
await event.reply(f"请输入你要{command[0][1:]}的关键词：")
|
||||
async with client.conversation(sender) as conv:
|
||||
response = await conv.get_response()
|
||||
await execute_keyword_command(event, command[0], response.text)
|
||||
if command in ['/add', '/delete', '/list']:
|
||||
await handle_keyword_command(event, command, args)
|
||||
elif command in ['/addwhite', '/delwhite', '/listwhite']:
|
||||
await handle_whitelist_command(event, command, args)
|
||||
|
||||
async def execute_keyword_command(event, command, keyword):
|
||||
async def handle_keyword_command(event, command, args):
|
||||
keywords = load_json(KEYWORDS_FILE)
|
||||
|
||||
if command.lower() == '/list':
|
||||
if keywords:
|
||||
await event.reply(f"当前关键词和语句列表：\n" + "\n".join(keywords))
|
||||
else:
|
||||
await event.reply("关键词列表为空。")
|
||||
return
|
||||
|
||||
if command.lower() == '/add':
|
||||
if keyword.lower() not in keywords:
|
||||
keywords.append(keyword.lower())
|
||||
if command == '/list':
|
||||
await event.reply("当前关键词列表：\n" + "\n".join(keywords) if keywords else "关键词列表为空。")
|
||||
elif command == '/add' and args:
|
||||
keyword = args[0].lower()
|
||||
if keyword not in keywords:
|
||||
keywords.append(keyword)
|
||||
save_json(KEYWORDS_FILE, keywords)
|
||||
await event.reply(f"关键词或语句 '{keyword}' 已添加到列表中。")
|
||||
await event.reply(f"关键词 '{keyword}' 已添加。")
|
||||
else:
|
||||
await event.reply(f"关键词或语句 '{keyword}' 已经在列表中。")
|
||||
|
||||
elif command.lower() == '/delete':
|
||||
if keyword.lower() in keywords:
|
||||
keywords.remove(keyword.lower())
|
||||
await event.reply(f"关键词 '{keyword}' 已存在。")
|
||||
elif command == '/delete' and args:
|
||||
keyword = args[0].lower()
|
||||
if keyword in keywords:
|
||||
keywords.remove(keyword)
|
||||
save_json(KEYWORDS_FILE, keywords)
|
||||
await event.reply(f"关键词或语句 '{keyword}' 已从列表中删除。")
|
||||
await event.reply(f"关键词 '{keyword}' 已删除。")
|
||||
else:
|
||||
await event.reply(f"关键词或语句 '{keyword}' 不在列表中。")
|
||||
|
||||
async def handle_whitelist_command(event, client):
|
||||
sender = await event.get_sender()
|
||||
if sender.id != ADMIN_ID:
|
||||
return
|
||||
|
||||
command = event.message.text.split(maxsplit=1)
|
||||
|
||||
if command[0].lower() == '/listwhite':
|
||||
await execute_whitelist_command(event, '/listwhite', '')
|
||||
elif len(command) > 1:
|
||||
await execute_whitelist_command(event, command[0], command[1])
|
||||
await event.reply(f"关键词 '{keyword}' 不存在。")
|
||||
else:
|
||||
await event.reply(f"请输入你要{command[0][1:]}的域名：")
|
||||
async with client.conversation(sender) as conv:
|
||||
response = await conv.get_response()
|
||||
await execute_whitelist_command(event, command[0], response.text)
|
||||
await event.reply("无效的命令或参数。")
|
||||
|
||||
async def execute_whitelist_command(event, command, domain):
|
||||
async def handle_whitelist_command(event, command, args):
|
||||
whitelist = load_json(WHITELIST_FILE)
|
||||
|
||||
if command.lower() == '/listwhite':
|
||||
if whitelist:
|
||||
await event.reply("白名单域名列表：\n" + "\n".join(whitelist))
|
||||
else:
|
||||
await event.reply("白名单为空。")
|
||||
return
|
||||
|
||||
if command.lower() == '/addwhite':
|
||||
if domain.lower() not in whitelist:
|
||||
whitelist.append(domain.lower())
|
||||
if command == '/listwhite':
|
||||
await event.reply("白名单域名列表：\n" + "\n".join(whitelist) if whitelist else "白名单为空。")
|
||||
elif command == '/addwhite' and args:
|
||||
domain = args[0].lower()
|
||||
if domain not in whitelist:
|
||||
whitelist.append(domain)
|
||||
save_json(WHITELIST_FILE, whitelist)
|
||||
await event.reply(f"域名 '{domain}' 已添加到白名单。")
|
||||
else:
|
||||
await event.reply(f"域名 '{domain}' 已经在白名单中。")
|
||||
|
||||
elif command.lower() == '/delwhite':
|
||||
if domain.lower() in whitelist:
|
||||
whitelist.remove(domain.lower())
|
||||
await event.reply(f"域名 '{domain}' 已在白名单中。")
|
||||
elif command == '/delwhite' and args:
|
||||
domain = args[0].lower()
|
||||
if domain in whitelist:
|
||||
whitelist.remove(domain)
|
||||
save_json(WHITELIST_FILE, whitelist)
|
||||
await event.reply(f"域名 '{domain}' 已从白名单中移除。")
|
||||
await event.reply(f"域名 '{domain}' 已从白名单中删除。")
|
||||
else:
|
||||
await event.reply(f"域名 '{domain}' 不在白名单中。")
|
||||
else:
|
||||
await event.reply("无效的命令或参数。")
|
||||
|
||||
def get_keywords():
|
||||
return load_json(KEYWORDS_FILE)
|
||||
|
||||
def get_whitelist():
|
||||
return load_json(WHITELIST_FILE)
|
||||
|
||||
async def handle_command(event, client):
|
||||
sender = await event.get_sender()
|
||||
if sender.id != ADMIN_ID:
|
||||
return
|
||||
|
||||
command = event.message.text.split(maxsplit=1)
|
||||
|
||||
if command[0].lower() in ['/add', '/delete', '/list']:
|
||||
await handle_keyword_command(event, client)
|
||||
elif command[0].lower() in ['/addwhite', '/delwhite', '/listwhite']:
|
||||
await handle_whitelist_command(event, client)
|
||||
|
||||
|
||||
|
||||
@ -84,6 +84,7 @@ async def command_handler(event):
|
||||
elif event.raw_text.startswith(('/addwhite', '/delwhite', '/listwhite')):
|
||||
link_filter.reload_whitelist()
|
||||
|
||||
|
||||
async def message_handler(event):
|
||||
if not event.is_private or event.sender_id != ADMIN_ID:
|
||||
async with rate_limiter:
|
||||
@ -99,9 +100,6 @@ async def start_bot():
|
||||
logger.info("TeleGuard is running...")
|
||||
await client.run_until_disconnected()
|
||||
|
||||
|
||||
|
||||
|
||||
# 主函数
|
||||
def run():
|
||||
while True:
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user