2025-08-12
索引可能需要更多的数据来体现差异，所以让Gemini帮我做了个数据脚本生成的代码。
import mysql.connector
from mysql.connector import errorcode
from faker import Faker
import random
from datetime import datetime, timedelta
import uuid # 导入 uuid 库以生成更唯一的字符串
# --- 数据库配置 ---
db_config = {
'host': 'localhost', # 你的MySQL主机地址
'user': '', # 你的MySQL用户名
'password': '', # 你的MySQL密码
'database': 'industrial_project_management' # 数据库名称
}
# --- 数据量配置 ---
NUM_USERS = 100
NUM_PROJECTS = 50
NUM_TASKS = 5000 # 增加任务数量以更好地测试索引
# 初始化 Faker
fake = Faker('zh_CN') # 你也可以选择 'en_US' 或其他语言
# --- 辅助函数：生成唯一数据 ---
def generate_unique_value(generator_func, existing_set, max_attempts=1000):
"""
尝试生成一个唯一的值。
:param generator_func: Faker 生成器函数 (例如: fake.user_name)
:param existing_set: 已经存在的唯一值集合
:param max_attempts: 最大尝试次数
:return: 唯一的生成值
:raises ValueError: 如果在 max_attempts 内无法生成唯一值
"""
for _ in range(max_attempts):
value = generator_func()
if value not in existing_set:
existing_set.add(value)
return value
raise ValueError(f"无法在 {max_attempts} 次尝试内生成唯一的 {generator_func.__name__}。可能生成器多样性不足。")
# --- 数据库操作函数 ---
def create_database(cursor, db_name):
try:
cursor.execute(f"CREATE DATABASE IF NOT EXISTS {db_name} DEFAULT CHARACTER SET 'utf8mb4'")
print(f"✅ 数据库 '{db_name}' 已创建或已存在。")
except mysql.connector.Error as err:
print(f"❌ 创建数据库失败: {err}")
exit()
def create_tables(cursor):
tables = {}
tables['users'] = (
"""
CREATE TABLE users (
user_id INT AUTO_INCREMENT PRIMARY KEY,
username VARCHAR(50) UNIQUE NOT NULL,
email VARCHAR(100) UNIQUE NOT NULL,
password_hash VARCHAR(255) NOT NULL,
role ENUM('Admin', 'Developer', 'ProjectManager', 'Guest') NOT NULL,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
"""
)
tables['projects'] = (
"""
CREATE TABLE projects (
project_id INT AUTO_INCREMENT PRIMARY KEY,
project_name VARCHAR(100) UNIQUE NOT NULL,
description TEXT,
start_date DATE NOT NULL,
end_date DATE NOT NULL,
status ENUM('Planning', 'Developing', 'Testing', 'Completed', 'Canceled') NOT NULL,
created_by_user_id INT,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (created_by_user_id) REFERENCES users(user_id) ON DELETE SET NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
"""
)
tables['tasks'] = (
"""
CREATE TABLE tasks (
task_id INT AUTO_INCREMENT PRIMARY KEY,
task_name VARCHAR(255) NOT NULL,
description TEXT,
project_id INT,
assigned_to_user_id INT,
status ENUM('Pending', 'In Progress', 'Completed', 'Blocked', 'Cancelled') NOT NULL,
priority ENUM('Low', 'Medium', 'High', 'Urgent') NOT NULL,
due_date DATE,
completed_at DATETIME,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
task_details TEXT, -- 用于模糊查询和前缀索引
FOREIGN KEY (project_id) REFERENCES projects(project_id) ON DELETE CASCADE,
FOREIGN KEY (assigned_to_user_id) REFERENCES users(user_id) ON DELETE SET NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
"""
)
for table_name, table_sql in tables.items():
try:
print(f"🗑️ 删除表 '{table_name}' (如果存在)...")
cursor.execute(f"DROP TABLE IF EXISTS {table_name}")
print(f"✨ 创建表 '{table_name}'...")
cursor.execute(table_sql)
print(f"✅ 表 '{table_name}' 创建成功。")
except mysql.connector.Error as err:
print(f"❌ 创建表 '{table_name}' 失败: {err}")
exit()
def generate_and_insert_data(cursor, cnx):
# --- 1. 插入用户数据 ---
print("\n--- 🚀 插入用户数据 ---")
users_to_insert = []
user_ids = []
user_roles = ['Admin', 'Developer', 'ProjectManager', 'Guest']
# 预先生成所有唯一的用户数据
unique_usernames = set()
unique_emails = set()
current_users_count = 0
while current_users_count < NUM_USERS:
try:
username = generate_unique_value(fake.user_name, unique_usernames)
email = generate_unique_value(fake.email, unique_emails)
password_hash = fake.sha256()
role = random.choice(user_roles)
created_at = fake.date_time_between(start_date='-2y', end_date='now')
users_to_insert.append((username, email, password_hash, role, created_at))
current_users_count += 1
except ValueError as e:
print(f"⚠️ 无法生成足够数量的唯一用户数据: {e}")
break # 退出循环，插入当前已生成的用户
try:
# 批量插入用户数据
if users_to_insert:
cursor.executemany(
"INSERT INTO users (username, email, password_hash, role, created_at) VALUES (%s, %s, %s, %s, %s)",
users_to_insert
)
cnx.commit()
# 获取插入的 user_id
cursor.execute("SELECT user_id FROM users ORDER BY user_id DESC LIMIT %s", (len(users_to_insert),))
user_ids = [row[0] for row in cursor.fetchall()]
print(f"✅ {len(user_ids)} 个用户数据插入成功。")
else:
print("⚠️ 未生成任何用户数据。")
except mysql.connector.Error as err:
print(f"❌ 批量插入用户数据失败: {err}")
cnx.rollback()
raise err
# --- 2. 插入项目数据 ---
print("\n--- 🚀 插入项目数据 ---")
projects_to_insert = []
project_ids = []
project_statuses = ['Planning', 'Developing', 'Testing', 'Completed', 'Canceled']
unique_project_names = set()
for _ in range(NUM_PROJECTS):
# 策略：使用 fake.company() 结合随机数或 UUID 确保唯一性
# 这比依赖 fake.unique() 对更简单的生成器更有保障
project_name = generate_unique_value(lambda: f"{fake.company()} {random.randint(1000, 9999)}",
unique_project_names)
description = fake.paragraph(nb_sentences=3)
start_date = fake.date_this_year()
end_date = start_date + timedelta(days=random.randint(90, 365))
status = random.choice(project_statuses)
created_by_user_id = random.choice(user_ids) if user_ids else None
projects_to_insert.append((project_name, description, start_date, end_date, status, created_by_user_id))
try:
# 批量插入项目数据
if projects_to_insert:
cursor.executemany(
"INSERT INTO projects (project_name, description, start_date, end_date, status, created_by_user_id) VALUES (%s, %s, %s, %s, %s, %s)",
projects_to_insert
)
cnx.commit()
# 获取插入的 project_id
cursor.execute("SELECT project_id FROM projects ORDER BY project_id DESC LIMIT %s",
(len(projects_to_insert),))
project_ids = [row[0] for row in cursor.fetchall()]
print(f"✅ {len(project_ids)} 个项目数据插入成功。")
else:
print("⚠️ 未生成任何项目数据。")
except mysql.connector.Error as err:
print(f"❌ 批量插入项目数据失败: {err}")
cnx.rollback()
raise err
# --- 3. 插入任务数据 ---
print("\n--- 🚀 插入任务数据 ---")
tasks_to_insert = []
task_statuses = ['Pending', 'In Progress', 'Completed', 'Blocked', 'Cancelled']
task_priorities = ['Low', 'Medium', 'High', 'Urgent']
if not project_ids or not user_ids:
print("⚠️ 项目或用户数据不足，无法生成任务数据。")
return
# 为避免单次事务过大，分批插入
batch_size = 1000
for i in range(NUM_TASKS):
task_name = fake.bs() + ' ' + fake.word()
description = fake.paragraph(nb_sentences=2)
project_id = random.choice(project_ids)
assigned_to_user_id = random.choice(user_ids)
status = random.choice(task_statuses)
priority = random.choice(task_priorities)
due_date_raw = fake.date_between(start_date='today', end_date='+1y')
due_date = due_date_raw
completed_at = None
if status == 'Completed':
completed_at = fake.date_time_between(start_date='-1y', end_date='now')
# 确保完成日期不晚于截止日期（如果截止日期存在）
if due_date and completed_at.date() > due_date:
completed_at = datetime.combine(due_date - timedelta(days=random.randint(1, 30)), datetime.min.time())
created_at = fake.date_time_between(start_date='-2y', end_date='now')
task_details = fake.text(max_nb_chars=500) # 模拟任务详细描述
tasks_to_insert.append((
task_name, description, project_id, assigned_to_user_id,
status, priority, due_date, completed_at, created_at, task_details
))
if (i + 1) % batch_size == 0 or i == NUM_TASKS - 1:
try:
cursor.executemany(
"""
INSERT INTO tasks (task_name, description, project_id, assigned_to_user_id,
status, priority, due_date, completed_at, created_at, task_details)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
""", tasks_to_insert
)
cnx.commit()
print(f"已插入 {i + 1} / {NUM_TASKS} 条任务数据...")
tasks_to_insert = []
except mysql.connector.Error as err:
print(f"❌ 批量插入任务数据失败: {err}")
cnx.rollback()
raise err
print(f"✅ {NUM_TASKS} 条任务数据插入成功。")
def create_initial_indexes(cursor, cnx):
print("\n--- 🛠️ 创建初始索引 ---")
indexes_to_create = [
# tasks 表
("idx_project_status_duedate", "tasks", "project_id, status, due_date"),
("idx_task_details_prefix", "tasks", "task_details(255)"),
("idx_assigned_to_user_status", "tasks", "assigned_to_user_id, status"),
# users 表
("idx_role_username", "users", "role, username"),
# projects 表
("idx_start_end_status", "projects", "start_date, end_date, status")
]
for index_name, table_name, columns in indexes_to_create:
try:
# 尝试删除现有同名索引
print(f"🗑️ 尝试删除旧索引 '{index_name}' ON '{table_name}'...")
try:
cursor.execute(f"DROP INDEX {index_name} ON {table_name};")
print(f"✅ 已删除旧索引 '{index_name}'。")
except mysql.connector.Error as err:
# 错误码 1091 (ER_CANT_DROP_FIELD_OR_KEY) 表示索引不存在，可以安全忽略
if err.errno == errorcode.ER_CANT_DROP_FIELD_OR_KEY:
print(f"ℹ️ 索引 '{index_name}' 不存在，无需删除。")
else:
raise err # 其他错误则抛出
# 创建新索引
index_sql = f"CREATE INDEX {index_name} ON {table_name} ({columns});"
print(f"✨ 创建索引: {index_sql.strip()}")
cursor.execute(index_sql)
print(f"✅ 索引 '{index_name}' 创建成功。")
cnx.commit() # 每次创建完一个索引就提交
except mysql.connector.Error as err:
print(f"❌ 创建索引失败: {index_name} - {err}")
cnx.rollback()
exit()
print("✅ 所有初始索引创建完成。")
# --- 主程序 ---
if __name__ == "__main__":
cnx = None
try:
# 连接到MySQL服务器（不指定数据库，先创建数据库）
print("--- 🔌 尝试连接到 MySQL 服务器 ---")
cnx = mysql.connector.connect(
host=db_config['host'],
user=db_config['user'],
password=db_config['password']
)
cursor = cnx.cursor()
print("✅ 连接成功。")
# 创建数据库
create_database(cursor, db_config['database'])
# 切换到指定数据库
cnx.database = db_config['database']
cursor.close() # 关闭旧游标
cursor = cnx.cursor() # 创建新游标，它现在知道数据库了
# 创建表
create_tables(cursor)
# 插入数据
generate_and_insert_data(cursor, cnx)
# 创建索引
create_initial_indexes(cursor, cnx)
print("\n🎉🎉🎉 数据库初始化、数据填充和索引创建全部完成！🎉🎉🎉")
except mysql.connector.Error as err:
if err.errno == errorcode.ER_ACCESS_DENIED_ERROR:
print("❌ 连接失败：用户名或密码错误。")
elif err.errno == errorcode.ER_BAD_DB_ERROR:
print("❌ 连接失败：数据库不存在。")
else:
print(f"❌ 发生未知错误: {err}")
finally:
if cnx and cnx.is_connected():
cursor.close()
cnx.close()
print("--- 🔌 数据库连接已关闭。---")