289 lines
9.0 KiB
Python
289 lines
9.0 KiB
Python
from requests import get, post
|
|
from faker import Faker
|
|
import random
|
|
import time
|
|
import uuid
|
|
import json
|
|
from datetime import datetime, timedelta
|
|
|
|
fake = Faker('zh_CN')
|
|
|
|
ip = "http://band.hxzhxy.cn"
|
|
param = {
|
|
"username": "xtl",
|
|
"password": "xia123456",
|
|
"client_id": "client",
|
|
"grant_type": "password",
|
|
"client_secret": "123456",
|
|
}
|
|
|
|
def get_token():
|
|
data = get(url=f"{ip}/oauth/token", params=param).json()
|
|
return data["access_token"]
|
|
|
|
token = get_token()
|
|
header = {
|
|
"content-type": "application/json",
|
|
"authorization": f"Bearer {token}",
|
|
}
|
|
teamId = 106
|
|
|
|
def generate_leave_data(stu_id, stu_name, stu_no, grade_name, class_name, grade_id, class_id, phone=None):
|
|
leave_types = ["事假", "病假"]
|
|
|
|
leave_type = random.choice(leave_types)
|
|
is_cycle = random.choice([1, 2, 3])
|
|
|
|
begin_date = datetime.now() - timedelta(days=random.randint(0,15))
|
|
end_date = begin_date + timedelta(days=random.randint(0, 120))
|
|
|
|
begin_section = random.randint(1, 6)
|
|
end_section = random.randint(begin_section, 12)
|
|
|
|
leave_reasons = [
|
|
"身体不适，需要休息",
|
|
"家中有事需要处理",
|
|
"去医院检查身体",
|
|
"发烧感冒，需要就医",
|
|
"参加重要家庭活动",
|
|
"个人事务需要处理",
|
|
"肠胃不适，需要就医",
|
|
"头痛发热，请假休息",
|
|
"参加校外培训活动",
|
|
"家庭紧急事务",
|
|
]
|
|
|
|
disease_date = begin_date.strftime("%Y-%m-%d")
|
|
|
|
if is_cycle == 3:
|
|
week_days = ["周一", "周二", "周三", "周四", "周五"]
|
|
selected_day = random.choice(week_days)
|
|
sections = random.sample(range(1, 13), random.randint(2, 4))
|
|
sections.sort()
|
|
week_info = json.dumps({selected_day: ",".join(str(s) for s in sections)})
|
|
end_date = begin_date
|
|
else:
|
|
week_info = None
|
|
|
|
data = {
|
|
"stuId": str(stu_id),
|
|
"leaveType": leave_type,
|
|
"isCycle": is_cycle,
|
|
"leaveReason": random.choice(leave_reasons),
|
|
"beginDate": begin_date.strftime("%Y-%m-%d"),
|
|
"endDate": end_date.strftime("%Y-%m-%d"),
|
|
"beginSection": str(begin_section),
|
|
"endSection": str(end_section),
|
|
"diseaseDate": disease_date,
|
|
"symptom": "",
|
|
"fever": "",
|
|
"hospital": "",
|
|
"treatmentDate": "",
|
|
"imgUrl": "[]",
|
|
"weekInfo": week_info,
|
|
}
|
|
|
|
return data
|
|
|
|
def get_grade_list():
|
|
url = f"{ip}/uc/grade/listSelect/{teamId}?access_token={token}"
|
|
response = get(url, headers=header)
|
|
result = response.json()
|
|
if result.get('code') == 0:
|
|
return result.get('data', [])
|
|
return []
|
|
|
|
def get_class_list(grade_id):
|
|
url = f"{ip}/uc/class/listClass"
|
|
params = {
|
|
"access_token": token,
|
|
"teamId": teamId,
|
|
"gradeId": grade_id
|
|
}
|
|
response = get(url, params=params, headers=header)
|
|
result = response.json()
|
|
if result.get('code') == 0:
|
|
return result.get('data', [])
|
|
return []
|
|
|
|
def get_students_from_db(grade_id=None, class_id=None):
|
|
import pymysql
|
|
conn = pymysql.connect(
|
|
host='192.168.0.244',
|
|
port=3306,
|
|
user='root',
|
|
password='Abcd@123456',
|
|
database='school_server',
|
|
charset='utf8'
|
|
)
|
|
cursor = conn.cursor()
|
|
|
|
sql = """
|
|
SELECT s.id, s.stu_no, s.stu_name, s.phone,
|
|
g.grade_name, c.class_name
|
|
FROM student s
|
|
LEFT JOIN grade g ON s.grade_id = g.id
|
|
LEFT JOIN tclass c ON s.class_id = c.id
|
|
WHERE s.team_id = %s
|
|
"""
|
|
params = [str(teamId)]
|
|
|
|
if grade_id:
|
|
sql += " AND s.grade_id = %s"
|
|
params.append(grade_id)
|
|
if class_id:
|
|
sql += " AND s.class_id = %s"
|
|
params.append(class_id)
|
|
|
|
sql += " ORDER BY RAND() LIMIT 100"
|
|
|
|
cursor.execute(sql, params)
|
|
students = cursor.fetchall()
|
|
conn.close()
|
|
return students
|
|
|
|
def add_leave_record(leave_data):
|
|
url = f"{ip}/studentLeave/add"
|
|
leave_data["access_token"] = token
|
|
response = post(url, json=leave_data, headers=header)
|
|
return response.json()
|
|
|
|
def batch_add_leaves(count=10):
|
|
print(f"正在获取年级列表...")
|
|
grades = get_grade_list()
|
|
if not grades:
|
|
print("未获取到年级数据，请检查 teamId 或 token")
|
|
return
|
|
|
|
grade = random.choice(grades)
|
|
grade_id = grade.get('id', grade.get('gradeId'))
|
|
grade_name = grade.get('gradeName', grade.get('name', ''))
|
|
print(f"随机选择年级: {grade_name}")
|
|
|
|
print(f"正在获取 [{grade_name}] 的班级列表...")
|
|
classes = get_class_list(grade_id)
|
|
if not classes:
|
|
print("未获取到班级数据")
|
|
return
|
|
|
|
cls = random.choice(classes)
|
|
class_id = cls.get('id', cls.get('classId'))
|
|
class_name = cls.get('className', cls.get('name', ''))
|
|
print(f"随机选择班级: {class_name}")
|
|
|
|
print(f"\n正在从数据库获取 [{grade_name} {class_name}] 的学生数据...")
|
|
students = get_students_from_db(grade_id, class_id)
|
|
print(f"获取到 {len(students)} 名学生")
|
|
|
|
if not students:
|
|
print("未获取到学生数据，请检查数据库连接")
|
|
return
|
|
|
|
success_count = 0
|
|
fail_count = 0
|
|
|
|
for i in range(min(count, len(students))):
|
|
stu = students[i]
|
|
stu_id, stu_no, stu_name, phone, db_grade_name, db_class_name = stu
|
|
|
|
leave_data = generate_leave_data(
|
|
stu_id, stu_name, stu_no,
|
|
grade_name, class_name, grade_id, class_id, phone
|
|
)
|
|
|
|
print(f"\n[{i+1}/{min(count, len(students))}] 为学生 {stu_name}({stu_no}) 添加请假记录...")
|
|
print(f" 请假类型: {leave_data['leaveType']}")
|
|
print(f" 请假时间: {leave_data['beginDate']} ~ {leave_data['endDate']}")
|
|
print(f" 请假节次: 第{leave_data['beginSection']}节 ~ 第{leave_data['endSection']}节")
|
|
print(f" 请假事由: {leave_data['leaveReason']}")
|
|
|
|
try:
|
|
result = add_leave_record(leave_data)
|
|
if result.get('code') == 0:
|
|
print(f" ✓ 添加成功")
|
|
success_count += 1
|
|
else:
|
|
print(f" ✗ 添加失败: {result.get('msg', '未知错误')}")
|
|
fail_count += 1
|
|
except Exception as e:
|
|
print(f" ✗ 请求异常: {e}")
|
|
fail_count += 1
|
|
|
|
time.sleep(0.5)
|
|
|
|
print(f"\n{'='*50}")
|
|
print(f"批量添加完成!")
|
|
print(f"成功: {success_count} 条")
|
|
print(f"失败: {fail_count} 条")
|
|
|
|
def generate_fake_students(count=20):
|
|
students = []
|
|
for i in range(count):
|
|
stu = {
|
|
"id": i + 1,
|
|
"stu_no": f"2024{str(i+1).zfill(4)}",
|
|
"stu_name": fake.name(),
|
|
"phone": fake.phone_number(),
|
|
"grade_name": random.choice(["初一", "初二", "初三", "高一", "高二", "高三"]),
|
|
"class_name": f"{random.randint(1, 10)}班",
|
|
}
|
|
students.append(stu)
|
|
return students
|
|
|
|
def batch_add_leaves_with_fake_students(count=10):
|
|
print(f"使用 Faker 生成 {count} 条请假记录...")
|
|
|
|
students = generate_fake_students(count)
|
|
success_count = 0
|
|
fail_count = 0
|
|
|
|
for i, stu in enumerate(students):
|
|
leave_data = generate_leave_data(
|
|
stu['id'], stu['stu_name'], stu['stu_no'],
|
|
stu['grade_name'], stu['class_name'],
|
|
None, None, stu['phone']
|
|
)
|
|
|
|
print(f"\n[{i+1}/{count}] 为学生 {stu['stu_name']}({stu['stu_no']}) 添加请假记录...")
|
|
print(f" 请假类型: {leave_data['leaveType']}")
|
|
print(f" 请假时间: {leave_data['beginDate']} ~ {leave_data['endDate']}")
|
|
print(f" 请假节次: 第{leave_data['beginSection']}节 ~ 第{leave_data['endSection']}节")
|
|
print(f" 请假事由: {leave_data['leaveReason']}")
|
|
|
|
try:
|
|
result = add_leave_record(leave_data)
|
|
if result.get('code') == 0:
|
|
print(f" ✓ 添加成功")
|
|
success_count += 1
|
|
else:
|
|
print(f" ✗ 添加失败: {result.get('msg', '未知错误')}")
|
|
fail_count += 1
|
|
except Exception as e:
|
|
print(f" ✗ 请求异常: {e}")
|
|
fail_count += 1
|
|
|
|
time.sleep(0.5)
|
|
|
|
print(f"\n{'='*50}")
|
|
print(f"批量添加完成!")
|
|
print(f"成功: {success_count} 条")
|
|
print(f"失败: {fail_count} 条")
|
|
|
|
if __name__ == "__main__":
|
|
print("="*50)
|
|
print("批量添加学生请假信息")
|
|
print("="*50)
|
|
print("1. 从数据库获取学生并添加请假记录")
|
|
print("2. 使用 Faker 生成虚拟学生并添加请假记录")
|
|
print("="*50)
|
|
|
|
choice = input("请选择模式 (1/2): ").strip()
|
|
count = int(input("请输入要生成的请假记录数量: ").strip() or "10")
|
|
|
|
if choice == "1":
|
|
batch_add_leaves(count)
|
|
elif choice == "2":
|
|
batch_add_leaves_with_fake_students(count)
|
|
else:
|
|
print("无效选择")
|