# Python開發(fā)機器人的核心優(yōu)勢
advantages = {
    "語法簡潔": "快速原型開發(fā)，代碼可讀性強",
    "生態(tài)豐富": "python-telegram-bot等成熟庫",
    "異步支持": "高效處理并發(fā)消息",
    "部署簡單": "多種部署方案可選",
    "社區(qū)活躍": "豐富的學習資源和解決方案"
}

#!/usr/bin/env python3
"""
Telegram機器人開發(fā)環(huán)境配置
"""

import os
from typing import Dict, Any

class BotConfig:
    """
    機器人配置管理類
    """
    
    def __init__(self):
        self.token = None
        self.webhook_url = None
        self.admin_ids = []
    
    def setup_environment(self) -> bool:
        """
        設置開發(fā)環(huán)境
        
        Returns:
            bool: 配置是否成功
        """
        # 從環(huán)境變量獲取Token
        self.token = os.getenv('TELEGRAM_BOT_TOKEN')
        
        if not self.token:
            print("? 未找到TELEGRAM_BOT_TOKEN環(huán)境變量")
            print("請按照以下步驟配置：")
            print("1. 在Telegram中搜索 @BotFather")
            print("2. 發(fā)送 /newbot 創(chuàng)建新機器人")
            print("3. 設置機器人名稱和用戶名")
            print("4. 復制獲得的Token")
            print("5. 設置環(huán)境變量: export TELEGRAM_BOT_TOKEN='你的token'")
            return False
        
        # 設置webhook URL（可選，用于生產(chǎn)環(huán)境）
        self.webhook_url = os.getenv('WEBHOOK_URL', '')
        
        # 管理員ID列表
        admin_env = os.getenv('ADMIN_IDS', '')
        if admin_env:
            self.admin_ids = [int(id.strip()) for id in admin_env.split(',')]
        
        print("? 環(huán)境配置完成")
        print(f"   機器人Token: {self.token[:10]}...")
        print(f"   管理員ID: {self.admin_ids}")
        
        return True
    
    def validate_token(self, token: str) -> bool:
        """
        驗證Token格式
        
        Args:
            token: 機器人Token
            
        Returns:
            bool: Token格式是否正確
        """
        if not token or not isinstance(token, str):
            return False
        
        # Telegram Bot Token格式: 數(shù)字:字母
        parts = token.split(':')
        if len(parts) != 2:
            return False
        
        # 第一部分應為純數(shù)字
        if not parts[0].isdigit():
            return False
        
        # 第二部分應包含字母數(shù)字
        if not parts[1].replace('_', '').isalnum():
            return False
        
        return True

def test_bot_connection(token: str) -> Dict[str, Any]:
    """
    測試機器人連接
    
    Args:
        token: 機器人Token
        
    Returns:
        dict: 連接測試結果
    """
    import requests
    
    url = f"https://api.telegram.org/bot{token}/getMe"
    
    try:
        response = requests.get(url, timeout=10)
        data = response.json()
        
        if data.get('ok'):
            bot_info = data['result']
            return {
                'success': True,
                'bot_username': bot_info.get('username'),
                'bot_name': bot_info.get('first_name'),
                'can_join_groups': bot_info.get('can_join_groups', False),
                'can_read_messages': bot_info.get('can_read_all_group_messages', False)
            }
        else:
            return {
                'success': False,
                'error': data.get('description', 'Unknown error')
            }
            
    except requests.exceptions.RequestException as e:
        return {
            'success': False,
            'error': f"網(wǎng)絡請求失敗: {e}"
        }

# 配置演示
if __name__ == "__main__":
    config = BotConfig()
    
    if config.setup_environment():
        # 測試連接
        result = test_bot_connection(config.token)
        
        if result['success']:
            print(f"? 機器人連接成功!")
            print(f"   用戶名: @{result['bot_username']}")
            print(f"   名稱: {result['bot_name']}")
            print(f"   可加入群組: {result['can_join_groups']}")
        else:
            print(f"? 連接失敗: {result['error']}")
    else:
        print("請先配置環(huán)境變量")

"""
依賴管理腳本
"""

import subprocess
import sys
import importlib

def install_package(package: str) -> bool:
    """
    安裝Python包
    
    Args:
        package: 包名稱
        
    Returns:
        bool: 安裝是否成功
    """
    try:
        subprocess.check_call([
            sys.executable, "-m", "pip", "install", package
        ])
        return True
    except subprocess.CalledProcessError:
        return False

def check_dependencies() -> Dict[str, bool]:
    """
    檢查依賴包是否已安裝
    
    Returns:
        dict: 依賴包安裝狀態(tài)
    """
    required_packages = {
        'python-telegram-bot': 'telegram',
        'requests': 'requests',
        'python-dotenv': 'dotenv',
        'aiohttp': 'aiohttp',
        'pytz': 'pytz',
        'schedule': 'schedule'
    }
    
    status = {}
    
    for package, import_name in required_packages.items():
        try:
            importlib.import_module(import_name)
            status[package] = True
            print(f"? {package} 已安裝")
        except ImportError:
            status[package] = False
            print(f"? {package} 未安裝")
    
    return status

def setup_dependencies() -> bool:
    """
    安裝所有必需的依賴包
    
    Returns:
        bool: 所有依賴是否安裝成功
    """
    print("開始安裝依賴包...")
    
    packages = [
        'python-telegram-bot[job-queue]==20.7',
        'requests==2.31.0',
        'python-dotenv==1.0.0',
        'aiohttp==3.8.5',
        'pytz==2023.3',
        'schedule==1.2.0'
    ]
    
    success_count = 0
    for package in packages:
        print(f"安裝 {package}...")
        if install_package(package):
            success_count += 1
            print(f"? {package} 安裝成功")
        else:
            print(f"? {package} 安裝失敗")
    
    print(f"\n安裝完成: {success_count}/{len(packages)}")
    return success_count == len(packages)

if __name__ == "__main__":
    print("檢查依賴環(huán)境...")
    status = check_dependencies()
    
    missing = [pkg for pkg, installed in status.items() if not installed]
    
    if missing:
        print(f"\n缺少 {len(missing)} 個依賴包")
        response = input("是否自動安裝? (y/n): ")
        if response.lower() == 'y':
            setup_dependencies()
    else:
        print("? 所有依賴包已安裝")

#!/usr/bin/env python3
"""
Telegram機器人核心架構
"""

import logging
from typing import Dict, List, Optional, Callable, Any
from abc import ABC, abstractmethod
from enum import Enum

class MessageType(Enum):
    """消息類型枚舉"""
    TEXT = "text"
    COMMAND = "command"
    CALLBACK_QUERY = "callback_query"
    PHOTO = "photo"
    DOCUMENT = "document"
    LOCATION = "location"

class BaseHandler(ABC):
    """處理器基類"""
    
    @abstractmethod
    def can_handle(self, update, context) -> bool:
        """判斷是否能處理該消息"""
        pass
    
    @abstractmethod
    def handle(self, update, context) -> Any:
        """處理消息"""
        pass

class TelegramBotCore:
    """
    機器人核心類
    負責消息路由和處理器管理
    """
    
    def __init__(self, token: str):
        self.token = token
        self.handlers: Dict[MessageType, List[BaseHandler]] = {}
        self.middlewares: List[Callable] = []
        self.logger = self._setup_logging()
        
        # 初始化處理器字典
        for msg_type in MessageType:
            self.handlers[msg_type] = []
    
    def _setup_logging(self) -> logging.Logger:
        """設置日志配置"""
        logging.basicConfig(
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
            level=logging.INFO
        )
        return logging.getLogger(__name__)
    
    def add_handler(self, handler: BaseHandler, message_type: MessageType) -> None:
        """
        添加消息處理器
        
        Args:
            handler: 處理器實例
            message_type: 消息類型
        """
        self.handlers[message_type].append(handler)
        self.logger.info(f"添加 {message_type.value} 處理器: {handler.__class__.__name__}")
    
    def add_middleware(self, middleware: Callable) -> None:
        """
        添加中間件
        
        Args:
            middleware: 中間件函數(shù)
        """
        self.middlewares.append(middleware)
        self.logger.info(f"添加中間件: {middleware.__name__}")
    
    def process_update(self, update, context) -> Optional[Any]:
        """
        處理更新消息
        
        Args:
            update: 更新對象
            context: 上下文對象
            
        Returns:
            處理結果
        """
        # 執(zhí)行中間件
        for middleware in self.middlewares:
            if not middleware(update, context):
                self.logger.debug("中間件攔截消息")
                return None
        
        # 確定消息類型
        message_type = self._determine_message_type(update)
        if not message_type:
            self.logger.warning("無法識別的消息類型")
            return None
        
        # 查找匹配的處理器
        for handler in self.handlers[message_type]:
            if handler.can_handle(update, context):
                try:
                    result = handler.handle(update, context)
                    self.logger.info(f"處理器 {handler.__class__.__name__} 處理成功")
                    return result
                except Exception as e:
                    self.logger.error(f"處理器執(zhí)行錯誤: {e}")
                    return None
        
        self.logger.debug(f"沒有找到合適的 {message_type.value} 處理器")
        return None
    
    def _determine_message_type(self, update) -> Optional[MessageType]:
        """確定消息類型"""
        if update.message:
            if update.message.text:
                if update.message.text.startswith('/'):
                    return MessageType.COMMAND
                else:
                    return MessageType.TEXT
            elif update.message.photo:
                return MessageType.PHOTO
            elif update.message.document:
                return MessageType.DOCUMENT
            elif update.message.location:
                return MessageType.LOCATION
        elif update.callback_query:
            return MessageType.CALLBACK_QUERY
        
        return None

class CommandHandler(BaseHandler):
    """命令處理器"""
    
    def __init__(self, command: str, handler_func: Callable):
        self.command = command
        self.handler_func = handler_func
    
    def can_handle(self, update, context) -> bool:
        """檢查是否為指定命令"""
        if (update.message and 
            update.message.text and 
            update.message.text.startswith('/')):
            
            command = update.message.text.split()[0].lower()
            return command == self.command.lower()
        
        return False
    
    def handle(self, update, context) -> Any:
        """處理命令"""
        return self.handler_func(update, context)

class TextMessageHandler(BaseHandler):
    """文本消息處理器"""
    
    def __init__(self, pattern: str, handler_func: Callable):
        self.pattern = pattern
        self.handler_func = handler_func
    
    def can_handle(self, update, context) -> bool:
        """檢查消息是否匹配模式"""
        if update.message and update.message.text:
            return self.pattern in update.message.text.lower()
        return False
    
    def handle(self, update, context) -> Any:
        """處理文本消息"""
        return self.handler_func(update, context)

#!/usr/bin/env python3
"""
多功能Telegram機器人實現(xiàn)
支持命令處理、消息回復、文件處理、定時任務等功能
"""

import os
import logging
import json
import asyncio
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Any
from dataclasses import dataclass
from enum import Enum

from telegram import (
    Update, 
    Bot, 
    InlineKeyboardButton, 
    InlineKeyboardMarkup,
    ReplyKeyboardMarkup,
    KeyboardButton,
    InputFile
)
from telegram.ext import (
    Application,
    CommandHandler,
    MessageHandler,
    CallbackQueryHandler,
    ContextTypes,
    filters,
    ConversationHandler
)
from telegram.error import TelegramError

# 配置日志
logging.basicConfig(
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    level=logging.INFO
)
logger = logging.getLogger(__name__)

# 對話狀態(tài)
class ConversationState(Enum):
    """對話狀態(tài)枚舉"""
    WAITING_FOR_NAME = 1
    WAITING_FOR_AGE = 2
    WAITING_FOR_FEEDBACK = 3

@dataclass
class UserData:
    """用戶數(shù)據(jù)結構"""
    user_id: int
    username: str
    first_name: str
    last_name: str
    join_date: datetime
    message_count: int = 0

class MultiFunctionBot:
    """
    多功能Telegram機器人
    """
    
    def __init__(self, token: str):
        self.token = token
        self.application = Application.builder().token(token).build()
        self.user_data: Dict[int, UserData] = {}
        self.setup_handlers()
        
        # 加載用戶數(shù)據(jù)
        self.load_user_data()
    
    def setup_handlers(self) -> None:
        """設置消息處理器"""
        
        # 命令處理器
        self.application.add_handler(CommandHandler("start", self.start_command))
        self.application.add_handler(CommandHandler("help", self.help_command))
        self.application.add_handler(CommandHandler("stats", self.stats_command))
        self.application.add_handler(CommandHandler("weather", self.weather_command))
        self.application.add_handler(CommandHandler("calc", self.calculator_command))
        self.application.add_handler(CommandHandler("remind", self.reminder_command))
        
        # 對話處理器
        conv_handler = ConversationHandler(
            entry_points=[CommandHandler("survey", self.survey_start)],
            states={
                ConversationState.WAITING_FOR_NAME: [
                    MessageHandler(filters.TEXT & ~filters.COMMAND, self.survey_name)
                ],
                ConversationState.WAITING_FOR_AGE: [
                    MessageHandler(filters.TEXT & ~filters.COMMAND, self.survey_age)
                ],
                ConversationState.WAITING_FOR_FEEDBACK: [
                    MessageHandler(filters.TEXT & ~filters.COMMAND, self.survey_feedback)
                ],
            },
            fallbacks=[CommandHandler("cancel", self.survey_cancel)],
        )
        self.application.add_handler(conv_handler)
        
        # 回調查詢處理器（按鈕點擊）
        self.application.add_handler(CallbackQueryHandler(self.button_handler))
        
        # 消息處理器
        self.application.add_handler(
            MessageHandler(filters.TEXT & ~filters.COMMAND, self.echo_message)
        )
        self.application.add_handler(
            MessageHandler(filters.PHOTO, self.handle_photo)
        )
        self.application.add_handler(
            MessageHandler(filters.DOCUMENT, self.handle_document)
        )
        
        # 錯誤處理器
        self.application.add_error_handler(self.error_handler)
    
    async def start_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        """處理 /start 命令"""
        user = update.effective_user
        chat = update.effective_chat
        
        # 記錄用戶信息
        self._record_user_activity(user)
        
        # 創(chuàng)建歡迎鍵盤
        keyboard = [
            [KeyboardButton("?? 查看統(tǒng)計"), KeyboardButton("??? 天氣查詢")],
            [KeyboardButton("?? 計算器"), KeyboardButton("? 設置提醒")],
            [KeyboardButton("?? 參與調查"), KeyboardButton("?? 幫助")]
        ]
        reply_markup = ReplyKeyboardMarkup(keyboard, resize_keyboard=True)
        
        welcome_text = f"""
?? 你好 {user.first_name}！

歡迎使用多功能機器人！我可以為你提供以下服務：

?? **數(shù)據(jù)統(tǒng)計** - 查看使用情況
??? **天氣查詢** - 獲取天氣信息  
?? **計算器** - 進行數(shù)學計算
? **提醒功能** - 設置定時提醒
?? **問卷調查** - 參與用戶調查
?? **文件處理** - 處理圖片和文檔

使用 /help 查看詳細命令列表，或者點擊下方按鈕開始使用！
        """
        
        await update.message.reply_text(welcome_text, reply_markup=reply_markup)
        logger.info(f"用戶 {user.id} 啟動了機器人")
    
    async def help_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        """處理 /help 命令"""
        help_text = """
?? **機器人命令列表**

**基礎命令:**
/start - 啟動機器人
/help - 顯示幫助信息
/stats - 查看使用統(tǒng)計

**實用工具:**
/weather <城市> - 查詢天氣
/calc <表達式> - 數(shù)學計算
/remind <時間> <消息> - 設置提醒

**交互功能:**
/survey - 參與用戶調查

**示例用法:**
/weather 北京
/calc 2+3*4
/remind 30 記得開會

?? 提示：你也可以使用下方的快捷按鈕進行操作！
        """
        
        await update.message.reply_text(help_text, parse_mode='Markdown')
    
    async def stats_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        """處理 /stats 命令"""
        user = update.effective_user
        
        if user.id not in self.user_data:
            await update.message.reply_text("? 未找到你的使用數(shù)據(jù)")
            return
        
        user_data = self.user_data[user.id]
        total_users = len(self.user_data)
        total_messages = sum(ud.message_count for ud in self.user_data.values())
        
        stats_text = f"""
?? **個人使用統(tǒng)計**

?? 用戶信息:
   ? 用戶名: {user_data.username or '未設置'}
   ? 姓名: {user_data.first_name} {user_data.last_name or ''}
   ? 加入時間: {user_data.join_date.strftime('%Y-%m-%d %H:%M')}

?? 使用數(shù)據(jù):
   ? 消息數(shù)量: {user_data.message_count}
   ? 總用戶數(shù): {total_users}
   ? 總消息數(shù): {total_messages}

?? 機器人運行:
   ? 運行狀態(tài): ? 正常
   ? 最后更新: {datetime.now().strftime('%Y-%m-%d %H:%M')}
        """
        
        await update.message.reply_text(stats_text, parse_mode='Markdown')
    
    async def weather_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        """處理 /weather 命令"""
        if not context.args:
            await update.message.reply_text("? 請指定城市名稱，例如: /weather 北京")
            return
        
        city = ' '.join(context.args)
        
        # 模擬天氣API調用
        weather_data = await self._get_weather_data(city)
        
        if weather_data:
            weather_text = f"""
??? **{city} 天氣信息**

?? 更新時間: {weather_data['time']}
??? 溫度: {weather_data['temp']}°C
?? 濕度: {weather_data['humidity']}%
??? 風速: {weather_data['wind']} km/h
?? 天氣: {weather_data['condition']}
?? 描述: {weather_data['description']}
            """
            
            # 添加建議按鈕
            keyboard = [
                [InlineKeyboardButton("?? 刷新天氣", callback_data=f"refresh_weather_{city}")],
                [InlineKeyboardButton("?? 其他城市", callback_data="other_city")]
            ]
            reply_markup = InlineKeyboardMarkup(keyboard)
            
            await update.message.reply_text(
                weather_text, 
                reply_markup=reply_markup,
                parse_mode='Markdown'
            )
        else:
            await update.message.reply_text(f"? 無法獲取 {city} 的天氣信息")
    
    async def calculator_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        """處理 /calc 命令"""
        if not context.args:
            # 顯示計算器鍵盤
            keyboard = [
                [
                    InlineKeyboardButton("7", callback_data="calc_7"),
                    InlineKeyboardButton("8", callback_data="calc_8"),
                    InlineKeyboardButton("9", callback_data="calc_9"),
                    InlineKeyboardButton("÷", callback_data="calc_divide")
                ],
                [
                    InlineKeyboardButton("4", callback_data="calc_4"),
                    InlineKeyboardButton("5", callback_data="calc_5"),
                    InlineKeyboardButton("6", callback_data="calc_6"),
                    InlineKeyboardButton("×", callback_data="calc_multiply")
                ],
                [
                    InlineKeyboardButton("1", callback_data="calc_1"),
                    InlineKeyboardButton("2", callback_data="calc_2"),
                    InlineKeyboardButton("3", callback_data="calc_3"),
                    InlineKeyboardButton("-", callback_data="calc_subtract")
                ],
                [
                    InlineKeyboardButton("0", callback_data="calc_0"),
                    InlineKeyboardButton(".", callback_data="calc_decimal"),
                    InlineKeyboardButton("=", callback_data="calc_equals"),
                    InlineKeyboardButton("+", callback_data="calc_add")
                ],
                [
                    InlineKeyboardButton("C", callback_data="calc_clear"),
                    InlineKeyboardButton("(", callback_data="calc_lparen"),
                    InlineKeyboardButton(")", callback_data="calc_rparen"),
                    InlineKeyboardButton("?", callback_data="calc_backspace")
                ]
            ]
            reply_markup = InlineKeyboardMarkup(keyboard)
            
            await update.message.reply_text(
                "?? **計算器**\n\n當前表達式: `0`\n結果: `0`",
                reply_markup=reply_markup,
                parse_mode='Markdown'
            )
            return
        
        # 計算表達式
        expression = ' '.join(context.args)
        try:
            # 安全地計算數(shù)學表達式
            result = self._safe_eval(expression)
            await update.message.reply_text(f"?? 計算結果:\n`{expression} = {result}`", 
                                         parse_mode='Markdown')
        except Exception as e:
            await update.message.reply_text(f"? 計算錯誤: {str(e)}")
    
    async def reminder_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        """處理 /remind 命令"""
        if len(context.args) < 2:
            await update.message.reply_text(
                "? 使用方法: /remind <分鐘> <提醒內容>\n"
                "示例: /remind 30 記得開會"
            )
            return
        
        try:
            minutes = int(context.args[0])
            message = ' '.join(context.args[1:])
            
            if minutes <= 0:
                await update.message.reply_text("? 時間必須大于0分鐘")
                return
            
            # 設置提醒
            remind_time = datetime.now() + timedelta(minutes=minutes)
            
            context.job_queue.run_once(
                self._send_reminder,
                minutes * 60,
                data={
                    'chat_id': update.effective_chat.id,
                    'message': message
                }
            )
            
            await update.message.reply_text(
                f"? 提醒已設置!\n"
                f"? 時間: {remind_time.strftime('%H:%M')}\n"
                f"?? 內容: {message}"
            )
            
        except ValueError:
            await update.message.reply_text("? 時間參數(shù)必須是數(shù)字")
    
    async def _send_reminder(self, context: ContextTypes.DEFAULT_TYPE) -> None:
        """發(fā)送提醒"""
        job_data = context.job.data
        await context.bot.send_message(
            chat_id=job_data['chat_id'],
            text=f"? **提醒**: {job_data['message']}",
            parse_mode='Markdown'
        )
    
    # 調查對話處理函數(shù)
    async def survey_start(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
        """開始調查"""
        await update.message.reply_text(
            "?? 歡迎參與用戶調查!\n\n"
            "請告訴我你的姓名:"
        )
        return ConversationState.WAITING_FOR_NAME
    
    async def survey_name(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
        """處理姓名輸入"""
        context.user_data['survey_name'] = update.message.text
        await update.message.reply_text("謝謝! 現(xiàn)在請告訴我你的年齡:")
        return ConversationState.WAITING_FOR_AGE
    
    async def survey_age(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
        """處理年齡輸入"""
        try:
            age = int(update.message.text)
            if age < 0 or age > 150:
                await update.message.reply_text("? 請輸入合理的年齡 (0-150):")
                return ConversationState.WAITING_FOR_AGE
            
            context.user_data['survey_age'] = age
            await update.message.reply_text(
                "很好! 最后請?zhí)峁┠愕姆答佉庖?"
            )
            return ConversationState.WAITING_FOR_FEEDBACK
            
        except ValueError:
            await update.message.reply_text("? 請輸入數(shù)字年齡:")
            return ConversationState.WAITING_FOR_AGE
    
    async def survey_feedback(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
        """處理反饋輸入"""
        context.user_data['survey_feedback'] = update.message.text
        
        # 保存調查結果
        survey_result = {
            'name': context.user_data['survey_name'],
            'age': context.user_data['survey_age'],
            'feedback': context.user_data['survey_feedback'],
            'timestamp': datetime.now().isoformat()
        }
        
        self._save_survey_result(update.effective_user.id, survey_result)
        
        await update.message.reply_text(
            "? 感謝你完成調查!\n\n"
            f"姓名: {survey_result['name']}\n"
            f"年齡: {survey_result['age']}\n"
            f"反饋: {survey_result['feedback']}\n\n"
            "你的意見對我們非常重要! ??"
        )
        
        # 清理用戶數(shù)據(jù)
        context.user_data.clear()
        return ConversationHandler.END
    
    async def survey_cancel(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
        """取消調查"""
        await update.message.reply_text("調查已取消。")
        context.user_data.clear()
        return ConversationHandler.END
    
    # 按鈕回調處理
    async def button_handler(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        """處理按鈕點擊"""
        query = update.callback_query
        await query.answer()
        
        data = query.data
        
        if data.startswith('refresh_weather_'):
            city = data.replace('refresh_weather_', '')
            weather_data = await self._get_weather_data(city)
            
            if weather_data:
                weather_text = f"""
??? **{city} 天氣信息** (已刷新)

?? 更新時間: {weather_data['time']}
??? 溫度: {weather_data['temp']}°C
?? 濕度: {weather_data['humidity']}%
??? 風速: {weather_data['wind']} km/h
?? 天氣: {weather_data['condition']}
                """
                await query.edit_message_text(
                    weather_text, 
                    parse_mode='Markdown'
                )
        
        elif data == 'other_city':
            await query.edit_message_text(
                "請使用 /weather <城市名> 查詢其他城市天氣\n"
                "例如: /weather 上海"
            )
    
    # 消息處理函數(shù)
    async def echo_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        """處理文本消息"""
        user = update.effective_user
        message_text = update.message.text
        
        self._record_user_activity(user)
        
        # 根據(jù)消息內容回復
        responses = {
            "你好": f"你好 {user.first_name}! ??",
            "謝謝": "不用客氣! ??",
            "時間": f"當前時間: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
            "幫助": "使用 /help 查看所有可用命令",
        }
        
        response = responses.get(message_text)
        if response:
            await update.message.reply_text(response)
        else:
            # 默認回復
            await update.message.reply_text(
                f"你說: {message_text}\n"
                f"使用 /help 查看我能做什么"
            )
    
    async def handle_photo(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        """處理圖片消息"""
        user = update.effective_user
        photo = update.message.photo[-1]  # 獲取最高質量的圖片
        
        await update.message.reply_text(
            f"?? 收到圖片!\n"
            f"文件ID: {photo.file_id}\n"
            f"大小: {photo.file_size} bytes\n\n"
            f"感謝分享圖片 {user.first_name}! ??"
        )
    
    async def handle_document(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        """處理文檔消息"""
        user = update.effective_user
        document = update.message.document
        
        file_info = await context.bot.get_file(document.file_id)
        
        await update.message.reply_text(
            f"?? 收到文檔!\n"
            f"文件名: {document.file_name}\n"
            f"類型: {document.mime_type}\n"
            f"大小: {document.file_size} bytes\n"
            f"文件ID: {document.file_id}\n\n"
            f"文檔已保存到服務器"
        )
    
    async def error_handler(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        """處理錯誤"""
        logger.error(f"機器人錯誤: {context.error}", exc_info=context.error)
        
        if update and update.effective_chat:
            await context.bot.send_message(
                chat_id=update.effective_chat.id,
                text="? 抱歉，發(fā)生了錯誤。請稍后重試。"
            )
    
    # 工具函數(shù)
    def _record_user_activity(self, user) -> None:
        """記錄用戶活動"""
        if user.id not in self.user_data:
            self.user_data[user.id] = UserData(
                user_id=user.id,
                username=user.username,
                first_name=user.first_name,
                last_name=user.last_name or '',
                join_date=datetime.now()
            )
        
        self.user_data[user.id].message_count += 1
        self.save_user_data()
    
    async def _get_weather_data(self, city: str) -> Optional[Dict[str, Any]]:
        """獲取天氣數(shù)據(jù)（模擬）"""
        # 在實際應用中，這里應該調用天氣API
        # 這里使用模擬數(shù)據(jù)
        
        import random
        weather_conditions = ["晴朗", "多云", "小雨", "陰天", "霧霾"]
        
        return {
            'city': city,
            'temp': random.randint(15, 35),
            'humidity': random.randint(30, 90),
            'wind': random.randint(5, 25),
            'condition': random.choice(weather_conditions),
            'description': f"{city}的天氣情況",
            'time': datetime.now().strftime('%Y-%m-%d %H:%M')
        }
    
    def _safe_eval(self, expression: str) -> float:
        """安全計算數(shù)學表達式"""
        import ast
        import operator as op
        
        # 支持的運算符
        operators = {
            ast.Add: op.add,
            ast.Sub: op.sub, 
            ast.Mult: op.mul,
            ast.Div: op.truediv,
            ast.Pow: op.pow,
            ast.USub: op.neg
        }
        
        def _eval(node):
            if isinstance(node, ast.Num):  # 數(shù)字
                return node.n
            elif isinstance(node, ast.BinOp):  # 二元操作
                return operators[type(node.op)](_eval(node.left), _eval(node.right))
            elif isinstance(node, ast.UnaryOp):  # 一元操作
                return operators[type(node.op)](_eval(node.operand))
            else:
                raise TypeError(f"不支持的表達式: {node}")
        
        # 解析表達式
        tree = ast.parse(expression, mode='eval').body
        return _eval(tree)
    
    def _save_survey_result(self, user_id: int, result: Dict) -> None:
        """保存調查結果"""
        filename = f"survey_results.json"
        try:
            if os.path.exists(filename):
                with open(filename, 'r', encoding='utf-8') as f:
                    data = json.load(f)
            else:
                data = []
            
            result['user_id'] = user_id
            data.append(result)
            
            with open(filename, 'w', encoding='utf-8') as f:
                json.dump(data, f, ensure_ascii=False, indent=2)
                
        except Exception as e:
            logger.error(f"保存調查結果失敗: {e}")
    
    def save_user_data(self) -> None:
        """保存用戶數(shù)據(jù)"""
        try:
            data = {
                'users': {
                    uid: {
                        'user_id': ud.user_id,
                        'username': ud.username,
                        'first_name': ud.first_name,
                        'last_name': ud.last_name,
                        'join_date': ud.join_date.isoformat(),
                        'message_count': ud.message_count
                    }
                    for uid, ud in self.user_data.items()
                },
                'last_updated': datetime.now().isoformat()
            }
            
            with open('user_data.json', 'w', encoding='utf-8') as f:
                json.dump(data, f, ensure_ascii=False, indent=2)
                
        except Exception as e:
            logger.error(f"保存用戶數(shù)據(jù)失敗: {e}")
    
    def load_user_data(self) -> None:
        """加載用戶數(shù)據(jù)"""
        try:
            if os.path.exists('user_data.json'):
                with open('user_data.json', 'r', encoding='utf-8') as f:
                    data = json.load(f)
                
                for uid, ud in data.get('users', {}).items():
                    self.user_data[int(uid)] = UserData(
                        user_id=ud['user_id'],
                        username=ud['username'],
                        first_name=ud['first_name'],
                        last_name=ud['last_name'],
                        join_date=datetime.fromisoformat(ud['join_date']),
                        message_count=ud['message_count']
                    )
                    
                logger.info(f"加載了 {len(self.user_data)} 個用戶數(shù)據(jù)")
                
        except Exception as e:
            logger.error(f"加載用戶數(shù)據(jù)失敗: {e}")
    
    def run(self) -> None:
        """運行機器人"""
        logger.info("啟動多功能Telegram機器人...")
        
        # 啟動輪詢
        self.application.run_polling(
            allowed_updates=Update.ALL_TYPES,
            poll_interval=1,
            timeout=20
        )

# 使用示例
if __name__ == "__main__":
    # 從環(huán)境變量獲取Token
    token = os.getenv('TELEGRAM_BOT_TOKEN')
    
    if not token:
        print("? 請設置 TELEGRAM_BOT_TOKEN 環(huán)境變量")
        print("export TELEGRAM_BOT_TOKEN='你的token'")
        exit(1)
    
    # 創(chuàng)建并運行機器人
    bot = MultiFunctionBot(token)
    bot.run()

#!/usr/bin/env python3
"""
機器人性能優(yōu)化工具
"""

import asyncio
import time
from typing import Dict, List
from dataclasses import dataclass
from concurrent.futures import ThreadPoolExecutor

@dataclass
class PerformanceMetrics:
    """性能指標"""
    response_time: float
    memory_usage: float
    active_users: int
    messages_processed: int

class BotOptimizer:
    """
    機器人性能優(yōu)化器
    """
    
    def __init__(self, max_workers: int = 10):
        self.max_workers = max_workers
        self.thread_pool = ThreadPoolExecutor(max_workers=max_workers)
        self.metrics: Dict[str, List[PerformanceMetrics]] = {}
    
    async def process_message_batch(self, messages: List) -> List:
        """
        批量處理消息以提高性能
        
        Args:
            messages: 消息列表
            
        Returns:
            處理結果列表
        """
        loop = asyncio.get_event_loop()
        
        # 使用線程池處理CPU密集型任務
        tasks = [
            loop.run_in_executor(
                self.thread_pool, 
                self._process_single_message, 
                message
            )
            for message in messages
        ]
        
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return results
    
    def _process_single_message(self, message) -> Dict:
        """
        處理單個消息（CPU密集型操作）
        
        Args:
            message: 消息對象
            
        Returns:
            處理結果
        """
        # 模擬處理時間
        time.sleep(0.01)
        return {"status": "processed", "message_id": getattr(message, 'id', 0)}
    
    def record_metrics(self, bot_name: str, metrics: PerformanceMetrics) -> None:
        """
        記錄性能指標
        
        Args:
            bot_name: 機器人名稱
            metrics: 性能指標
        """
        if bot_name not in self.metrics:
            self.metrics[bot_name] = []
        
        self.metrics[bot_name].append(metrics)
        
        # 保持最近100條記錄
        if len(self.metrics[bot_name]) > 100:
            self.metrics[bot_name].pop(0)
    
    def get_performance_report(self, bot_name: str) -> Dict:
        """
        獲取性能報告
        
        Args:
            bot_name: 機器人名稱
            
        Returns:
            性能報告
        """
        if bot_name not in self.metrics:
            return {}
        
        metrics_list = self.metrics[bot_name]
        if not metrics_list:
            return {}
        
        avg_response_time = sum(m.response_time for m in metrics_list) / len(metrics_list)
        avg_memory_usage = sum(m.memory_usage for m in metrics_list) / len(metrics_list)
        max_users = max(m.active_users for m in metrics_list)
        total_messages = sum(m.messages_processed for m in metrics_list)
        
        return {
            'avg_response_time': avg_response_time,
            'avg_memory_usage': avg_memory_usage,
            'peak_users': max_users,
            'total_messages_processed': total_messages,
            'sample_size': len(metrics_list)
        }

# 性能測試
async def performance_test():
    """性能測試函數(shù)"""
    optimizer = BotOptimizer()
    
    # 模擬消息處理
    test_messages = [f"message_{i}" for i in range(100)]
    
    start_time = time.time()
    results = await optimizer.process_message_batch(test_messages)
    end_time = time.time()
    
    print(f"處理 {len(test_messages)} 條消息耗時: {end_time - start_time:.2f}秒")
    print(f"平均每條消息: {(end_time - start_time) / len(test_messages) * 1000:.2f}毫秒")

if __name__ == "__main__":
    asyncio.run(performance_test())

#!/usr/bin/env python3
"""
生產(chǎn)環(huán)境部署配置
"""

import os
import logging
from typing import Optional

class DeploymentConfig:
    """
    部署配置管理
    """
    
    def __init__(self):
        self.environment = os.getenv('ENVIRONMENT', 'development')
        self.webhook_url = os.getenv('WEBHOOK_URL', '')
        self.port = int(os.getenv('PORT', '8443'))
        self.host = os.getenv('HOST', '0.0.0.0')
        
    def setup_production(self, application, token: str) -> None:
        """
        設置生產(chǎn)環(huán)境配置
        
        Args:
            application: Telegram應用實例
            token: 機器人Token
        """
        if not self.webhook_url:
            logging.warning("未設置WEBHOOK_URL，使用輪詢模式")
            return
        
        # 設置webhook
        webhook_url = f"{self.webhook_url}/{token}"
        
        try:
            application.run_webhook(
                listen=self.host,
                port=self.port,
                url_path=token,
                webhook_url=webhook_url,
                cert=None  # 如果有SSL證書可以在這里指定
            )
            logging.info(f"Webhook設置成功: {webhook_url}")
        except Exception as e:
            logging.error(f"Webhook設置失敗: {e}")
            logging.info("回退到輪詢模式")
            application.run_polling()

def setup_logging():
    """設置生產(chǎn)環(huán)境日志"""
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
        handlers=[
            logging.FileHandler('bot.log'),
            logging.StreamHandler()
        ]
    )

# 健康檢查端點
def create_health_check():
    """創(chuàng)建健康檢查（用于Webhook部署）"""
    from http.server import HTTPServer, BaseHTTPRequestHandler
    
    class HealthHandler(BaseHTTPRequestHandler):
        def do_GET(self):
            self.send_response(200)
            self.send_header('Content-type', 'text/plain')
            self.end_headers()
            self.wfile.write(b'OK')
        
        def log_message(self, format, *args):
            # 減少健康檢查日志噪音
            if '/health' not in self.path:
                logging.info(format % args)
    
    return HealthHandler

def start_health_server(port: int = 8080):
    """啟動健康檢查服務器"""
    handler = create_health_check()
    httpd = HTTPServer(('0.0.0.0', port), handler)
    
    logging.info(f"健康檢查服務器運行在端口 {port}")
    httpd.serve_forever()