""" 种子数据初始化模块 在 main.py 启动时自动调用，向数据库注入测试数据 幂等设计：已存在数据则跳过，不重复插入 """ from datetime import date from sqlalchemy import text from databases import engine_all, Base_all, Session def seed_test_data(): """启动时自动注入测试数据（幂等）""" db = Session() try: # 临时关闭外键检查，避免 class_management_model.py 的 FK bug db.execute(text("SET FOREIGN_KEY_CHECKS = 0")) # 导入所有模型以获取 metadata（触发 Base_all.metadata.create_all） import students.model.students_model import Teachers.model.tea_model import scores.model.score_model import stu_jiuye.model # 创建所有表（临时关闭外键检查） db.execute(text("SET FOREIGN_KEY_CHECKS = 0")) Base_all.metadata.create_all(bind=engine_all) # 手动建 classes 表（因 FK bug 无法通过 ORM 建表） db.execute(text(""" CREATE TABLE IF NOT EXISTS classes ( id INT AUTO_INCREMENT PRIMARY KEY, num VARCHAR(50) NOT NULL UNIQUE, name VARCHAR(50) NOT NULL, head_teacher_id INT, coach_teacher_id INT, tutor_teacher_id INT, class_start_time DATE, class_end_time DATE, is_delete INT DEFAULT 0, create_time DATETIME, update_time DATETIME, FOREIGN KEY (head_teacher_id) REFERENCES teachers(id), FOREIGN KEY (coach_teacher_id) REFERENCES teachers(id), FOREIGN KEY (tutor_teacher_id) REFERENCES teachers(id) ) """)) # 恢复外键检查 db.execute(text("SET FOREIGN_KEY_CHECKS = 1")) db.commit() # 幂等检查：科目表有数据就跳过全部 if db.execute(text("SELECT COUNT(*) FROM subject")).scalar() > 0: return print(" [种子] 开始注入测试数据...") # ── 1. 科目 ────────────────────────────────────── subjects = ["Python程序设计", "数据结构", "数据库原理", "人工智能导论", "计算机网络", "操作系统"] sub_ids = {} for name in subjects: result = db.execute( text("INSERT INTO subject (name) VALUES (:name)"), {"name": name} ) sub_ids[name] = result.lastrowid # ── 2. 教师 ────────────────────────────────────── teachers = [ ("张伟", "m", 1, sub_ids["Python程序设计"], "北京", "Python", "博士", "清华", date(2018, 3, 1), "6年"), ("李芳", "f", 1, sub_ids["数据结构"], "上海", "算法", "博士", "北大", date(2019, 9, 1), "4年"), ("王强", "m", 2, sub_ids["数据库原理"], "广州", "MySQL", "硕士", "中山大学", date(2020, 2, 1), "3年"), ("陈静", "f", 2, sub_ids["人工智能导论"], "深圳", "机器学习", "博士", "华南理工", date(2017, 7, 1), "7年"), ("刘洋", "m", 3, sub_ids["计算机网络"], "杭州", "网络", "硕士", "浙江大学", date(2021, 1, 1), "2年"), ("赵敏", "f", 3, sub_ids["操作系统"], "南京", "Linux", "博士", "南京大学", date(2019, 3, 1), "5年"), ] teacher_ids = [] for t in teachers: phone = f"1380000{len(teacher_ids):04d}" result = db.execute(text(""" INSERT INTO teachers (name, sex, class_id, phone, subject_id, home_place, specialty, education, collage, start_time, work_experience, is_delete) VALUES (:name, :sex, :class_id, :phone, :subject_id, :home_place, :specialty, :education, :collage, :start_time, :work_experience, 0) """), { "name": t[0], "sex": t[1], "class_id": t[2], "phone": phone, "subject_id": t[3], "home_place": t[4], "specialty": t[5], "education": t[6], "collage": t[7], "start_time": t[8], "work_experience": t[9], }) teacher_ids.append(result.lastrowid) # ── 3. 班级 ────────────────────────────────────── classes = [ ("CS202401", "计算机科学2401班", teacher_ids[0], teacher_ids[1], teacher_ids[2], date(2024, 9, 1), date(2028, 6, 30)), ("CS202402", "计算机科学2402班", teacher_ids[3], teacher_ids[4], teacher_ids[5], date(2024, 9, 1), date(2028, 6, 30)), ("AI202501", "人工智能2501班", teacher_ids[0], teacher_ids[3], teacher_ids[5], date(2025, 9, 1), date(2029, 6, 30)), ] class_ids = [] for c in classes: result = db.execute(text(""" INSERT INTO classes (num, name, head_teacher_id, coach_teacher_id, tutor_teacher_id, class_start_time, class_end_time, is_delete) VALUES (:num, :name, :head_teacher_id, :coach_teacher_id, :tutor_teacher_id, :class_start_time, :class_end_time, 0) """), { "num": c[0], "name": c[1], "head_teacher_id": c[2], "coach_teacher_id": c[3], "tutor_teacher_id": c[4], "class_start_time": c[5], "class_end_time": c[6], }) class_ids.append(result.lastrowid) # ── 4. 学生 ────────────────────────────────────── first_names = ["志强", "丽华", "建国", "美玲", "文博", "晓燕", "浩然", "雨桐", "子轩", "思涵", "俊杰", "雅琪", "宇航", "欣怡", "天翊", "诗涵", "一诺", "浩宇", "梓萱", "博文"] home_places = ["北京", "上海", "广州", "深圳", "杭州", "成都", "武汉", "西安"] colleges = ["清华大学", "北京大学", "复旦大学", "上海交通大学", "浙江大学", "南京大学", "武汉大学", "华中科技大学"] specialties = ["计算机科学", "软件工程", "数据科学", "人工智能"] for i in range(20): class_idx = i % 3 enrollment = date(2024, 9, 1) if class_idx != 2 else date(2025, 9, 1) graduate = date(2028, 6, 30) if class_idx != 2 else date(2029, 6, 30) phone = f"1390000{1000 + i}" db.execute(text(""" INSERT INTO students (num, name, age, sex, home_Place, college, specialty, enrollment_time, graduate_time, education, class_id, advisor_id, phone, is_deleted) VALUES (:num, :name, :age, :sex, :home_Place, :college, :specialty, :enrollment_time, :graduate_time, '本科', :class_id, :advisor_id, :phone, 0) """), { "num": f"2024{1001 + i}", "name": first_names[i], "age": 18 + (i % 4), "sex": "男" if i % 2 == 0 else "女", "home_Place": home_places[i % 8], "college": colleges[i % 8], "specialty": specialties[i % 4], "enrollment_time": enrollment, "graduate_time": graduate, "class_id": class_ids[class_idx], "advisor_id": teacher_ids[i % 6], "phone": phone, }) # ── 5. 成绩 ────────────────────────────────────── all_students = db.execute( text("SELECT id, class_id FROM students WHERE is_deleted = 0") ).fetchall() all_subjects = db.execute( text("SELECT id FROM subject") ).fetchall() for idx, st in enumerate(all_students): for subj_idx, sub in enumerate(all_subjects): score_val = 60 + (idx * 7 + subj_idx * 13) % 41 # 前3名学生部分课程不及格（供统计接口测试） if idx < 3 and subj_idx < 2: score_val = 30 + (idx * 5) % 25 db.execute(text(""" INSERT INTO scores (sid, cid, num, score, t_subject, is_deleted) VALUES (:sid, :cid, 1, :score, :t_subject, 0) """), { "sid": st[0], "cid": st[1], "score": score_val, "t_subject": sub[0], }) # ── 6. 地址 ────────────────────────────────────── addresses = ["北京市海淀区", "上海市浦东新区", "深圳市南山区", "杭州市西湖区", "广州市天河区"] address_ids = [] for addr in addresses: result = db.execute( text("INSERT INTO Address (somewhere) VALUES (:where)"), {"where": addr} ) address_ids.append(result.lastrowid) # ── 7. 公司 ────────────────────────────────────── companies = ["字节跳动", "阿里巴巴", "腾讯", "华为", "百度"] company_ids = [] for idx, comp_name in enumerate(companies): result = db.execute( text("INSERT INTO Company (employment_company) VALUES (:name)"), {"name": comp_name} ) company_ids.append(result.lastrowid) # ── 8. 就业信息 ─────────────────────────────────── employment_data = [ (all_students[0][0], all_students[0][1], company_ids[0], 35000, date(2028, 3, 1), date(2028, 4, 15)), (all_students[1][0], all_students[1][1], company_ids[1], 32000, date(2028, 3, 1), date(2028, 5, 1)), (all_students[2][0], all_students[2][1], company_ids[2], 30000, date(2028, 3, 1), date(2028, 4, 20)), (all_students[3][0], all_students[3][1], company_ids[3], 28000, date(2028, 3, 1), date(2028, 5, 10)), (all_students[4][0], all_students[4][1], company_ids[4], 25000, date(2028, 3, 1), date(2028, 6, 1)), (all_students[5][0], all_students[5][1], company_ids[0], 33000, date(2028, 3, 1), date(2028, 4, 25)), (all_students[6][0], all_students[6][1], company_ids[1], 29000, date(2028, 3, 1), None), (all_students[7][0], all_students[7][1], company_ids[2], 31000, date(2028, 3, 1), date(2028, 5, 5)), ] for idx, emp in enumerate(employment_data): db.execute(text(""" INSERT INTO Employments (stuid, class_id, company_id, address_id, employment_salary, employment_open_time, offer_recived_time, is_deleted) VALUES (:stuid, :class_id, :company_id, :addr_id, :salary, :open_time, :offer_time, 0) """), { "stuid": emp[0], "class_id": emp[1], "company_id": emp[2], "addr_id": address_ids[idx % len(address_ids)], "salary": emp[3], "open_time": emp[4], "offer_time": emp[5], }) db.commit() print(f" [种子] 完成！插入 {len(subjects)} 科目、{len(teachers)} 教师、" f"{len(classes)} 班级、20 学生、{len(all_students)*len(all_subjects)} 成绩、" f"{len(companies)} 公司、{len(employment_data)} 就业") except Exception as e: db.rollback() print(f" [种子] 失败: {e}") raise finally: db.close()