aigc-test/mysql_connect.py

import mysql.connector
from mysql.connector import Error

MYSQL_CONFIG = {
    'host': 'rm-bp1k5853td1r25g3n690.mysql.rds.aliyuncs.com',
    'database': 'incentive',
    '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,tone,target_audience,target_age,target_gender,address,theme)
            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
            """
            # 准备要插入的数据，转换字典列表为元组列表
            data_to_insert = [(gpt_res['video_id'], gpt_res['key_words'], gpt_res['search_keys'], gpt_res['extra_keys'],
                               gpt_res['tone'], gpt_res['target_audience'], gpt_res['target_age'], gpt_res['target_gender'],
                               gpt_res['address'], gpt_res['theme'])]

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

            # 检查插入结果
            # cursor.execute('SELECT * FROM employees')
            # records = cursor.fetchall()
            # print('插入的数据:', records)

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


# 执行函数
# connect_insert()