|
|
import json
|
|
|
from typing import List, Optional
|
|
|
from .student_dal import IStudentDAL
|
|
|
from ..model.student import Student
|
|
|
from datetime import datetime
|
|
|
|
|
|
|
|
|
class JsonStudentDAL(IStudentDAL):
|
|
|
"""学生信息的JSON文件存储实现，实现了IStudentDAL接口，提供基于JSON文件的学生数据增删改查功能"""
|
|
|
|
|
|
def __init__(self, file_path: str = 'students.json'):
|
|
|
"""初始化JSON数据访问层
|
|
|
:param file_path: JSON数据文件路径，默认使用当前目录下的students.json"""
|
|
|
self.file_path = file_path
|
|
|
self._ensure_file_exists() # 调用私有方法确保数据文件存在，避免后续操作报错
|
|
|
|
|
|
def _ensure_file_exists(self):
|
|
|
"""确保JSON数据文件存在且格式正确
|
|
|
处理两种异常情况：
|
|
|
1. 文件不存在时创建空JSON数组文件
|
|
|
2. 文件存在但非JSON格式时（如损坏）重建空文件"""
|
|
|
try:
|
|
|
with open(self.file_path, 'r', encoding='utf-8') as file:
|
|
|
json.load(file) # 尝试解析文件，验证是否为合法JSON
|
|
|
except (FileNotFoundError, json.JSONDecodeError):
|
|
|
# 异常处理：文件不存在或解析失败时，创建包含空数组的新文件
|
|
|
with open(self.file_path, 'w', encoding='utf-8') as file:
|
|
|
json.dump([], file) # 写入空数组，确保文件格式正确
|
|
|
|
|
|
def _student_to_dict(self, student: Student) -> dict:
|
|
|
"""将Student实体对象转换为JSON兼容的字典格式
|
|
|
:param student: 学生实体对象
|
|
|
:return: 可序列化的字典，日期类型转换为字符串
|
|
|
处理逻辑：
|
|
|
- 日期字段enrollment_date转为%Y-%m-%d格式字符串
|
|
|
- None值保持为None，确保JSON序列化合法"""
|
|
|
return {
|
|
|
'name': student.name, # 直接存储字符串属性
|
|
|
'id_card': student.id_card, # 身份证号（字符串）
|
|
|
'stu_id': student.stu_id, # 学号（字符串）
|
|
|
'gender': student.gender, # 性别（布尔值）
|
|
|
'height': student.height, # 身高（整数）
|
|
|
'weight': student.weight, # 体重（整数）
|
|
|
# 日期转字符串：若存在日期则格式化为%Y-%m-%d，否则保持None
|
|
|
'enrollment_date': str(student.enrollment_date) if student.enrollment_date else None,
|
|
|
'class_name': student.class_name, # 班级（字符串）
|
|
|
'major': student.major, # 专业（字符串）
|
|
|
'email': student.email, # 邮箱（字符串）
|
|
|
'phone': student.phone # 电话（字符串）
|
|
|
}
|
|
|
|
|
|
def _dict_to_student(self, data: dict) -> Student:
|
|
|
"""将JSON字典转换为Student实体对象
|
|
|
:param data: 从JSON文件读取的字典数据
|
|
|
:return: 初始化的Student对象，字符串日期转换为date对象
|
|
|
处理逻辑：
|
|
|
- enrollment_date字段从字符串解析为datetime.date对象
|
|
|
- 缺失字段使用None或默认值初始化"""
|
|
|
enrollment_date = data.get('enrollment_date')
|
|
|
if enrollment_date:
|
|
|
# 字符串转日期：使用strptime解析为datetime对象，再提取date部分
|
|
|
enrollment_date = datetime.strptime(enrollment_date, '%Y-%m-%d').date()
|
|
|
|
|
|
return Student(
|
|
|
name=data.get('name'), # 获取姓名，缺失时为None
|
|
|
id_card=data.get('id_card'), # 获取身份证号
|
|
|
stu_id=data.get('stu_id'), # 获取学号
|
|
|
gender=data.get('gender'), # 获取性别
|
|
|
height=data.get('height'), # 获取身高
|
|
|
weight=data.get('weight'), # 获取体重
|
|
|
enrollment_date=enrollment_date, # 转换后的入学日期
|
|
|
class_name=data.get('class_name'), # 获取班级
|
|
|
major=data.get('major'), # 获取专业
|
|
|
email=data.get('email'), # 获取邮箱
|
|
|
phone=data.get('phone') # 获取电话
|
|
|
)
|
|
|
|
|
|
def _load_data(self) -> List[dict]:
|
|
|
"""从JSON文件加载学生数据
|
|
|
:return: 包含学生字典的列表
|
|
|
封装文件读取逻辑，统一处理编码和JSON解析"""
|
|
|
with open(self.file_path, 'r', encoding='utf-8') as file:
|
|
|
return json.load(file) # 读取文件并解析为Python列表
|
|
|
|
|
|
def _save_data(self, data: List[dict]):
|
|
|
"""将学生数据保存至JSON文件
|
|
|
:param data: 包含学生字典的列表
|
|
|
保存逻辑：
|
|
|
- ensure_ascii=False确保中文正常存储
|
|
|
- indent=2添加缩进，提升文件可读性"""
|
|
|
with open(self.file_path, 'w', encoding='utf-8') as file:
|
|
|
json.dump(data, file, ensure_ascii=False, indent=2) # 写入文件，禁用ASCII转义
|
|
|
|
|
|
def get_by_id(self, id_card: str) -> Optional[Student]:
|
|
|
"""根据身份证号查询学生信息
|
|
|
:param id_card: 待查询的身份证号
|
|
|
:return: 匹配的Student对象或None
|
|
|
实现逻辑：
|
|
|
- 加载全部数据遍历匹配
|
|
|
- 找到后转换为Student对象返回"""
|
|
|
for data in self._load_data():
|
|
|
if data.get('id_card') == id_card: # 精确匹配身份证号
|
|
|
return self._dict_to_student(data) # 转换为实体对象
|
|
|
return None # 未找到返回None
|
|
|
|
|
|
def get_by_stu_id(self, stu_id: str) -> Optional[Student]:
|
|
|
"""根据学号查询学生信息，逻辑同get_by_id
|
|
|
:param stu_id: 待查询的学号
|
|
|
:return: 匹配的Student对象或None"""
|
|
|
for data in self._load_data():
|
|
|
if data.get('stu_id') == stu_id: # 精确匹配学号
|
|
|
return self._dict_to_student(data)
|
|
|
return None
|
|
|
|
|
|
def get_all(self) -> List[Student]:
|
|
|
"""获取所有学生信息
|
|
|
:return: 包含所有Student对象的列表
|
|
|
实现逻辑：
|
|
|
- 加载全部字典数据
|
|
|
- 逐个转换为Student对象"""
|
|
|
return [self._dict_to_student(data) for data in self._load_data()] # 列表推导式批量转换
|
|
|
|
|
|
def add(self, student: Student) -> bool:
|
|
|
"""添加新学生信息
|
|
|
:param student: 待添加的Student对象
|
|
|
:return: 添加成功返回True，失败返回False
|
|
|
业务逻辑：
|
|
|
1. 先校验身份证号和学号的唯一性
|
|
|
2. 校验通过后追加数据并保存
|
|
|
3. 唯一性冲突时返回False"""
|
|
|
# 检查身份证号或学号是否已存在
|
|
|
if self.get_by_id(student.id_card) or self.get_by_stu_id(student.stu_id):
|
|
|
return False # 已存在相同标识，添加失败
|
|
|
|
|
|
data = self._load_data() # 加载当前数据
|
|
|
data.append(self._student_to_dict(student)) # 转换并追加
|
|
|
self._save_data(data) # 保存更新后的数据
|
|
|
return True # 添加成功
|
|
|
|
|
|
def update(self, student: Student) -> bool:
|
|
|
"""更新学生信息
|
|
|
:param student: 包含更新后信息的Student对象
|
|
|
:return: 更新成功返回True，失败返回False
|
|
|
实现逻辑：
|
|
|
1. 按身份证号定位记录
|
|
|
2. 找到后替换为新数据
|
|
|
3. 未找到或无变更时返回False"""
|
|
|
data = self._load_data() # 加载当前数据
|
|
|
updated = False # 标记是否更新
|
|
|
|
|
|
for i, item in enumerate(data):
|
|
|
if item.get('id_card') == student.id_card: # 匹配身份证号
|
|
|
data[i] = self._student_to_dict(student) # 替换为新数据
|
|
|
updated = True
|
|
|
break
|
|
|
|
|
|
if updated:
|
|
|
self._save_data(data) # 保存更新后的数据
|
|
|
return updated # 返回是否更新成功
|
|
|
|
|
|
def delete_by_id(self, id_card: str) -> bool:
|
|
|
"""根据身份证号删除学生信息
|
|
|
:param id_card: 待删除学生的身份证号
|
|
|
:return: 删除成功返回True，失败返回False
|
|
|
实现逻辑：
|
|
|
1. 过滤掉目标身份证号的记录
|
|
|
2. 比较删除前后数据长度
|
|
|
3. 长度变化时保存并返回True"""
|
|
|
data = self._load_data() # 加载当前数据
|
|
|
original_length = len(data) # 记录原始长度
|
|
|
|
|
|
# 过滤掉目标身份证号的记录
|
|
|
data = [item for item in data if item.get('id_card') != id_card]
|
|
|
|
|
|
if len(data) < original_length: # 长度减少表示删除成功
|
|
|
self._save_data(data) # 保存更新后的数据
|
|
|
return True
|
|
|
return False # 未找到目标记录
|
|
|
|
|
|
def delete_by_stu_id(self, stu_id: str) -> bool:
|
|
|
"""根据学号删除学生信息，逻辑同delete_by_id
|
|
|
:param stu_id: 待删除学生的学号
|
|
|
:return: 删除成功返回True，失败返回False"""
|
|
|
data = self._load_data()
|
|
|
original_length = len(data)
|
|
|
|
|
|
data = [item for item in data if item.get('stu_id') != stu_id]
|
|
|
|
|
|
if len(data) < original_length:
|
|
|
self._save_data(data)
|
|
|
return True
|
|
|
return False
|
|
|
|
|
|
def search_by_name(self, name: str) -> List[Student]:
|
|
|
"""根据姓名模糊查询学生
|
|
|
:param name: 待查询的姓名片段
|
|
|
:return: 包含匹配学生的列表
|
|
|
实现逻辑：
|
|
|
- 遍历所有数据
|
|
|
- 使用in操作符实现包含匹配
|
|
|
- 处理name字段为None的情况"""
|
|
|
return [
|
|
|
self._dict_to_student(data) for data in self._load_data()
|
|
|
if name in data.get('name', '') # data.get('name', '')避免KeyError，空字符串确保in操作合法
|
|
|
]
|
|
|
|
|
|
def search_by_class(self, class_name: str) -> List[Student]:
|
|
|
"""根据班级模糊查询学生
|
|
|
:param class_name: 待查询的班级片段
|
|
|
:return: 包含匹配学生的列表
|
|
|
实现逻辑：
|
|
|
- 处理class_name为None的情况（替换为空字符串）
|
|
|
- 使用in操作符实现包含匹配"""
|
|
|
return [
|
|
|
self._dict_to_student(data) for data in self._load_data()
|
|
|
if class_name in (data.get('class_name') or '') # data.get('class_name')为None时替换为空字符串
|
|
|
]
|
|
|
|
|
|
def search_by_major(self, major: str) -> List[Student]:
|
|
|
"""根据专业模糊查询学生，逻辑同search_by_class
|
|
|
:param major: 待查询的专业片段
|
|
|
:return: 包含匹配学生的列表"""
|
|
|
return [
|
|
|
self._dict_to_student(data) for data in self._load_data()
|
|
|
if major in (data.get('major') or '') # 处理major为None的情况
|
|
|
]
|
