import os import logging import aiosqlite from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup from telegram.ext import Application, MessageHandler, CommandHandler, filters, ContextTypes # --- 从环境变量读取配置 --- BOT_TOKEN = os.getenv('BOT_TOKEN') ADMIN_ID = int(os.getenv('ADMIN_ID', '0')) DB_FILE = os.getenv('DB_PATH', '/app/data/messages.db') GITHUB_URL = os.getenv('GITHUB_URL', 'https://github.com/EchoZenith/TelegramContactBot') # 启用日志 logging.basicConfig( format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO ) logger = logging.getLogger(__name__) # 初始化数据库 async def init_db(): async with aiosqlite.connect(DB_FILE) as db: # 记录：管理员端消息ID <-> 用户端消息ID <-> 用户ID await db.execute('''CREATE TABLE IF NOT EXISTS msg_pairs (admin_msg_id INTEGER PRIMARY KEY, user_msg_id INTEGER, user_id INTEGER)''') await db.execute('''CREATE TABLE IF NOT EXISTS blacklist (user_id INTEGER PRIMARY KEY)''') await db.commit() # 处理 /start 命令 async def handle_start(update: Update, context: ContextTypes.DEFAULT_TYPE): user_id = update.effective_chat.id keyboard = [[InlineKeyboardButton("🌟 查看项目源码", url=GITHUB_URL)]] reply_markup = InlineKeyboardMarkup(keyboard) if user_id != ADMIN_ID: await update.message.reply_text("Hello!\n\nYou can contact us using this bot.", reply_markup=reply_markup) else: await update.message.reply_text("你好，管理员！有人给机器人发消息时，我会转发给你。你直接【回复】该消息即可回信。") # 处理 /ban 命令 async def handle_ban(update: Update, context: ContextTypes.DEFAULT_TYPE): if update.effective_chat.id != ADMIN_ID: return msg = update.message if not msg.reply_to_message: await msg.reply_text("❌ 请回复一条要封禁的用户消息并输入 /ban") return target_id = msg.reply_to_message.message_id async with aiosqlite.connect(DB_FILE) as db: # 通过消息 ID 找到对应的用户 ID async with db.execute("SELECT user_id FROM msg_pairs WHERE admin_msg_id = ?", (target_id,)) as cursor: row = await cursor.fetchone() if row: user_id = row[0] try: await db.execute("INSERT OR IGNORE INTO blacklist VALUES (?)", (user_id,)) await db.commit() await msg.reply_text(f"🚫 用户 `{user_id}` 已被加入黑名单，后续消息将不再转发。", parse_mode='Markdown') except Exception as e: logger.error(f"Ban failed: {e}") else: await msg.reply_text("⚠️ 找不到该消息的用户记录，无法执行封禁。") # 处理 /unban 命令 async def handle_unban(update: Update, context: ContextTypes.DEFAULT_TYPE): if update.effective_chat.id != ADMIN_ID: return msg = update.message if not msg.reply_to_message: await msg.reply_text("❌ 请回复一条要解封的用户消息并输入 /unban") return target_id = msg.reply_to_message.message_id async with aiosqlite.connect(DB_FILE) as db: # 同样先通过消息 ID 找到对应的用户 ID async with db.execute("SELECT user_id FROM msg_pairs WHERE admin_msg_id = ?", (target_id,)) as cursor: row = await cursor.fetchone() if row: user_id = row[0] try: # 从黑名单表中移除 await db.execute("DELETE FROM blacklist WHERE user_id = ?", (user_id,)) await db.commit() await msg.reply_text(f"✅ 用户 `{user_id}` 已解封，现在可以正常接收其消息。", parse_mode='Markdown') except Exception as e: logger.error(f"Unban failed: {e}") else: await msg.reply_text("⚠️ 找不到该消息的用户记录，无法执行解封。") # 处理 /del 命令 async def handle_delete(update: Update, context: ContextTypes.DEFAULT_TYPE): # 权限检查：仅限管理员 if update.effective_chat.id != ADMIN_ID: return message = update.message # 必须是回复某条消息才能执行删除 if not message.reply_to_message: await message.reply_text("请回复一条你想删除的消息并输入 /del") return target_id = message.reply_to_message.message_id async with aiosqlite.connect(DB_FILE) as db: # 查找这条消息对应的用户ID和用户端消息ID async with db.execute( "SELECT user_id, user_msg_id FROM msg_pairs WHERE admin_msg_id = ?", (target_id,) ) as cursor: row = await cursor.fetchone() if row: user_id, user_msg_id = row try: # 1. 删除用户端的消息 await context.bot.delete_message(chat_id=user_id, message_id=user_msg_id) # 2. 删除管理员端的消息 await context.bot.delete_message(chat_id=ADMIN_ID, message_id=target_id) # 3. 删除管理员刚才发的 "/del" 命令消息，保持界面整洁 await message.delete() # 4. 从数据库中移除这条记录 await db.execute("DELETE FROM msg_pairs WHERE admin_msg_id = ?", (target_id,)) await db.commit() except Exception as e: logger.error(f"同步删除失败: {e}") await message.reply_text(f"删除失败，可能消息已超过48小时或机器人无权限：{e}") else: await message.reply_text("未找到该消息的转发记录，无法同步删除。") # 处理同步修改逻辑 async def handle_edit(update: Update, context: ContextTypes.DEFAULT_TYPE): edited_msg = update.edited_message if not edited_msg: return user_id = edited_msg.chat.id async with aiosqlite.connect(DB_FILE) as db: if user_id != ADMIN_ID: # 1. 查找旧的转发记录 async with db.execute("SELECT admin_msg_id FROM msg_pairs WHERE user_msg_id = ? AND user_id = ?", (edited_msg.message_id, user_id)) as cursor: row = await cursor.fetchone() if row: old_admin_msg_id = row[0] # 获取新内容（文字或媒体说明） content = edited_msg.text or edited_msg.caption or "" new_text = f"【此消息已修改】\n{content}" try: # 2. 判断是纯文本还是带媒体的消息 if edited_msg.text: # 发送纯文字消息，并回复在旧消息上 new_msg = await context.bot.send_message( chat_id=ADMIN_ID, text=new_text, reply_to_message_id=old_admin_msg_id ) else: # 如果是图片/视频等媒体，复制它并修改其 Caption new_msg = await context.bot.copy_message( chat_id=ADMIN_ID, from_chat_id=user_id, message_id=edited_msg.message_id, caption=new_text, reply_to_message_id=old_admin_msg_id ) # 3. 更新映射关系，确保管理员回复这条带“【新消息】”提示的消息时，也能回传 await db.execute("INSERT OR REPLACE INTO msg_pairs VALUES (?, ?, ?)", (new_msg.message_id, edited_msg.message_id, user_id)) await db.commit() except Exception as e: logger.error(f"处理用户修改回传失败: {e}") else: # 管理员修改逻辑保持不变（原地编辑用户收到的消息） async with db.execute("SELECT user_id, user_msg_id FROM msg_pairs WHERE admin_msg_id = ?", (edited_msg.message_id,)) as cursor: row = await cursor.fetchone() if row: try: new_content = edited_msg.text or edited_msg.caption await context.bot.edit_message_text(chat_id=row[0], message_id=row[1], text=new_content) except Exception as e: logger.warning(f"管理员修改同步略过: {e}") # 处理所有普通消息 async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE): if not update.message: return user_id = update.effective_chat.id message = update.message # 1. 用户 -> 管理员 if user_id != ADMIN_ID: # 检查黑名单 async with aiosqlite.connect(DB_FILE) as db: async with db.execute("SELECT 1 FROM blacklist WHERE user_id = ?", (user_id,)) as cursor: if await cursor.fetchone(): return # 如果在黑名单里，直接无视，不执行后续转发 try: forwarded_msg = await context.bot.forward_message(chat_id=ADMIN_ID, from_chat_id=user_id, message_id=message.message_id) async with aiosqlite.connect(DB_FILE) as db: await db.execute("INSERT INTO msg_pairs VALUES (?, ?, ?)", (forwarded_msg.message_id, message.message_id, user_id)) await db.commit() except Exception as e: logger.error(f"转发失败: {e}") # 2. 管理员回复 -> 用户 elif message.reply_to_message: async with aiosqlite.connect(DB_FILE) as db: async with db.execute("SELECT user_id FROM msg_pairs WHERE admin_msg_id = ?", (message.reply_to_message.message_id,)) as cursor: row = await cursor.fetchone() if row: target_user_id = row[0] try: sent_msg = await context.bot.copy_message(chat_id=target_user_id, from_chat_id=ADMIN_ID, message_id=message.message_id) # 记录管理员发出的回复，以便管理员后续修改这条回复时能同步给用户 async with aiosqlite.connect(DB_FILE) as db: await db.execute("INSERT INTO msg_pairs (admin_msg_id, user_msg_id, user_id) VALUES (?, ?, ?)", (message.message_id, sent_msg.message_id, target_user_id)) await db.commit() except Exception as e: await message.reply_text(f"回复失败: {e}") async def post_init(application: Application): await init_db() if __name__ == '__main__': # 检查配置 if not BOT_TOKEN or ADMIN_ID == 0: print("错误: 请确保环境变量 BOT_TOKEN 和 ADMIN_ID 已正确设置！") exit(1) # 构建应用 application = Application.builder().token(BOT_TOKEN).post_init(post_init).build() # 处理器注册 application.add_handler(CommandHandler("start", handle_start)) # 监听封禁 application.add_handler(CommandHandler("ban", handle_ban)) # 监听解禁 application.add_handler(CommandHandler("unban", handle_unban)) # 监听管理员删除消息 application.add_handler(CommandHandler("del", handle_delete)) # 监听修改消息的更新 application.add_handler(MessageHandler(filters.UpdateType.EDITED_MESSAGE & filters.TEXT, handle_edit)) # 监听普通消息 application.add_handler(MessageHandler(filters.ALL & ~filters.COMMAND, handle_message)) print(f"机器人已启动... 管理员ID: {ADMIN_ID}") application.run_polling()