chat-ai-api/app/db/tencent_im.py

"""
腾讯云IM API客户端

用于获取群聊消息历史，不在MongoDB中冗余存储群消息。

API文档：https://cloud.tencent.com/document/product/269/2738
"""

import httpx
import random
import time
from typing import List, Optional
from dotenv import load_dotenv
import os
from TLSSigAPIv2 import TLSSigAPIv2

load_dotenv()

# 腾讯云IM配置
TENCENT_IM_APP_ID = os.getenv("TENCENT_IM_APP_ID")
TENCENT_IM_ADMIN_ID = os.getenv("TENCENT_IM_ADMIN_ID")
TENCENT_IM_SECRET_KEY = os.getenv("TENCENT_IM_SECRET_KEY")


class TencentIMClient:
    """腾讯云IM API客户端"""

    def __init__(self):
        self.app_id =  TENCENT_IM_APP_ID
        self.admin_id =  TENCENT_IM_ADMIN_ID
        self.secret_key =  TENCENT_IM_SECRET_KEY

        if not all([self.app_id, self.admin_id, self.secret_key]):
            print("警告：腾讯云IM配置不完整，请检查环境变量")

        self.base_url = "https://console.tim.qq.com/v4"

        # 初始化 TLSSigAPIv2 实例
        if self.app_id and self.secret_key:
            self.sig_api = TLSSigAPIv2(int(self.app_id), self.secret_key)
        else:
            self.sig_api = None

    def _generate_usersig(self, expire: int = 86400 * 180) -> str:
        """
        生成UserSig（使用腾讯云官方SDK）

        Args:
            expire: 有效期（秒），默认180天

        Returns:
            UserSig字符串
        """
        if not self.sig_api:
            raise ValueError("腾讯云IM配置不完整，无法生成UserSig")

        return self.sig_api.gen_sig(self.admin_id, expire)

    async def get_group_messages(
        self,
        group_id: str,
        req_msg_number: int = 20,
        req_msg_seq: int = None
    ) -> List[dict]:
        """
        获取群消息历史

        API: group_open_http_svc/group_msg_get_simple

        Args:
            group_id: 群ID
            req_msg_number: 拉取消息数量（1-20）
            req_msg_seq: 翻页标识（上一页返回的最小MsgSeq）

        Returns:
            消息列表，每条消息格式：
            {
                "From_Account": "user123",
                "MsgSeq": 12345,
                "MsgRandom": 12345,
                "MsgTimeStamp": 1609459200,
                "MsgBody": [
                    {
                        "MsgType": "TIMTextElem",
                        "MsgContent": {"Text": "消息内容"}
                    }
                ]
            }
        """
        if not self.app_id or not self.secret_key:
            print("腾讯云IM配置不完整，返回空消息列表")
            return []

        try:
            # 生成UserSig
            user_sig = self._generate_usersig()

            # 构造请求URL
            random_num = random.randint(0, 4294967295)
            url = (
                f"{self.base_url}/group_open_http_svc/group_msg_get_simple"
                f"?sdkappid={self.app_id}"
                f"&identifier={self.admin_id}"
                f"&usersig={user_sig}"
                f"&random={random_num}"
                f"&contenttype=json"
            )

            # 构造请求体
            payload = {
                "GroupId": group_id,
                "ReqMsgNumber": min(max(1, req_msg_number), 20),  # 限制1-20
            }
            if req_msg_seq is not None:
                payload["ReqMsgSeq"] = req_msg_seq

            # 发送请求
            async with httpx.AsyncClient(timeout=10.0) as client:
                response = await client.post(url, json=payload)
                response.raise_for_status()
                result = response.json()

            # 检查返回状态
            if result.get("ActionStatus") == "OK":
                return result.get("RspMsgList", [])
            else:
                error_info = result.get("ErrorInfo", "未知错误")
                print(f"获取群消息失败: {error_info}")
                return []

        except Exception as e:
            print(f"调用腾讯云IM API失败: {e}")
            return []

    def parse_message_content(self, msg: dict) -> dict:
        """
        解析消息体，提取内容（支持文本、图片、视频、分享）

        过滤规则：
        - IsSystemMsg == 1: 系统消息 → 返回None
        - IsPlaceMsg != 0: 占位消息 → 返回None
        - From_Account == "@TIM#SYSTEM": 系统账号 → 返回None

        Args:
            msg: 消息对象（从get_group_messages返回）

        Returns:
            {
                "type": "text" | "image" | "video" | "share" | "mixed",
                "text": "文本内容",
                "images": ["url1", "url2"],  # 图片URL列表
                "videos": [{"url": "...", "thumb": "..."}],  # 视频信息
                "share_title": "分享标题"
            }
            如果是系统消息，返回None
        """
        # 1. 过滤系统消息
        if msg.get("IsSystemMsg") == 1:
            return None

        # 2. 过滤占位消息
        if msg.get("IsPlaceMsg", 0) != 0:
            return None

        # 3. 过滤系统账号
        if msg.get("From_Account") == "@TIM#SYSTEM":
            return None

        # 4. 解析消息内容
        result = {
            "type": "text",
            "text": "",
            "images": [],
            "videos": [],
            "share_title": ""
        }

        msg_body = msg.get("MsgBody", [])
        if not msg_body:
            return None

        for body in msg_body:
            msg_type = body.get("MsgType")
            content = body.get("MsgContent", {})

            if msg_type == "TIMTextElem":
                # 文本消息
                text = content.get("Text", "")
                if text:
                    result["text"] += text

            elif msg_type == "TIMImageElem":
                # 图片消息
                image_info = content.get("ImageInfoArray", [])
                if image_info:
                    # 取第一个（原图）
                    result["images"].append(image_info[0].get("URL", ""))
                    result["type"] = "image" if not result["text"] else "mixed"

            elif msg_type == "TIMVideoFileElem":
                # 视频消息
                video_url = content.get("VideoUrl", "")
                thumb_url = content.get("ThumbUrl", "")
                if video_url:
                    result["videos"].append({
                        "url": video_url,
                        "thumb": thumb_url
                    })
                    result["type"] = "video" if not result["text"] else "mixed"

            elif msg_type == "TIMCustomElem":
                # 自定义消息（链接分享等）
                data_str = content.get("Data", "")
                try:
                    data = json.loads(data_str)
                    title = data.get("title", "")
                    if title:
                        result["share_title"] = title
                        result["type"] = "share" if not result["text"] else "mixed"
                except:
                    pass

        # 如果没有任何有效内容，返回None
        if not result["text"] and not result["images"] and not result["videos"] and not result["share_title"]:
            return None

        return result

    def parse_message_text(self, msg: dict) -> str:
        """
        解析消息体，提取文本内容（兼容旧接口）

        Args:
            msg: 消息对象（从get_group_messages返回）

        Returns:
            文本内容（多个文本元素会拼接）
        """
        parsed = self.parse_message_content(msg)
        if not parsed:
            return ""
        return parsed.get("text", "")

    async def get_group_context(
        self,
        group_id: str,
        max_messages: int = 20,
        ai_account_map: dict = None
    ) -> List[dict]:
        """
        获取群聊上下文（最近N条消息）

        Args:
            group_id: 群ID
            max_messages: 最多获取多少条消息
            ai_account_map: AI账号映射 {"6537328": "AI企业家", "6587745": "AI知心姐姐"}

        Returns:
            消息列表，格式化后的结构：
            [
                {
                    "user_id": "6485208",  # 或 "[AI企业家]"
                    "content": "消息文本",
                    "images": ["url1"],
                    "videos": [{"url": "...", "thumb": "..."}],
                    "share_title": "分享标题",
                    "timestamp": 1790827391,
                    "msg_seq": 482
                },
                ...
            ]
        """
        if ai_account_map is None:
            ai_account_map = {
                "6537328": "AI企业家",
                "6587745": "AI知心姐姐"
            }

        raw_messages = await self.get_group_messages(group_id, max_messages)

        # 格式化消息（过滤掉系统消息）
        formatted = []
        for msg in raw_messages:
            parsed = self.parse_message_content(msg)
            if parsed:  # 只保留有有效内容的消息
                user_id = msg.get("From_Account", "")

                # 如果是AI账号，标注为AI角色
                display_id = ai_account_map.get(user_id, user_id)
                if display_id != user_id:
                    display_id = f"[{display_id}]"

                formatted.append({
                    "user_id": display_id,
                    "content": parsed.get("text", ""),
                    "images": parsed.get("images", []),
                    "videos": parsed.get("videos", []),
                    "share_title": parsed.get("share_title", ""),
                    "timestamp": msg.get("MsgTimeStamp", 0),
                    "msg_seq": msg.get("MsgSeq", 0),
                })

        # 过滤最近30分钟的消息
        current_time = int(time.time())
        time_threshold = current_time - 7200  # 30分钟 = 1800秒
        formatted = [msg for msg in formatted if msg["timestamp"] >= time_threshold]

        # 按时间升序排列（最早的在前）
        formatted.sort(key=lambda x: x["timestamp"])


        return formatted


# 全局实例
im_client = TencentIMClient()