#!/usr/bin/env python """ 轻医美连锁岗位与课程种子数据 """ import asyncio import sys from pathlib import Path import aiomysql import os import json from datetime import datetime # 添加项目根目录到 Python 路径 project_root = Path(__file__).parent.parent sys.path.append(str(project_root)) from app.core.config import settings async def execute_seed(): """执行种子数据插入""" try: # 从环境变量或配置中获取数据库连接信息 database_url = os.getenv("DATABASE_URL", settings.DATABASE_URL) # 解析数据库连接字符串 if database_url.startswith("mysql+aiomysql://"): url = database_url.replace("mysql+aiomysql://", "") else: url = database_url # 解析连接参数 auth_host = url.split("@")[1] user_pass = url.split("@")[0] host_port_db = auth_host.split("/") host_port = host_port_db[0].split(":") user = user_pass.split(":")[0] password = user_pass.split(":")[1] host = host_port[0] port = int(host_port[1]) if len(host_port) > 1 else 3306 database = host_port_db[1].split("?")[0] if len(host_port_db) > 1 else "kaopeilian" print(f"连接数据库: {host}:{port}/{database}") # 创建数据库连接 conn = await aiomysql.connect( host=host, port=port, user=user, password=password, db=database, charset='utf8mb4' ) async with conn.cursor() as cursor: print("\n🎯 开始插入轻医美连锁业务数据...") # 1. 插入轻医美相关岗位 print("\n📌 插入岗位数据...") positions_data = [ { 'name': '区域经理', 'code': 'region_manager', 'description': '负责多家门店的运营管理和业绩达成', 'status': 'active', 'skills': json.dumps(['团队管理', '业绩分析', '战略规划', '客户关系'], ensure_ascii=False), 'level': 'expert', 'sort_order': 10 }, { 'name': '店长', 'code': 'store_manager', 'description': '负责门店日常运营管理，团队建设和业绩达成', 'status': 'active', 'skills': json.dumps(['门店管理', '团队建设', '销售管理', '客户维护'], ensure_ascii=False), 'level': 'senior', 'sort_order': 20, 'parent_code': 'region_manager' }, { 'name': '美容顾问', 'code': 'beauty_consultant', 'description': '为客户提供专业的美容咨询和方案设计', 'status': 'active', 'skills': json.dumps(['产品知识', '销售技巧', '方案设计', '客户沟通'], ensure_ascii=False), 'level': 'intermediate', 'sort_order': 30, 'parent_code': 'store_manager' }, { 'name': '美容技师', 'code': 'beauty_therapist', 'description': '为客户提供专业的美容护理服务', 'status': 'active', 'skills': json.dumps(['护肤技术', '仪器操作', '手法技巧', '服务意识'], ensure_ascii=False), 'level': 'intermediate', 'sort_order': 40, 'parent_code': 'store_manager' }, { 'name': '医美咨询师', 'code': 'medical_beauty_consultant', 'description': '提供医疗美容项目咨询和方案制定', 'status': 'active', 'skills': json.dumps(['医美知识', '风险评估', '方案设计', '合规意识'], ensure_ascii=False), 'level': 'senior', 'sort_order': 35, 'parent_code': 'store_manager' }, { 'name': '护士', 'code': 'nurse', 'description': '协助医生进行医美项目操作，负责术后护理', 'status': 'active', 'skills': json.dumps(['护理技术', '无菌操作', '应急处理', '医疗知识'], ensure_ascii=False), 'level': 'intermediate', 'sort_order': 45, 'parent_code': 'store_manager' }, { 'name': '前台接待', 'code': 'receptionist', 'description': '负责客户接待、预约管理和前台事务', 'status': 'active', 'skills': json.dumps(['接待礼仪', '沟通能力', '信息管理', '服务意识'], ensure_ascii=False), 'level': 'junior', 'sort_order': 50, 'parent_code': 'store_manager' }, { 'name': '市场专员', 'code': 'marketing_specialist', 'description': '负责门店营销活动策划和执行', 'status': 'active', 'skills': json.dumps(['活动策划', '社媒运营', '数据分析', '创意设计'], ensure_ascii=False), 'level': 'intermediate', 'sort_order': 60, 'parent_code': 'store_manager' } ] # 先获取已存在的岗位ID映射 position_id_map = {} # 插入岗位（处理层级关系） for position in positions_data: # 检查是否已存在 check_sql = "SELECT id FROM positions WHERE code = %s AND is_deleted = FALSE" await cursor.execute(check_sql, (position['code'],)) existing = await cursor.fetchone() if existing: position_id_map[position['code']] = existing[0] print(f" ⚠️ 岗位 '{position['name']}' 已存在，ID: {existing[0]}") else: # 获取parent_id parent_id = None if 'parent_code' in position: parent_code = position.pop('parent_code') if parent_code in position_id_map: parent_id = position_id_map[parent_code] # 插入岗位 insert_sql = """ INSERT INTO positions (name, code, description, parent_id, status, skills, level, sort_order, created_at, updated_at) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW()) """ await cursor.execute(insert_sql, ( position['name'], position['code'], position['description'], parent_id, position['status'], position['skills'], position['level'], position['sort_order'] )) position_id = cursor.lastrowid position_id_map[position['code']] = position_id print(f" ✅ 插入岗位: {position['name']} (ID: {position_id})") await conn.commit() # 2. 插入轻医美相关课程 print("\n📚 插入课程数据...") courses_data = [ { 'name': '皮肤生理学基础', 'description': '学习皮肤结构、功能和常见问题，为专业护理打下基础', 'category': 'technology', 'status': 'published', 'duration_hours': 16, 'difficulty_level': 2, 'tags': json.dumps(['皮肤学', '基础理论', '必修课'], ensure_ascii=False), 'is_featured': True, 'sort_order': 100 }, { 'name': '医美产品知识与应用', 'description': '全面了解各类医美产品的成分、功效和适用人群', 'category': 'technology', 'status': 'published', 'duration_hours': 20, 'difficulty_level': 3, 'tags': json.dumps(['产品知识', '医美', '专业技能'], ensure_ascii=False), 'is_featured': True, 'sort_order': 110 }, { 'name': '美容仪器操作与维护', 'description': '掌握各类美容仪器的操作方法、注意事项和日常维护', 'category': 'technology', 'status': 'published', 'duration_hours': 24, 'difficulty_level': 3, 'tags': json.dumps(['仪器操作', '实操技能', '设备维护'], ensure_ascii=False), 'is_featured': False, 'sort_order': 120 }, { 'name': '轻医美销售技巧', 'description': '学习专业的销售话术、客户需求分析和成交技巧', 'category': 'business', 'status': 'published', 'duration_hours': 16, 'difficulty_level': 2, 'tags': json.dumps(['销售技巧', '客户沟通', '业绩提升'], ensure_ascii=False), 'is_featured': True, 'sort_order': 130 }, { 'name': '客户服务与投诉处理', 'description': '提升服务意识，掌握客户投诉处理的方法和技巧', 'category': 'business', 'status': 'published', 'duration_hours': 12, 'difficulty_level': 2, 'tags': json.dumps(['客户服务', '危机处理', '沟通技巧'], ensure_ascii=False), 'is_featured': False, 'sort_order': 140 }, { 'name': '卫生消毒与感染控制', 'description': '学习医美机构的卫生标准和消毒流程，确保服务安全', 'category': 'general', 'status': 'published', 'duration_hours': 8, 'difficulty_level': 1, 'tags': json.dumps(['卫生安全', '消毒规范', '合规管理'], ensure_ascii=False), 'is_featured': True, 'sort_order': 150 }, { 'name': '门店运营管理', 'description': '学习门店日常管理、团队建设和业绩管理', 'category': 'management', 'status': 'published', 'duration_hours': 20, 'difficulty_level': 3, 'tags': json.dumps(['门店管理', '团队管理', '运营策略'], ensure_ascii=False), 'is_featured': False, 'sort_order': 160 }, { 'name': '医美项目介绍与咨询', 'description': '详细了解各类医美项目的原理、效果和适应症', 'category': 'technology', 'status': 'published', 'duration_hours': 30, 'difficulty_level': 4, 'tags': json.dumps(['医美项目', '专业咨询', '风险告知'], ensure_ascii=False), 'is_featured': True, 'sort_order': 170 }, { 'name': '社媒营销与私域运营', 'description': '学习如何通过社交媒体进行品牌推广和客户维护', 'category': 'business', 'status': 'published', 'duration_hours': 16, 'difficulty_level': 2, 'tags': json.dumps(['社媒营销', '私域流量', '客户维护'], ensure_ascii=False), 'is_featured': False, 'sort_order': 180 } ] course_id_map = {} for course in courses_data: # 检查是否已存在 check_sql = "SELECT id FROM courses WHERE name = %s AND is_deleted = FALSE" await cursor.execute(check_sql, (course['name'],)) existing = await cursor.fetchone() if existing: course_id_map[course['name']] = existing[0] print(f" ⚠️ 课程 '{course['name']}' 已存在，ID: {existing[0]}") else: # 插入课程 insert_sql = """ INSERT INTO courses ( name, description, category, status, duration_hours, difficulty_level, tags, is_featured, sort_order, published_at, created_at, updated_at ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW(), NOW()) """ await cursor.execute(insert_sql, ( course['name'], course['description'], course['category'], course['status'], course['duration_hours'], course['difficulty_level'], course['tags'], course['is_featured'], course['sort_order'] )) course_id = cursor.lastrowid course_id_map[course['name']] = course_id print(f" ✅ 插入课程: {course['name']} (ID: {course_id})") await conn.commit() # 3. 设置岗位与课程的关联 print("\n🔗 设置岗位课程关联...") position_courses = [ # 店长必修课程 ('store_manager', '门店运营管理', 'required', 1), ('store_manager', '轻医美销售技巧', 'required', 2), ('store_manager', '客户服务与投诉处理', 'required', 3), ('store_manager', '卫生消毒与感染控制', 'required', 4), # 美容顾问必修课程 ('beauty_consultant', '皮肤生理学基础', 'required', 1), ('beauty_consultant', '医美产品知识与应用', 'required', 2), ('beauty_consultant', '轻医美销售技巧', 'required', 3), ('beauty_consultant', '客户服务与投诉处理', 'required', 4), ('beauty_consultant', '社媒营销与私域运营', 'optional', 5), # 美容技师必修课程 ('beauty_therapist', '皮肤生理学基础', 'required', 1), ('beauty_therapist', '美容仪器操作与维护', 'required', 2), ('beauty_therapist', '卫生消毒与感染控制', 'required', 3), ('beauty_therapist', '医美产品知识与应用', 'optional', 4), # 医美咨询师必修课程 ('medical_beauty_consultant', '医美项目介绍与咨询', 'required', 1), ('medical_beauty_consultant', '皮肤生理学基础', 'required', 2), ('medical_beauty_consultant', '医美产品知识与应用', 'required', 3), ('medical_beauty_consultant', '轻医美销售技巧', 'required', 4), ('medical_beauty_consultant', '客户服务与投诉处理', 'required', 5), # 护士必修课程 ('nurse', '卫生消毒与感染控制', 'required', 1), ('nurse', '医美项目介绍与咨询', 'required', 2), ('nurse', '皮肤生理学基础', 'required', 3), # 前台接待必修课程 ('receptionist', '客户服务与投诉处理', 'required', 1), ('receptionist', '医美产品知识与应用', 'optional', 2), # 市场专员必修课程 ('marketing_specialist', '社媒营销与私域运营', 'required', 1), ('marketing_specialist', '医美产品知识与应用', 'optional', 2), ('marketing_specialist', '轻医美销售技巧', 'optional', 3), ] for pos_code, course_name, course_type, priority in position_courses: if pos_code in position_id_map and course_name in course_id_map: position_id = position_id_map[pos_code] course_id = course_id_map[course_name] # 检查是否已存在 check_sql = """ SELECT id FROM position_courses WHERE position_id = %s AND course_id = %s AND is_deleted = FALSE """ await cursor.execute(check_sql, (position_id, course_id)) existing = await cursor.fetchone() if not existing: insert_sql = """ INSERT INTO position_courses (position_id, course_id, course_type, priority, is_deleted, created_at, updated_at) VALUES (%s, %s, %s, %s, FALSE, NOW(), NOW()) """ await cursor.execute(insert_sql, (position_id, course_id, course_type, priority)) print(f" ✅ 关联: {pos_code} - {course_name} ({course_type})") await conn.commit() # 4. 显示统计信息 print("\n📊 数据统计:") # 统计岗位 await cursor.execute("SELECT COUNT(*) FROM positions WHERE is_deleted = FALSE") total_positions = (await cursor.fetchone())[0] print(f" 岗位总数: {total_positions}") # 统计课程 await cursor.execute("SELECT COUNT(*) FROM courses WHERE is_deleted = FALSE") total_courses = (await cursor.fetchone())[0] print(f" 课程总数: {total_courses}") # 统计岗位课程关联 await cursor.execute("SELECT COUNT(*) FROM position_courses WHERE is_deleted = FALSE") total_pc = (await cursor.fetchone())[0] print(f" 岗位课程关联数: {total_pc}") print("\n🎉 轻医美连锁业务数据插入完成！") print("\n💡 提示：") print(" 1. 可以登录系统查看岗位管理页面") print(" 2. 每个岗位都配置了相应的必修和选修课程") print(" 3. 课程涵盖了技术、管理、业务和通用等各个分类") conn.close() except Exception as e: print(f"❌ 执行失败: {e}") import traceback traceback.print_exc() sys.exit(1) if __name__ == "__main__": print("🌟 开始插入轻医美连锁岗位与课程数据...") asyncio.run(execute_seed())