102 lines
3.8 KiB
Python
102 lines
3.8 KiB
Python
import logging
|
||
import os
|
||
import asyncio
|
||
import time
|
||
from telegram import Update
|
||
from telegram.ext import ApplicationBuilder, ContextTypes, CommandHandler, MessageHandler, filters
|
||
from autoDabing import process_verification
|
||
|
||
# --- 配置 ---
|
||
BOT_TOKEN = "" # 替换为你的 Bot Token
|
||
ALLOWED_USER_IDS = [] # 可选：限制允许使用的用户ID，为空则不限制
|
||
|
||
# --- 日志 ---
|
||
logging.basicConfig(
|
||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
|
||
level=logging.INFO
|
||
)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
await update.message.reply_text(
|
||
"欢迎使用 SheerID 验证 Bot！\n"
|
||
"请直接发送验证 URL 给我也开始验证。\n"
|
||
"格式: https://services.sheerid.com/verify/..."
|
||
)
|
||
|
||
async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
user_id = update.effective_user.id
|
||
if ALLOWED_USER_IDS and user_id not in ALLOWED_USER_IDS:
|
||
await update.message.reply_text("⛔️ 您没有权限使用此 Bot。\n")
|
||
return
|
||
|
||
text = update.message.text.strip()
|
||
|
||
# 简单检查是否包含 sheerid URL
|
||
if "sheerid.com/verify" not in text:
|
||
await update.message.reply_text("⚠️ 请发送有效的 SheerID 验证链接。\n")
|
||
return
|
||
|
||
await update.message.reply_text("🚀 开始验证，请稍候...")
|
||
|
||
# 定义日志回调，发送消息给用户
|
||
# 注意：在多线程中调用 async 函数比较麻烦，这里我们简化处理，
|
||
# 只在关键节点发送消息，或者使用 asyncio.run_coroutine_threadsafe (如果需要实时反馈)
|
||
# 为了避免消息过多触发限制，我们可以收集日志或者只发送重要信息。
|
||
# 这里演示实时发送，但要注意频率。
|
||
|
||
last_msg_time = 0
|
||
main_loop = asyncio.get_running_loop()
|
||
|
||
def log_callback(msg):
|
||
# 这是一个同步回调，运行在 worker 线程中
|
||
# 我们通过 asyncio.run_coroutine_threadsafe 将发送消息的任务提交回主循环
|
||
# 为了防止 flood limit，这里可以做一些过滤或缓冲，这里暂且直接发送
|
||
try:
|
||
nonlocal last_msg_time
|
||
now = time.time()
|
||
if now - last_msg_time < 0.8:
|
||
return
|
||
# 简单的过滤，只发送带 emoji 的或者关键信息
|
||
if any(char in msg for char in ['✅', '❌', '>>>', '🛑', '邮箱']):
|
||
last_msg_time = now
|
||
asyncio.run_coroutine_threadsafe(
|
||
context.bot.send_message(chat_id=update.effective_chat.id, text=msg),
|
||
main_loop,
|
||
)
|
||
except Exception as e:
|
||
logger.error(f"发送日志消息失败: {e}")
|
||
|
||
# 在执行器中运行阻塞的验证函数
|
||
loop = asyncio.get_running_loop()
|
||
result = await loop.run_in_executor(None, process_verification, text, log_callback)
|
||
|
||
if result['success']:
|
||
await update.message.reply_text(
|
||
f"🎉 验证成功！\n"
|
||
f"邮箱: `{result.get('email')}`\n"
|
||
f"消息: {result.get('message')}",
|
||
parse_mode='Markdown'
|
||
)
|
||
else:
|
||
await update.message.reply_text(f"❌ 验证失败: {result.get('message')}")
|
||
|
||
def main():
|
||
if BOT_TOKEN == "YOUR_BOT_TOKEN_HERE":
|
||
print("请在 bot.py 中设置 BOT_TOKEN")
|
||
return
|
||
|
||
application = ApplicationBuilder().token(BOT_TOKEN).build()
|
||
|
||
start_handler = CommandHandler('start', start)
|
||
msg_handler = MessageHandler(filters.TEXT & (~filters.COMMAND), handle_message)
|
||
|
||
application.add_handler(start_handler)
|
||
application.add_handler(msg_handler)
|
||
|
||
print("Bot 已启动...")
|
||
application.run_polling()
|
||
|
||
if __name__ == '__main__':
|
||
main()
|