abc/student/dal/sqlite_student_dal.py

import sqlite3
from typing import List, Optional
from .student_dal import IStudentDAL
from ..model.student import Student
from datetime import datetime


class SQLiteStudentDAL(IStudentDAL):
    """学生信息的SQLite数据库存储实现"""

    def __init__(self, db_path: str = 'students.db'):
        self.db_path = db_path
        self._create_table()

    def _create_table(self):
        """创建学生表（如果不存在）"""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute('''
            CREATE TABLE IF NOT EXISTS students (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT NOT NULL,
                id_card TEXT UNIQUE NOT NULL,
                stu_id TEXT UNIQUE NOT NULL,
                gender INTEGER,
                height INTEGER,
                weight REAL,
                enrollment_date TEXT,
                class_name TEXT,
                major TEXT,
                email TEXT,
                phone TEXT
            )
            ''')
            conn.commit()

    def _row_to_student(self, row: tuple) -> Optional[Student]:
        """将数据库行转换为Student对象"""
        if not row:
            return None

        # 处理日期类型
        enrollment_date = row[7]
        if enrollment_date:
            enrollment_date = datetime.strptime(enrollment_date, '%Y-%m-%d').date()

        # 处理布尔类型
        gender = row[3]
        if gender is not None:
            gender = bool(gender)

        return Student(
            name=row[1],
            id_card=row[2],
            stu_id=row[3],
            gender=gender,
            height=row[4],
            weight=row[5],
            enrollment_date=enrollment_date,
            class_name=row[8],
            major=row[9],
            email=row[10],
            phone=row[11]
        )

    def get_by_id(self, id_card: str) -> Optional[Student]:
        """根据身份证号获取学生信息"""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT * FROM students WHERE id_card = ?", (id_card,))
            row = cursor.fetchone()
            return self._row_to_student(row)

    def get_by_stu_id(self, stu_id: str) -> Optional[Student]:
        """根据学号获取学生信息"""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT * FROM students WHERE stu_id = ?", (stu_id,))
            row = cursor.fetchone()
            return self._row_to_student(row)

    def get_all(self) -> List[Student]:
        """获取所有学生信息"""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT * FROM students")
            rows = cursor.fetchall()
            return [self._row_to_student(row) for row in rows]

    def add(self, student: Student) -> bool:
        """添加学生信息"""
        try:
            with sqlite3.connect(self.db_path) as conn:
                cursor = conn.cursor()
                cursor.execute(
                    """INSERT INTO students 
                    (name, id_card, stu_id, gender, height, weight, enrollment_date, class_name, major, email, phone)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
                    (
                        student.name,
                        student.id_card,
                        student.stu_id,
                        student.gender,
                        student.height,
                        student.weight,
                        str(student.enrollment_date) if student.enrollment_date else None,
                        student.class_name,
                        student.major,
                        student.email,
                        student.phone
                    )
                )
                conn.commit()
            return True
        except sqlite3.IntegrityError:  # 违反唯一约束
            return False

    def update(self, student: Student) -> bool:
        """更新学生信息"""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute(
                """UPDATE students 
                SET name=?, gender=?, height=?, weight=?, enrollment_date=?, 
                class_name=?, major=?, email=?, phone=?
                WHERE id_card=?""",
                (
                    student.name,
                    student.gender,
                    student.height,
                    student.weight,
                    str(student.enrollment_date) if student.enrollment_date else None,
                    student.class_name,
                    student.major,
                    student.email,
                    student.phone,
                    student.id_card
                )
            )
            conn.commit()
            return cursor.rowcount > 0

    def delete_by_id(self, id_card: str) -> bool:
        """根据身份证号删除学生信息"""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute("DELETE FROM students WHERE id_card = ?", (id_card,))
            conn.commit()
            return cursor.rowcount > 0

    def delete_by_stu_id(self, stu_id: str) -> bool:
        """根据学号删除学生信息"""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute("DELETE FROM students WHERE stu_id = ?", (stu_id,))
            conn.commit()
            return cursor.rowcount > 0

    def search_by_name(self, name: str) -> List[Student]:
        """根据姓名模糊查询学生信息"""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT * FROM students WHERE name LIKE ?", ('%' + name + '%',))
            rows = cursor.fetchall()
            return [self._row_to_student(row) for row in rows]

    def search_by_class(self, class_name: str) -> List[Student]:
        """根据班级模糊查询学生信息"""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT * FROM students WHERE class_name LIKE ?", ('%' + class_name + '%',))
            rows = cursor.fetchall()
            return [self._row_to_student(row) for row in rows]

    def search_by_major(self, major: str) -> List[Student]:
        """根据专业模糊查询学生信息"""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT * FROM students WHERE major LIKE ?", ('%' + major + '%',))
            rows = cursor.fetchall()
            return [self._row_to_student(row) for row in rows]