Telegram Bot 整合 Claude CLI + MCP 圖片生成

def is_user_allowed(user_id: int) -> bool:
    """檢查用戶是否被允許使用 bot"""
    allowed = get_allowed_users()
    # 如果沒有設定白名單，則拒絕所有人
    if not allowed:
        return False
    return user_id in allowed

def is_group_allowed(chat_id: int) -> bool:
    """檢查群組是否被允許使用 bot"""
    allowed = get_allowed_groups()
    if not allowed:
        return False
    return chat_id in allowed

def check_permission(update: Update, context: ContextTypes.DEFAULT_TYPE) -> bool:
    """統一檢查權限（用戶 + 群組）"""
    user = update.effective_user
    chat = update.effective_chat

    if not is_user_allowed(user.id):
        return False

    # 私人對話直接允許
    if chat.type == "private":
        return True

    # 群組對話檢查群組白名單
    return is_group_allowed(chat.id)

async def status_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    user = update.effective_user
    # 未授權用戶：只顯示自己的 ID
    if not is_user_allowed(user.id):
        status_text = (
            f"你的用戶 ID: {user.id}\n\n"
            f"你尚未被授權使用此 Bot。\n"
            f"請將此 ID 提供給管理員以申請存取權限。"
        )
        await update.message.reply_text(status_text, parse_mode="HTML")
        return
    # ... 已授權用戶顯示完整狀態

BOT_START_TIME = None

async def post_init(application: Application) -> None:
    """Bot 啟動後的初始化"""
    global BOT_USERNAME, BOT_START_TIME
    bot = await application.bot.get_me()
    BOT_USERNAME = bot.username
    BOT_START_TIME = datetime.now()

    # 通知管理員 Bot 已啟動
    admin_id = get_admin_id()
    if admin_id:
        await application.bot.send_message(
            chat_id=admin_id,
            text=f"Bot 已上線\n@{BOT_USERNAME}\n{BOT_START_TIME.strftime('%Y-%m-%d %H:%M:%S')}",
            parse_mode="HTML",
        )

async def ping_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    if BOT_START_TIME:
        uptime = datetime.now() - BOT_START_TIME
        hours, remainder = divmod(int(uptime.total_seconds()), 3600)
        minutes, seconds = divmod(remainder, 60)
        uptime_str = f"{hours}h {minutes}m {seconds}s"
    else:
        uptime_str = "未知"

    await update.message.reply_text(f"Pong! Bot 運行中\n已運行: {uptime_str}")

# 舊版：直接呼叫 Claude CLI（已棄用）
process = await asyncio.create_subprocess_exec(
    "claude", "-p", prompt, "--model", model,
    stdout=asyncio.subprocess.PIPE,
    stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=120)

from copilot import CopilotClient

# 全域 Client，只需啟動一次
_client: CopilotClient | None = None

async def _get_client() -> CopilotClient:
    """取得或初始化全域 CopilotClient"""
    global _client
    async with _client_lock:
        if _client is None:
            _client = CopilotClient({"cwd": PROJECT_DIR})
            await _client.start()
        return _client

async def call_claude(
    prompt: str,
    model: str = "haiku",
    system_prompt: str | None = None,
    timeout: int = DEFAULT_TIMEOUT,
    on_tool_start: ToolNotifyCallback | None = None,
    on_tool_end: ToolNotifyCallback | None = None,
    allowed_tools: list[str] | None = None,
) -> ClaudeResponse:
    client = await _get_client()

    session_config = {
        "model": cli_model,
        "cli_path": "claude",
        "cli_args": "--experimental-acp",
        "streaming": False,
    }
    if system_prompt:
        session_config["system_message"] = {"content": system_prompt}
    if tools:
        session_config["allowed_tools"] = tools

    session = await client.create_session(session_config)

def on_event(event):
    event_type = event.type.value

    if event_type == "assistant.message":
        # AI 回應了一段文字
        response_content = event.data.content or ""

    elif event_type == "assistant.tool_use":
        # Tool 開始執行
        tool_name = getattr(event.data, "name", "")
        tool_input = getattr(event.data, "input", {})
        # 通知使用者

    elif event_type == "tool.result":
        # Tool 執行完成
        tool_output = getattr(event.data, "content", "")
        # 計算耗時，通知使用者

    elif event_type == "session.idle":
        # 整個對話完成
        done.set()

session.on(on_event)
await session.send({"prompt": prompt})

MODEL_MAP = {
    "opus": "claude-sonnet-4.5",
    "sonnet": "claude-sonnet-4",
    "haiku": "claude-haiku-4.5",
}

async def on_tool_start(tool_name: str, tool_input: dict):
    """Tool 開始執行時的回調"""
    status_line = f"tool_name\n   執行中..."
    tool_status_lines.append({"name": tool_name, "status": "running", "line": status_line})

    full_text = "AI 處理中\n\n" + "\n\n".join(t["line"] for t in tool_status_lines)

    if notify_message_id is None:
        msg = await bot.send_message(chat_id=chat_id, text=full_text, parse_mode="HTML")
        notify_message_id = msg.message_id
    else:
        await bot.edit_message_text(
            chat_id=chat_id, message_id=notify_message_id,
            text=full_text, parse_mode="HTML",
        )

async def on_tool_end(tool_name: str, result: dict):
    duration_ms = result.get("duration_ms", 0)
    duration_str = f"{duration_ms}ms" if duration_ms < 1000 else f"{duration_ms/1000:.1f}s"

    for tool in tool_status_lines:
        if tool["name"] == tool_name and tool["status"] == "running":
            tool["status"] = "done"
            tool["line"] = tool["line"].replace("執行中...", f"完成 ({duration_str})")
            break

MAX_TURNS = 2

# 工具完成後檢查是否超過 turn 限制
if turn_count >= MAX_TURNS and not pending_tools:
    logger.info(f"已達 MAX_TURNS={MAX_TURNS} 且所有工具已完成，中止 session")
    session.abort()

def extract_image_paths_from_tool_calls(tool_calls: list) -> list[str]:
    """從 tool_calls 中提取 nanobanana 生成的圖片路徑"""
    nanobanana_tools = {
        "mcp__nanobanana__generate_image",
        "mcp__nanobanana__edit_image",
        "mcp__nanobanana__restore_image",
        "mcp__nanobanana__generate_icon",
        "mcp__nanobanana__generate_pattern",
        "mcp__nanobanana__generate_story",
        "mcp__nanobanana__generate_diagram",
    }

    generated_files = []
    for tc in tool_calls:
        if tc.name not in nanobanana_tools:
            continue

        output_data = json.loads(tc.output)
        # 格式: [{"text": '{"success": true, "generatedFiles": [...]}', "type": "text"}]
        if isinstance(output_data, list):
            for item in output_data:
                if item.get("type") == "text":
                    inner_data = json.loads(item["text"])
                    if inner_data.get("success") and inner_data.get("generatedFiles"):
                        generated_files.extend(inner_data["generatedFiles"])

    # 去重複並過濾存在的檔案
    return [p for p in set(generated_files) if os.path.exists(p)]

# 發送圖片
for img_path in image_paths:
    with open(img_path, "rb") as img_file:
        await update.message.reply_photo(photo=img_file)

async def get_reply_context(update: Update, context: ContextTypes.DEFAULT_TYPE) -> str | None:
    reply = update.message.reply_to_message
    if not reply:
        return None

    parts = []

    # 回覆的文字（圖片的 caption 或文字訊息）
    reply_text = reply.text or reply.caption
    if reply_text:
        if len(reply_text) > 500:
            reply_text = reply_text[:500] + "..."
        parts.append(f"[回覆訊息: {reply_text}]")

    return "\n".join(parts) if parts else None

# 回覆的圖片
if reply.photo:
    # 取得最大尺寸的圖片
    photo = reply.photo[-1]
    file = await context.bot.get_file(photo.file_id)

    # 下載到暫存目錄
    file_path = os.path.join(REPLY_IMAGE_DIR, f"{photo.file_unique_id}.jpg")
    await file.download_to_drive(file_path)

    parts.append(f"[回覆圖片: {file_path}]")

def extract_image_urls(text: str) -> list[str]:
    """從文字中提取圖片 URL"""
    pattern = r'https?://[^\s\n\[\]()<>\"\']+\.(?:jpg|jpeg|png|gif|webp)(?:\?[^\s\n\[\]()<>\"\']*)?'
    urls = re.findall(pattern, text, re.IGNORECASE)
    return list(dict.fromkeys(urls))  # 去重保留順序

async def download_image_from_url(url: str) -> str | None:
    """下載圖片 URL 到暫存目錄"""
    async with httpx.AsyncClient(follow_redirects=True, timeout=30) as client:
        resp = await client.get(url)
        if resp.status_code != 200:
            return None

        content_type = resp.headers.get("content-type", "")
        if not content_type.startswith("image/"):
            return None

        # 用 MD5 hash 作為檔名避免重複
        filename = hashlib.md5(url.encode()).hexdigest()[:12] + ext
        file_path = os.path.join(download_dir, filename)

        with open(file_path, "wb") as f:
            f.write(resp.content)

        return file_path

image_urls = extract_image_urls(response)
if image_urls:
    for url in image_urls[:5]:  # 最多下載 5 張
        local_path = await download_image_from_url(url)
        if local_path:
            image_paths.append(local_path)

def is_mentioned(update: Update, context: ContextTypes.DEFAULT_TYPE) -> bool:
    message = update.message

    # 檢查是否為回覆 Bot 的訊息
    if message.reply_to_message and message.reply_to_message.from_user:
        if message.reply_to_message.from_user.id == context.bot.id:
            return True

    # 檢查訊息中是否有 @Bot
    if message.entities:
        for entity in message.entities:
            if entity.type == "mention":
                mention_text = message.text[entity.offset:entity.offset + entity.length]
                if BOT_USERNAME and mention_text.lower() == f"@{BOT_USERNAME.lower()}":
                    return True

    return False

if BOT_USERNAME:
    text = text.replace(f"@{BOT_USERNAME}", "").strip()

# 舊版做法
cmd = ["claude", "-p", prompt, "--model", model, "--output-format", "json"]
process = await asyncio.create_subprocess_exec(
    *cmd,
    stdout=asyncio.subprocess.PIPE,
    stderr=asyncio.subprocess.PIPE,
    cwd=PROJECT_DIR,
)
stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=120)
result = json.loads(stdout.decode())

from copilot import CopilotClient

# 全域 Client，只需啟動一次
client = CopilotClient({"cwd": PROJECT_DIR})
await client.start()

# 每次對話建立新 Session
session = await client.create_session({
    "model": "claude-haiku-4.5",
    "cli_path": "claude",
    "cli_args": "--experimental-acp",
    "allowed_tools": ["WebSearch", "mcp__nanobanana__generate_image"],
})

# 事件驅動
session.on(on_event)
await session.send({"prompt": "畫一隻貓"})

async def post_shutdown(application: Application) -> None:
    """Bot 關閉時清理 CopilotClient"""
    await shutdown_client()