aigc-test/mysql_connect.py

import mysql.connector
from mysql.connector import Error
import json

MYSQL_CONFIG = {
    'host': 'rm-bp19uc56sud25ag4o.mysql.rds.aliyuncs.com',
    'database': 'longvideo',
    'port': 3306,
    'user': 'wx2016_longvideo',
    'password': 'wx2016_longvideoP@assword1234',
}


def insert_content(gpt_res):
    """ 连接MySQL数据库并插入一行数据 """
    try:
        # 连接MySQL数据库
        conn = mysql.connector.connect(
            host=MYSQL_CONFIG['host'],
            database=MYSQL_CONFIG['database'],
            user=MYSQL_CONFIG['user'],
            password=MYSQL_CONFIG['password'],
        )

        if conn.is_connected():
            print('成功连接到数据库')
            cursor = conn.cursor()

            # 插入数据的SQL语句
            insert_query = """
            INSERT INTO video_content(video_id,key_words,search_keys,extra_keys,category_list,tone,target_audience,target_age,target_gender,address,theme)
            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
            """
            # 准备要插入的数据，转换字典列表为元组列表
            data_to_insert = [
                (
                    gpt_res.get('video_id', ''),
                    json.dumps(gpt_res.get('key_words', ''),
                               sort_keys=True, ensure_ascii=False),
                    json.dumps(gpt_res.get('search_keys', ''),
                               sort_keys=True, ensure_ascii=False),
                    json.dumps(gpt_res.get('extra_keys', ''),
                               sort_keys=True, ensure_ascii=False),
                    json.dumps(gpt_res.get('category', ''),
                               sort_keys=True, ensure_ascii=False),
                    str(gpt_res.get('tone', '')),
                    str(gpt_res.get('target_audience', '')),
                    str(gpt_res.get('target_age', '')),
                    str(gpt_res.get('target_gender', '')),
                    str(gpt_res.get('address', '')),
                    str(gpt_res.get('theme', ''))
                )
            ]

            # 执行批量插入操作
            cursor.executemany(insert_query, data_to_insert)
            print('数据插入成功')

            # 提交事务
            conn.commit()
    except Error as e:
        print('数据库连接或操作出错:', e)
    finally:
        if conn.is_connected():
            cursor.close()
            conn.close()
            print('数据库连接已关闭')


# 执行函数
# connect_insert()