all-project/test/batch_create_students.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
批量创建学生信息脚本
为每个班级创建30个学生的测试数据
"""

import requests
import json
import random
import time
from datetime import datetime, timedelta
import logging

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class SimpleDataGenerator:
    """简单的数据生成器，替代faker"""
    
    def __init__(self):
        # 姓氏
        self.surnames = ['李', '王', '张', '刘', '陈', '杨', '黄', '赵', '周', '吴', 
                        '徐', '孙', '朱', '马', '胡', '郭', '林', '何', '高', '梁']
        
        # 男性名字
        self.male_names = ['伟', '强', '磊', '军', '勇', '涛', '明', '超', '辉', '华',
                          '建', '文', '健', '云', '峰', '浩', '鹏', '飞', '宇', '杰']
        
        # 女性名字
        self.female_names = ['丽', '娜', '敏', '静', '佳', '雅', '莉', '萍', '红', '梅',
                           '芳', '燕', '玲', '霞', '琳', '晶', '欣', '宁', '蕾', '洁']
        
        # 省份
        self.provinces = ['北京市', '上海市', '天津市', '重庆市', '河北省', '山西省', '辽宁省', 
                         '吉林省', '黑龙江省', '江苏省', '浙江省', '安徽省', '福建省', '江西省',
                         '山东省', '河南省', '湖北省', '湖南省', '广东省', '海南省', '四川省',
                         '贵州省', '云南省', '陕西省', '甘肃省', '青海省', '台湾省', '内蒙古自治区',
                         '广西壮族自治区', '西藏自治区', '宁夏回族自治区', '新疆维吾尔自治区']
    
    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)
        
        # 有时候生成双字名
        if random.random() < 0.3:
            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',
                   '150', '151', '152', '153', '155', '156', '157', '158', '159',
                   '180', '181', '182', '183', '184', '185', '186', '187', '188', '189']
        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', '110105', '110106', '110107', '110108',
                     '310101', '310104', '310105', '310106', '310107', '310109',
                     '440101', '440103', '440104', '440105', '440106', '440111']
        
        area_code = random.choice(area_codes)
        
        # 出生日期（YYYYMMDD格式）
        birth_str = birth_date.replace('-', '')
        
        # 顺序码（3位）+ 校验码（1位）
        sequence = str(random.randint(100, 999))
        check_digit = str(random.randint(0, 9))
        
        return area_code + birth_str + sequence + check_digit
    
    def generate_address(self):
        """生成地址"""
        province = random.choice(self.provinces)
        cities = ['市', '县']
        city = f"某{random.choice(cities)}"
        district = f"某区某街道{random.randint(1, 999)}号"
        return province + city + district
    
    def generate_province(self):
        """生成省份"""
        return random.choice(self.provinces)

class StudentBatchCreator:
    def __init__(self, base_url="http://localhost:8080", token=None):
        """
        初始化学生批量创建器
        
        Args:
            base_url: 后端API基础URL
            token: 认证token
        """
        self.base_url = base_url
        self.token = token
        self.session = requests.Session()
        self.data_gen = SimpleDataGenerator()
        
        # 设置请求头
        if token:
            self.session.headers.update({
                'Authorization': f'Bearer {token}',
                'Content-Type': 'application/json'
            })
    
    def set_token(self, token):
        """设置认证token"""
        self.token = token
        self.session.headers.update({
            'Authorization': f'Bearer {token}'
        })
    
    def get_class_list(self):
        """获取班级列表"""
        try:
            url = f"{self.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_birth_date(self):
        """生成出生日期（18-22岁）"""
        today = datetime.now()
        min_age = 18
        max_age = 22
        
        # 计算出生年份范围
        birth_year = today.year - random.randint(min_age, max_age)
        birth_month = random.randint(1, 12)
        birth_day = random.randint(1, 28)  # 使用28避免月份天数问题
        
        return f"{birth_year:04d}-{birth_month:02d}-{birth_day:02d}"
    
    def generate_student_data(self, class_info, student_index):
        """
        生成单个学生的测试数据
        
        Args:
            class_info: 班级信息
            student_index: 学生编号（1-30）
        
        Returns:
            dict: 学生数据
        """
        # 生成基础信息
        gender = random.choice(['0', '1'])  # 0-男，1-女
        name = self.data_gen.generate_name(gender)
        
        # 生成学号：班级代码 + 学生编号
        class_code = class_info.get('classCode', class_info.get('class_code', ''))
        if not class_code:
            # 如果没有班级代码，使用班级ID
            class_code = f"CLS{class_info.get('id', '')}"
        
        student_no = f"{class_code}{student_index:02d}"
        
        # 生成出生日期
        birth_date = self.generate_birth_date()
        
        # 生成身份证号
        id_card = self.data_gen.generate_id_card(birth_date)
        
        # 生成手机号
        phone = self.data_gen.generate_phone()
        
        # 民族选择
        nations = ['01', '02', '03', '04', '05']  # 汉族、蒙古族、回族、藏族、维吾尔族等
        nationality = random.choice(nations)
        
        # 政治面貌选择
        political_statuses = ['01', '02', '03', '13']  # 群众、团员、党员等
        political_status = random.choice(political_statuses)
        
        # 构建学生数据
        student_data = {
            # 基础信息
            "studentNo": student_no,
            "idCard": id_card,
            "name": name,
            "gender": gender,
            "birthDate": birth_date,
            "nationality": nationality,
            "politicalStatus": political_status,
            "phone": 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": self.data_gen.generate_province(),
            "sourceRegion": self.data_gen.generate_province(),
            "homeAddress": self.data_gen.generate_address(),
            "currentAddress": "学校宿舍",
            
            # 家庭信息
            "fatherName": self.data_gen.generate_name('0'),
            "fatherPhone": self.data_gen.generate_phone(),
            "motherName": self.data_gen.generate_name('1'),
            "motherPhone": self.data_gen.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": 1 if nationality != '01' else 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):
        """
        创建单个学生
        
        Args:
            student_data: 学生数据
        
        Returns:
            tuple: (success, result)
        """
        try:
            url = f"{self.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):
        """
        为指定班级批量创建学生
        
        Args:
            class_info: 班级信息
            student_count: 要创建的学生数量
        
        Returns:
            dict: 创建结果统计
        """
        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)}")
                failed_students.append({
                    'student_no': f"未知_{i}",
                    'name': '未知',
                    'error': 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 batch_create_students_for_all_classes(self, student_count_per_class=30):
        """
        为所有班级批量创建学生
        
        Args:
            student_count_per_class: 每个班级创建的学生数量
        
        Returns:
            list: 所有班级的创建结果
        """
        # 获取班级列表
        classes = self.get_class_list()
        if not classes:
            logger.error("没有获取到班级信息，无法创建学生")
            return []
        
        logger.info(f"开始为 {len(classes)} 个班级批量创建学生，每个班级 {student_count_per_class} 个学生")
        
        all_results = []
        total_success = 0
        total_failed = 0
        
        for class_info in 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"处理班级 {class_info} 时发生异常: {str(e)}")
                all_results.append({
                    'class_name': f"班级{class_info.get('id', '未知')}",
                    'total': student_count_per_class,
                    'success': 0,
                    'failed': student_count_per_class,
                    'error': str(e)
                })
                total_failed += student_count_per_class
        
        # 输出总结
        logger.info("="*50)
        logger.info("批量创建学生完成！")
        logger.info(f"总班级数: {len(classes)}")
        logger.info(f"总学生数: {len(classes) * student_count_per_class}")
        logger.info(f"成功创建: {total_success}")
        logger.info(f"创建失败: {total_failed}")
        logger.info("="*50)
        
        return all_results

def main():
    """主函数"""
    print("学生批量创建工具")
    print("="*50)
    
    # 配置参数
    base_url = input("请输入后端API地址 (默认: http://localhost:8080): ").strip()
    if not base_url:
        base_url = "http://localhost:8080"
    
    token = input("请输入认证Token (可为空): ").strip()
    
    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 = StudentBatchCreator(base_url, token)
    
    # 确认操作
    print(f"\n配置信息:")
    print(f"API地址: {base_url}")
    print(f"每个班级学生数: {student_count}")
    
    confirm = input("\n确认开始批量创建学生？(y/N): ").strip().lower()
    if confirm not in ['y', 'yes']:
        print("操作已取消")
        return
    
    # 开始批量创建
    start_time = time.time()
    results = creator.batch_create_students_for_all_classes(student_count)
    end_time = time.time()
    
    # 保存结果到文件
    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(results, f, ensure_ascii=False, indent=2)
    
    print(f"\n创建结果已保存到: {result_file}")
    print(f"总耗时: {end_time - start_time:.2f} 秒")

if __name__ == "__main__":
    main()