FlaskProject/db_utils/files.py

# -*- coding: utf-8 -*-
from unicodedata import category

import mysql.connector
from mysql.connector import Error

# 配置数据库连接
DB_CONFIG = {
    'host': '60.204.139.57',
    'user': 'root',
    'password': '1',
    'port': 6003,
    'database': 'files'  # 替换为数据库名称
}

# 初始化数据库
def initialize_files_database():
    try:
        # 连接 MySQL 服务器（不指定数据库）
        connection = mysql.connector.connect(
            host=DB_CONFIG['host'],
            user=DB_CONFIG['user'],
            password=DB_CONFIG['password'],
            port=DB_CONFIG['port']
        )

        if connection.is_connected():
            print("成功连接到 MySQL 服务器")
            cursor = connection.cursor()

            # 检查数据库是否存在
            cursor.execute(f"SHOW DATABASES LIKE '{DB_CONFIG['database']}';")
            result = cursor.fetchone()

            if result:
                print(f"数据库 {DB_CONFIG['database']} 已存在。")
            else:
                print(f"数据库 {DB_CONFIG['database']} 不存在，正在创建...")
                cursor.execute(f"CREATE DATABASE {DB_CONFIG['database']};")
                print(f"数据库 {DB_CONFIG['database']} 创建成功。")

    except Error as err:
        print(f"连接 MySQL 时发生错误：{err}")
    finally:
        if connection.is_connected():
            cursor.close()
            connection.close()
            print("MySQL 服务器连接已关闭")

# 初始化表
def initialize_files_table(table_name = "category"):
    try:
        connection = mysql.connector.connect(**DB_CONFIG)
        if connection.is_connected():
            cursor = connection.cursor()

            # 动态创建表
            create_table_query = f'''
                CREATE TABLE IF NOT EXISTS `{table_name}` (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    name VARCHAR(255),
                    url VARCHAR(255)
                )
            '''

            cursor.execute(create_table_query)
            print(f"表 {table_name} 初始化成功。")
    except Error as err:
        print(f"初始化表时发生错误：{err}")
    finally:
        if connection.is_connected():
            cursor.close()
            connection.close()
            print("MySQL 连接已关闭")

# 插入数据
def insert_url(name, url, table_name="category"):
    try:
        connection = mysql.connector.connect(**DB_CONFIG)
        if connection.is_connected():
            cursor = connection.cursor()

            # 确保目标文件表存在
            initialize_files_table(table_name)

            # 插入文件数据到目标表
            insert_query = f"INSERT INTO `{table_name}` (name, url) VALUES (%s, %s)"
            cursor.execute(insert_query, (name, url))
            connection.commit()
            print(f"插入文件数据成功：({name}, {url}) 到表 {table_name}")

            # 确保 category 表存在
            initialize_files_table("category")

            # 插入表名到 category 表
            insert_query = "INSERT IGNORE INTO `category` (name) VALUES (%s)"
            cursor.execute(insert_query, (table_name,))
            connection.commit()
            print(f"插入表名成功：({table_name}) 到表 category")

    except Error as err:
        print(f"插入数据时发生错误：{err}")
    finally:
        if connection.is_connected():
            cursor.close()
            connection.close()
            print("MySQL 连接已关闭")


# 查询数据
def fetch_data(table_name):
    try:
        connection = mysql.connector.connect(**DB_CONFIG)
        if connection.is_connected():
            cursor = connection.cursor()

            # 查询数据
            query = f"SELECT * FROM `{table_name}`"
            cursor.execute(query)
            rows = cursor.fetchall()
            print(f"--- 表 {table_name} 的查询结果 ---")
            for row in rows:
                print(row)
            return rows
    except Error as err:
        print(f"查询数据时发生错误：{err}")
    finally:
        if connection.is_connected():
            cursor.close()
            connection.close()
            print("MySQL 连接已关闭")

# 主函数
if __name__ == "__main__":
    # 初始化数据库
    initialize_files_database()

    # 示例：插入和查询数据
    table_name = "filePath"
    insert_url("example_file.mp4", table_name)
    fetch_data(table_name)