run/telegram-bot-demo/demo_multi_instance.py

#!/usr/bin/env python3
"""
多实例Telegram Bot演示脚本
展示如何同时运行多个不同的Bot实例
"""

import asyncio
import logging
import sys
import os
import tempfile
from pathlib import Path

# 添加src目录到Python路径
current_dir = Path(__file__).parent
project_root = current_dir.parent.parent
src_dir = project_root / "src"
sys.path.insert(0, str(src_dir))

from claude_agent.storage.persistence import PersistenceManager


def setup_logging():
    """设置日志配置"""
    logging.basicConfig(
        level=logging.INFO,
        format='[%(asctime)s] %(levelname)s %(name)s: %(message)s',
        datefmt='%Y-%m-%d %H:%M:%S'
    )


async def demo_storage_isolation():
    """演示存储隔离功能"""
    print("\n🗃️ 演示多实例存储隔离")
    print("=" * 50)

    with tempfile.TemporaryDirectory() as temp_dir:
        print(f"临时存储目录: {temp_dir}")

        # 创建三个不同Bot的存储管理器
        bot_configs = [
            {"id": "123456789", "name": "ChatBot"},
            {"id": "987654321", "name": "HelpBot"},
            {"id": "555666777", "name": "NewsBot"}
        ]

        managers = {}
        for config in bot_configs:
            bot_id = config["id"]
            bot_name = config["name"]
            managers[bot_id] = PersistenceManager(temp_dir, bot_id=bot_id)
            print(f"✅ 创建 {bot_name} (ID: {bot_id}) 的存储管理器")

        # 模拟各Bot保存不同的对话历史
        print("\n📝 模拟各Bot保存对话历史...")
        for i, (bot_id, manager) in enumerate(managers.items()):
            chat_id = "user_123"  # 相同的用户ID
            history = [
                {"message": f"这是Bot {bot_id}的第{j+1}条消息", "timestamp": f"2024-01-{i+1:02d}"}
                for j in range(3)
            ]
            manager.save_conversation_history(chat_id, history)
            print(f"  Bot {bot_id}: 保存了 {len(history)} 条消息")

        # 验证存储隔离
        print("\n🔍 验证存储隔离...")
        for bot_id, manager in managers.items():
            history = manager.load_conversation_history("user_123")
            stats = manager.get_storage_stats()
            print(f"  Bot {bot_id}: 加载到 {len(history)} 条消息，存储目录: {Path(stats['storage_dir']).name}")

        # 显示目录结构
        print("\n📁 存储目录结构:")
        for root, dirs, files in os.walk(temp_dir):
            level = root.replace(temp_dir, '').count(os.sep)
            indent = '  ' * level
            print(f'{indent}{os.path.basename(root)}/')
            sub_indent = '  ' * (level + 1)
            for file in files:
                print(f'{sub_indent}{file}')


async def demo_lock_isolation():
    """演示文件锁隔离功能"""
    print("\n🔒 演示多实例文件锁隔离")
    print("=" * 50)

    # 直接导入ProcessLock类
    scripts_dir = project_root / "scripts"
    sys.path.insert(0, str(scripts_dir))
    from start_telegram_bot import ProcessLock

    # 模拟不同Bot的锁
    lock_names = ["telegram_bot_123456789", "telegram_bot_987654321", "telegram_bot_555666777"]

    print("尝试同时获取不同Bot的锁...")

    locks = []
    try:
        for lock_name in lock_names:
            lock = ProcessLock(lock_name)
            lock.__enter__()
            locks.append(lock)
            print(f"✅ 成功获取锁: {lock_name}")

        print("✅ 所有不同Bot的锁都可以同时持有")

    except Exception as e:
        print(f"❌ 锁获取失败: {e}")
    finally:
        # 清理锁
        for lock in locks:
            try:
                lock.__exit__(None, None, None)
            except:
                pass

    # 测试相同Bot的锁互斥
    print("\n测试相同Bot的锁互斥...")
    import threading
    import time

    def try_same_lock():
        try:
            with ProcessLock("telegram_bot_123456789"):
                print("❌ 线程2意外获取到相同的锁!")
        except RuntimeError:
            print("✅ 线程2正确被阻止获取相同的锁")

    with ProcessLock("telegram_bot_123456789"):
        print("主线程持有锁: telegram_bot_123456789")
        thread = threading.Thread(target=try_same_lock)
        thread.start()
        thread.join()


async def demo_configuration_examples():
    """演示配置示例"""
    print("\n⚙️ 多实例配置示例")
    print("=" * 50)

    examples = [
        {
            "name": "ChatBot配置",
            "config": """# configs/chatbot.toml
[telegram]
bot_token = "123456789:AbCdEfGhIjKlMnOpQrStUvWxYz0123456789"
allowed_users = [111111111, 222222222]
allowed_groups = [-333333333]

[telegram.group_participation]
bot_names = ["chatbot", "小聊", "机器人"]
random_participation_range = [3, 8]
""",
            "command": "CLAUDE_CONFIG=chatbot python scripts/start_telegram_bot.py"
        },
        {
            "name": "HelpBot配置",
            "config": """# configs/helpbot.toml
[telegram]
bot_token = "987654321:XyZaBcDeFgHiJkLmNoPqRsTuVwXyZ9876543"
allowed_users = [444444444, 555555555]
allowed_groups = [-666666666]

[telegram.group_participation]
bot_names = ["helpbot", "助手", "客服", "help"]
random_participation_range = [10, 20]
""",
            "command": "CLAUDE_CONFIG=helpbot python scripts/start_telegram_bot.py"
        }
    ]

    for example in examples:
        print(f"\n📋 {example['name']}:")
        print(example['config'])
        print(f"🚀 启动命令: {example['command']}")


async def main():
    """主演示函数"""
    setup_logging()

    print("🤖 Telegram Bot 多实例功能演示")
    print("=" * 60)

    await demo_storage_isolation()
    await demo_lock_isolation()
    await demo_configuration_examples()

    print("\n" + "=" * 60)
    print("✅ 多实例功能演示完成!")
    print("\n💡 总结:")
    print("  - 不同Bot使用不同的bot_id进行存储隔离")
    print("  - 不同Bot使用不同的锁文件防止冲突")
    print("  - 相同Bot不能同时运行多个实例")
    print("  - 可以通过不同配置文件同时运行多个不同的Bot")
    print("  - 每个Bot可以有独立的名字和参与行为配置")


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\n\n演示被用户中断")
    except Exception as e:
        print(f"\n演示发生错误: {e}")