dsProject/dsRag/Util/MySQLUtil.py

"""
pip install aiomysql
"""
import logging
from typing import Optional, Dict, List

from aiomysql import create_pool
from Config.Config import *
# 配置日志
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

# MySQL 配置
MYSQL_CONFIG = {
    "host": MYSQL_HOST,
    "port": MYSQL_PORT,
    "user": MYSQL_USER,
    "password": MYSQL_PASSWORD,
    "db": MYSQL_DB_NAME,
    "minsize": 1,
    "maxsize": 20,
}


# 初始化 MySQL 连接池
async def init_mysql_pool():
    return await create_pool(**MYSQL_CONFIG)


# 保存聊天记录到 MySQL
async def save_chat_to_mysql(mysql_pool, person_id, prompt, result, audio_url, duration, input_type=1, output_type=1,
                             input_image_type=0, image_width=0, image_height=0):
    async with mysql_pool.acquire() as conn:
        await conn.ping()  # 重置连接
        async with conn.cursor() as cur:
            await cur.execute(
                "INSERT INTO t_chat_log (person_id, user_input, model_response,audio_url,duration,input_type,output_type,input_image_type,image_width,image_height,create_time) VALUES (%s, %s, %s, %s, %s, %s, %s,%s,%s,%s,NOW())",
                (person_id, prompt, result, audio_url, duration, input_type, output_type, input_image_type, image_width,
                 image_height)
            )
            await conn.commit()


# 清空表
async def truncate_chat_log(mysql_pool):
    async with mysql_pool.acquire() as conn:
        await conn.ping()  # 重置连接
        async with conn.cursor() as cur:
            await cur.execute("TRUNCATE TABLE t_chat_log")
            await conn.commit()
            logger.info("表 t_chat_log 已清空。")


from aiomysql import DictCursor


# 分页查询聊天记录
async def get_chat_log_by_session(mysql_pool, person_id, page=1, page_size=10):
    """
    根据 person_id 查询聊天记录，并按 id 降序分页
    :param mysql_pool: MySQL 连接池
    :param person_id: 用户会话 ID
    :param page: 当前页码（默认值为 1，但会动态计算为最后一页）
    :param page_size: 每页记录数
    :return: 分页数据
    """
    if not mysql_pool:
        raise ValueError("MySQL 连接池未初始化")

    async with mysql_pool.acquire() as conn:
        await conn.ping()  # 重置连接
        async with conn.cursor(DictCursor) as cur:  # 使用 DictCursor
            # 查询总记录数
            await cur.execute(
                "SELECT COUNT(*) FROM t_chat_log WHERE person_id = %s",
                (person_id,)
            )
            total = (await cur.fetchone())['COUNT(*)']

            # 计算总页数
            total_pages = (total + page_size - 1) // page_size

            # 计算偏移量
            offset = (page - 1) * page_size

            # 查询分页数据，按 id 降序排列
            await cur.execute(
                "SELECT id, person_id, user_input, model_response, audio_url, duration,input_type,output_type,input_image_type,image_width,image_height, create_time "
                "FROM t_chat_log WHERE person_id = %s ORDER BY id DESC LIMIT %s OFFSET %s",
                (person_id, page_size, offset)
            )
            records = await cur.fetchall()

            # 将查询结果反转，确保最新消息显示在最后
            if page==1 and records:
                records.reverse()

            # 将查询结果转换为字典列表
            result = [
                {
                    "id": record['id'],
                    "person_id": record['person_id'],
                    "user_input": record['user_input'],
                    "model_response": record['model_response'],
                    "audio_url": record['audio_url'],
                    "duration": record['duration'],
                    "input_type": record['input_type'],
                    "output_type": record['output_type'],
                    "image_width": record['image_width'],
                    "image_height": record['image_height'],
                    "input_image_type": record['input_image_type'],
                    "create_time": record['create_time'].strftime("%Y-%m-%d %H:%M:%S")
                }
                for record in records
            ]

            return {
                "data": result,  # 按 id 升序排列的数据
                "total": total,
                "page": page,
                "page_size": page_size,
                "total_pages": total_pages
            }


# 获取指定会话的最后一条记录的 id
async def get_last_chat_log_id(mysql_pool, person_id):
    """
    获取指定会话的最后一条记录的 id
    :param mysql_pool: MySQL 连接池
    :param session_id: 用户会话 ID
    :return: 最后一条记录的 id，如果未找到则返回 None
    """
    async with mysql_pool.acquire() as conn:
        await conn.ping()  # 重置连接
        async with conn.cursor() as cur:
            await cur.execute(
                "SELECT id FROM t_chat_log WHERE person_id = %s ORDER BY id DESC LIMIT 1",
                (person_id,)
            )
            result = await cur.fetchone()
            return result[0] if result else None


# 更新为危险的记录
async def update_risk(mysql_pool, person_id, risk_memo):
    async with mysql_pool.acquire() as conn:
        await conn.ping()  # 重置连接
        async with conn.cursor() as cur:
            # 1. 获取此人员的最后一条记录 id
            last_id = await get_last_chat_log_id(mysql_pool, person_id)

            if last_id:
                # 2. 更新 risk_flag 和 risk_memo
                await cur.execute(
                    "UPDATE t_chat_log SET risk_flag = 1, risk_memo = %s WHERE id = %s",
                    (risk_memo.replace('\n', '').replace("NO", ""), last_id)
                )
                await conn.commit()
                logger.info(f"已更新 person_id={person_id} 的最后一条记录 (id={last_id}) 的 risk_flag 和 risk_memo。")
            else:
                logger.warning(f"未找到 person_id={person_id} 的记录。")


# 查询用户信息
async def get_user_by_login_name(mysql_pool, login_name: str) -> Optional[Dict]:
    """
    根据用户名查询用户信息
    :param pool: MySQL 连接池
    :param login_name: 用户名
    :return: 用户信息（字典形式）
    """
    async with mysql_pool.acquire() as conn:
        await conn.ping()  # 重置连接
        async with conn.cursor() as cursor:
            sql = "SELECT * FROM t_base_person WHERE login_name = %s"
            await cursor.execute(sql, (login_name,))
            row = await cursor.fetchone()
            if not row:
                return None

            # 将元组转换为字典
            columns = [column[0] for column in cursor.description]
            return dict(zip(columns, row))


# 显示统计分析页面
async def get_chat_logs_summary(mysql_pool, risk_flag: int, offset: int, page_size: int) -> (List[Dict], int):
    """
    获取聊天记录的统计分析结果
    :param mysql_pool: MySQL 连接池
    :param risk_flag: 风险标志
    :param offset: 偏移量
    :param page_size: 每页记录数
    :return: 日志列表和总记录数
    """
    async with mysql_pool.acquire() as conn:
        await conn.ping()  # 重置连接
        async with conn.cursor() as cursor:
            # 查询符合条件的记录
            sql = """
                SELECT tbp.*, COUNT(*) AS cnt 
                FROM t_chat_log AS tcl 
                INNER JOIN t_base_person AS tbp ON tcl.person_id = tbp.person_id
                WHERE tcl.risk_flag = %s 
                GROUP BY tcl.person_id 
                ORDER BY COUNT(*) DESC
                LIMIT %s OFFSET %s
            """
            await cursor.execute(sql, (risk_flag, page_size, offset))
            rows = await cursor.fetchall()

            # 获取列名
            columns = [column[0] for column in cursor.description]

            # 查询总记录数
            count_sql = """
                SELECT COUNT(DISTINCT tcl.person_id) 
                FROM t_chat_log AS tcl 
                INNER JOIN t_base_person AS tbp ON tcl.person_id = tbp.person_id 
                WHERE tcl.risk_flag = %s
            """
            await cursor.execute(count_sql, (risk_flag,))
            total = (await cursor.fetchone())[0]

            # 将元组转换为字典
            logs = [dict(zip(columns, row)) for row in rows] if rows else []

            return logs, total


async def get_chat_logs_by_risk_flag(mysql_pool, risk_flag: int, person_id: str, offset: int, page_size: int) -> (
        List[Dict], int):
    """
    根据风险标志查询聊天记录
    :param mysql_pool: MySQL 连接池
    :param risk_flag: 风险标志
    :param offset: 分页偏移量
    :param page_size: 每页记录数
    :return: 聊天记录列表和总记录数
    """
    async with mysql_pool.acquire() as conn:
        await conn.ping()  # 重置连接
        async with conn.cursor() as cursor:
            # 查询符合条件的记录
            sql = """
                SELECT  tcl.id, tcl.user_input, tcl.model_response, tcl.audio_url, tcl.duration, 
                    tcl.create_time, tcl.risk_flag, tcl.risk_memo, tcl.risk_result, 
                    tbp.person_id, tbp.login_name, tbp.person_name 
                FROM t_chat_log AS tcl 
                INNER JOIN t_base_person AS tbp ON tcl.person_id = tbp.person_id 
                WHERE tcl.risk_flag = %s and tcl.person_id=%s ORDER BY TCL.ID DESC 
                LIMIT %s OFFSET %s
            """
            await cursor.execute(sql, (risk_flag, person_id, page_size, offset))
            rows = await cursor.fetchall()

            # 在 count_sql 执行前获取列名
            columns = [column[0] for column in cursor.description]

            # 查询总记录数
            count_sql = """
                SELECT COUNT(*) 
                FROM t_chat_log AS tcl 
                INNER JOIN t_base_person AS tbp ON tcl.person_id = tbp.person_id 
                WHERE tcl.risk_flag = %s and tcl.person_id=%s
            """
            await cursor.execute(count_sql, (risk_flag, person_id))
            total = (await cursor.fetchone())[0]

            # 将元组转换为字典，并格式化 create_time
            if rows:
                logs = []
                for row in rows:
                    log = dict(zip(columns, row))
                    # 格式化 create_time
                    if log["create_time"]:
                        log["create_time"] = log["create_time"].strftime("%Y-%m-%d %H:%M:%S")
                    logs.append(log)
                return logs, total
            return [], 0