398 lines
15 KiB
Python
398 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
登录并批量创建学生信息的完整脚本
|
|
包含登录获取Token和批量创建学生两个功能
|
|
"""
|
|
|
|
import requests
|
|
import json
|
|
import random
|
|
import time
|
|
from datetime import datetime
|
|
import logging
|
|
|
|
# 配置日志
|
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class LoginAndCreateStudents:
|
|
def __init__(self, base_url="http://localhost", api_prefix="/dev-api"):
|
|
"""
|
|
初始化登录和学生创建器
|
|
|
|
Args:
|
|
base_url: 基础URL
|
|
api_prefix: API前缀
|
|
"""
|
|
self.base_url = base_url
|
|
self.api_prefix = api_prefix
|
|
self.full_base_url = base_url + api_prefix
|
|
self.token = None
|
|
self.session = requests.Session()
|
|
|
|
# 简单的数据生成器
|
|
self.surnames = ['李', '王', '张', '刘', '陈', '杨', '黄', '赵', '周', '吴']
|
|
self.male_names = ['伟', '强', '磊', '军', '勇', '涛', '明', '超', '辉', '华']
|
|
self.female_names = ['丽', '娜', '敏', '静', '佳', '雅', '莉', '萍', '红', '梅']
|
|
self.provinces = ['北京市', '上海市', '广东省', '江苏省', '浙江省', '山东省', '河南省', '湖北省']
|
|
|
|
def login(self, username="admin", password="admin123"):
|
|
"""
|
|
尝试登录获取Token
|
|
"""
|
|
try:
|
|
# 尝试标准登录
|
|
login_data = {
|
|
"tenantId": "000000",
|
|
"username": username,
|
|
"password": password,
|
|
"clientId": "e5cd7e4891bf95d1d19206ce24a7b32e",
|
|
"grantType": "password"
|
|
}
|
|
|
|
headers = {
|
|
'Content-Type': 'application/json;charset=utf-8',
|
|
'clientid': 'e5cd7e4891bf95d1d19206ce24a7b32e',
|
|
'referer': f'{self.base_url}/login?redirect=%252Findex',
|
|
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8'
|
|
}
|
|
|
|
logger.info("正在尝试登录...")
|
|
response = self.session.post(
|
|
f"{self.full_base_url}/auth/login",
|
|
json=login_data,
|
|
headers=headers,
|
|
timeout=10
|
|
)
|
|
|
|
response.raise_for_status()
|
|
result = response.json()
|
|
|
|
logger.info(f"登录响应: {json.dumps(result, ensure_ascii=False, indent=2)}")
|
|
|
|
if result.get('code') == 200 and result.get('data', {}).get('access_token'):
|
|
self.token = result['data']['access_token']
|
|
# 更新session的默认headers
|
|
self.session.headers.update({
|
|
'Authorization': f'Bearer {self.token}',
|
|
'Content-Type': 'application/json'
|
|
})
|
|
logger.info(f"✅ 登录成功，Token: {self.token[:50]}...")
|
|
return True
|
|
else:
|
|
logger.error(f"❌ 登录失败: {result.get('msg')}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
logger.error(f"❌ 登录异常: {str(e)}")
|
|
return False
|
|
|
|
def get_class_list(self):
|
|
"""获取班级列表"""
|
|
if not self.token:
|
|
logger.error("未登录，无法获取班级列表")
|
|
return []
|
|
|
|
try:
|
|
url = f"{self.full_base_url}/classMsg/classMsg/list"
|
|
params = {
|
|
'pageNum': 1,
|
|
'pageSize': 1000
|
|
}
|
|
|
|
response = self.session.get(url, params=params)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
if data.get('code') == 200:
|
|
classes = data.get('rows', [])
|
|
logger.info(f"获取到 {len(classes)} 个班级")
|
|
return classes
|
|
else:
|
|
logger.error(f"获取班级列表失败: {data.get('msg')}")
|
|
return []
|
|
|
|
except Exception as e:
|
|
logger.error(f"获取班级列表异常: {str(e)}")
|
|
return []
|
|
|
|
def generate_name(self, gender):
|
|
"""生成姓名"""
|
|
surname = random.choice(self.surnames)
|
|
if gender == '0': # 男性
|
|
given_name = random.choice(self.male_names)
|
|
else: # 女性
|
|
given_name = random.choice(self.female_names)
|
|
return surname + given_name
|
|
|
|
def generate_phone(self):
|
|
"""生成手机号"""
|
|
prefixes = ['130', '131', '132', '133', '134', '135', '136', '137', '138', '139']
|
|
prefix = random.choice(prefixes)
|
|
suffix = ''.join([str(random.randint(0, 9)) for _ in range(8)])
|
|
return prefix + suffix
|
|
|
|
def generate_id_card(self, birth_date):
|
|
"""生成身份证号"""
|
|
area_codes = ['110101', '110102', '310101', '310104', '440101', '440103']
|
|
area_code = random.choice(area_codes)
|
|
birth_str = birth_date.replace('-', '')
|
|
sequence = str(random.randint(100, 999))
|
|
check_digit = str(random.randint(0, 9))
|
|
return area_code + birth_str + sequence + check_digit
|
|
|
|
def generate_birth_date(self):
|
|
"""生成出生日期（18-22岁）"""
|
|
today = datetime.now()
|
|
birth_year = today.year - random.randint(18, 22)
|
|
birth_month = random.randint(1, 12)
|
|
birth_day = random.randint(1, 28)
|
|
return f"{birth_year:04d}-{birth_month:02d}-{birth_day:02d}"
|
|
|
|
def generate_student_data(self, class_info, student_index):
|
|
"""生成学生数据"""
|
|
gender = random.choice(['0', '1'])
|
|
name = self.generate_name(gender)
|
|
|
|
# 生成学号
|
|
class_code = class_info.get('classCode', class_info.get('class_code', ''))
|
|
if not class_code:
|
|
class_code = f"CLS{class_info.get('id', '')}"
|
|
|
|
student_no = f"{class_code}{student_index:02d}"
|
|
birth_date = self.generate_birth_date()
|
|
|
|
student_data = {
|
|
# 基础信息
|
|
"studentNo": student_no,
|
|
"idCard": self.generate_id_card(birth_date),
|
|
"name": name,
|
|
"gender": gender,
|
|
"birthDate": birth_date,
|
|
"nationality": random.choice(['01', '02', '03', '04', '05']),
|
|
"politicalStatus": random.choice(['01', '02', '03', '13']),
|
|
"phone": self.generate_phone(),
|
|
"email": f"{student_no}@student.edu.cn",
|
|
|
|
# 学籍信息
|
|
"classId": class_info.get('id'),
|
|
"className": class_info.get('className', class_info.get('class_name', '')),
|
|
"majorId": class_info.get('majorId', class_info.get('major_id')),
|
|
"majorName": class_info.get('majorName', class_info.get('major_name', '')),
|
|
"collegeId": class_info.get('collegeId', class_info.get('college_id')),
|
|
"collegeName": class_info.get('collegeName', class_info.get('college_name', '')),
|
|
"grade": class_info.get('grade', ''),
|
|
"enrollmentYear": class_info.get('enrollmentYear', class_info.get('enrollment_year')),
|
|
"academicLevel": class_info.get('classType', class_info.get('class_type', 1)),
|
|
"classTeacherId": class_info.get('classTeacherId', class_info.get('class_teacher_id')),
|
|
"classTeacherName": class_info.get('classTeacherName', class_info.get('class_teacher_name', '')),
|
|
"schoolSystem": class_info.get('schoolSystem', class_info.get('school_system', 3)),
|
|
|
|
# 默认状态
|
|
"studentStatus": "1",
|
|
"trainingType": "全日制",
|
|
"admissionCategory": "统招",
|
|
"householdRegisterType": "非农业",
|
|
"educationLevel": "2",
|
|
"maritalStatus": "1",
|
|
|
|
# 地址信息
|
|
"nativePlace": random.choice(self.provinces),
|
|
"sourceRegion": random.choice(self.provinces),
|
|
"homeAddress": f"{random.choice(self.provinces)}某市某区某街道{random.randint(1, 999)}号",
|
|
"currentAddress": "学校宿舍",
|
|
|
|
# 家庭信息
|
|
"fatherName": self.generate_name('0'),
|
|
"fatherPhone": self.generate_phone(),
|
|
"motherName": self.generate_name('1'),
|
|
"motherPhone": self.generate_phone(),
|
|
|
|
# 健康信息
|
|
"height": random.randint(155, 185),
|
|
"weight": random.randint(45, 80),
|
|
"bloodType": random.choice(['A', 'B', 'AB', 'O']),
|
|
"healthStatus": "良好",
|
|
|
|
# 布尔字段
|
|
"isBoarder": 1,
|
|
"isPoorStudent": random.randint(0, 1),
|
|
"isMinority": 0,
|
|
"isDisabled": 0,
|
|
"isOrphan": 0,
|
|
"isLeftBehindChild": 0,
|
|
"isSingleChild": random.randint(0, 1),
|
|
"isLowIncome": 0,
|
|
"hasLoan": random.randint(0, 1),
|
|
|
|
"remark": f"批量生成的测试学生数据 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
|
|
}
|
|
|
|
return student_data
|
|
|
|
def create_student(self, student_data):
|
|
"""创建单个学生"""
|
|
try:
|
|
url = f"{self.full_base_url}/studentMsg/studentMsg"
|
|
|
|
response = self.session.post(url, json=student_data)
|
|
response.raise_for_status()
|
|
|
|
result = response.json()
|
|
if result.get('code') == 200:
|
|
logger.info(f"成功创建学生: {student_data['name']} ({student_data['studentNo']})")
|
|
return True, result
|
|
else:
|
|
logger.error(f"创建学生失败: {student_data['name']} - {result.get('msg')}")
|
|
return False, result
|
|
|
|
except Exception as e:
|
|
logger.error(f"创建学生异常: {student_data['name']} - {str(e)}")
|
|
return False, str(e)
|
|
|
|
def batch_create_students_for_class(self, class_info, student_count=30):
|
|
"""为指定班级批量创建学生"""
|
|
class_name = class_info.get('className', class_info.get('class_name', f"班级{class_info.get('id')}"))
|
|
logger.info(f"开始为班级 [{class_name}] 创建 {student_count} 个学生...")
|
|
|
|
success_count = 0
|
|
failed_count = 0
|
|
failed_students = []
|
|
|
|
for i in range(1, student_count + 1):
|
|
try:
|
|
student_data = self.generate_student_data(class_info, i)
|
|
success, result = self.create_student(student_data)
|
|
|
|
if success:
|
|
success_count += 1
|
|
else:
|
|
failed_count += 1
|
|
failed_students.append({
|
|
'student_no': student_data['studentNo'],
|
|
'name': student_data['name'],
|
|
'error': result
|
|
})
|
|
|
|
time.sleep(0.1) # 短暂延迟
|
|
|
|
except Exception as e:
|
|
failed_count += 1
|
|
logger.error(f"处理第 {i} 个学生时发生异常: {str(e)}")
|
|
|
|
result = {
|
|
'class_name': class_name,
|
|
'total': student_count,
|
|
'success': success_count,
|
|
'failed': failed_count,
|
|
'failed_students': failed_students
|
|
}
|
|
|
|
logger.info(f"班级 [{class_name}] 学生创建完成: 成功 {success_count}，失败 {failed_count}")
|
|
return result
|
|
|
|
def run(self, username="admin", password="admin123", student_count_per_class=30):
|
|
"""运行完整流程"""
|
|
print("🚀 开始执行批量创建学生流程...")
|
|
|
|
# 1. 登录
|
|
if not self.login(username, password):
|
|
print("❌ 登录失败，无法继续")
|
|
return
|
|
|
|
# 2. 获取班级列表
|
|
classes = self.get_class_list()
|
|
if not classes:
|
|
print("❌ 没有获取到班级信息")
|
|
return
|
|
|
|
print(f"\n📋 找到 {len(classes)} 个班级，每个班级将创建 {student_count_per_class} 个学生")
|
|
|
|
# 显示班级信息
|
|
for i, class_info in enumerate(classes[:5], 1):
|
|
print(f" {i}. {class_info.get('className', class_info.get('class_name', '未知'))} "
|
|
f"({class_info.get('collegeName', class_info.get('college_name', '未知'))})")
|
|
if len(classes) > 5:
|
|
print(f" ... 还有 {len(classes) - 5} 个班级")
|
|
|
|
# 3. 确认操作
|
|
confirm = input(f"\n确认为所有 {len(classes)} 个班级创建学生？(y/N): ").strip().lower()
|
|
if confirm not in ['y', 'yes']:
|
|
print("操作已取消")
|
|
return
|
|
|
|
# 4. 批量创建学生
|
|
start_time = time.time()
|
|
all_results = []
|
|
total_success = 0
|
|
total_failed = 0
|
|
|
|
for i, class_info in enumerate(classes, 1):
|
|
print(f"\n📚 处理第 {i}/{len(classes)} 个班级...")
|
|
try:
|
|
result = self.batch_create_students_for_class(class_info, student_count_per_class)
|
|
all_results.append(result)
|
|
total_success += result['success']
|
|
total_failed += result['failed']
|
|
time.sleep(0.5) # 班级间延迟
|
|
except Exception as e:
|
|
logger.error(f"处理班级时发生异常: {str(e)}")
|
|
total_failed += student_count_per_class
|
|
|
|
end_time = time.time()
|
|
|
|
# 5. 输出结果
|
|
print("\n" + "="*50)
|
|
print("🎉 批量创建学生完成！")
|
|
print(f"📊 总班级数: {len(classes)}")
|
|
print(f"👥 预计学生数: {len(classes) * student_count_per_class}")
|
|
print(f"✅ 成功创建: {total_success}")
|
|
print(f"❌ 创建失败: {total_failed}")
|
|
print(f"⏱️ 总耗时: {end_time - start_time:.2f} 秒")
|
|
print("="*50)
|
|
|
|
# 保存结果
|
|
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
|
|
result_file = f"student_creation_result_{timestamp}.json"
|
|
|
|
with open(result_file, 'w', encoding='utf-8') as f:
|
|
json.dump(all_results, f, ensure_ascii=False, indent=2)
|
|
|
|
print(f"📄 详细结果已保存到: {result_file}")
|
|
|
|
def main():
|
|
"""主函数"""
|
|
print("🏫 学生批量创建工具")
|
|
print("="*50)
|
|
|
|
# 配置参数
|
|
base_url = input("请输入系统地址 (默认: http://localhost): ").strip()
|
|
if not base_url:
|
|
base_url = "http://localhost"
|
|
|
|
username = input("请输入用户名 (默认: admin): ").strip()
|
|
if not username:
|
|
username = "admin"
|
|
|
|
password = input("请输入密码 (默认: admin123): ").strip()
|
|
if not password:
|
|
password = "admin123"
|
|
|
|
student_count = input("请输入每个班级创建的学生数量 (默认: 30): ").strip()
|
|
if not student_count:
|
|
student_count = 30
|
|
else:
|
|
try:
|
|
student_count = int(student_count)
|
|
except ValueError:
|
|
print("输入的学生数量无效，使用默认值 30")
|
|
student_count = 30
|
|
|
|
# 创建并运行
|
|
creator = LoginAndCreateStudents(base_url)
|
|
creator.run(username, password, student_count)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|