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

import threading
from pymongo import MongoClient
from bson import ObjectId
from datetime import datetime
from dotenv import load_dotenv
import os

load_dotenv()

MONGO_URI = os.getenv("ARK_LOGS_MONGO_URI")

client = MongoClient(MONGO_URI, serverSelectionTimeoutMS=5000)
# 数据库
db = client["arklogs"]
# 豆包大模型的对话日志
chat_logs = db["chat_logs"]
# 聊天历史记录
chat_history_col = db["chat_history"]
# 兴趣圈集合
circle_prompts = db["circle_prompt"]
# 历史人物集合
historical_figures = db["historical_figures"]

def _ensure_index():
    try:
        chat_logs.create_index([("user_id", 1), ("asked_at", -1)])
        chat_history_col.create_index([("user_id", 1), ("session_id", 1), ("timestamp", -1)])
    except Exception:
        pass


def _do_save_chat_log(doc: dict):
    try:
        _ensure_index()
        chat_logs.insert_one(doc)
    except Exception as e:
        print(f"MongoDB 日志写入失败: {e}")


def save_chat_log(
    user_id: str,
    question: str,
    stream_mode: bool,
    raw_response: str = None,
    status: str = "success",
    error: str = None,
):
    """
    运维功能:保存聊天原始响应日志到 MongoDB（异步写入，不阻塞调用方）【日志审计】

    Args:
        user_id: 提问人
        question: 提问的问题
        stream_mode: 回答方式（流式或非流式）
        raw_response: API 原始响应的 repr 字符串
        status: 响应状态 success | error
        error: 异常时的错误信息
    """
    doc = {
        "user_id": user_id,
        "question": question,
        "stream_mode": stream_mode,
        "raw_response": raw_response,
        "status": status,
        "error": error,
        "asked_at": datetime.now(),
    }
    threading.Thread(target=_do_save_chat_log, args=(doc,), daemon=True).start()


def save_chat_history(
    user_id: str,
    session_id: str,
    role: str,
    content: str,
    timestamp: datetime,
    response_id: str = None,
    thinking: str = None,
    searching: str = None,
    attachments: list = None,
    input_tokens: int = None,
    output_tokens: int = None,
    total_tokens: int = None,
    app_name:str=None,
):
    """
    业务功能: 保存用户跟chat机器人的聊天记录，用于会话存储
    """

    try:
        _ensure_index()
        doc = {
            "user_id": user_id,
            "session_id": session_id,
            "role": role,
            "content": content,
            "thinking": thinking,
            "searching": searching,
            "response_id": response_id,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "total_tokens": total_tokens,
            "timestamp": timestamp,
            "app_name": app_name,
        }
        if attachments:
            doc["attachments"] = attachments
        chat_history_col.insert_one(doc)
    except Exception as e:
        print(f"MongoDB 聊天历史写入失败: {e}")


def get_chat_history(user_id: str, session_id: str) -> list:
    try:
        docs = chat_history_col.find(
            {"user_id": user_id, "session_id": session_id},
            {"_id": 0}
        ).sort("timestamp", 1)
        return list(docs)
    except Exception as e:
        print(f"MongoDB 聊天历史读取失败: {e}")
        return []


def get_last_response_id(user_id: str, session_id: str) -> str | None:
    try:
        doc = chat_history_col.find_one(
            {"user_id": user_id, "session_id": session_id, "role": "assistant", "response_id": {"$ne": None}},
            {"response_id": 1, "_id": 0},
            sort=[("timestamp", -1)]
        )
        return doc["response_id"] if doc else None
    except Exception as e:
        print(f"MongoDB 查询 response_id 失败: {e}")
        return None


def delete_chat_history(user_id: str, session_id: str) -> int:
    try:
        result = chat_history_col.delete_many({"user_id": user_id, "session_id": session_id})
        return result.deleted_count
    except Exception as e:
        print(f"MongoDB 聊天历史删除失败: {e}")
        return 0


def get_sessions(user_id: str) -> list:
    try:
        pipeline = [
            {"$match": {"user_id": user_id, "role": "user"}},
            {"$sort": {"timestamp": 1}},
            {"$group": {
                "_id": "$session_id",
                "createdAt": {"$first": "$timestamp"},
                "preview": {"$first": "$content"},
            }},
            {"$sort": {"createdAt": -1}},
            {"$project": {
                "_id": 0,
                "sessionId": "$_id",
                "createdAt": 1,
                "preview": {"$substrCP": ["$preview", 0, 20]},
            }},
        ]
        return list(chat_history_col.aggregate(pipeline))
    except Exception as e:
        print(f"MongoDB 会话列表查询失败: {e}")
        return []


_DEFAULT_PROMPT_CONFIG = {
    "name": "兴趣圈",
    "role": "活跃用户",
    "style": "自然亲切，有活人感",
    "keywords": [],
    "forbidden": [],
}


def get_circle_prompt(app_name: str) -> dict:
    try:
        doc = circle_prompts.find_one({"appName": app_name})
        return doc if doc else _DEFAULT_PROMPT_CONFIG
    except Exception:
        return _DEFAULT_PROMPT_CONFIG


def upsert_circle_prompt(data: dict) -> None:
    circle_prompts.update_one(
        {"appName": data["appName"]},
        {"$set": data},
        upsert=True,
    )


# ===================== 历史人物 =====================

def get_all_figures() -> list:
    try:
        docs = historical_figures.find({})
        return [{"_id": str(doc["_id"]), **{k: v for k, v in doc.items() if k != "_id"}} for doc in docs]
    except Exception as e:
        print(f"MongoDB 历史人物列表查询失败: {e}")
        return []


def get_figure_by_id(figure_id: str) -> dict | None:
    try:
        doc = historical_figures.find_one({"_id": ObjectId(figure_id)})
        if doc:
            doc["_id"] = str(doc["_id"])
        return doc
    except Exception as e:
        print(f"MongoDB 历史人物查询失败: {e}")
        return None


def insert_figure(data: dict) -> str:
    try:
        result = historical_figures.insert_one(data)
        return str(result.inserted_id)
    except Exception as e:
        print(f"MongoDB 历史人物新增失败: {e}")
        return None


def update_figure(figure_id: str, data: dict) -> int:
    try:
        result = historical_figures.update_one(
            {"_id": ObjectId(figure_id)},
            {"$set": data},
        )
        return result.matched_count
    except Exception as e:
        print(f"MongoDB 历史人物修改失败: {e}")
        return 0


def delete_figure(figure_id: str) -> int:
    try:
        result = historical_figures.delete_one({"_id": ObjectId(figure_id)})
        return result.deleted_count
    except Exception as e:
        print(f"MongoDB 历史人物删除失败: {e}")
        return 0


# ===================== 群聊AI回复历史 =====================

# 群聊AI回复记录集合（只存AI回复，不存群消息）
group_chat_history_col = db["group_chat_history"]


def _ensure_group_chat_index():
    try:
        group_chat_history_col.create_index([("session_id", 1), ("timestamp", -1)])
        group_chat_history_col.create_index([("group_id", 1), ("ai_role", 1), ("timestamp", -1)])
    except Exception:
        pass


def save_group_ai_reply(
    session_id: str,
    group_id: str,
    ai_role: str,
    content: str,
    response_id: str,
    trigger_message: str,
    trigger_user_id: str,
    timestamp: datetime,
    app_name: str = "souyue_im",
):
    """
    业务功能：保存AI在群聊中的回复

    注意：只存储AI回复，不存储群消息（群消息通过腾讯云API获取）

    Args:
        session_id: 会话ID（群ID + 小时窗口）
        group_id: 群ID
        ai_role: AI角色（entrepreneur | femininity）
        content: AI回复内容
        response_id: 大模型response ID
        trigger_message: 触发AI的原始消息
        trigger_user_id: 触发用户ID
        timestamp: 时间戳
        app_name: 应用名称
    """
    try:
        _ensure_group_chat_index()
        doc = {
            "session_id": session_id,
            "group_id": group_id,
            "role": "assistant",
            "user_id": f"ai_{ai_role}",
            "content": content,
            "ai_role": ai_role,
            "response_id": response_id,
            "trigger_message": trigger_message,
            "trigger_user_id": trigger_user_id,
            "timestamp": timestamp,
            "app_name": app_name,
        }
        group_chat_history_col.insert_one(doc)
    except Exception as e:
        print(f"MongoDB 群聊AI回复写入失败: {e}")


def get_last_group_response_id(session_id: str, ai_role: str) -> str | None:
    """
    获取指定AI的最后一次回复的response_id

    Args:
        session_id: 会话ID
        ai_role: AI角色（必须指定，防止串线）

    Returns:
        response_id 或 None
    """
    try:
        doc = group_chat_history_col.find_one(
            {
                "session_id": session_id,
                "ai_role": ai_role,
                "response_id": {"$ne": None}
            },
            {"response_id": 1, "_id": 0},
            sort=[("timestamp", -1)]
        )
        return doc["response_id"] if doc else None
    except Exception as e:
        print(f"MongoDB 查询群聊 response_id 失败: {e}")
        return None


def get_group_ai_reply_count(session_id: str, ai_role: str = None) -> int:
    """
    获取会话中的AI回复次数

    Args:
        session_id: 会话ID
        ai_role: AI角色（可选，不传则统计所有AI）

    Returns:
        回复次数
    """
    try:
        query = {"session_id": session_id}
        if ai_role:
            query["ai_role"] = ai_role
        return group_chat_history_col.count_documents(query)
    except Exception as e:
        print(f"MongoDB 查询AI回复次数失败: {e}")
        return 0