all-project/test/login_and_create_students.py

#!/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()