QingLong/AI/Text2Sql/Model/biModel.py

from Text2Sql.Util.PostgreSQLUtil import PostgreSQLUtil, postgresql_pool


# 删除数据
def delete_question(question_id: str):
    with PostgreSQLUtil(postgresql_pool.getconn()) as db:
        # 删除 t_bi_question 表中的数据
        delete_sql = """
                    DELETE FROM t_bi_question WHERE id = %s
                    """
        db.execute_query(delete_sql, (question_id,))


# 插入数据
def insert_question(question_id: str, question: str):
    # 向 t_bi_question 表插入数据
    with PostgreSQLUtil(postgresql_pool.getconn()) as db:
        insert_sql = """
                   INSERT INTO t_bi_question (id,question, state_id, is_system, is_collect)
                   VALUES (%s,%s, %s, %s, %s)
               """
        db.execute_query(insert_sql, (question_id, question, 0, 0, 0))


# 修改数据
'''
示例:
# 更新 question 和 state_id 字段
update_question_by_id(db, question_id=1, question="新的问题描述", state_id=1)

# 只更新 excel_file_name 字段
update_question_by_id(db, question_id=1, excel_file_name="new_excel.xlsx")

# 只更新 is_collect 字段
update_question_by_id(db, question_id=1, is_collect=1)

# 不更新任何字段（因为所有参数都是 None）
update_question_by_id(db, question_id=1, question=None, state_id=None)
'''


def update_question_by_id(question_id: str, **kwargs):
    """
    根据主键更新 t_bi_question 表，只更新非 None 的字段
    :param db: PostgreSQLUtil 实例
    :param question_id: 主键 id
    :param kwargs: 需要更新的字段和值
    :return: 更新是否成功
    """
    # 过滤掉值为 None 的字段
    update_fields = {k: v for k, v in kwargs.items() if v is not None}
    if not update_fields:
        return False  # 没有需要更新的字段

    # 动态构建 SET 子句
    set_clause = ", ".join([f"{field} = %s" for field in update_fields.keys()])

    with PostgreSQLUtil(postgresql_pool.getconn()) as db:
        # 构建完整 SQL
        sql = f"""
            UPDATE t_bi_question
            SET {set_clause}
            WHERE id = %s
        """
        # 参数列表
        params = list(update_fields.values()) + [question_id]

        # 执行更新
        try:
            db.execute_query(sql, params)
            return True
        except Exception as e:
            print(f"更新失败: {e}")
            return False


# 根据 问题id 查询 sql
def get_question_by_id(question_id: str):
    with PostgreSQLUtil(postgresql_pool.getconn()) as db:
        select_sql = """
                            select * from t_bi_question where id=%s
                            """
        _data = db.execute_query(select_sql, (question_id,))
        return _data


# 保存系统推荐
def set_system_recommend_questions(question_id: str, flag: str):
    with PostgreSQLUtil(postgresql_pool.getconn()) as db:
        sql = f"""
                UPDATE t_bi_question
                SET is_system =%s  WHERE id = %s
            """
        # 执行更新
        try:
            db.execute_query(sql, int(flag), question_id)
            return True
        except Exception as e:
            print(f"更新失败: {e}")
            return False

# 设置用户收藏
def set_user_collect_questions(question_id: str, flag: str):
    sql = f"""
            UPDATE t_bi_question
            SET is_collect =%s  WHERE id = %s
        """
    with PostgreSQLUtil(postgresql_pool.getconn()) as db:
        # 执行更新
        try:
            db.execute_query(sql, int(flag), question_id)
            return True
        except Exception as e:
            print(f"更新失败: {e}")
            return False

# 查询有哪些系统推荐问题
def get_system_recommend_questions():
    sql = """
    SELECT * FROM t_bi_question WHERE is_system = 1
    """
    with PostgreSQLUtil(postgresql_pool.getconn()) as db:
        _data = db.execute_query(sql)
        return _data

# 查询有哪些用户收藏问题
def get_user_collect_questions():
    # 从t_bi_question表中获取所有is_collect=1的数据
    sql="""
    SELECT * FROM t_bi_question WHERE is_collect = 1
    """
    with PostgreSQLUtil(postgresql_pool.getconn()) as db:
        _data = db.execute_query(sql)
        return _data