引言:数据管理的完美结合

在实际应用中,我们经常需要处理多种数据格式和存储方式。CSV文件因其简单通用而成为数据交换的常见格式,而SQLite数据库则提供了强大的数据管理和查询能力。本项目将结合两者的优势,构建一个功能完整的学生成绩管理系统,展示如何在实际项目中综合运用文件处理和数据库技术。

本章将通过一个完整的实战项目,演示如何设计、实现一个基于CSVSQLite的学生成绩管理系统,涵盖数据导入导出、CRUD操作、数据统计和报表生成等功能。


第一部分:系统设计与规划

1.1 系统需求分析

学生成绩管理系统需要满足以下核心功能:

  1. 学生管理:添加、删除、修改、查询学生信息
  2. 课程管理:管理课程信息,包括课程名称、学分等
  3. 成绩管理:录入、修改、查询学生成绩
  4. 数据导入导出:从CSV文件导入数据,导出数据到CSV文件
  5. 统计分析:计算平均分、排名、成绩分布等统计信息
  6. 报表生成:生成成绩单、统计报表等

1.2 数据库设计

系统使用SQLite数据库,设计以下三张表:

-- 学生表
CREATE TABLE students (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    student_id TEXT UNIQUE NOT NULL,  -- 学号
    name TEXT NOT NULL,               -- 姓名
    gender TEXT,                      -- 性别
    class TEXT,                       -- 班级
    major TEXT                        -- 专业
);

-- 课程表
CREATE TABLE courses (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    course_id TEXT UNIQUE NOT NULL,   -- 课程编号
    name TEXT NOT NULL,               -- 课程名称
    credit REAL DEFAULT 0,            -- 学分
    teacher TEXT                      -- 授课教师
);

-- 成绩表
CREATE TABLE scores (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    student_id TEXT NOT NULL,         -- 学号
    course_id TEXT NOT NULL,          -- 课程编号
    score REAL,                       -- 成绩
    semester TEXT,                    -- 学期
    FOREIGN KEY (student_id) REFERENCES students (student_id),
    FOREIGN KEY (course_id) REFERENCES courses (course_id),
    UNIQUE(student_id, course_id, semester)  -- 防止重复录入
);

1.3 系统架构设计

系统采用分层架构:

  • 数据访问层:负责数据库操作和CSV文件处理
  • 业务逻辑层:实现核心业务功能
  • 表示层:提供命令行或图形用户界面

第二部分:核心代码实现

2.1 数据库初始化与连接

# database.py
import sqlite3
import os
from typing import List, Dict, Any, Optional

class DatabaseManager:
    def __init__(self, db_path: str = "student_management.db"):
        self.db_path = db_path
        self.init_database()
    
    def init_database(self):
        """初始化数据库表结构"""
        if not os.path.exists(self.db_path):
            print("创建新的数据库...")
        
        conn = self.get_connection()
        cursor = conn.cursor()
        
        # 创建学生表
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS students (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                student_id TEXT UNIQUE NOT NULL,
                name TEXT NOT NULL,
                gender TEXT,
                class TEXT,
                major TEXT
            )
        ''')
        
        # 创建课程表
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS courses (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                course_id TEXT UNIQUE NOT NULL,
                name TEXT NOT NULL,
                credit REAL DEFAULT 0,
                teacher TEXT
            )
        ''')
        
        # 创建成绩表
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS scores (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                student_id TEXT NOT NULL,
                course_id TEXT NOT NULL,
                score REAL,
                semester TEXT,
                FOREIGN KEY (student_id) REFERENCES students (student_id),
                FOREIGN KEY (course_id) REFERENCES courses (course_id),
                UNIQUE(student_id, course_id, semester)
            )
        ''')
        
        # 创建索引以提高查询性能
        cursor.execute('CREATE INDEX IF NOT EXISTS idx_students_id ON students(student_id)')
        cursor.execute('CREATE INDEX IF NOT EXISTS idx_courses_id ON courses(course_id)')
        cursor.execute('CREATE INDEX IF NOT EXISTS idx_scores_student ON scores(student_id)')
        cursor.execute('CREATE INDEX IF NOT EXISTS idx_scores_course ON scores(course_id)')
        cursor.execute('CREATE INDEX IF NOT EXISTS idx_scores_semester ON scores(semester)')
        
        conn.commit()
        conn.close()
        print("数据库初始化完成")
    
    def get_connection(self):
        """获取数据库连接"""
        return sqlite3.connect(self.db_path)
    
    def execute_query(self, query: str, params: tuple = None, fetch: bool = False):
        """执行SQL查询"""
        conn = self.get_connection()
        cursor = conn.cursor()
        
        try:
            if params:
                cursor.execute(query, params)
            else:
                cursor.execute(query)
            
            if fetch:
                result = cursor.fetchall()
            else:
                result = None
            
            conn.commit()
            return result
        
        except sqlite3.Error as e:
            conn.rollback()
            raise e
        
        finally:
            conn.close()
    
    def execute_many(self, query: str, params_list: List[tuple]):
        """批量执行SQL语句"""
        conn = self.get_connection()
        cursor = conn.cursor()
        
        try:
            cursor.executemany(query, params_list)
            conn.commit()
        
        except sqlite3.Error as e:
            conn.rollback()
            raise e
        
        finally:
            conn.close()

2.2 学生管理模块

# student_manager.py
from database import DatabaseManager
from typing import List, Dict, Any, Optional

class StudentManager:
    def __init__(self, db_manager: DatabaseManager):
        self.db = db_manager
    
    def add_student(self, student_id: str, name: str, gender: str = None, 
                   class_name: str = None, major: str = None) -> bool:
        """添加学生"""
        try:
            query = '''
                INSERT INTO students (student_id, name, gender, class, major)
                VALUES (?, ?, ?, ?, ?)
            '''
            self.db.execute_query(query, (student_id, name, gender, class_name, major))
            print(f"学生 {name} ({student_id}) 添加成功")
            return True
        except sqlite3.IntegrityError:
            print(f"学生ID {student_id} 已存在")
            return False
        except Exception as e:
            print(f"添加学生时发生错误: {e}")
            return False
    
    def get_student(self, student_id: str) -> Optional[Dict[str, Any]]:
        """获取学生信息"""
        try:
            query = 'SELECT * FROM students WHERE student_id = ?'
            result = self.db.execute_query(query, (student_id,), fetch=True)
            
            if result:
                columns = ['id', 'student_id', 'name', 'gender', 'class', 'major']
                return dict(zip(columns, result[0]))
            else:
                return None
        except Exception as e:
            print(f"获取学生信息时发生错误: {e}")
            return None
    
    def update_student(self, student_id: str, **kwargs) -> bool:
        """更新学生信息"""
        if not kwargs:
            return False
        
        try:
            set_clause = ', '.join([f"{key} = ?" for key in kwargs.keys()])
            values = list(kwargs.values())
            values.append(student_id)
            
            query = f'UPDATE students SET {set_clause} WHERE student_id = ?'
            self.db.execute_query(query, tuple(values))
            
            print(f"学生 {student_id} 信息更新成功")
            return True
        except Exception as e:
            print(f"更新学生信息时发生错误: {e}")
            return False
    
    def delete_student(self, student_id: str) -> bool:
        """删除学生"""
        try:
            # 先删除相关的成绩记录
            self.db.execute_query('DELETE FROM scores WHERE student_id = ?', (student_id,))
            
            # 再删除学生记录
            query = 'DELETE FROM students WHERE student_id = ?'
            self.db.execute_query(query, (student_id,))
            
            print(f"学生 {student_id} 删除成功")
            return True
        except Exception as e:
            print(f"删除学生时发生错误: {e}")
            return False
    
    def get_all_students(self) -> List[Dict[str, Any]]:
        """获取所有学生"""
        try:
            query = 'SELECT * FROM students ORDER BY student_id'
            result = self.db.execute_query(query, fetch=True)
            
            columns = ['id', 'student_id', 'name', 'gender', 'class', 'major']
            return [dict(zip(columns, row)) for row in result]
        except Exception as e:
            print(f"获取学生列表时发生错误: {e}")
            return []
    
    def search_students(self, **kwargs) -> List[Dict[str, Any]]:
        """搜索学生"""
        if not kwargs:
            return self.get_all_students()
        
        try:
            conditions = []
            values = []
            
            for key, value in kwargs.items():
                if value:  # 只添加非空的搜索条件
                    conditions.append(f"{key} LIKE ?")
                    values.append(f"%{value}%")
            
            if not conditions:
                return self.get_all_students()
            
            query = f'SELECT * FROM students WHERE {" AND ".join(conditions)} ORDER BY student_id'
            result = self.db.execute_query(query, tuple(values), fetch=True)
            
            columns = ['id', 'student_id', 'name', 'gender', 'class', 'major']
            return [dict(zip(columns, row)) for row in result]
        except Exception as e:
            print(f"搜索学生时发生错误: {e}")
            return []

2.3 课程管理模块

# course_manager.py
from database import DatabaseManager
from typing import List, Dict, Any, Optional

class CourseManager:
    def __init__(self, db_manager: DatabaseManager):
        self.db = db_manager
    
    def add_course(self, course_id: str, name: str, credit: float = 0, teacher: str = None) -> bool:
        """添加课程"""
        try:
            query = '''
                INSERT INTO courses (course_id, name, credit, teacher)
                VALUES (?, ?, ?, ?)
            '''
            self.db.execute_query(query, (course_id, name, credit, teacher))
            print(f"课程 {name} ({course_id}) 添加成功")
            return True
        except sqlite3.IntegrityError:
            print(f"课程ID {course_id} 已存在")
            return False
        except Exception as e:
            print(f"添加课程时发生错误: {e}")
            return False
    
    def get_course(self, course_id: str) -> Optional[Dict[str, Any]]:
        """获取课程信息"""
        try:
            query = 'SELECT * FROM courses WHERE course_id = ?'
            result = self.db.execute_query(query, (course_id,), fetch=True)
            
            if result:
                columns = ['id', 'course_id', 'name', 'credit', 'teacher']
                return dict(zip(columns, result[0]))
            else:
                return None
        except Exception as e:
            print(f"获取课程信息时发生错误: {e}")
            return None
    
    def update_course(self, course_id: str, **kwargs) -> bool:
        """更新课程信息"""
        if not kwargs:
            return False
        
        try:
            set_clause = ', '.join([f"{key} = ?" for key in kwargs.keys()])
            values = list(kwargs.values())
            values.append(course_id)
            
            query = f'UPDATE courses SET {set_clause} WHERE course_id = ?'
            self.db.execute_query(query, tuple(values))
            
            print(f"课程 {course_id} 信息更新成功")
            return True
        except Exception as e:
            print(f"更新课程信息时发生错误: {e}")
            return False
    
    def delete_course(self, course_id: str) -> bool:
        """删除课程"""
        try:
            # 先删除相关的成绩记录
            self.db.execute_query('DELETE FROM scores WHERE course_id = ?', (course_id,))
            
            # 再删除课程记录
            query = 'DELETE FROM courses WHERE course_id = ?'
            self.db.execute_query(query, (course_id,))
            
            print(f"课程 {course_id} 删除成功")
            return True
        except Exception as e:
            print(f"删除课程时发生错误: {e}")
            return False
    
    def get_all_courses(self) -> List[Dict[str, Any]]:
        """获取所有课程"""
        try:
            query = 'SELECT * FROM courses ORDER BY course_id'
            result = self.db.execute_query(query, fetch=True)
            
            columns = ['id', 'course_id', 'name', 'credit', 'teacher']
            return [dict(zip(columns, row)) for row in result]
        except Exception as e:
            print(f"获取课程列表时发生错误: {e}")
            return []

2.4 成绩管理模块

# score_manager.py
from database import DatabaseManager
from typing import List, Dict, Any, Optional

class ScoreManager:
    def __init__(self, db_manager: DatabaseManager):
        self.db = db_manager
    
    def add_score(self, student_id: str, course_id: str, score: float, semester: str) -> bool:
        """添加成绩"""
        try:
            # 检查学生和课程是否存在
            student_query = 'SELECT 1 FROM students WHERE student_id = ?'
            course_query = 'SELECT 1 FROM courses WHERE course_id = ?'
            
            if not self.db.execute_query(student_query, (student_id,), fetch=True):
                print(f"学生ID {student_id} 不存在")
                return False
            
            if not self.db.execute_query(course_query, (course_id,), fetch=True):
                print(f"课程ID {course_id} 不存在")
                return False
            
            # 添加成绩记录
            query = '''
                INSERT INTO scores (student_id, course_id, score, semester)
                VALUES (?, ?, ?, ?)
            '''
            self.db.execute_query(query, (student_id, course_id, score, semester))
            print(f"成绩记录添加成功: 学生 {student_id}, 课程 {course_id}, 成绩 {score}")
            return True
        except sqlite3.IntegrityError:
            print(f"该学生本学期该课程成绩已存在")
            return False
        except Exception as e:
            print(f"添加成绩时发生错误: {e}")
            return False
    
    def update_score(self, student_id: str, course_id: str, semester: str, score: float) -> bool:
        """更新成绩"""
        try:
            query = '''
                UPDATE scores SET score = ?
                WHERE student_id = ? AND course_id = ? AND semester = ?
            '''
            self.db.execute_query(query, (score, student_id, course_id, semester))
            
            print(f"成绩更新成功: 学生 {student_id}, 课程 {course_id}, 成绩 {score}")
            return True
        except Exception as e:
            print(f"更新成绩时发生错误: {e}")
            return False
    
    def get_student_scores(self, student_id: str, semester: str = None) -> List[Dict[str, Any]]:
        """获取学生成绩"""
        try:
            if semester:
                query = '''
                    SELECT s.course_id, c.name, s.score, c.credit, s.semester
                    FROM scores s
                    JOIN courses c ON s.course_id = c.course_id
                    WHERE s.student_id = ? AND s.semester = ?
                    ORDER BY s.semester, c.course_id
                '''
                result = self.db.execute_query(query, (student_id, semester), fetch=True)
            else:
                query = '''
                    SELECT s.course_id, c.name, s.score, c.credit, s.semester
                    FROM scores s
                    JOIN courses c ON s.course_id = c.course_id
                    WHERE s.student_id = ?
                    ORDER BY s.semester, c.course_id
                '''
                result = self.db.execute_query(query, (student_id,), fetch=True)
            
            columns = ['course_id', 'course_name', 'score', 'credit', 'semester']
            return [dict(zip(columns, row)) for row in result]
        except Exception as e:
            print(f"获取学生成绩时发生错误: {e}")
            return []
    
    def get_course_scores(self, course_id: str, semester: str = None) -> List[Dict[str, Any]]:
        """获取课程成绩"""
        try:
            if semester:
                query = '''
                    SELECT s.student_id, st.name, s.score, st.class, s.semester
                    FROM scores s
                    JOIN students st ON s.student_id = st.student_id
                    WHERE s.course_id = ? AND s.semester = ?
                    ORDER BY s.score DESC
                '''
                result = self.db.execute_query(query, (course_id, semester), fetch=True)
            else:
                query = '''
                    SELECT s.student_id, st.name, s.score, st.class, s.semester
                    FROM scores s
                    JOIN students st ON s.student_id = st.student_id
                    WHERE s.course_id = ?
                    ORDER BY s.semester, s.score DESC
                '''
                result = self.db.execute_query(query, (course_id,), fetch=True)
            
            columns = ['student_id', 'student_name', 'score', 'class', 'semester']
            return [dict(zip(columns, row)) for row in result]
        except Exception as e:
            print(f"获取课程成绩时发生错误: {e}")
            return []
    
    def calculate_student_gpa(self, student_id: str, semester: str = None) -> Dict[str, Any]:
        """计算学生GPA"""
        scores = self.get_student_scores(student_id, semester)
        
        if not scores:
            return {"gpa": 0, "total_credit": 0, "passed_credit": 0}
        
        total_credit = 0
        total_grade_point = 0
        passed_credit = 0
        
        for score in scores:
            credit = score['credit']
            score_value = score['score']
            
            # 计算绩点(常见算法:90-100:4.0, 80-89:3.0, 70-79:2.0, 60-69:1.0, 0-59:0)
            if score_value >= 90:
                grade_point = 4.0
            elif score_value >= 80:
                grade_point = 3.0
            elif score_value >= 70:
                grade_point = 2.0
            elif score_value >= 60:
                grade_point = 1.0
            else:
                grade_point = 0
            
            total_credit += credit
            total_grade_point += grade_point * credit
            
            if score_value >= 60:  # 及格
                passed_credit += credit
        
        gpa = total_grade_point / total_credit if total_credit > 0 else 0
        
        return {
            "gpa": round(gpa, 2),
            "total_credit": total_credit,
            "passed_credit": passed_credit
        }

2.5 CSV导入导出模块

# csv_handler.py
import csv
import os
from database import DatabaseManager
from typing import List, Dict, Any

class CSVHandler:
    def __init__(self, db_manager: DatabaseManager):
        self.db = db_manager
    
    def import_students_from_csv(self, file_path: str) -> int:
        """从CSV文件导入学生数据"""
        if not os.path.exists(file_path):
            print(f"文件不存在: {file_path}")
            return 0
        
        imported_count = 0
        
        try:
            with open(file_path, 'r', encoding='utf-8') as file:
                reader = csv.DictReader(file)
                
                for row in reader:
                    student_id = row.get('student_id', '').strip()
                    name = row.get('name', '').strip()
                    gender = row.get('gender', '').strip()
                    class_name = row.get('class', '').strip()
                    major = row.get('major', '').strip()
                    
                    if student_id and name:
                        query = '''
                            INSERT OR IGNORE INTO students (student_id, name, gender, class, major)
                            VALUES (?, ?, ?, ?, ?)
                        '''
                        self.db.execute_query(query, (student_id, name, gender, class_name, major))
                        imported_count += 1
            
            print(f"从 {file_path} 导入了 {imported_count} 条学生记录")
            return imported_count
        
        except Exception as e:
            print(f"导入学生数据时发生错误: {e}")
            return 0
    
    def import_courses_from_csv(self, file_path: str) -> int:
        """从CSV文件导入课程数据"""
        if not os.path.exists(file_path):
            print(f"文件不存在: {file_path}")
            return 0
        
        imported_count = 0
        
        try:
            with open(file_path, 'r', encoding='utf-8') as file:
                reader = csv.DictReader(file)
                
                for row in reader:
                    course_id = row.get('course_id', '').strip()
                    name = row.get('name', '').strip()
                    credit_str = row.get('credit', '0').strip()
                    teacher = row.get('teacher', '').strip()
                    
                    try:
                        credit = float(credit_str) if credit_str else 0
                    except ValueError:
                        credit = 0
                    
                    if course_id and name:
                        query = '''
                            INSERT OR IGNORE INTO courses (course_id, name, credit, teacher)
                            VALUES (?, ?, ?, ?)
                        '''
                        self.db.execute_query(query, (course_id, name, credit, teacher))
                        imported_count += 1
            
            print(f"从 {file_path} 导入了 {imported_count} 条课程记录")
            return imported_count
        
        except Exception as e:
            print(f"导入课程数据时发生错误: {e}")
            return 0
    
    def import_scores_from_csv(self, file_path: str) -> int:
        """从CSV文件导入成绩数据"""
        if not os.path.exists(file_path):
            print(f"文件不存在: {file_path}")
            return 0
        
        imported_count = 0
        
        try:
            with open(file_path, 'r', encoding='utf-8') as file:
                reader = csv.DictReader(file)
                
                for row in reader:
                    student_id = row.get('student_id', '').strip()
                    course_id = row.get('course_id', '').strip()
                    score_str = row.get('score', '').strip()
                    semester = row.get('semester', '').strip()
                    
                    try:
                        score = float(score_str) if score_str else 0
                    except ValueError:
                        score = 0
                    
                    if student_id and course_id and semester:
                        query = '''
                            INSERT OR REPLACE INTO scores (student_id, course_id, score, semester)
                            VALUES (?, ?, ?, ?)
                        '''
                        self.db.execute_query(query, (student_id, course_id, score, semester))
                        imported_count += 1
            
            print(f"从 {file_path} 导入了 {imported_count} 条成绩记录")
            return imported_count
        
        except Exception as e:
            print(f"导入成绩数据时发生错误: {e}")
            return 0
    
    def export_students_to_csv(self, file_path: str) -> int:
        """导出学生数据到CSV文件"""
        try:
            query = 'SELECT student_id, name, gender, class, major FROM students ORDER BY student_id'
            students = self.db.execute_query(query, fetch=True)
            
            with open(file_path, 'w', encoding='utf-8', newline='') as file:
                writer = csv.writer(file)
                writer.writerow(['student_id', 'name', 'gender', 'class', 'major'])
                writer.writerows(students)
            
            print(f"导出了 {len(students)} 条学生记录到 {file_path}")
            return len(students)
        
        except Exception as e:
            print(f"导出学生数据时发生错误: {e}")
            return 0
    
    def export_courses_to_csv(self, file_path: str) -> int:
        """导出课程数据到CSV文件"""
        try:
            query = 'SELECT course_id, name, credit, teacher FROM courses ORDER BY course_id'
            courses = self.db.execute_query(query, fetch=True)
            
            with open(file_path, 'w', encoding='utf-8', newline='') as file:
                writer = csv.writer(file)
                writer.writerow(['course_id', 'name', 'credit', 'teacher'])
                writer.writerows(courses)
            
            print(f"导出了 {len(courses)} 条课程记录到 {file_path}")
            return len(courses)
        
        except Exception as e:
            print(f"导出课程数据时发生错误: {e}")
            return 0
    
    def export_scores_to_csv(self, file_path: str, semester: str = None) -> int:
        """导出成绩数据到CSV文件"""
        try:
            if semester:
                query = '''
                    SELECT student_id, course_id, score, semester 
                    FROM scores 
                    WHERE semester = ?
                    ORDER BY student_id, course_id
                '''
                scores = self.db.execute_query(query, (semester,), fetch=True)
            else:
                query = 'SELECT student_id, course_id, score, semester FROM scores ORDER BY semester, student_id, course_id'
                scores = self.db.execute_query(query, fetch=True)
            
            with open(file_path, 'w', encoding='utf-8', newline='') as file:
                writer = csv.writer(file)
                writer.writerow(['student_id', 'course_id', 'score', 'semester'])
                writer.writerows(scores)
            
            print(f"导出了 {len(scores)} 条成绩记录到 {file_path}")
            return len(scores)
        
        except Exception as e:
            print(f"导出成绩数据时发生错误: {e}")
            return 0

2.6 报表生成模块

# report_generator.py
from database import DatabaseManager
from typing import List, Dict, Any
import csv

class ReportGenerator:
    def __init__(self, db_manager: DatabaseManager):
        self.db = db_manager
    
    def generate_student_report(self, student_id: str, semester: str = None) -> Dict[str, Any]:
        """生成学生成绩报告"""
        try:
            # 获取学生基本信息
            student_query = 'SELECT name, gender, class, major FROM students WHERE student_id = ?'
            student_info = self.db.execute_query(student_query, (student_id,), fetch=True)
            
            if not student_info:
                return {"error": "学生不存在"}
            
            name, gender, class_name, major = student_info[0]
            
            # 获取学生成绩
            if semester:
                score_query = '''
                    SELECT c.course_id, c.name, s.score, c.credit, s.semester
                    FROM scores s
                    JOIN courses c ON s.course_id = c.course_id
                    WHERE s.student_id = ? AND s.semester = ?
                    ORDER BY c.course_id
                '''
                scores = self.db.execute_query(score_query, (student_id, semester), fetch=True)
            else:
                score_query = '''
                    SELECT c.course_id, c.name, s.score, c.credit, s.semester
                    FROM scores s
                    JOIN courses c ON s.course_id = c.course_id
                    WHERE s.student_id = ?
                    ORDER BY s.semester, c.course_id
                '''
                scores = self.db.execute_query(score_query, (student_id,), fetch=True)
            
            # 计算统计信息
            total_credit = 0
            total_grade_point = 0
            passed_credit = 0
            course_count = len(scores)
            
            for course in scores:
                course_id, course_name, score, credit, semester = course
                total_credit += credit
                
                # 计算绩点
                if score >= 90:
                    grade_point = 4.0
                elif score >= 80:
                    grade_point = 3.0
                elif score >= 70:
                    grade_point = 2.0
                elif score >= 60:
                    grade_point = 1.0
                else:
                    grade_point = 0
                
                total_grade_point += grade_point * credit
                
                if score >= 60:
                    passed_credit += credit
            
            gpa = total_grade_point / total_credit if total_credit > 0 else 0
            
            # 组织报告数据
            report = {
                "student_id": student_id,
                "name": name,
                "gender": gender,
                "class": class_name,
                "major": major,
                "semester": semester or "所有学期",
                "courses": [],
                "statistics": {
                    "course_count": course_count,
                    "total_credit": total_credit,
                    "passed_credit": passed_credit,
                    "gpa": round(gpa, 2)
                }
            }
            
            # 添加课程详情
            for course in scores:
                course_id, course_name, score, credit, semester = course
                report["courses"].append({
                    "course_id": course_id,
                    "course_name": course_name,
                    "score": score,
                    "credit": credit,
                    "semester": semester,
                    "status": "及格" if score >= 60 else "不及格"
                })
            
            return report
        
        except Exception as e:
            return {"error": f"生成报告时发生错误: {e}"}
    
    def generate_course_report(self, course_id: str, semester: str = None) -> Dict[str, Any]:
        """生成课程成绩报告"""
        try:
            # 获取课程基本信息
            course_query = 'SELECT name, credit, teacher FROM courses WHERE course_id = ?'
            course_info = self.db.execute_query(course_query, (course_id,), fetch=True)
            
            if not course_info:
                return {"error": "课程不存在"}
            
            name, credit, teacher = course_info[0]
            
            # 获取课程成绩
            if semester:
                score_query = '''
                    SELECT s.student_id, st.name, s.score, st.class, s.semester
                    FROM scores s
                    JOIN students st ON s.student_id = st.student_id
                    WHERE s.course_id = ? AND s.semester = ?
                    ORDER BY s.score DESC
                '''
                scores = self.db.execute_query(score_query, (course_id, semester), fetch=True)
            else:
                score_query = '''
                    SELECT s.student_id, st.name, s.score, st.class, s.semester
                    FROM scores s
                    JOIN students st ON s.student_id = st.student_id
                    WHERE s.course_id = ?
                    ORDER BY s.semester, s.score DESC
                '''
                scores = self.db.execute_query(score_query, (course_id,), fetch=True)
            
            # 计算统计信息
            student_count = len(scores)
            avg_score = 0
            max_score = 0
            min_score = 100
            pass_count = 0
            
            if student_count > 0:
                total_score = sum(score[2] for score in scores)
                avg_score = total_score / student_count
                max_score = max(score[2] for score in scores)
                min_score = min(score[2] for score in scores)
                pass_count = sum(1 for score in scores if score[2] >= 60)
            
            pass_rate = (pass_count / student_count * 100) if student_count > 0 else 0
            
            # 组织报告数据
            report = {
                "course_id": course_id,
                "course_name": name,
                "credit": credit,
                "teacher": teacher,
                "semester": semester or "所有学期",
                "scores": [],
                "statistics": {
                    "student_count": student_count,
                    "avg_score": round(avg_score, 2),
                    "max_score": max_score,
                    "min_score": min_score,
                    "pass_count": pass_count,
                    "pass_rate": round(pass_rate, 2)
                }
            }
            
            # 添加学生成绩详情
            for score in scores:
                student_id, student_name, score_value, class_name, semester = score
                report["scores"].append({
                    "student_id": student_id,
                    "student_name": student_name,
                    "score": score_value,
                    "class": class_name,
                    "semester": semester,
                    "status": "及格" if score_value >= 60 else "不及格"
                })
            
            return report
        
        except Exception as e:
            return {"error": f"生成报告时发生错误: {e}"}
    
    def export_student_report_to_csv(self, student_id: str, file_path: str, semester: str = None) -> bool:
        """导出学生成绩报告到CSV文件"""
        report = self.generate_student_report(student_id, semester)
        
        if "error" in report:
            print(report["error"])
            return False
        
        try:
            with open(file_path, 'w', encoding='utf-8', newline='') as file:
                writer = csv.writer(file)
                
                # 写入学生基本信息
                writer.writerow(["学号", report["student_id"]])
                writer.writerow(["姓名", report["name"]])
                writer.writerow(["性别", report["gender"]])
                writer.writerow(["班级", report["class"]])
                writer.writerow(["专业", report["major"]])
                writer.writerow(["学期", report["semester"]])
                writer.writerow([])
                
                # 写入统计信息
                stats = report["statistics"]
                writer.writerow(["课程总数", stats["course_count"]])
                writer.writerow(["总学分", stats["total_credit"]])
                writer.writerow(["获得学分", stats["passed_credit"]])
                writer.writerow(["平均绩点", stats["gpa"]])
                writer.writerow([])
                
                # 写入课程成绩
                writer.writerow(["课程编号", "课程名称", "成绩", "学分", "学期", "状态"])
                for course in report["courses"]:
                    writer.writerow([
                        course["course_id"],
                        course["course_name"],
                        course["score"],
                        course["credit"],
                        course["semester"],
                        course["status"]
                    ])
            
            print(f"学生成绩报告已导出到 {file_path}")
            return True
        
        except Exception as e:
            print(f"导出报告时发生错误: {e}")
            return False
    
    def export_course_report_to_csv(self, course_id: str, file_path: str, semester: str = None) -> bool:
        """导出课程成绩报告到CSV文件"""
        report = self.generate_course_report(course_id, semester)
        
        if "error" in report:
            print(report["error"])
            return False
        
        try:
            with open(file_path, 'w', encoding='utf-8', newline='') as file:
                writer = csv.writer(file)
                
                # 写入课程基本信息
                writer.writerow(["课程编号", report["course_id"]])
                writer.writerow(["课程名称", report["course_name"]])
                writer.writerow(["学分", report["credit"]])
                writer.writerow(["授课教师", report["teacher"]])
                writer.writerow(["学期", report["semester"]])
                writer.writerow([])
                
                # 写入统计信息
                stats = report["statistics"]
                writer.writerow(["学生总数", stats["student_count"]])
                writer.writerow(["平均分", stats["avg_score"]])
                writer.writerow(["最高分", stats["max_score"]])
                writer.writerow(["最低分", stats["min_score"]])
                writer.writerow(["及格人数", stats["pass_count"]])
                writer.writerow(["及格率", f"{stats['pass_rate']}%"])
                writer.writerow([])
                
                # 写入学生成绩
                writer.writerow(["学号", "姓名", "成绩", "班级", "学期", "状态"])
                for score in report["scores"]:
                    writer.writerow([
                        score["student_id"],
                        score["student_name"],
                        score["score"],
                        score["class"],
                        score["semester"],
                        score["status"]
                    ])
            
            print(f"课程成绩报告已导出到 {file_path}")
            return True
        
        except Exception as e:
            print(f"导出报告时发生错误: {e}")
            return False

第三部分:主程序与用户界面

3.1 命令行界面

# main.py
import sys
import os
from database import DatabaseManager
from student_manager import StudentManager
from course_manager import CourseManager
from score_manager import ScoreManager
from csv_handler import CSVHandler
from report_generator import ReportGenerator

class StudentManagementSystem:
    def __init__(self):
        self.db = DatabaseManager()
        self.student_manager = StudentManager(self.db)
        self.course_manager = CourseManager(self.db)
        self.score_manager = ScoreManager(self.db)
        self.csv_handler = CSVHandler(self.db)
        self.report_generator = ReportGenerator(self.db)
    
    def display_menu(self):
        """显示主菜单"""
        print("\n" + "=" * 50)
        print("学生成绩管理系统")
        print("=" * 50)
        print("1. 学生管理")
        print("2. 课程管理")
        print("3. 成绩管理")
        print("4. 数据导入导出")
        print("5. 报表生成")
        print("0. 退出系统")
        print("=" * 50)
    
    def student_management_menu(self):
        """学生管理子菜单"""
        while True:
            print("\n" + "-" * 30)
            print("学生管理")
            print("-" * 30)
            print("1. 添加学生")
            print("2. 查看学生信息")
            print("3. 修改学生信息")
            print("4. 删除学生")
            print("5. 列出所有学生")
            print("6. 搜索学生")
            print("0. 返回主菜单")
            print("-" * 30)
            
            choice = input("请选择操作: ").strip()
            
            if choice == "1":
                self.add_student()
            elif choice == "2":
                self.view_student()
            elif choice == "3":
                self.update_student()
            elif choice == "4":
                self.delete_student()
            elif choice == "5":
                self.list_students()
            elif choice == "6":
                self.search_students()
            elif choice == "0":
                break
            else:
                print("无效选择,请重新输入")
    
    def course_management_menu(self):
        """课程管理子菜单"""
        while True:
            print("\n" + "-" * 30)
            print("课程管理")
            print("-" * 30)
            print("1. 添加课程")
            print("2. 查看课程信息")
            print("3. 修改课程信息")
            print("4. 删除课程")
            print("5. 列出所有课程")
            print("0. 返回主菜单")
            print("-" * 30)
            
            choice = input("请选择操作: ").strip()
            
            if choice == "1":
                self.add_course()
            elif choice == "2":
                self.view_course()
            elif choice == "3":
                self.update_course()
            elif choice == "4":
                self.delete_course()
            elif choice == "5":
                self.list_courses()
            elif choice == "0":
                break
            else:
                print("无效选择,请重新输入")
    
    def score_management_menu(self):
        """成绩管理子菜单"""
        while True:
            print("\n" + "-" * 30)
            print("成绩管理")
            print("-" * 30)
            print("1. 添加成绩")
            print("2. 修改成绩")
            print("3. 查看学生成绩")
            print("4. 查看课程成绩")
            print("5. 计算学生GPA")
            print("0. 返回主菜单")
            print("-" * 30)
            
            choice = input("请选择操作: ").strip()
            
            if choice == "1":
                self.add_score()
            elif choice == "2":
                self.update_score()
            elif choice == "3":
                self.view_student_scores()
            elif choice == "4":
                self.view_course_scores()
            elif choice == "5":
                self.calculate_student_gpa()
            elif choice == "0":
                break
            else:
                print("无效选择,请重新输入")
    
    def data_io_menu(self):
        """数据导入导出子菜单"""
        while True:
            print("\n" + "-" * 30)
            print("数据导入导出")
            print("-" * 30)
            print("1. 导入学生数据(CSV)")
            print("2. 导入课程数据(CSV)")
            print("3. 导入成绩数据(CSV)")
            print("4. 导出学生数据(CSV)")
            print("5. 导出课程数据(CSV)")
            print("6. 导出成绩数据(CSV)")
            print("0. 返回主菜单")
            print("-" * 30)
            
            choice = input("请选择操作: ").strip()
            
            if choice == "1":
                self.import_students()
            elif choice == "2":
                self.import_courses()
            elif choice == "3":
                self.import_scores()
            elif choice == "4":
                self.export_students()
            elif choice == "5":
                self.export_courses()
            elif choice == "6":
                self.export_scores()
            elif choice == "0":
                break
            else:
                print("无效选择,请重新输入")
    
    def report_menu(self):
        """报表生成子菜单"""
        while True:
            print("\n" + "-" * 30)
            print("报表生成")
            print("-" * 30)
            print("1. 生成学生成绩报告")
            print("2. 生成课程成绩报告")
            print("3. 导出学生成绩报告(CSV)")
            print("4. 导出课程成绩报告(CSV)")
            print("0. 返回主菜单")
            print("-" * 30)
            
            choice = input("请选择操作: ").strip()
            
            if choice == "1":
                self.generate_student_report()
            elif choice == "2":
                self.generate_course_report()
            elif choice == "3":
                self.export_student_report()
            elif choice == "4":
                self.export_course_report()
            elif choice == "0":
                break
            else:
                print("无效选择,请重新输入")
    
    # 学生管理功能实现
    def add_student(self):
        print("\n添加学生")
        student_id = input("学号: ").strip()
        name = input("姓名: ").strip()
        gender = input("性别(可选): ").strip()
        class_name = input("班级(可选): ").strip()
        major = input("专业(可选): ").strip()
        
        if not student_id or not name:
            print("学号和姓名不能为空")
            return
        
        self.student_manager.add_student(student_id, name, gender, class_name, major)
    
    def view_student(self):
        print("\n查看学生信息")
        student_id = input("请输入学号: ").strip()
        
        if not student_id:
            print("学号不能为空")
            return
        
        student = self.student_manager.get_student(student_id)
        if student:
            print(f"\n学号: {student['student_id']}")
            print(f"姓名: {student['name']}")
            print(f"性别: {student['gender'] or '未设置'}")
            print(f"班级: {student['class'] or '未设置'}")
            print(f"专业: {student['major'] or '未设置'}")
        else:
            print("找不到该学生")
    
    def update_student(self):
        print("\n修改学生信息")
        student_id = input("请输入学号: ").strip()
        
        if not student_id:
            print("学号不能为空")
            return
        
        student = self.student_manager.get_student(student_id)
        if not student:
            print("找不到该学生")
            return
        
        print("留空表示不修改该字段")
        name = input(f"姓名({student['name']}): ").strip()
        gender = input(f"性别({student['gender'] or '未设置'}): ").strip()
        class_name = input(f"班级({student['class'] or '未设置'}): ").strip()
        major = input(f"专业({student['major'] or '未设置'}): ").strip()
        
        update_data = {}
        if name: update_data['name'] = name
        if gender: update_data['gender'] = gender
        if class_name: update_data['class'] = class_name
        if major: update_data['major'] = major
        
        if update_data:
            self.student_manager.update_student(student_id, **update_data)
        else:
            print("没有提供任何修改")
    
    def delete_student(self):
        print("\n删除学生")
        student_id = input("请输入学号: ").strip()
        
        if not student_id:
            print("学号不能为空")
            return
        
        confirm = input(f"确定要删除学号为 {student_id} 的学生吗?(y/N): ").strip().lower()
        if confirm == 'y':
            self.student_manager.delete_student(student_id)
        else:
            print("取消删除")
    
    def list_students(self):
        print("\n所有学生列表")
        students = self.student_manager.get_all_students()
        
        if not students:
            print("没有学生记录")
            return
        
        print(f"{'学号':<12} {'姓名':<10} {'性别':<4} {'班级':<10} {'专业':<10}")
        print("-" * 50)
        for student in students:
            print(f"{student['student_id']:<12} {student['name']:<10} "
                  f"{student['gender'] or '':<4} {student['class'] or '':<10} "
                  f"{student['major'] or '':<10}")
    
    def search_students(self):
        print("\n搜索学生")
        print("留空表示不限制该条件")
        student_id = input("学号: ").strip()
        name = input("姓名: ").strip()
        gender = input("性别: ").strip()
        class_name = input("班级: ").strip()
        major = input("专业: ").strip()
        
        search_criteria = {}
        if student_id: search_criteria['student_id'] = student_id
        if name: search_criteria['name'] = name
        if gender: search_criteria['gender'] = gender
        if class_name: search_criteria['class'] = class_name
        if major: search_criteria['major'] = major
        
        students = self.student_manager.search_students(**search_criteria)
        
        if not students:
            print("没有找到符合条件的学生")
            return
        
        print(f"\n找到 {len(students)} 个符合条件的学生:")
        print(f"{'学号':<12} {'姓名':<10} {'性别':<4} {'班级':<10} {'专业':<10}")
        print("-" * 50)
        for student in students:
            print(f"{student['student_id']:<12} {student['name']:<10} "
                  f"{student['gender'] or '':<4} {student['class'] or '':<10} "
                  f"{student['major'] or '':<10}")
    
    # 课程管理功能实现(类似学生管理,篇幅限制省略部分代码)
    def add_course(self):
        print("\n添加课程")
        course_id = input("课程编号: ").strip()
        name = input("课程名称: ").strip()
        credit = input("学分(默认0): ").strip()
        teacher = input("授课教师(可选): ").strip()
        
        if not course_id or not name:
            print("课程编号和名称不能为空")
            return
        
        try:
            credit_val = float(credit) if credit else 0
        except ValueError:
            print("学分必须是数字")
            return
        
        self.course_manager.add_course(course_id, name, credit_val, teacher)
    
    # 其他课程管理方法类似,篇幅限制省略...
    
    # 成绩管理功能实现
    def add_score(self):
        print("\n添加成绩")
        student_id = input("学号: ").strip()
        course_id = input("课程编号: ").strip()
        score = input("成绩: ").strip()
        semester = input("学期: ").strip()
        
        if not student_id or not course_id or not score or not semester:
            print("所有字段都不能为空")
            return
        
        try:
            score_val = float(score)
            if score_val < 0 or score_val > 100:
                print("成绩必须在0-100之间")
                return
        except ValueError:
            print("成绩必须是数字")
            return
        
        self.score_manager.add_score(student_id, course_id, score_val, semester)
    
    def update_score(self):
        print("\n修改成绩")
        student_id = input("学号: ").strip()
        course_id = input("课程编号: ").strip()
        semester = input("学期: ").strip()
        score = input("新成绩: ").strip()
        
        if not student_id or not course_id or not semester or not score:
            print("所有字段都不能为空")
            return
        
        try:
            score_val = float(score)
            if score_val < 0 or score_val > 100:
                print("成绩必须在0-100之间")
                return
        except ValueError:
            print("成绩必须是数字")
            return
        
        self.score_manager.update_score(student_id, course_id, semester, score_val)
    
    def view_student_scores(self):
        print("\n查看学生成绩")
        student_id = input("学号: ").strip()
        semester = input("学期(留空查看所有学期): ").strip() or None
        
        if not student_id:
            print("学号不能为空")
            return
        
        scores = self.score_manager.get_student_scores(student_id, semester)
        
        if not scores:
            print("没有找到成绩记录")
            return
        
        student = self.student_manager.get_student(student_id)
        if student:
            print(f"\n学生: {student['name']} ({student_id})")
        
        total_credit = sum(score['credit'] for score in scores)
        passed_credit = sum(score['credit'] for score in scores if score['score'] >= 60)
        
        print(f"{'课程编号':<10} {'课程名称':<20} {'成绩':<6} {'学分':<6} {'学期':<10} {'状态':<6}")
        print("-" * 60)
        for score in scores:
            status = "及格" if score['score'] >= 60 else "不及格"
            print(f"{score['course_id']:<10} {score['course_name']:<20} "
                  f"{score['score']:<6} {score['credit']:<6} "
                  f"{score['semester']:<10} {status:<6}")
        
        print("-" * 60)
        print(f"总学分: {total_credit}, 获得学分: {passed_credit}")
    
    def view_course_scores(self):
        print("\n查看课程成绩")
        course_id = input("课程编号: ").strip()
        semester = input("学期(留空查看所有学期): ").strip() or None
        
        if not course_id:
            print("课程编号不能为空")
            return
        
        scores = self.score_manager.get_course_scores(course_id, semester)
        
        if not scores:
            print("没有找到成绩记录")
            return
        
        course = self.course_manager.get_course(course_id)
        if course:
            print(f"\n课程: {course['name']} ({course_id})")
        
        print(f"{'学号':<12} {'姓名':<10} {'成绩':<6} {'班级':<10} {'学期':<10} {'状态':<6}")
        print("-" * 60)
        for score in scores:
            status = "及格" if score['score'] >= 60 else "不及格"
            print(f"{score['student_id']:<12} {score['student_name']:<10} "
                  f"{score['score']:<6} {score['class']:<10} "
                  f"{score['semester']:<10} {status:<6}")
    
    def calculate_student_gpa(self):
        print("\n计算学生GPA")
        student_id = input("学号: ").strip()
        semester = input("学期(留空计算所有学期): ").strip() or None
        
        if not student_id:
            print("学号不能为空")
            return
        
        result = self.score_manager.calculate_student_gpa(student_id, semester)
        
        student = self.student_manager.get_student(student_id)
        if student:
            print(f"\n学生: {student['name']} ({student_id})")
        
        print(f"学期: {semester or '所有学期'}")
        print(f"平均绩点(GPA): {result['gpa']}")
        print(f"总学分: {result['total_credit']}")
        print(f"获得学分: {result['passed_credit']}")
    
    # 数据导入导出功能实现
    def import_students(self):
        print("\n导入学生数据")
        file_path = input("CSV文件路径: ").strip()
        
        if not file_path:
            print("文件路径不能为空")
            return
        
        self.csv_handler.import_students_from_csv(file_path)
    
    def import_courses(self):
        print("\n导入课程数据")
        file_path = input("CSV文件路径: ").strip()
        
        if not file_path:
            print("文件路径不能为空")
            return
        
        self.csv_handler.import_courses_from_csv(file_path)
    
    def import_scores(self):
        print("\n导入成绩数据")
        file_path = input("CSV文件路径: ").strip()
        
        if not file_path:
            print("文件路径不能为空")
            return
        
        self.csv_handler.import_scores_from_csv(file_path)
    
    def export_students(self):
        print("\n导出学生数据")
        file_path = input("CSV文件路径: ").strip()
        
        if not file_path:
            print("文件路径不能为空")
            return
        
        self.csv_handler.export_students_to_csv(file_path)
    
    def export_courses(self):
        print("\n导出课程数据")
        file_path = input("CSV文件路径: ").strip()
        
        if not file_path:
            print("文件路径不能为空")
            return
        
        self.csv_handler.export_courses_to_csv(file_path)
    
    def export_scores(self):
        print("\n导出成绩数据")
        file_path = input("CSV文件路径: ").strip()
        semester = input("学期(留空导出所有学期): ").strip() or None
        
        if not file_path:
            print("文件路径不能为空")
            return
        
        if semester:
            self.csv_handler.export_scores_to_csv(file_path, semester)
        else:
            self.csv_handler.export_scores_to_csv(file_path)
    
    # 报表生成功能实现
    def generate_student_report(self):
        print("\n生成学生成绩报告")
        student_id = input("学号: ").strip()
        semester = input("学期(留空查看所有学期): ").strip() or None
        
        if not student_id:
            print("学号不能为空")
            return
        
        report = self.report_generator.generate_student_report(student_id, semester)
        
        if "error" in report:
            print(report["error"])
            return
        
        student = report
        print(f"\n学生成绩报告")
        print("=" * 50)
        print(f"学号: {student['student_id']}")
        print(f"姓名: {student['name']}")
        print(f"性别: {student['gender'] or '未设置'}")
        print(f"班级: {student['class'] or '未设置'}")
        print(f"专业: {student['major'] or '未设置'}")
        print(f"学期: {student['semester']}")
        print()
        
        stats = student['statistics']
        print(f"课程总数: {stats['course_count']}")
        print(f"总学分: {stats['total_credit']}")
        print(f"获得学分: {stats['passed_credit']}")
        print(f"平均绩点(GPA): {stats['gpa']}")
        print()
        
        print(f"{'课程编号':<10} {'课程名称':<20} {'成绩':<6} {'学分':<6} {'学期':<10} {'状态':<6}")
        print("-" * 60)
        for course in student['courses']:
            print(f"{course['course_id']:<10} {course['course_name']:<20} "
                  f"{course['score']:<6} {course['credit']:<6} "
                  f"{course['semester']:<10} {course['status']:<6}")
    
    def generate_course_report(self):
        print("\n生成课程成绩报告")
        course_id = input("课程编号: ").strip()
        semester = input("学期(留空查看所有学期): ").strip() or None
        
        if not course_id:
            print("课程编号不能为空")
            return
        
        report = self.report_generator.generate_course_report(course_id, semester)
        
        if "error" in report:
            print(report["error"])
            return
        
        course = report
        print(f"\n课程成绩报告")
        print("=" * 50)
        print(f"课程编号: {course['course_id']}")
        print(f"课程名称: {course['course_name']}")
        print(f"学分: {course['credit']}")
        print(f"授课教师: {course['teacher'] or '未设置'}")
        print(f"学期: {course['semester']}")
        print()
        
        stats = course['statistics']
        print(f"学生总数: {stats['student_count']}")
        print(f"平均分: {stats['avg_score']}")
        print(f"最高分: {stats['max_score']}")
        print(f"最低分: {stats['min_score']}")
        print(f"及格人数: {stats['pass_count']}")
        print(f"及格率: {stats['pass_rate']}%")
        print()
        
        print(f"{'学号':<12} {'姓名':<10} {'成绩':<6} {'班级':<10} {'学期':<10} {'状态':<6}")
        print("-" * 60)
        for score in course['scores']:
            print(f"{score['student_id']:<12} {score['student_name']:<10} "
                  f"{score['score']:<6} {score['class']:<10} "
                  f"{score['semester']:<10} {score['status']:<6}")
    
    def export_student_report(self):
        print("\n导出学生成绩报告")
        student_id = input("学号: ").strip()
        file_path = input("CSV文件路径: ").strip()
        semester = input("学期(留空导出所有学期): ").strip() or None
        
        if not student_id or not file_path:
            print("学号和文件路径不能为空")
            return
        
        self.report_generator.export_student_report_to_csv(student_id, file_path, semester)
    
    def export_course_report(self):
        print("\n导出课程成绩报告")
        course_id = input("课程编号: ").strip()
        file_path = input("CSV文件路径: ").strip()
        semester = input("学期(留空导出所有学期): ").strip() or None
        
        if not course_id or not file_path:
            print("课程编号和文件路径不能为空")
            return
        
        self.report_generator.export_course_report_to_csv(course_id, file_path, semester)
    
    def run(self):
        """运行系统主循环"""
        while True:
            self.display_menu()
            choice = input("请选择操作: ").strip()
            
            if choice == "1":
                self.student_management_menu()
            elif choice == "2":
                self.course_management_menu()
            elif choice == "3":
                self.score_management_menu()
            elif choice == "4":
                self.data_io_menu()
            elif choice == "5":
                self.report_menu()
            elif choice == "0":
                print("感谢使用学生成绩管理系统,再见!")
                break
            else:
                print("无效选择,请重新输入")

# 启动系统
if __name__ == "__main__":
    system = StudentManagementSystem()
    system.run()

3.2 示例数据文件

创建示例CSV文件用于测试:

students.csv

student_id,name,gender,class,major
2023001,张三,男,计算机1班,计算机科学
2023002,李四,女,计算机1班,计算机科学
2023003,王五,男,计算机2班,计算机科学
2023004,赵六,女,数学1班,数学
2023005,钱七,男,数学1班,数学

courses.csv

course_id,name,credit,teacher
CS101,Python程序设计,3.0,张老师
CS102,数据结构,4.0,李老师
MATH201,高等数学,5.0,王老师
ENGL101,大学英语,2.0,赵老师

scores.csv

student_id,course_id,score,semester
2023001,CS101,95,2023-秋季
2023001,CS102,88,2023-秋季
2023001,MATH201,92,2023-秋季
2023002,CS101,78,2023-秋季
2023002,CS102,85,2023-秋季
2023002,MATH201,90,2023-秋季
2023003,CS101,82,2023-秋季
2023003,CS102,79,2023-秋季
2023003,MATH201,86,2023-秋季
2023004,MATH201,95,2023-秋季
2023004,ENGL101,88,2023-秋季
2023005,MATH201,92,2023-秋季
2023005,ENGL101,85,2023-秋季

第四部分:系统测试与使用

4.1 系统测试

  1. 初始化数据库:运行系统时会自动创建数据库和表结构
  2. 导入示例数据:使用数据导入功能导入示例CSV文件
  3. 测试各项功能:逐一测试学生管理、课程管理、成绩管理等功能
  4. 生成报表:测试报表生成和导出功能
  5. 验证数据一致性:确保数据库和CSV文件数据一致

4.2 使用示例

  1. 启动系统

    python main.py
    
  2. 导入数据

    • 选择菜单选项"4. 数据导入导出"
    • 选择"1. 导入学生数据(CSV)“,输入文件路径"students.csv”
    • 同样方式导入课程和成绩数据
  3. 查询学生成绩

    • 选择菜单选项"3. 成绩管理"
    • 选择"3. 查看学生成绩",输入学号"2023001"
  4. 生成报表

    • 选择菜单选项"5. 报表生成"
    • 选择"1. 生成学生成绩报告",输入学号"2023001"
    • 选择"3. 导出学生成绩报告(CSV)",导出报告到文件
  5. 导出数据

    • 选择菜单选项"4. 数据导入导出"
    • 选择"4. 导出学生数据(CSV)",导出到新文件

第五部分:项目总结与扩展

5.1 项目总结

通过本项目,我们实现了一个完整的学生成绩管理系统,具有以下特点:

  1. 模块化设计:将系统分为数据库管理、学生管理、课程管理、成绩管理、CSV处理和报表生成等模块
  2. 数据持久化:使用SQLite数据库持久化存储数据
  3. 数据交换:支持CSV格式的数据导入导出
  4. 丰富功能:实现了增删改查、搜索、统计、报表等完整功能
  5. 用户友好:提供命令行界面,操作简单直观

5.2 技术要点

  1. SQLite数据库操作:使用SQLite进行数据存储和管理
  2. CSV文件处理:使用Python标准库csv模块处理CSV文件
  3. 面向对象编程:采用面向对象方式组织代码,提高可维护性
  4. 异常处理:完善的异常处理机制,提高系统稳定性
  5. 数据验证:对输入数据进行验证,确保数据完整性

5.3 扩展思路

  1. Web界面:使用Flask或Django开发Web界面
  2. 用户认证:添加用户登录和权限管理功能
  3. 数据可视化:使用matplotlib或plotly生成数据图表
  4. API接口:提供RESTful API供其他系统调用
  5. 云存储:支持将数据备份到云存储服务
  6. 移动应用:开发移动端APP方便随时查询
  7. 成绩预测:使用机器学习算法预测学生成绩
  8. 智能推荐:根据学生成绩推荐适合的课程

总结

本章通过一个完整的学生成绩管理系统项目,展示了如何综合运用CSV文件处理和SQLite数据库技术。项目涵盖了数据导入导出、CRUD操作、数据统计和报表生成等核心功能,体现了Python在数据处理和管理方面的强大能力。

通过本项目的学习,你应该掌握了:

  1. SQLite数据库的设计和操作
  2. CSV文件的读写和处理
  3. 模块化编程和代码组织
  4. 命令行界面的设计和实现
  5. 数据验证和异常处理

本项目不仅是一个学习示例,也是一个可以进一步扩展和完善的实际应用基础。你可以根据自己的需求,添加更多功能或改进现有实现,构建更强大的数据管理系统。

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐