Hermes Qq

"""
Hermes QQ Bot Platform Adapter

Fully aligned with Hermes gateway specification + Tencent OpenClaw QQ Bot official implementation.
QQ Open Platform documentation: https://q.qq.com/wiki/

This adapter should be placed in:
  ~/.hermes/plugins/hermes-qqbot/qqbot/adapter.py

Or integrated into core with 2 minimal modifications:
  1. gateway/config.py - Add Platform.QQ = "qq"
  2. gateway/run.py   - Add QQ adapter registration in _create_adapter()
"""

import asyncio
import hashlib
import json
import logging
import time
import uuid
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional, Any, Tuple

import sys
import os

try:
    import aiohttp
    AIOHTTP_AVAILABLE = True
except ImportError:
    aiohttp = None
    AIOHTTP_AVAILABLE = False

from gateway.config import Platform, PlatformConfig
from gateway.platforms.base import (
    BasePlatformAdapter,
    MessageEvent,
    MessageType,
    SendResult,
    cache_image_from_bytes,
    cache_audio_from_bytes,
    cache_document_from_bytes,
    cache_image_from_url,
    get_image_cache_dir,
    get_audio_cache_dir,
)

logger = logging.getLogger(__name__)

QQ_API_BASE = "https://api.sgroup.qq.com"
QQ_TOKEN_URL = "https://bots.qq.com/app/getAppAccessToken"
QQ_WS_GATEWAY_URL = "https://api.sgroup.qq.com/gateway/bot"

QQ_INTENTS = {
    "GUILDS": 1 << 0,
    "GUILD_MEMBERS": 1 << 1,
    "PUBLIC_GUILD_MESSAGES": 1 << 30,
    "DIRECT_MESSAGE": 1 << 12,
    "GROUP_AT_MESSAGE_CREATE": 1 << 29,
}

# 计算FULL_INTENTS：直接使用整数值
FULL_INTENTS = (
    (1 << 25)  # PUBLIC_MESSAGES (包含C2C消息和群聊AT消息)
    | (1 << 30)  # PUBLIC_GUILD_MESSAGES
    | (1 << 12)  # DIRECT_MESSAGE
)

RECONNECT_DELAYS = [1000, 2000, 5000, 10000, 30000, 60000]
MAX_RECONNECT_ATTEMPTS = 100


def check_qq_requirements() -> bool:
    """Check if QQ platform dependencies are available."""
    return AIOHTTP_AVAILABLE


@dataclass
class QQBotConfig:
    """QQ Bot configuration, fully compatible with OpenClaw format."""
    app_id: str
    app_secret: str
    allowed_users: List[str] = field(default_factory=list)
    allowed_groups: List[str] = field(default_factory=list)
    at_required: bool = True
    auto_quote: bool = False
    markdown_support: bool = False


class RefIndexStore:
    """In-memory store for message reference index (ref_idx)."""
    
    def __init__(self):
        self._store: Dict[str, Dict[str, Any]] = {}
    
    def set(self, ref_idx: str, entry: Dict[str, Any]) -> None:
        self._store[ref_idx] = entry
    
    def get(self, ref_idx: str) -> Optional[Dict[str, Any]]:
        return self._store.get(ref_idx)
    
    def clear(self) -> None:
        self._store.clear()


class MessageQueue:
    """Per-user concurrent message queue."""
    
    def __init__(self, account_id: str, log: Optional[logging.Logger] = None):
        self.account_id = account_id
        self.log = log
        self._queues: Dict[str, List[Dict[str, Any]]] = {}
    
    def enqueue(self, msg: Dict[str, Any]) -> None:
        peer_id = msg.get("peer_id", msg.get("sender_id", "unknown"))
        if peer_id not in self._queues:
            self._queues[peer_id] = []
        self._queues[peer_id].append(msg)
    
    def get_next(self, peer_id: str) -> Optional[Dict[str, Any]]:
        if peer_id in self._queues and self._queues[peer_id]:
            return self._queues[peer_id].pop(0)
        return None
    
    def clear_user_queue(self, peer_id: str) -> int:
        count = len(self._queues.get(peer_id, []))
        if peer_id in self._queues:
            self._queues[peer_id] = []
        return count


class QQAdapter(BasePlatformAdapter):
    """
    Hermes QQ Official Bot Adapter.
    
    Fully implements BasePlatformAdapter interface, aligned with official platform specification.
    """
    
    MAX_MESSAGE_LENGTH = 500
    
    def __init__(self, config: PlatformConfig):
        super().__init__(config, Platform.QQ)
        
        if not check_qq_requirements():
            logger.warning("QQ: aiohttp not installed. QQ adapter will not be available.")
        
        extra = config.extra or {}
        
        token_parts = (self.config.token or "").split(":", 1)
        app_id = token_parts[0] if token_parts else ""
        app_secret = token_parts[1] if len(token_parts) > 1 else extra.get("app_secret", "")
        
        self.bot_config = QQBotConfig(
            app_id=app_id,
            app_secret=app_secret,
            allowed_users=self._parse_allowlist(extra.get("allowed_users", "")),
            allowed_groups=self._parse_allowlist(extra.get("allowed_groups", "")),
            at_required=extra.get("at_required", True),
            auto_quote=extra.get("auto_quote", False),
            markdown_support=extra.get("markdown_support", False),
        )
        
        self._session: Optional[aiohttp.ClientSession] = None
        self._ws: Optional[aiohttp.ClientWebSocketResponse] = None
        self._access_token: str = ""
        self._token_expire_at: float = 0
        self._bot_openid: str = ""
        self._connected: bool = False
        self._heartbeat_task: Optional[asyncio.Task] = None
        self._reconnect_task: Optional[asyncio.Task] = None
        self._running: bool = False
        self._reconnect_attempts: int = 0
        self._session_id: Optional[str] = None
        self._last_seq: Optional[int] = None
        self._ref_index: RefIndexStore = RefIndexStore()
        self._msg_queue: MessageQueue = MessageQueue(self.name, logger)
        self._image_cache_dir = get_image_cache_dir()
        self._audio_cache_dir = get_audio_cache_dir()
    
    def _mark_connected(self) -> None:
        """Override to set _connected flag checked by connect()."""
        self._connected = True
        super()._mark_connected()

    def _parse_allowlist(self, value: Any) -> List[str]:
        """Parse allowlist from string (comma-separated) or list."""
        if not value:
            return []
        if isinstance(value, list):
            return value
        if isinstance(value, str):
            return [v.strip() for v in value.split(",") if v.strip()]
        return []
    
    async def _refresh_access_token(self, force: bool = False) -> bool:
        """Refresh AccessToken with background refresh logic."""
        if not self.bot_config.app_id or not self.bot_config.app_secret:
            logger.error("QQ Bot app_id/app_secret not configured")
            return False
        
        if not force and self._access_token and time.time() < self._token_expire_at - 60:
            return True
        
        try:
            async with self._session.post(
                QQ_TOKEN_URL,
                json={
                    "appId": self.bot_config.app_id,
                    "clientSecret": self.bot_config.app_secret
                },
                timeout=aiohttp.ClientTimeout(total=10)
            ) as resp:
                resp.raise_for_status()
                data = await resp.json()
                
                if "access_token" not in data:
                    logger.error(f"Failed to get QQ AccessToken: {data}")
                    return False
                
                self._access_token = data["access_token"]
                expires_in = data.get("expires_in", 7200)
                # 确保 expires_in 是数字类型
                if isinstance(expires_in, str):
                    try:
                        expires_in = int(expires_in)
                    except ValueError:
                        expires_in = 7200
                self._token_expire_at = time.time() + expires_in
                logger.info("QQ AccessToken refreshed successfully")
                return True
                
        except Exception as e:
            logger.exception(f"Failed to refresh QQ AccessToken: {e}")
            return False
    
    @property
    def _auth_headers(self) -> Dict[str, str]:
        """Official required authentication headers."""
        return {
            "Authorization": f"QQBot {self._access_token}",
            "Content-Type": "application/json",
            "X-Union-Appid": self.bot_config.app_id
        }
    
    async def _get_gateway_url(self) -> str:
        """Get WebSocket gateway URL."""
        await self._refresh_access_token()
        
        async with self._session.get(
            QQ_WS_GATEWAY_URL,
            headers=self._auth_headers,
            timeout=aiohttp.ClientTimeout(total=10)
        ) as resp:
            resp.raise_for_status()
            data = await resp.json()
            return data["url"]
    
    async def _start_heartbeat(self, interval: int) -> None:
        """WebSocket heartbeat loop."""
        try:
            seq = self._last_seq or 0
            while self._running and self._ws and not self._ws.closed:
                await self._ws.send_json({"op": 1, "d": seq})
                logger.debug(f"QQ Heartbeat sent, seq={seq}")
                await asyncio.sleep(interval / 1000)
        except asyncio.CancelledError:
            pass
        except Exception as e:
            logger.warning(f"QQ Heartbeat loop error: {e}")
    
    async def _download_and_cache_attachment(
        self, 
        url: str, 
        content_type: str,
        msg_id: str
    ) -> Tuple[Optional[str], Optional[str]]:
        """Download attachment and cache locally."""
        if not url:
            return None, None
        
        try:
            ext = ".jpg"
            if "image" in content_type:
                if ".png" in url.lower():
                    ext = ".png"
                elif ".gif" in url.lower():
                    ext = ".gif"
                elif ".webp" in url.lower():
                    ext = ".webp"
            elif "audio" in content_type or "voice" in content_type:
                ext = ".ogg"
            elif "video" in content_type:
                ext = ".mp4"
            
            cache_dir = self._image_cache_dir if "image" in content_type else self._audio_cache_dir
            
            filename = f"qq_{msg_id[:8]}_{uuid.uuid4().hex[:8]}{ext}"
            filepath = cache_dir / filename
            
            async with self._session.get(url, timeout=aiohttp.ClientTimeout(total=30)) as resp:
                resp.raise_for_status()
                data = await resp.read()
                filepath.write_bytes(data)
                
            return str(filepath), content_type
                
        except Exception as e:
            logger.warning(f"Failed to download attachment: {e}")
            return None, None
    
    async def _parse_event(self, event: Dict[str, Any]) -> Optional[MessageEvent]:
        """Parse QQ event into Hermes MessageEvent."""
        op = event.get("op")
        t = event.get("t")
        d = event.get("d", {})
        s = event.get("s")
        
        # 🔍 调试：记录收到的原始事件结构
        logger.info(f"[DEBUG] QQ event received: op={op}, t={t!r}, keys={list(event.keys())}, d_keys={list(d.keys()) if d else None}")
        
        if s is not None:
            self._last_seq = int(s) if isinstance(s, (int, str)) else s
        
        # op=0 是 DISPATCH 事件（包含消息等业务事件）
        # op=1 是心跳响应，忽略
        if op == 1:
            # 心跳确认（我们发出的心跳被服务器确认），忽略
            return None
        
        # op=11 是服务器发送的心跳请求（PING），必须立即回复心跳响应（PONG含 seq）
        if op == 11:
            seq = event.get("s") or self._last_seq or 0
            try:
                await self._ws.send_json({"op": 1, "d": seq})
                logger.debug(f"QQ heartbeat response sent (seq={seq})")
            except Exception as e:
                logger.warning(f"Failed to send heartbeat response: {e}")
            return None
        
        if t == "READY":
            self._bot_openid = d.get("user", {}).get("id", "")
            logger.info(f"QQ Bot connected! ID: {self._bot_openid}")
            self._session_id = d.get("session_id")
            self._mark_connected()
            return None
        
        if t == "RESUMED":
            logger.info("QQ Session resumed successfully")
            self._mark_connected()
            return None
        
        if t == "C2C_MESSAGE_CREATE":
            logger.info(f"[C2C] Processing message from user_openid={d.get('author',{}).get('user_openid')}, content={d.get('content','')[:50]}")
            return await self._handle_c2c_message(d)
        
        if t == "GROUP_AT_MESSAGE_CREATE":
            logger.info(f"[GROUP] Processing message from group={d.get('group_openid')}, member={d.get('author',{}).get('member_openid')}")
            return await self._handle_group_message(d)
        
        # 其他事件类型记录 debug
        logger.debug(f"QQ received unhandled event type: {t}")
        return None

    async def _handle_c2c_message(self, d: Dict[str, Any]) -> Optional[MessageEvent]:
        """Handle C2C (private) message."""
        sender = d.get("author", {})
        user_openid = sender.get("user_openid", "")
        content = d.get("content", "").strip()
        message_id = d.get("id", "")

        # 解析timestamp，兼容ISO 8601格式（如"2026-03-22T08:55:03+08:00"）
        ts_raw = d.get("timestamp", time.time() * 1000)
        if isinstance(ts_raw, str):
            try:
                timestamp = int(ts_raw)
            except ValueError:
                try:
                    from datetime import datetime
                    dt = datetime.strptime(ts_raw.replace('+08:00', '+0000'), "%Y-%m-%dT%H:%M:%S%z")
                    timestamp = int(dt.timestamp() * 1000)
                except ValueError:
                    timestamp = int(time.time() * 1000)
        else:
            timestamp = int(ts_raw)

        if self.bot_config.allowed_users and user_openid not in self.bot_config.allowed_users:
            logger.warning(f"Rejecting non-whitelisted user C2C: {user_openid}")
            return None

        if not content:
            return None

        media_urls = []
        media_types = []

        for att in d.get("attachments", []):
            url = att.get("url", "")
            content_type = att.get("content_type", "image/jpeg")

            cached_path, _ = await self._download_and_cache_attachment(url, content_type, message_id)
            if cached_path:
                media_urls.append(cached_path)
                media_types.append(content_type)
            else:
                media_urls.append(url)
                media_types.append(content_type)

        logger.info(f"QQ C2C message from {user_openid}: {content[:50]}")

        source = self.build_source(
            chat_id=f"qq:c2c:{user_openid}",
            chat_name=user_openid,
            chat_type="dm",
            user_id=user_openid,
            user_name=sender.get("username", ""),
        )

        return MessageEvent(
            text=content,
            message_type=MessageType.TEXT,
            source=source,
            raw_message=d,
            message_id=message_id,
            media_urls=media_urls,
            media_types=media_types,
            timestamp=datetime.fromtimestamp(timestamp / 1000),
        )

    async def _handle_group_message(self, d: Dict[str, Any]) -> Optional[MessageEvent]:
        """Handle group @ message."""
        group_openid = d.get("group_openid", "")
        sender = d.get("author", {})
        user_openid = sender.get("member_openid", sender.get("user_openid", ""))
        content = d.get("content", "").strip()
        message_id = d.get("id", "")

        # 解析timestamp，兼容ISO 8601格式（如"2026-03-22T08:55:03+08:00"）
        ts_raw = d.get("timestamp", time.time() * 1000)
        if isinstance(ts_raw, str):
            try:
                timestamp = int(ts_raw)
            except ValueError:
                try:
                    from datetime import datetime
                    dt = datetime.strptime(ts_raw.replace('+08:00', '+0000'), "%Y-%m-%dT%H:%M:%S%z")
                    timestamp = int(dt.timestamp() * 1000)
                except ValueError:
                    timestamp = int(time.time() * 1000)
        else:
            timestamp = int(ts_raw)

        if self.bot_config.allowed_groups and group_openid not in self.bot_config.allowed_groups:
            return None

        if self.bot_config.at_required:
            at_mark = f"<@!{self._bot_openid}>"
            if at_mark not in content:
                return None
            content = content.replace(at_mark, "").strip()

        if not content:
            return None

        media_urls = []
        media_types = []

        for att in d.get("attachments", []):
            url = att.get("url", "")
            content_type = att.get("content_type", "image/jpeg")

            cached_path, _ = await self._download_and_cache_attachment(url, content_type, message_id)
            if cached_path:
                media_urls.append(cached_path)
                media_types.append(content_type)

        source = self.build_source(
            chat_id=f"qq:group:{group_openid}",
            chat_name=group_openid,
            chat_type="group",
            user_id=user_openid,
            user_name=sender.get("member_name", sender.get("username", "")),
        )

        return MessageEvent(
            text=content,
            message_type=MessageType.TEXT,
            source=source,
            raw_message=d,
            message_id=message_id,
            media_urls=media_urls,
            media_types=media_types,
            timestamp=datetime.fromtimestamp(timestamp / 1000),
        )

    async def _ws_receive_loop(self) -> None:
        """WebSocket event receiving loop."""
        while self._running:
            try:
                if not self._ws or self._ws.closed:
                    await self._connect_ws()
                    continue
                
                async for msg in self._ws:
                    if not self._running:
                        break
                    
                    if msg.type == aiohttp.WSMsgType.TEXT:
                        try:
                            event = json.loads(msg.data)
                            
                            # 调试：记录所有收到的原始事件（截断避免日志过大）
                            d_val = event.get('d', {})
                            d_keys = list(d_val.keys()) if isinstance(d_val, dict) else str(d_val)[:50]
                            logger.debug(f"QQ raw event: op={event.get('op')}, t={event.get('t')}, s={event.get('s')}, d_keys={d_keys}")
                            
                            if event.get("op") == 10:
                                interval = event.get("d", {}).get("heartbeat_interval", 45000)
                                logger.info(f"QQ WebSocket ready, heartbeat interval: {interval}ms")
                                if self._heartbeat_task:
                                    self._heartbeat_task.cancel()
                                self._heartbeat_task = asyncio.create_task(self._start_heartbeat(interval))
                            
                            msg_event = await self._parse_event(event)
                            if msg_event:
                                logger.info(f"QQ message parsed: type={msg_event.message_type}, user={msg_event.source.user_id}, content_len={len(msg_event.text)}")
                                await self.handle_message(msg_event)
                                logger.info(f"QQ message handled and dispatched to gateway")
                                
                        except json.JSONDecodeError:
                            logger.warning(f"Invalid JSON received: {msg.data[:100]}")
                        except Exception as e:
                            logger.exception(f"Error parsing event: {e}")
                            
                    elif msg.type in (aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.ERROR):
                        logger.warning(f"WebSocket closed or error: {msg.type}")
                        break
                        
            except asyncio.CancelledError:
                break
            except Exception as e:
                logger.exception(f"WebSocket receive loop error: {e}")
            
            if self._running:
                await self._schedule_reconnect()
    
    async def _connect_ws(self) -> None:
        """Establish WebSocket connection."""
        try:
            gateway_url = await self._get_gateway_url()
            
            headers = {
                "User-Agent": "QQBot/Hermes/1.0"
            }
            
            self._ws = await self._session.ws_connect(
                gateway_url,
                headers=headers,
                timeout=aiohttp.ClientTimeout(total=30),
                heartbeat=30,
            )
            
            # 强制IDENTIFY而不是RESUMED，确保正确声明intents
            # 清空旧的session信息，强制重新IDENTIFY
            self._session_id = None
            self._last_seq = None
            
            await self._ws.send_json({
                "op": 2,
                "d": {
                    "token": f"QQBot {self._access_token}",
                    "intents": FULL_INTENTS,
                    "shard": [0, 1],
                }
            })
            
            logger.info("QQ WebSocket connected (IDENTIFY sent)")
            
        except Exception as e:
            logger.exception(f"Failed to connect WebSocket: {e}")
            raise
    
    async def _schedule_reconnect(self) -> None:
        """Schedule reconnection with exponential backoff."""
        if not self._running:
            return
        
        self._reconnect_attempts += 1
        delay_idx = min(self._reconnect_attempts, len(RECONNECT_DELAYS) - 1)
        delay = RECONNECT_DELAYS[delay_idx]
        
        logger.info(f"QQ Reconnecting in {delay}ms (attempt {self._reconnect_attempts})")
        
        await asyncio.sleep(delay / 1000)
        
        if self._running:
            try:
                await self._connect_ws()
                self._reconnect_attempts = 0
            except Exception as e:
                logger.warning(f"Reconnect failed: {e}")
    
    async def connect(self) -> bool:
        """Connect to QQ Bot platform."""
        if not check_qq_requirements():
            logger.error("QQ: Missing required dependencies (aiohttp)")
            return False
        
        logger.info("Connecting to QQ Bot...")
        
        self._session = aiohttp.ClientSession()
        
        if not await self._refresh_access_token():
            logger.error("QQ: Failed to get initial access token")
            await self._session.close()
            return False
        
        self._running = True
        self._reconnect_task = asyncio.create_task(self._ws_receive_loop())
        
        for _ in range(30):
            if self._connected:
                logger.info("QQ Bot connected successfully")
                return True
            await asyncio.sleep(0.5)
        
        logger.error("QQ Bot connection timeout")
        return False
    
    async def disconnect(self) -> None:
        """Disconnect from QQ Bot platform."""
        logger.info("Disconnecting QQ Bot...")
        self._running = False
        
        if self._heartbeat_task:
            self._heartbeat_task.cancel()
            try:
                await self._heartbeat_task
            except asyncio.CancelledError:
                pass
        
        if self._reconnect_task:
            self._reconnect_task.cancel()
            try:
                await self._reconnect_task
            except asyncio.CancelledError:
                pass
        
        if self._ws:
            await self._ws.close()
        
        if self._session:
            await self._session.close()
        
        self._connected = False
        self._mark_disconnected()
        logger.info("QQ Bot disconnected")
    
    def _get_next_msg_seq(self) -> int:
        """生成 QQ API 要求的消息序列号 (msg_seq)"""
        # 使用时间戳微秒 + 随机数保证唯一性
        return int(time.time() * 1000000) % 2147483647

    async def send(
        self,
        chat_id: str,
        content: str,
        reply_to: Optional[str] = None,
        metadata: Optional[Dict[str, Any]] = None,
    ) -> SendResult:
        """Send a text message."""
        if not self._connected:
            return SendResult(success=False, error="Not connected")
        
        try:
            await self._refresh_access_token()
            
            parts = chat_id.split(":", 2)
            if len(parts) != 3:
                return SendResult(success=False, error=f"Invalid chat_id format: {chat_id}")
            
            _, chat_type, target_id = parts
            
            # 构建 payload，使用 v2 API 格式
            payload: Dict[str, Any] = {
                "content": content,
                "msg_seq": self._get_next_msg_seq()
            }
            
            if reply_to and self.bot_config.auto_quote:
                payload["msg_id"] = reply_to
            
            if chat_type == "c2c":
                # V2 API: /v2/users/{openid}/messages
                url = f"{QQ_API_BASE}/v2/users/{target_id}/messages"
            elif chat_type == "group":
                # V2 API: /v2/groups/{group_openid}/messages
                url = f"{QQ_API_BASE}/v2/groups/{target_id}/messages"
            else:
                return SendResult(success=False, error=f"Unsupported chat type: {chat_type}")
            
            async with self._session.post(
                url,
                headers=self._auth_headers,
                json=payload,
                timeout=aiohttp.ClientTimeout(total=10)
            ) as resp:
                if resp.status == 200:
                    data = await resp.json()
                    msg_id = data.get("msg_id", data.get("id", ""))
                    logger.info(f"QQ message sent to {chat_id}: {msg_id}")
                    return SendResult(success=True, message_id=msg_id)
                else:
                    error_text = await resp.text()
                    logger.error(f"QQ send failed ({resp.status}): {error_text}")
                    return SendResult(success=False, error=f"HTTP {resp.status}: {error_text}")
                    
        except Exception as e:
            logger.exception(f"QQ send error: {e}")
            return SendResult(success=False, error=str(e))
    
    async def send_image(
        self,
        chat_id: str,
        image_url: str,
        caption: Optional[str] = None,
        reply_to: Optional[str] = None,
        metadata: Optional[Dict[str, Any]] = None,
    ) -> SendResult:
        """Send an image message."""
        if not self._connected:
            return SendResult(success=False, error="Not connected")
        
        try:
            await self._refresh_access_token()
            
            parts = chat_id.split(":", 2)
            if len(parts) != 3:
                return SendResult(success=False, error=f"Invalid chat_id format: {chat_id}")
            
            _, chat_type, target_id = parts
            
            payload: Dict[str, Any] = {
                "image_url": image_url,
            }
            if caption:
                payload["content"] = caption
            
            if chat_type == "c2c":
                url = f"{QQ_API_BASE}/v2/users/{target_id}/messages"
                payload["msg_seq"] = self._get_next_msg_seq()
            elif chat_type == "group":
                url = f"{QQ_API_BASE}/v2/groups/{target_id}/messages"
                payload["msg_seq"] = self._get_next_msg_seq()
            else:
                return SendResult(success=False, error=f"Unsupported chat type: {chat_type}")
            
            async with self._session.post(
                url,
                headers=self._auth_headers,
                json=payload,
                timeout=aiohttp.ClientTimeout(total=30)
            ) as resp:
                if resp.status == 200:
                    data = await resp.json()
                    msg_id = data.get("msg_id", data.get("id", ""))
                    return SendResult(success=True, message_id=msg_id)
                else:
                    error_text = await resp.text()
                    return SendResult(success=False, error=f"HTTP {resp.status}: {error_text}")
                    
        except Exception as e:
            logger.exception(f"QQ send image error: {e}")
            return SendResult(success=False, error=str(e))
    
    async def send_typing(self, chat_id: str, metadata=None) -> None:
        """Send typing indicator (not directly supported by QQ API)."""
        pass
    
    async def get_chat_info(self, chat_id: str) -> Dict[str, Any]:
        """Get information about a chat."""
        parts = chat_id.split(":", 2)
        if len(parts) != 3:
            return {"name": chat_id, "type": "unknown", "chat_id": chat_id}
        
        _, chat_type, target_id = parts
        
        return {
            "name": target_id,
            "type": "group" if chat_type == "group" else "dm",
            "chat_id": chat_id,
        }
    
    def format_message(self, content: str) -> str:
        """Format message for QQ platform."""
        return content

    # ======================================================================
    # Media file sending via QQ chunked upload API
    # ======================================================================

    # QQ MediaFileType enum values
    _MEDIA_FILE_TYPE_IMAGE = 1
    _MEDIA_FILE_TYPE_VIDEO = 2
    _MEDIA_FILE_TYPE_VOICE = 3
    _MEDIA_FILE_TYPE_FILE = 4

    # Chunked upload constants (aligned with OpenClaw implementation)
    _DEFAULT_CONCURRENT_PARTS = 1
    _MAX_CONCURRENT_PARTS = 10
    _MAX_PART_FINISH_RETRY_TIMEOUT_S = 600  # 10 minutes
    _PART_UPLOAD_TIMEOUT_S = 300  # 5 minutes per part
    _PART_UPLOAD_MAX_RETRIES = 2
    _COMPLETE_UPLOAD_MAX_RETRIES = 2
    _COMPLETE_UPLOAD_BASE_DELAY_S = 2.0
    _PART_FINISH_MAX_RETRIES = 2
    _PART_FINISH_BASE_DELAY_S = 1.0
    _PART_FINISH_RETRYABLE_CODES = {40093001}
    _UPLOAD_PREPARE_FALLBACK_CODE = 40093002
    _PART_FINISH_RETRYABLE_DEFAULT_TIMEOUT_S = 120  # 2 minutes

    # File hash: first 10002432 bytes for md5_10m (aligned with QQ protocol)
    _MD5_10M_SIZE = 10002432

    async def _compute_file_hashes(self, file_path: str) -> Dict[str, str]:
        """
        Stream-compute MD5, SHA1, and MD5_10M (first 10002432 bytes) in a single pass.
        Returns dict with keys: md5, sha1, md5_10m
        """
        md5_h = hashlib.md5()
        sha1_h = hashlib.sha1()
        md5_10m_h = hashlib.md5()

        file_size = Path(file_path).stat().st_size
        need_10m = file_size > self._MD5_10M_SIZE
        bytes_read = 0

        with open(file_path, "rb") as f:
            while True:
                chunk = f.read(65536)
                if not chunk:
                    break
                md5_h.update(chunk)
                sha1_h.update(chunk)

                if need_10m:
                    remaining = self._MD5_10M_SIZE - bytes_read
                    if remaining > 0:
                        md5_10m_h.update(chunk if len(chunk) <= remaining else chunk[:remaining])

                bytes_read += len(chunk)

        md5 = md5_h.hexdigest()
        md5_10m = md5_10m_h.hexdigest() if need_10m else md5

        return {
            "md5": md5,
            "sha1": sha1_h.hexdigest(),
            "md5_10m": md5_10m,
        }

    def _classify_file_type(self, file_path: str) -> int:
        """Classify file type based on extension. Returns QQ MediaFileType value."""
        ext = Path(file_path).suffix.lower()
        if ext in (".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp"):
            return self._MEDIA_FILE_TYPE_IMAGE
        if ext in (".mp4", ".avi", ".mov", ".mkv", ".webm", ".flv"):
            return self._MEDIA_FILE_TYPE_VIDEO
        if ext in (".silk", ".ogg", ".mp3", ".wav", ".amr", ".aac", ".m4a"):
            return self._MEDIA_FILE_TYPE_VOICE
        return self._MEDIA_FILE_TYPE_FILE

    async def _put_to_presigned_url(self, url: str, data: bytes) -> None:
        """PUT chunk data to a COS presigned URL with retry and timeout."""
        last_error = None
        for attempt in range(self._PART_UPLOAD_MAX_RETRIES + 1):
            try:
                async with self._session.put(
                    url,
                    data=data,
                    headers={"Content-Length": str(len(data))},
                    timeout=aiohttp.ClientTimeout(total=self._PART_UPLOAD_TIMEOUT_S),
                ) as resp:
                    if resp.status == 200:
                        return
                    body = await resp.text()
                    raise Exception(f"COS PUT failed: {resp.status} {resp.status} - {body}")
            except asyncio.CancelledError:
                raise
            except Exception as e:
                last_error = e
                if attempt < self._PART_UPLOAD_MAX_RETRIES:
                    delay = 1.0 * (2 ** attempt)
                    logger.warning(f"[chunked-upload] Part upload attempt {attempt + 1} failed: {e}, retrying in {delay}s")
                    await asyncio.sleep(delay)

        raise last_error  # type: ignore

    async def _api_post(self, path: str, body: Dict[str, Any]) -> Dict[str, Any]:
        """POST to QQ API and return JSON response. Raises on non-200."""
        url = f"{QQ_API_BASE}{path}"
        async with self._session.post(
            url,
            headers=self._auth_headers,
            json=body,
            timeout=aiohttp.ClientTimeout(total=30),
        ) as resp:
            if resp.status == 200:
                return await resp.json()
            error_text = await resp.text()
            raise Exception(f"QQ API {resp.status} on {path}: {error_text}")

    async def _part_finish_with_retry(
        self,
        path: str,
        body: Dict[str, Any],
        retry_timeout_s: Optional[float] = None,
    ) -> None:
        """
        Call upload_part_finish with retry strategy:
        - Retryable biz code (40093001): persistent retry at 1s interval until timeout
        - Other errors: up to PART_FINISH_MAX_RETRIES with exponential backoff
        """
        max_attempts = self._PART_FINISH_MAX_RETRIES + 1
        for attempt in range(max_attempts):
            try:
                await self._api_post(path, body)
                return
            except Exception as e:
                err_msg = str(e)
                biz_code = self._extract_biz_code(err_msg)

                if biz_code in self._PART_FINISH_RETRYABLE_CODES:
                    timeout_s = retry_timeout_s if retry_timeout_s is not None else self._PART_FINISH_RETRYABLE_DEFAULT_TIMEOUT_S
                    timeout_s = min(timeout_s, self._MAX_PART_FINISH_RETRY_TIMEOUT_S)
                    logger.warning(f"[chunked-upload] PartFinish hit retryable bizCode={biz_code}, entering persistent retry (timeout={timeout_s}s)")
                    await self._part_finish_persistent_retry(path, body, timeout_s)
                    return

                if attempt < max_attempts - 1:
                    delay = self._PART_FINISH_BASE_DELAY_S * (2 ** attempt)
                    logger.warning(f"[chunked-upload] PartFinish attempt {attempt + 1} failed: {e}, retrying in {delay}s")

        raise Exception(f"PartFinish failed after {max_attempts} attempts")

    async def _part_finish_persistent_retry(
        self,
        path: str,
        body: Dict[str, Any],
        timeout_s: float,
    ) -> None:
        """Persistent retry at 1s interval until success or timeout."""
        deadline = time.monotonic() + timeout_s
        attempt = 0
        last_error = None

        while time.monotonic() < deadline:
            attempt += 1
            try:
                await self._api_post(path, body)
                logger.info(f"[chunked-upload] PartFinish persistent retry succeeded after {attempt} retries")
                return
            except Exception as e:
                last_error = e
                biz_code = self._extract_biz_code(str(e))
                if biz_code not in self._PART_FINISH_RETRYABLE_CODES:
                    raise  # Different error type, bail out

            await asyncio.sleep(1.0)

        raise Exception(f"PartFinish persistent retry timed out after {timeout_s}s: {last_error}")

    @staticmethod
    def _extract_biz_code(error_msg: str) -> Optional[int]:
        """Try to extract a QQ biz error code from error message string."""
        # QQ API returns {"code": NNNNN, "message": "..."} in error responses
        import re
        m = re.search(r'"code"\s*:\s*(\d+)', error_msg)
        if m:
            return int(m.group(1))
        return None

    async def _chunked_upload_media(
        self,
        chat_id: str,
        file_path: str,
        file_type: Optional[int] = None,
    ) -> Dict[str, Any]:
        """
        Upload a local file via QQ chunked upload API and return file_info.

        Flow:
        1. Compute file hashes (MD5, SHA1, MD5_10M)
        2. upload_prepare -> get upload_id + block_size + presigned URLs
        3. Upload all parts in parallel (with concurrency control)
        4. complete_upload -> get file_info

        Returns dict with: file_uuid, file_info, ttl
        """
        if file_type is None:
            file_type = self._classify_file_type(file_path)

        file_stat = Path(file_path).stat()
        file_size = file_stat.st_size
        file_name = Path(file_path).name

        parts = chat_id.split(":", 2)
        if len(parts) != 3:
            raise Exception(f"Invalid chat_id format: {chat_id}")
        _, chat_type, target_id = parts

        # Build API path prefix
        if chat_type == "c2c":
            api_prefix = f"/v2/users/{target_id}"
        elif chat_type == "group":
            api_prefix = f"/v2/groups/{target_id}"
        else:
            raise Exception(f"Unsupported chat type: {chat_type}")

        # 1. Compute hashes
        logger.info(f"[chunked-upload] Computing hashes for {file_name} ({file_size} bytes, type={file_type})")
        hashes = await asyncio.get_event_loop().run_in_executor(None, self._compute_file_hashes_sync, file_path)
        logger.info(f"[chunked-upload] Hashes: md5={hashes['md5'][:16]}..., sha1={hashes['sha1'][:16]}...")

        # 2. upload_prepare
        await self._refresh_access_token()
        prepare_body = {
            "file_type": file_type,
            "file_name": file_name,
            "file_size": file_size,
            "md5": hashes["md5"],
            "sha1": hashes["sha1"],
            "md5_10m": hashes["md5_10m"],
        }

        try:
            prepare_resp = await self._api_post(f"{api_prefix}/upload_prepare", prepare_body)
        except Exception as e:
            biz_code = self._extract_biz_code(str(e))
            if biz_code == self._UPLOAD_PREPARE_FALLBACK_CODE:
                raise Exception(f"Upload daily limit exceeded for {file_name} ({file_size} bytes): {e}")
            raise

        upload_id = prepare_resp["upload_id"]
        block_size = int(prepare_resp["block_size"])
        upload_parts = prepare_resp["parts"]  # list of {index, presigned_url}
        concurrency = min(
            int(prepare_resp.get("concurrency", self._DEFAULT_CONCURRENT_PARTS)),
            self._MAX_CONCURRENT_PARTS,
        )
        retry_timeout_s = None
        if "retry_timeout" in prepare_resp:
            retry_timeout_s = min(
                float(prepare_resp["retry_timeout"]),
                self._MAX_PART_FINISH_RETRY_TIMEOUT_S,
            )

        logger.info(
            f"[chunked-upload] Prepared: upload_id={upload_id}, block_size={block_size}, "
            f"parts={len(upload_parts)}, concurrency={concurrency}"
        )

        # 3. Upload parts with concurrency control (batch mode)
        semaphore = asyncio.Semaphore(concurrency)

        async def upload_one_part(part_info: Dict[str, Any]) -> None:
            async with semaphore:
                part_index = int(part_info["index"])  # 1-based
                presigned_url = part_info["presigned_url"]

                # Calculate offset and length for this part
                offset = (part_index - 1) * block_size
                length = min(block_size, file_size - offset)

                # Read chunk from file
                loop = asyncio.get_event_loop()
                chunk_data = await loop.run_in_executor(
                    None, self._read_file_chunk, file_path, offset, length
                )

                # Compute part MD5
                part_md5 = hashlib.md5(chunk_data).hexdigest()

                logger.info(
                    f"[chunked-upload] Part {part_index}/{len(upload_parts)}: "
                    f"uploading {length} bytes (offset={offset}, md5={part_md5[:16]}...)"
                )

                # PUT to COS presigned URL
                await self._put_to_presigned_url(presigned_url, chunk_data)

                # Notify platform that part is done (refresh token to avoid expiry)
                await self._refresh_access_token()
                await self._part_finish_with_retry(
                    f"{api_prefix}/upload_part_finish",
                    {
                        "upload_id": upload_id,
                        "part_index": part_index,
                        "block_size": length,
                        "md5": part_md5,
                    },
                    retry_timeout_s=retry_timeout_s,
                )

                logger.info(f"[chunked-upload] Part {part_index}/{len(upload_parts)}: done")

        # Execute all parts (semaphore controls concurrency)
        tasks = [upload_one_part(p) for p in upload_parts]
        await asyncio.gather(*tasks)

        logger.info(f"[chunked-upload] All {len(upload_parts)} parts uploaded, completing...")

        # 4. Complete upload
        await self._refresh_access_token()
        complete_resp = await self._complete_upload_with_retry(
            f"{api_prefix}/files",
            {"upload_id": upload_id},
        )

        logger.info(f"[chunked-upload] Upload complete: file_uuid={complete_resp.get('file_uuid')}")
        return complete_resp

    def _compute_file_hashes_sync(self, file_path: str) -> Dict[str, str]:
        """Synchronous version of hash computation (for thread pool executor)."""
        md5_h = hashlib.md5()
        sha1_h = hashlib.sha1()
        md5_10m_h = hashlib.md5()

        file_size = Path(file_path).stat().st_size
        need_10m = file_size > self._MD5_10M_SIZE
        bytes_read = 0

        with open(file_path, "rb") as f:
            while True:
                chunk = f.read(65536)
                if not chunk:
                    break
                md5_h.update(chunk)
                sha1_h.update(chunk)

                if need_10m:
                    remaining = self._MD5_10M_SIZE - bytes_read
                    if remaining > 0:
                        md5_10m_h.update(chunk if len(chunk) <= remaining else chunk[:remaining])

                bytes_read += len(chunk)

        md5 = md5_h.hexdigest()
        return {
            "md5": md5,
            "sha1": sha1_h.hexdigest(),
            "md5_10m": md5_10m_h.hexdigest() if need_10m else md5,
        }

    @staticmethod
    def _read_file_chunk(file_path: str, offset: int, length: int) -> bytes:
        """Read a specific byte range from a file."""
        with open(file_path, "rb") as f:
            f.seek(offset)
            return f.read(length)

    async def _complete_upload_with_retry(
        self,
        path: str,
        body: Dict[str, Any],
    ) -> Dict[str, Any]:
        """Call complete_upload with unconditional retry on any error."""
        last_error = None
        for attempt in range(self._COMPLETE_UPLOAD_MAX_RETRIES + 1):
            try:
                return await self._api_post(path, body)
            except Exception as e:
                last_error = e
                if attempt < self._COMPLETE_UPLOAD_MAX_RETRIES:
                    delay = self._COMPLETE_UPLOAD_BASE_DELAY_S * (2 ** attempt)
                    logger.warning(f"[chunked-upload] CompleteUpload attempt {attempt + 1} failed: {e}, retrying in {delay}s")
                    await asyncio.sleep(delay)

        raise last_error  # type: ignore

    async def _upload_and_send_media(
        self,
        chat_id: str,
        file_path: str,
        file_type: Optional[int] = None,
        caption: Optional[str] = None,
        reply_to: Optional[str] = None,
    ) -> SendResult:
        """
        Upload a local file and send it as a media message (msg_type=7).

        This is the unified entry point for all media types (image, video, voice, file).
        """
        if not self._connected:
            return SendResult(success=False, error="Not connected")

        if not Path(file_path).exists():
            return SendResult(success=False, error=f"File not found: {file_path}")

        try:
            # Upload file
            upload_result = await self._chunked_upload_media(chat_id, file_path, file_type)
            file_info = upload_result["file_info"]

            # Send media message with msg_type=7
            parts = chat_id.split(":", 2)
            _, chat_type, target_id = parts

            if chat_type == "c2c":
                url = f"{QQ_API_BASE}/v2/users/{target_id}/messages"
            elif chat_type == "group":
                url = f"{QQ_API_BASE}/v2/groups/{target_id}/messages"
            else:
                return SendResult(success=False, error=f"Unsupported chat type: {chat_type}")

            payload: Dict[str, Any] = {
                "msg_type": 7,
                "media": {"file_info": file_info},
                "msg_seq": self._get_next_msg_seq(),
            }
            if caption:
                payload["content"] = caption
            if reply_to and self.bot_config.auto_quote:
                payload["msg_id"] = reply_to

            await self._refresh_access_token()
            async with self._session.post(
                url,
                headers=self._auth_headers,
                json=payload,
                timeout=aiohttp.ClientTimeout(total=30),
            ) as resp:
                if resp.status == 200:
                    data = await resp.json()
                    msg_id = data.get("msg_id", data.get("id", ""))
                    return SendResult(success=True, message_id=msg_id)
                else:
                    error_text = await resp.text()
                    logger.error(f"QQ send media failed ({resp.status}): {error_text}")
                    return SendResult(success=False, error=f"HTTP {resp.status}: {error_text}")

        except Exception as e:
            logger.exception(f"QQ upload and send media error: {e}")
            return SendResult(success=False, error=str(e))

    # ======================================================================
    # Override BasePlatformAdapter media sending methods
    # ======================================================================

    async def send_image_file(
        self,
        chat_id: str,
        image_path: str,
        caption: Optional[str] = None,
        reply_to: Optional[str] = None,
        metadata: Optional[Dict[str, Any]] = None,
        **kwargs,
    ) -> SendResult:
        """Send a local image file via chunked upload."""
        return await self._upload_and_send_media(
            chat_id=chat_id,
            file_path=image_path,
            file_type=self._MEDIA_FILE_TYPE_IMAGE,
            caption=caption,
            reply_to=reply_to,
        )

    async def send_voice(
        self,
        chat_id: str,
        audio_path: str,
        caption: Optional[str] = None,
        reply_to: Optional[str] = None,
        metadata: Optional[Dict[str, Any]] = None,
        **kwargs,
    ) -> SendResult:
        """Send an audio file as voice message via chunked upload."""
        return await self._upload_and_send_media(
            chat_id=chat_id,
            file_path=audio_path,
            file_type=self._MEDIA_FILE_TYPE_VOICE,
            caption=caption,
            reply_to=reply_to,
        )

    async def send_video(
        self,
        chat_id: str,
        video_path: str,
        caption: Optional[str] = None,
        reply_to: Optional[str] = None,
        metadata: Optional[Dict[str, Any]] = None,
        **kwargs,
    ) -> SendResult:
        """Send a video file via chunked upload."""
        return await self._upload_and_send_media(
            chat_id=chat_id,
            file_path=video_path,
            file_type=self._MEDIA_FILE_TYPE_VIDEO,
            caption=caption,
            reply_to=reply_to,
        )

    async def send_document(
        self,
        chat_id: str,
        file_path: str,
        caption: Optional[str] = None,
        file_name: Optional[str] = None,
        reply_to: Optional[str] = None,
        metadata: Optional[Dict[str, Any]] = None,
        **kwargs,
    ) -> SendResult:
        """Send a document/file via chunked upload."""
        return await self._upload_and_send_media(
            chat_id=chat_id,
            file_path=file_path,
            file_type=self._MEDIA_FILE_TYPE_FILE,
            caption=caption,
            reply_to=reply_to,
        )

#!/usr/bin/env python3
from __future__ import annotations

import shutil
import sys
from pathlib import Path


def replace_once_or_skip(text: str, old: str, new: str, label: str) -> str:
    if new in text:
        return text
    if old not in text:
        raise RuntimeError(f"patch anchor not found for {label}")
    return text.replace(old, new, 1)


def insert_after_once(text: str, anchor: str, block: str, label: str) -> str:
    if block in text:
        return text
    if anchor not in text:
        raise RuntimeError(f"insert anchor not found for {label}")
    return text.replace(anchor, anchor + block, 1)


def patch_config_py(path: Path) -> None:
    text = path.read_text(encoding="utf-8")

    text = replace_once_or_skip(
        text,
        '    DINGTALK = "dingtalk"\n    API_SERVER = "api_server"\n',
        '    DINGTALK = "dingtalk"\n    QQ = "qq"\n    API_SERVER = "api_server"\n',
        "config platform enum",
    )

    text = replace_once_or_skip(
        text,
        '            # WeCom uses extra dict for bot credentials\n'
        '            elif platform == Platform.WECOM and config.extra.get("bot_id"):\n'
        '                connected.append(platform)\n'
        '            # BlueBubbles uses extra dict for local server config\n',
        '            # WeCom uses extra dict for bot credentials\n'
        '            elif platform == Platform.WECOM and config.extra.get("bot_id"):\n'
        '                connected.append(platform)\n'
        '            # QQ bot uses app_id/app_secret, stored in token as "app_id:app_secret"\n'
        '            elif platform == Platform.QQ and config.token:\n'
        '                connected.append(platform)\n'
        '            # BlueBubbles uses extra dict for local server config\n',
        "config connected platforms",
    )

    text = replace_once_or_skip(
        text,
        '        Platform.SLACK: "SLACK_BOT_TOKEN",\n'
        '        Platform.MATTERMOST: "MATTERMOST_TOKEN",\n'
        '        Platform.MATRIX: "MATRIX_ACCESS_TOKEN",\n',
        '        Platform.SLACK: "SLACK_BOT_TOKEN",\n'
        '        Platform.MATTERMOST: "MATTERMOST_TOKEN",\n'
        '        Platform.MATRIX: "MATRIX_ACCESS_TOKEN",\n'
        '        Platform.QQ: "QQ_APP_ID",\n',
        "config token env map",
    )

    text = replace_once_or_skip(
        text,
        '    # BlueBubbles (iMessage)\n',
        '    # QQ Bot\n'
        '    qq_app_id = os.getenv("QQ_APP_ID")\n'
        '    qq_app_secret = os.getenv("QQ_APP_SECRET")\n'
        '    if qq_app_id and qq_app_secret:\n'
        '        if Platform.QQ not in config.platforms:\n'
        '            config.platforms[Platform.QQ] = PlatformConfig()\n'
        '        config.platforms[Platform.QQ].enabled = True\n'
        '        config.platforms[Platform.QQ].token = f"{qq_app_id}:{qq_app_secret}"\n'
        '\n'
        '        qq_extra = config.platforms[Platform.QQ].extra\n'
        '        qq_allowed_users = os.getenv("QQ_ALLOWED_USERS", "")\n'
        '        qq_allowed_groups = os.getenv("QQ_ALLOWED_GROUPS", "")\n'
        '        if qq_allowed_users:\n'
        '            qq_extra["allowed_users"] = qq_allowed_users\n'
        '        if qq_allowed_groups:\n'
        '            qq_extra["allowed_groups"] = qq_allowed_groups\n'
        '        if os.getenv("QQ_AT_REQUIRED"):\n'
        '            qq_extra["at_required"] = os.getenv("QQ_AT_REQUIRED", "true").lower() in ("true", "1", "yes")\n'
        '        if os.getenv("QQ_AUTO_QUOTE"):\n'
        '            qq_extra["auto_quote"] = os.getenv("QQ_AUTO_QUOTE", "false").lower() in ("true", "1", "yes")\n'
        '        if os.getenv("QQ_MARKDOWN_SUPPORT"):\n'
        '            qq_extra["markdown_support"] = os.getenv("QQ_MARKDOWN_SUPPORT", "false").lower() in ("true", "1", "yes")\n'
        '\n'
        '        qq_home = os.getenv("QQ_HOME_CHANNEL")\n'
        '        if qq_home:\n'
        '            config.platforms[Platform.QQ].home_channel = HomeChannel(\n'
        '                platform=Platform.QQ,\n'
        '                chat_id=qq_home,\n'
        '                name=os.getenv("QQ_HOME_CHANNEL_NAME", "Home"),\n'
        '            )\n'
        '\n'
        '    # BlueBubbles (iMessage)\n',
        "config qq env overrides",
    )

    path.write_text(text, encoding="utf-8")


def patch_base_py(path: Path) -> None:
    text = path.read_text(encoding="utf-8")
    text = replace_once_or_skip(
        text,
        """            r'''[`"']?MEDIA:\\s*(?P<path>`[^`\\n]+`|"[^"\\n]+"|'[^'\\n]+'|(?:~/|/)\\S+(?:[^\\S\\n]+\\S+)*?\\.(?:png|jpe?g|gif|webp|mp4|mov|avi|mkv|webm|ogg|opus|mp3|wav|m4a)(?=[\\s`"',;:)\\]}]|$)|\\S+)[`"']?'''""",
        """            r'''[`"']?MEDIA:\\s*(?P<path>`[^`\\n]+`|"[^"\\n]+"|'[^'\\n]+'|(?:~/|/)\\S+(?:[^\\S\\n]+\\S+)*?\\.(?:png|jpe?g|gif|webp|mp4|mov|avi|mkv|webm|ogg|opus|mp3|wav|m4a|md|pdf|txt|docx?|xlsx?|pptx?|csv|json|zip)(?=[\\s`"',;:)\\]}]|$)|\\S+)[`"']?'''""",
        "base media regex",
    )
    text = replace_once_or_skip(
        text,
        "        _LOCAL_MEDIA_EXTS = (\n"
        "            '.png', '.jpg', '.jpeg', '.gif', '.webp',\n"
        "            '.mp4', '.mov', '.avi', '.mkv', '.webm',\n"
        "        )\n"
        "        ext_part = '|'.join(e.lstrip('.') for e in _LOCAL_MEDIA_EXTS)\n",
        "        _LOCAL_FILE_EXTS = (\n"
        "            '.png', '.jpg', '.jpeg', '.gif', '.webp',\n"
        "            '.mp4', '.mov', '.avi', '.mkv', '.webm',\n"
        "            '.md', '.pdf', '.txt', '.doc', '.docx', '.xlsx', '.pptx',\n"
        "            '.csv', '.json', '.zip',\n"
        "        )\n"
        "        ext_part = '|'.join(e.lstrip('.') for e in _LOCAL_FILE_EXTS)\n",
        "base local file extensions",
    )
    path.write_text(text, encoding="utf-8")


def patch_run_py(path: Path) -> None:
    text = path.read_text(encoding="utf-8")

    text = replace_once_or_skip(
        text,
        "        elif platform == Platform.API_SERVER:\n"
        "            from gateway.platforms.api_server import APIServerAdapter, check_api_server_requirements\n"
        "            if not check_api_server_requirements():\n"
        "                logger.warning(\"API Server: aiohttp not installed\")\n"
        "                return None\n"
        "            return APIServerAdapter(config)\n"
        "\n"
        "        elif platform == Platform.WEBHOOK:\n",
        "        elif platform == Platform.API_SERVER:\n"
        "            from gateway.platforms.api_server import APIServerAdapter, check_api_server_requirements\n"
        "            if not check_api_server_requirements():\n"
        "                logger.warning(\"API Server: aiohttp not installed\")\n"
        "                return None\n"
        "            return APIServerAdapter(config)\n"
        "\n"
        "        elif platform == Platform.QQ:\n"
        "            from gateway.platforms.qq import QQAdapter, check_qq_requirements\n"
        "            if not check_qq_requirements():\n"
        "                logger.warning(\"QQ: aiohttp not installed. Run: pip install aiohttp\")\n"
        "                return None\n"
        "            return QQAdapter(config)\n"
        "\n"
        "        elif platform == Platform.WEBHOOK:\n",
        "run create_adapter qq",
    )

    text = replace_once_or_skip(
        text,
        '            Platform.DINGTALK: "DINGTALK_ALLOWED_USERS",\n'
        '            Platform.FEISHU: "FEISHU_ALLOWED_USERS",\n'
        '            Platform.WECOM: "WECOM_ALLOWED_USERS",\n'
        '            Platform.BLUEBUBBLES: "BLUEBUBBLES_ALLOWED_USERS",\n',
        '            Platform.DINGTALK: "DINGTALK_ALLOWED_USERS",\n'
        '            Platform.FEISHU: "FEISHU_ALLOWED_USERS",\n'
        '            Platform.WECOM: "WECOM_ALLOWED_USERS",\n'
        '            Platform.QQ: "QQ_ALLOWED_USERS",\n'
        '            Platform.BLUEBUBBLES: "BLUEBUBBLES_ALLOWED_USERS",\n',
        "run allowlist map",
    )

    text = replace_once_or_skip(
        text,
        '            Platform.DINGTALK: "DINGTALK_ALLOW_ALL_USERS",\n'
        '            Platform.FEISHU: "FEISHU_ALLOW_ALL_USERS",\n'
        '            Platform.WECOM: "WECOM_ALLOW_ALL_USERS",\n'
        '            Platform.BLUEBUBBLES: "BLUEBUBBLES_ALLOW_ALL_USERS",\n',
        '            Platform.DINGTALK: "DINGTALK_ALLOW_ALL_USERS",\n'
        '            Platform.FEISHU: "FEISHU_ALLOW_ALL_USERS",\n'
        '            Platform.WECOM: "WECOM_ALLOW_ALL_USERS",\n'
        '            Platform.QQ: "QQ_ALLOW_ALL_USERS",\n'
        '            Platform.BLUEBUBBLES: "BLUEBUBBLES_ALLOW_ALL_USERS",\n',
        "run allow-all map",
    )

    old_deliver = """        from pathlib import Path\n\n        try:\n            media_files, _ = adapter.extract_media(response)\n            _, cleaned = adapter.extract_images(response)\n            local_files, _ = adapter.extract_local_files(cleaned)\n\n            _thread_meta = {\"thread_id\": event.source.thread_id} if event.source.thread_id else None\n\n            _AUDIO_EXTS = {'.ogg', '.opus', '.mp3', '.wav', '.m4a'}\n            _VIDEO_EXTS = {'.mp4', '.mov', '.avi', '.mkv', '.webm', '.3gp'}\n            _IMAGE_EXTS = {'.jpg', '.jpeg', '.png', '.webp', '.gif'}\n\n            for media_path, is_voice in media_files:\n                try:\n                    ext = Path(media_path).suffix.lower()\n                    if ext in _AUDIO_EXTS:\n                        await adapter.send_voice(\n                            chat_id=event.source.chat_id,\n                            audio_path=media_path,\n                            metadata=_thread_meta,\n                        )\n                    elif ext in _VIDEO_EXTS:\n                        await adapter.send_video(\n                            chat_id=event.source.chat_id,\n                            video_path=media_path,\n                            metadata=_thread_meta,\n                        )\n                    elif ext in _IMAGE_EXTS:\n                        await adapter.send_image_file(\n                            chat_id=event.source.chat_id,\n                            image_path=media_path,\n                            metadata=_thread_meta,\n                        )\n                    else:\n                        await adapter.send_document(\n                            chat_id=event.source.chat_id,\n                            file_path=media_path,\n                            metadata=_thread_meta,\n                        )\n                except Exception as e:\n                    logger.warning(\"[%s] Post-stream media delivery failed: %s\", adapter.name, e)\n\n            for file_path in local_files:\n                try:\n                    ext = Path(file_path).suffix.lower()\n                    if ext in _IMAGE_EXTS:\n                        await adapter.send_image_file(\n                            chat_id=event.source.chat_id,\n                            image_path=file_path,\n                            metadata=_thread_meta,\n                        )\n                    else:\n                        await adapter.send_document(\n                            chat_id=event.source.chat_id,\n                            file_path=file_path,\n                            metadata=_thread_meta,\n                        )\n                except Exception as e:\n                    logger.warning(\"[%s] Post-stream file delivery failed: %s\", adapter.name, e)\n\n        except Exception as e:\n            logger.warning(\"Post-stream media extraction failed: %s\", e)\n"""

    new_deliver = """        from pathlib import Path\n\n        async def _send_and_check(send_coro, *, kind: str) -> None:\n            result = await send_coro\n            if isinstance(result, SendResult) and not result.success:\n                raise RuntimeError(result.error or f\"unknown {kind} delivery failure\")\n\n        try:\n            media_files, cleaned = adapter.extract_media(response)\n            local_files, _ = adapter.extract_local_files(cleaned)\n\n            _thread_meta = {\"thread_id\": event.source.thread_id} if event.source.thread_id else None\n\n            _AUDIO_EXTS = {'.ogg', '.opus', '.mp3', '.wav', '.m4a'}\n            _VIDEO_EXTS = {'.mp4', '.mov', '.avi', '.mkv', '.webm', '.3gp'}\n            _IMAGE_EXTS = {'.jpg', '.jpeg', '.png', '.webp', '.gif'}\n\n            for media_path, _is_voice in media_files:\n                try:\n                    ext = Path(media_path).suffix.lower()\n                    if ext in _AUDIO_EXTS:\n                        await _send_and_check(\n                            adapter.send_voice(\n                                chat_id=event.source.chat_id,\n                                audio_path=media_path,\n                                metadata=_thread_meta,\n                            ),\n                            kind=\"media\",\n                        )\n                    elif ext in _VIDEO_EXTS:\n                        await _send_and_check(\n                            adapter.send_video(\n                                chat_id=event.source.chat_id,\n                                video_path=media_path,\n                                metadata=_thread_meta,\n                            ),\n                            kind=\"media\",\n                        )\n                    elif ext in _IMAGE_EXTS:\n                        await _send_and_check(\n                            adapter.send_image_file(\n                                chat_id=event.source.chat_id,\n                                image_path=media_path,\n                                metadata=_thread_meta,\n                            ),\n                            kind=\"media\",\n                        )\n                    else:\n                        await _send_and_check(\n                            adapter.send_document(\n                                chat_id=event.source.chat_id,\n                                file_path=media_path,\n                                metadata=_thread_meta,\n                            ),\n                            kind=\"media\",\n                        )\n                except Exception as e:\n                    logger.warning(\"[%s] Post-stream media delivery failed for %s: %s\", adapter.name, media_path, e)\n                    try:\n                        await adapter.send(\n                            chat_id=event.source.chat_id,\n                            content=f\"⚠️ 文件发送失败：{Path(media_path).name}\\n原因：{e}\",\n                            metadata=_thread_meta,\n                        )\n                    except Exception:\n                        logger.debug(\"[%s] Failed to send post-stream media failure notice\", adapter.name, exc_info=True)\n\n            for file_path in local_files:\n                try:\n                    ext = Path(file_path).suffix.lower()\n                    if ext in _IMAGE_EXTS:\n                        await _send_and_check(\n                            adapter.send_image_file(\n                                chat_id=event.source.chat_id,\n                                image_path=file_path,\n                                metadata=_thread_meta,\n                            ),\n                            kind=\"file\",\n                        )\n                    else:\n                        await _send_and_check(\n                            adapter.send_document(\n                                chat_id=event.source.chat_id,\n                                file_path=file_path,\n                                metadata=_thread_meta,\n                            ),\n                            kind=\"file\",\n                        )\n                except Exception as e:\n                    logger.warning(\"[%s] Post-stream file delivery failed for %s: %s\", adapter.name, file_path, e)\n                    try:\n                        await adapter.send(\n                            chat_id=event.source.chat_id,\n                            content=f\"⚠️ 文件发送失败：{Path(file_path).name}\\n原因：{e}\",\n                            metadata=_thread_meta,\n                        )\n                    except Exception:\n                        logger.debug(\"[%s] Failed to send post-stream file failure notice\", adapter.name, exc_info=True)\n\n        except Exception as e:\n            logger.warning(\"Post-stream media extraction failed: %s\", e)\n"""
    text = replace_once_or_skip(text, old_deliver, new_deliver, "run post-stream delivery")

    old_bg = """                # Send media files\n                for media_path in (media_files or []):\n                    try:\n                        await adapter.send_document(\n                            chat_id=source.chat_id,\n                            file_path=media_path,\n                        )\n                    except Exception:\n                        pass\n"""
    new_bg = """                # Send media files\n                for media_path in (media_files or []):\n                    try:\n                        result = await adapter.send_document(\n                            chat_id=source.chat_id,\n                            file_path=media_path,\n                        )\n                        if isinstance(result, SendResult) and not result.success:\n                            raise RuntimeError(result.error or \"unknown background media delivery failure\")\n                    except Exception as exc:\n                        logger.warning(\"[%s] Background media delivery failed for %s: %s\", adapter.name, media_path, exc)\n                        try:\n                            await adapter.send(\n                                chat_id=source.chat_id,\n                                content=f\"⚠️ 后台任务文件发送失败：{Path(media_path).name}\\n原因：{exc}\",\n                                metadata=_thread_metadata,\n                            )\n                        except Exception:\n                            logger.debug(\"[%s] Failed to send background media failure notice\", adapter.name, exc_info=True)\n"""
    text = replace_once_or_skip(text, old_bg, new_bg, "run background media delivery")

    path.write_text(text, encoding="utf-8")


def patch_tools_config_py(path: Path) -> None:
    text = path.read_text(encoding="utf-8")
    text = replace_once_or_skip(
        text,
        '    "mattermost": {"label": "💬 Mattermost", "default_toolset": "hermes-mattermost"},\n'
        '    "webhook": {"label": "🔗 Webhook", "default_toolset": "hermes-webhook"},\n'
        '}\n',
        '    "mattermost": {"label": "💬 Mattermost", "default_toolset": "hermes-mattermost"},\n'
        '    "webhook": {"label": "🔗 Webhook", "default_toolset": "hermes-webhook"},\n'
        '    "qq":       {"label": "🐧 QQ",         "default_toolset": "hermes-qq"},\n'
        '}\n',
        "tools config qq platform",
    )
    path.write_text(text, encoding="utf-8")


def patch_toolsets_py(path: Path) -> None:
    text = path.read_text(encoding="utf-8")
    text = replace_once_or_skip(
        text,
        '    "hermes-webhook": {\n'
        '        "description": "Webhook toolset - receive and process external webhook events",\n'
        '        "tools": _HERMES_CORE_TOOLS,\n'
        '        "includes": []\n'
        '    },\n'
        '\n'
        '    "hermes-gateway": {\n',
        '    "hermes-webhook": {\n'
        '        "description": "Webhook toolset - receive and process external webhook events",\n'
        '        "tools": _HERMES_CORE_TOOLS,\n'
        '        "includes": []\n'
        '    },\n'
        '\n'
        '    "hermes-qq": {\n'
        '        "description": "QQ bot toolset - Tencent QQ messaging platform (full access)",\n'
        '        "tools": _HERMES_CORE_TOOLS,\n'
        '        "includes": []\n'
        '    },\n'
        '\n'
        '    "hermes-gateway": {\n',
        "toolsets hermes-qq block",
    )
    text = replace_once_or_skip(
        text,
        '"includes": ["hermes-telegram", "hermes-discord", "hermes-whatsapp", "hermes-slack", "hermes-signal", "hermes-bluebubbles", "hermes-homeassistant", "hermes-email", "hermes-sms", "hermes-mattermost", "hermes-matrix", "hermes-dingtalk", "hermes-feishu", "hermes-wecom", "hermes-webhook"]',
        '"includes": ["hermes-telegram", "hermes-discord", "hermes-whatsapp", "hermes-slack", "hermes-signal", "hermes-bluebubbles", "hermes-homeassistant", "hermes-email", "hermes-sms", "hermes-mattermost", "hermes-matrix", "hermes-dingtalk", "hermes-feishu", "hermes-wecom", "hermes-webhook", "hermes-qq"]',
        "toolsets gateway includes",
    )
    path.write_text(text, encoding="utf-8")


def patch_tests(root: Path) -> None:
    extract_path = root / "tests/gateway/test_extract_local_files.py"
    if extract_path.exists():
        text = extract_path.read_text(encoding="utf-8")
        text = replace_once_or_skip(
            text,
            '    def test_no_media_extensions(self):\n'
            '        """Non-media extensions should not be matched."""\n'
            '        paths, _ = _extract("See /tmp/data.csv and /tmp/script.py and /tmp/notes.txt")\n'
            '        assert paths == []\n',
            '    def test_no_media_extensions(self):\n'
            '        """Unsupported extensions should not be matched, supported docs should."""\n'
            '        paths, _ = _extract("See /tmp/data.csv and /tmp/script.py and /tmp/notes.txt")\n'
            '        assert "/tmp/script.py" not in paths\n'
            '        assert "/tmp/data.csv" in paths\n'
            '        assert "/tmp/notes.txt" in paths\n',
            "test extract local files edge case",
        )
        extract_path.write_text(text, encoding="utf-8")

    platform_base_path = root / "tests/gateway/test_platform_base.py"
    if platform_base_path.exists():
        text = platform_base_path.read_text(encoding="utf-8")
        marker = (
            '    def test_media_tag_supports_quoted_paths_with_spaces(self):\n'
            '        content = "Here\\nMEDIA: \'/tmp/my image.png\'\\nAfter"\n'
            '        media, cleaned = BasePlatformAdapter.extract_media(content)\n'
            '        assert media == [("/tmp/my image.png", False)]\n'
            '        assert "Here" in cleaned\n'
            '        assert "After" in cleaned\n'
        )
        block = (
            '\n'
            '    def test_media_tag_extracts_document_paths(self):\n'
            '        content = "Report ready\\nMEDIA:\'/tmp/quarterly report.pdf\'"\n'
            '        media, cleaned = BasePlatformAdapter.extract_media(content)\n'
            '        assert media == [("/tmp/quarterly report.pdf", False)]\n'
            '        assert "Report ready" in cleaned\n'
            '        assert "MEDIA:" not in cleaned\n'
        )
        text = insert_after_once(text, marker, block, "test platform base document media")
        platform_base_path.write_text(text, encoding="utf-8")


def copy_asset(skill_dir: Path, repo_root: Path) -> None:
    src = skill_dir / "assets/qq.py"
    dst = repo_root / "gateway/platforms/qq.py"
    dst.parent.mkdir(parents=True, exist_ok=True)
    shutil.copyfile(src, dst)


def validate_repo(repo_root: Path) -> None:
    required = [
        repo_root / "gateway/config.py",
        repo_root / "gateway/platforms/base.py",
        repo_root / "gateway/run.py",
        repo_root / "hermes_cli/tools_config.py",
        repo_root / "toolsets.py",
    ]
    missing = [str(p) for p in required if not p.exists()]
    if missing:
        raise RuntimeError("not a hermes-agent repo or files missing: " + ", ".join(missing))


def main() -> int:
    skill_dir = Path(__file__).resolve().parents[1]
    repo_root = Path(sys.argv[1]).resolve() if len(sys.argv) > 1 else Path.cwd().resolve()
    validate_repo(repo_root)

    copy_asset(skill_dir, repo_root)
    patch_config_py(repo_root / "gateway/config.py")
    patch_base_py(repo_root / "gateway/platforms/base.py")
    patch_run_py(repo_root / "gateway/run.py")
    patch_tools_config_py(repo_root / "hermes_cli/tools_config.py")
    patch_toolsets_py(repo_root / "toolsets.py")
    patch_tests(repo_root)

    print(f"hermes-qq installed into {repo_root}")
    print("Next steps:")
    print("1. source venv/bin/activate")
    print("2. python -m py_compile gateway/config.py gateway/platforms/base.py gateway/platforms/qq.py gateway/run.py hermes_cli/tools_config.py toolsets.py")
    print("3. python -m pytest tests/gateway/test_platform_base.py tests/gateway/test_extract_local_files.py tests/gateway/test_send_image_file.py tests/cron/test_scheduler.py tests/gateway/test_background_command.py tests/gateway/test_internal_event_bypass_pairing.py -q -n 0")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())