统计分析

_debug.py

@@ -0,0 +1,15 @@
+from database import engine
+from sqlalchemy import text
+conn = engine.connect()
+
+tables = conn.execute(text("SHOW TABLES")).fetchall()
+print('Tables:', tables)
+
+for t in tables:
+    tname = t[0]
+    rows = conn.execute(text(f"SHOW COLUMNS FROM `{tname}` WHERE Field = 'is_deleted'")).fetchall()
+    if rows:
+        vals = conn.execute(text(f"SELECT is_deleted, COUNT(*) FROM `{tname}` GROUP BY is_deleted")).fetchall()
+        print(f"  {tname}: type={rows[0][1]}, values={vals}")
+
+conn.close()

api/course_api.py

@@ -0,0 +1,29 @@
+from fastapi import APIRouter, Depends, HTTPException
+from schema.course_request import CourseRequest, CourseResponse
+from database import get_db
+from dao.course_dao import (
+    add_course_dao, get_all_courses_dao, get_course_by_id_dao
+)
+
+course_api = APIRouter()
+
+
+@course_api.post('', response_model=CourseResponse, summary='添加课程')
+def add_course(course: CourseRequest, db=Depends(get_db)):
+    d = course.model_dump()
+    r = add_course_dao(o=d, db=db)
+    if not r:
+        raise HTTPException(status_code=500, detail='添加失败！')
+    return CourseResponse(totals=1, data=d)
+
+
+@course_api.get('/{course_id}', summary='按编号查课程')
+def get_course(course_id: str, db=Depends(get_db)):
+    c = get_course_by_id_dao(course_id=course_id, db=db)
+    if not c:
+        raise HTTPException(status_code=404, detail='课程不存在')
+    return {
+        'course_id': c.course_id,
+        'course_name': c.course_name,
+        'teacher_id': c.teacher_id,
+    }

api/score_api.py

@@ -0,0 +1,70 @@
+from fastapi import APIRouter, Depends, HTTPException
+from schema.score_request import ScoreRequest, ScoreUpdateRequest, ScoreResponse
+from database import get_db
+from dao.score_dao import (
+    calc_level, calc_pass,
+    add_score_dao, update_score_dao, delete_score_dao, get_scores_dao
+)
+
+score_api = APIRouter()
+
+
+# ========== 1. 添加成绩（按考核序次录入） ==========
+@score_api.post('', response_model=ScoreResponse, summary='添加成绩')
+def add_score(score: ScoreRequest, db=Depends(get_db)):
+    d = score.model_dump()
+    d['score_level'] = calc_level(d['score'])     # ← 调 DAO 的工具函数
+    d['is_pass'] = calc_pass(d['score'])
+
+    r = add_score_dao(o=d, db=db)
+    if not r:
+        raise HTTPException(
+            status_code=500,
+            detail='添加失败！'
+        )
+    return ScoreResponse(totals=1, data=d)
+
+
+# ========== 2. 查询成绩（动态条件 + 分页） ==========
+@score_api.get('', summary='查询成绩（支持多条件+分页）')
+def get_scores(
+    score_id: int = None,
+    student_id: str = None,
+    course_id: str = None,
+    teacher_id: str = None,
+    exam_order: int = None,
+    page: int = 1,
+    page_size: int = 10,
+    db=Depends(get_db)
+):
+    r = get_scores_dao(
+        score_id=score_id, student_id=student_id, course_id=course_id,
+        teacher_id=teacher_id, exam_order=exam_order,
+        page=page, page_size=page_size, db=db
+    )
+    if r:
+        return r
+    raise HTTPException(status_code=404, detail='成绩不存在！')
+
+
+# ========== 3. 修改成绩 ==========
+@score_api.put('/{score_id}', summary='修改成绩')
+def update_score(score_id: int, score: ScoreUpdateRequest, db=Depends(get_db)):
+    d = score.model_dump(exclude_unset=True)
+    if 'score' in d:
+        d['score_level'] = calc_level(d['score'])
+        d['is_pass'] = calc_pass(d['score'])
+
+    r = update_score_dao(score_id=score_id, update_data=d, db=db)
+    if not r:
+        raise HTTPException(status_code=500, detail='没有更新！')
+    return {'code': 200, 'totals': r, 'detail': '更新成功'}
+
+
+# ========== 4. 删除成绩（逻辑删除） ==========
+@score_api.delete('/{score_id}', summary='删除成绩(逻辑删除)')
+def delete_score(score_id: int, db=Depends(get_db)):
+    r = delete_score_dao(score_id=score_id, db=db)
+    if not r:
+        raise HTTPException(status_code=500, detail='删除异常，稍后操作！')
+    return {'code': 200, 'totals': r, 'detail': '删除成功'}

api/statistics_api.py

@@ -0,0 +1,126 @@
+from fastapi import APIRouter, Depends, Query
+from database import get_db
+from dao.statistics_dao import StatisticsDao
+from schema.statistics_schema import (
+    ResponseModel, ResponseModel1, ResponseModel2, ResponseModel3,
+    ResponseModel4, ResponseModel5, ResponseModel6, ClassStatResponse
+)
+from math import ceil
+
+BasicInformationAPI = APIRouter(tags=['统计分析'])
+
+
+@BasicInformationAPI.get('/basic-information', summary='查询年龄区间内的学员信息')
+def get_students_by_age_range_api(
+    n: int = Query(..., description='页码，从1开始', ge=1),
+    m: int = Query(..., description='每页条数', ge=1),
+    min_age: int = Query(..., description='最小年龄', ge=0),
+    max_age: int = Query(..., description='最大年龄', le=100),
+    db = Depends(get_db)):
+    if min_age > max_age:
+        return ResponseModel(code=400,
+                             message='最小年龄不能大于最大年龄',
+                             total=0,
+                             total_pages=0,
+                             data=[])
+
+    req, total = StatisticsDao.get_students_by_age_range_dao(n, m, min_age, max_age, db)
+
+    return ResponseModel(code=200,
+                         message='查询成功',
+                         total=total,
+                         total_pages=ceil(total / m) if total > 0 else 0,
+                         data=req)
+
+
+@BasicInformationAPI.get('/basic-information/{class_id}', summary='统计每个班级的人数以及男生/女生的人数')
+def get_students_by_class_id_api(class_id: str,
+                                 db = Depends(get_db)):
+    total_count, male_count, female_count = StatisticsDao.get_students_by_class_id_dao(class_id, db)
+    return ClassStatResponse(class_id=class_id,
+                             total_count=total_count,
+                             male_count=male_count,
+                             female_count=female_count)
+
+
+@BasicInformationAPI.get('/scores', summary='在某个分数段的学生')
+def get_students_by_score_api(
+        n: int = Query(..., description='页码，从1开始', ge=1),
+        m: int = Query(..., description='每页条数', ge=1),
+        score: float = Query(ge=0, le=100),
+        db = Depends(get_db)):
+    if score < 0 or score > 100:
+        return ResponseModel1(code=400,
+                              message='分数需要在0~100！',
+                              total=0,
+                              total_pages=0,
+                              data=[])
+    req, total = StatisticsDao.get_students_by_score_dao(n, m, score, db)
+    return ResponseModel1(code=200,
+                          message='查询成功',
+                          total=total,
+                          total_pages=ceil(total / m) if total > 0 else 0,
+                          data=req)
+
+
+@BasicInformationAPI.get('/scores_no_pass', summary='查询有n门成绩不合格的学生')
+def get_students_by_no_pass_api(
+        n: int = Query(..., description='页码,从1开始', ge=1),
+        m: int = Query(..., description='每页条数', ge=1),
+        fail_count: int = Query(..., description='不及格门数', ge=1),
+        db = Depends(get_db)):
+    req, total = StatisticsDao.get_student_by_no_pass_dao(n, m, fail_count, db)
+    return ResponseModel2(code=200,
+                          message='查询成功',
+                          total=total,
+                          total_pages=ceil(total / m) if total > 0 else 0,
+                          data=req)
+
+
+@BasicInformationAPI.get('/scores_avg', summary='统计每次考试每个班级的平均分（从高到低排序）')
+def get_class_exam_avg_api(
+        n: int = Query(..., description='页码,从1开始', ge=1),
+        m: int = Query(..., description='每页条数', ge=1),
+        db = Depends(get_db)):
+    req, total = StatisticsDao.get_class_exam_avg_score_dao(n, m, db)
+    return ResponseModel3(code=200,
+                          message='查询成功',
+                          total=total,
+                          total_pages=ceil(total / m) if total > 0 else 0,
+                          data=req)
+
+
+@BasicInformationAPI.get('/salary_top', summary='查询就业薪资最高的前m名学生')
+def get_salary_top_api(
+        m: int = Query(5, description='查询前m名', ge=1),
+        db = Depends(get_db)):
+    req = StatisticsDao.get_salary_top_dao(m, db)
+    return ResponseModel4(code=200,
+                          message='查询成功',
+                          data=req)
+
+
+@BasicInformationAPI.get('/time_size', summary='统计每个学生的就业时长（offer下发时间-就业开放时间）')
+def get_time_size_api(
+        n: int = Query(..., description='页码,从1开始', ge=1),
+        m: int = Query(..., description='每页条数', ge=1),
+        db = Depends(get_db)):
+    req, total = StatisticsDao.get_time_size_dao(n, m, db)
+    return ResponseModel5(code=200,
+                          message='查询成功',
+                          total=total,
+                          total_pages=ceil(total / m) if total > 0 else 0,
+                          data=req)
+
+
+@BasicInformationAPI.get('/class_avg_time_size', summary='统计每个班级的平均就业时长（只统计进入就业阶段的学生）')
+def get_class_avg_time_size_api(
+        n: int = Query(..., description='页码,从1开始', ge=1),
+        m: int = Query(..., description='每页条数', ge=1),
+        db = Depends(get_db)):
+    req, total = StatisticsDao.get_class_avg_time_size_dao(n, m, db)
+    return ResponseModel6(code=200,
+                          message='查询成功',
+                          total=total,
+                          total_pages=ceil(total / m) if total > 0 else 0,
+                          data=req)

dao/course_dao.py

@@ -0,0 +1,24 @@
+from model.course_model import Course
+
+
+def add_course_dao(o, db):
+    try:
+        c = Course(**o)
+        db.add(c)
+    except:
+        db.rollback()
+        return False
+    else:
+        db.commit()
+        return True
+
+
+def get_all_courses_dao(db):
+    return db.query(Course).filter(Course.is_deleted == 0).all()
+
+
+def get_course_by_id_dao(course_id, db):
+    return db.query(Course).filter(
+        Course.course_id == course_id,
+        Course.is_deleted == 0
+    ).first()

dao/score_dao.py

@@ -0,0 +1,98 @@
+from model.score_model import ScoreRecord
+
+
+# ========== 工具函数：算等级 ==========
+def calc_level(score):
+    if score is None: return None
+    if score >= 95: return 'A+'
+    if score >= 90: return 'A'
+    if score >= 80: return 'B'
+    if score >= 70: return 'C'
+    if score >= 60: return 'D'
+    return 'E'
+
+
+# ========== 工具函数：算及格 ==========
+def calc_pass(score):
+    if score is None: return None
+    return 1 if score >= 60 else 0
+
+
+# ========== 1. 添加成绩 ==========
+def add_score_dao(o, db):
+    try:
+        # 【业务校验】该学生该课程该序次是否已存在
+        existing = db.query(ScoreRecord).filter(
+            ScoreRecord.student_id == o['student_id'],
+            ScoreRecord.course_id == o['course_id'],
+            ScoreRecord.exam_order == o['exam_order'],
+            ScoreRecord.is_deleted == 0
+        ).all()
+
+        if existing:
+            raise ValueError('该学生该课程该序次成绩已存在')
+
+        o1 = ScoreRecord(**o)
+        db.add(o1)
+    except:
+        db.rollback()
+        return False
+    else:
+        db.commit()
+        return True
+
+
+# ========== 2. 修改成绩 ==========
+def update_score_dao(score_id, update_data, db):
+    try:
+        rows = db.query(ScoreRecord).filter(
+            ScoreRecord.score_id == score_id,
+            ScoreRecord.is_deleted == 0
+        ).update(update_data)
+    except:
+        db.rollback()
+        return False
+    else:
+        db.commit()
+        return rows
+
+
+# ========== 3. 查询成绩（动态条件 + 分页） ==========
+def get_scores_dao(score_id, student_id, course_id, teacher_id,
+                   exam_order, page, page_size, db):
+    q = db.query(ScoreRecord).filter(ScoreRecord.is_deleted == 0)
+
+    if score_id:
+        q = q.filter(ScoreRecord.score_id == score_id)
+    if student_id:
+        q = q.filter(ScoreRecord.student_id == student_id)
+    if course_id:
+        q = q.filter(ScoreRecord.course_id == course_id)
+    if teacher_id:
+        q = q.filter(ScoreRecord.teacher_id == teacher_id)
+    if exam_order:
+        q = q.filter(ScoreRecord.exam_order == exam_order)
+
+    r = q.order_by(ScoreRecord.exam_order.asc()) \
+         .offset((page - 1) * page_size) \
+         .limit(page_size).all()
+
+    if r:
+        return [item.to_dict() for item in r]
+    return []
+
+
+# ========== 4. 删除成绩 ==========
+def delete_score_dao(score_id, db):
+    try:
+        rows = db.query(ScoreRecord).filter(
+            ScoreRecord.score_id == score_id,
+            ScoreRecord.is_deleted == 0
+        ).update({'is_deleted': 1})
+    except:
+        db.rollback()
+        rows = 0
+    else:
+        db.commit()
+    finally:
+        return rows

dao/statistics_dao.py

@@ -0,0 +1,222 @@
+from model.student_info_region_model import Student_info
+from model.score_model import ScoreRecord
+from model.class_model import Class_info
+from model.employment_model import Employment
+from sqlalchemy import func, text
+
+GENDER_MAP = {0: '女', 1: '男'}
+STATUS_MAP = {0: '在读', 1: '休学', 2: '退学', 3: '毕业', 4: '结业'}
+EDUCATION_MAP = {1: '大专', 2: '本科', 3: '硕士研究生', 4: '博士研究生'}
+
+
+class StatisticsDao:
+    @staticmethod
+    def get_students_by_age_range_dao(n: int, m: int, min_age: int, max_age: int, db):
+        try:
+            age = func.timestampdiff(text('YEAR'), Student_info.birthday, func.curdate())
+            q = db.query(Student_info). \
+                filter(Student_info.is_deleted == '0'). \
+                filter(age >= min_age). \
+                filter(age <= max_age)
+            total = q.count()
+            req = q.offset((n - 1) * m).limit(m).all()
+        except Exception as e:
+            db.rollback()
+            raise e
+        else:
+            data = [
+                {
+                    "student_id": s.student_id,
+                    "student_name": s.student_name,
+                    "gender": GENDER_MAP.get(s.gender, str(s.gender)),
+                    "id_card": s.id_card,
+                    "birthday": s.birthday,
+                    "ethnicity": s.ethnicity,
+                    "region_id": s.region_id,
+                    "phone": s.phone,
+                    "major": s.major,
+                    "class_id": s.class_id,
+                    "enrollment_date": s.enrollment_date,
+                    "graduation_date": s.graduation_date,
+                    "student_status": STATUS_MAP.get(s.student_status, str(s.student_status)),
+                    "education_level": EDUCATION_MAP.get(s.education_level, str(s.education_level)),
+                }
+                for s in req
+            ]
+            return data, total
+
+    @staticmethod
+    def get_students_by_class_id_dao(class_id: str, db):
+        try:
+            q = db.query(Student_info).filter(Student_info.class_id == class_id). \
+                filter(Student_info.is_deleted == '0')
+            total_count = q.count()
+            male_count = q.filter(Student_info.gender == 1).count()
+            female_count = q.filter(Student_info.gender == 0).count()
+        except Exception as e:
+            db.rollback()
+            raise e
+        else:
+            return total_count, male_count, female_count
+
+    @staticmethod
+    def get_students_by_score_dao(n: int, m: int, score: float, db):
+        try:
+            q = (db.query(
+                Student_info.student_name,
+                Student_info.student_id,
+                func.min(ScoreRecord.score).label('score')
+            ).join(ScoreRecord, Student_info.student_id == ScoreRecord.student_id)
+                 .filter(Student_info.is_deleted == '0')
+                 .filter(ScoreRecord.is_deleted == 0)
+                 .group_by(Student_info.student_id, Student_info.student_name)
+                 .having(func.min(ScoreRecord.score) > score))
+            total = q.count()
+            req = q.offset((n - 1) * m).limit(m).all()
+        except Exception as e:
+            db.rollback()
+            raise e
+        else:
+            return [{"student_name": i.student_name,
+                     'student_id': i.student_id,
+                     "score": float(i.score) if i.score is not None else None} for i in req], total
+
+    @staticmethod
+    def get_student_by_no_pass_dao(n: int, m: int, fail_count: int, db):
+        try:
+            q = (db.query(
+                Student_info.student_id,
+                Student_info.student_name,
+                func.count(ScoreRecord.score_id).label('fail_count')
+            ).join(ScoreRecord, Student_info.student_id == ScoreRecord.student_id)
+                 .filter(Student_info.is_deleted == '0')
+                 .filter(ScoreRecord.is_deleted == 0)
+                 .filter(ScoreRecord.is_pass == 0)
+                 .group_by(Student_info.student_id, Student_info.student_name)
+                 .having(func.count(ScoreRecord.score_id) >= fail_count))
+            total = q.count()
+            req = q.offset((n - 1) * m).limit(m).all()
+        except Exception as e:
+            db.rollback()
+            raise e
+        else:
+            return [{"student_id": i.student_id,
+                     "student_name": i.student_name,
+                     "fail_count": i.fail_count} for i in req], total
+
+    @staticmethod
+    def get_class_exam_avg_score_dao(n: int, m: int, db):
+        try:
+            q = (db.query(
+                ScoreRecord.course_id,
+                Student_info.class_id,
+                Class_info.class_name,
+                func.avg(func.ifnull(ScoreRecord.score, 0)).label('avg_score')
+            ).join(Student_info, ScoreRecord.student_id == Student_info.student_id)
+                 .join(Class_info, Student_info.class_id == Class_info.class_id)
+                 .filter(Student_info.is_deleted == '0')
+                 .filter(ScoreRecord.is_deleted == 0)
+                 .filter(Class_info.is_deleted == False)
+                 .group_by(ScoreRecord.course_id,
+                           Student_info.class_id,
+                           Class_info.class_name)
+                 .order_by(func.avg(func.ifnull(ScoreRecord.score, 0)).desc()))
+            total = q.count()
+            req = q.offset((n - 1) * m).limit(m).all()
+        except Exception as e:
+            db.rollback()
+            raise e
+        else:
+            return [{"course_id": i.course_id,
+                     "class_id": i.class_id,
+                     "class_name": i.class_name,
+                     "avg_score": round(float(i.avg_score), 2)} for i in req], total
+
+    @staticmethod
+    def get_salary_top_dao(m: int, db):
+        try:
+            sq = (db.query(
+                Employment.student_id,
+                func.max(Employment.salary).label('max_salary')
+            ).filter(Employment.is_deleted == 0)
+             .group_by(Employment.student_id)
+             .order_by(func.max(Employment.salary).desc())
+             .limit(m)
+             .subquery())
+
+            q = (db.query(
+                Student_info.student_name,
+                Class_info.class_name,
+                Employment.offer_date,
+                Employment.company_name,
+                Employment.salary
+            ).join(Employment, Student_info.student_id == Employment.student_id)
+             .join(Class_info, Student_info.class_id == Class_info.class_id)
+             .join(sq,
+                   (Employment.student_id == sq.c.student_id) &
+                   (Employment.salary == sq.c.max_salary))
+             .filter(Student_info.is_deleted == '0')
+             .filter(Class_info.is_deleted == False)
+             .filter(Employment.is_deleted == 0))
+
+            req = q.all()
+        except Exception as e:
+            db.rollback()
+            raise e
+        else:
+            return [{"student_name": i.student_name,
+                     "class_name": i.class_name,
+                     "offer_date": i.offer_date,
+                     "company_name": i.company_name,
+                     "salary": float(i.salary) if i.salary is not None else None} for i in req]
+
+    @staticmethod
+    def get_time_size_dao(n: int, m: int, db):
+        try:
+            q = (db.query(
+                Student_info.student_id,
+                Student_info.student_name,
+                Employment.offer_date,
+                Employment.resume_open_date,
+                func.datediff(Employment.offer_date, Employment.resume_open_date).label('time_size')
+            ).join(Employment, Student_info.student_id == Employment.student_id)
+             .filter(Student_info.is_deleted == '0')
+             .filter(Employment.is_deleted == 0)
+             .filter(Employment.resume_open_date.isnot(None)))
+
+            total = q.count()
+            req = q.offset((n - 1) * m).limit(m).all()
+        except Exception as e:
+            db.rollback()
+            raise e
+        else:
+            return [{"student_id": i.student_id,
+                     "student_name": i.student_name,
+                     "offer_date": i.offer_date,
+                     "resume_open_date": i.resume_open_date,
+                     "time_size": i.time_size} for i in req], total
+
+    @staticmethod
+    def get_class_avg_time_size_dao(n: int, m: int, db):
+        try:
+            q = (db.query(
+                Class_info.class_id,
+                Class_info.class_name,
+                func.avg(func.datediff(Employment.offer_date, Employment.resume_open_date)).label('avg_time_size')
+            ).join(Student_info, Class_info.class_id == Student_info.class_id)
+             .join(Employment, Student_info.student_id == Employment.student_id)
+             .filter(Class_info.is_deleted == False)
+             .filter(Student_info.is_deleted == '0')
+             .filter(Employment.is_deleted == 0)
+             .filter(Employment.resume_open_date.isnot(None))
+             .group_by(Class_info.class_id, Class_info.class_name))
+
+            total = q.count()
+            req = q.offset((n - 1) * m).limit(m).all()
+        except Exception as e:
+            db.rollback()
+            raise e
+        else:
+            return [{"class_id": i.class_id,
+                     "class_name": i.class_name,
+                     "avg_time_size": round(float(i.avg_time_size), 2)} for i in req], total

main.py

@@ -7,6 +7,9 @@ from api.student_info_api import student_info_api
 from api.region_api import region_api
 from api.teacher_api import TeacherAPI
 from api.teacher_course_api import TeacherCourseAPI
+from api.course_api import course_api
+from api.score_api import score_api
+from api.statistics_api import BasicInformationAPI
 from starlette.middleware.cors import CORSMiddleware
 
 
@@ -31,6 +34,10 @@ app.include_router(TeacherAPI,tags=['老师信息'])
 
 app.include_router(TeacherCourseAPI,tags=['老师课程信息'])
 
+app.include_router(score_api,tags=['成绩信息'],prefix='/score')
+app.include_router(course_api,tags=['课程信息'],prefix='/course')
+
+app.include_router(BasicInformationAPI)
 app.include_router(employment_router)
 
 

model/course_model.py

@@ -0,0 +1,13 @@
+from database import DATETIME, Base, Column, Integer, String
+from datetime import datetime
+
+
+class Course(Base):
+    __tablename__ = 'student_course'
+
+    course_id = Column(String(20), primary_key=True, comment='课程编号')
+    course_name = Column(String(50), nullable=False, comment='课程名称')
+    teacher_id = Column(String(20), nullable=True, comment='授课教师编号')
+    is_deleted = Column(Integer, nullable=False, default=0, comment='0未删1已删')
+    create_time = Column(DATETIME, default=datetime.now, nullable=False)
+    update_time = Column(DATETIME, default=datetime.now, onupdate=datetime.now, nullable=False)

model/score_model.py

@@ -0,0 +1,79 @@
+from database import DATETIME, Base, Column, Integer, String, Numeric, Date
+from datetime import datetime
+from sqlalchemy import ForeignKey
+
+
+# 考试类型映射
+EXAM_TYPE_MAP = {
+    1: '期中',
+    2: '期末',
+    3: '月考',
+    4: '模拟',
+}
+
+
+class ScoreRecord(Base):
+    __tablename__ = 'student_score'
+
+    score_id = Column(Integer, primary_key=True, autoincrement=True,
+                      comment='成绩记录自增主键')
+    student_id = Column(
+        String(20),
+        # ForeignKey("student_info.student_id", ondelete="RESTRICT"),  # 等组员表定稿再加
+        nullable=False,
+        comment='学号'
+    )
+
+    exam_order = Column(
+        Integer,
+        nullable=False,
+        default=1,
+        comment='考核序次（第几次）'
+    )
+
+    course_id = Column(
+        String(20),
+        # ForeignKey("student_course.course_id", ondelete="RESTRICT"),
+        nullable=False,
+        comment='课程编号'
+    )
+    teacher_id = Column(
+        String(20),
+        # ForeignKey("teacher_info.teacher_id", ondelete="RESTRICT"),  # 等组员表定稿再加
+        nullable=False,comment='教师编号'
+    )
+    exam_type = Column(Integer, nullable=False, comment='1期中 2期末 3月考 4模拟')
+
+    score = Column(Numeric(5, 2), nullable=True, comment='分数0~100，缺考NULL')
+
+    score_level = Column(String(2), nullable=True, comment='等级A+~E')
+
+    is_pass = Column(Integer, nullable=True, default=0, comment='1及格 0不及格')
+
+    exam_date = Column(Date, nullable=False, comment='考试日期')
+
+    remark = Column(String(200), nullable=True, comment='缺考,缓考')
+
+    is_deleted = Column(Integer, nullable=False, default=0, comment='0未删 1已删')
+
+    create_time = Column(DATETIME, default=datetime.now, nullable=False)
+
+    update_time = Column(DATETIME, default=datetime.now,
+                         onupdate=datetime.now, nullable=False)
+
+    # ========== 实例方法：转字典 ==========
+    def to_dict(self):
+        return {
+            'score_id': self.score_id,
+            'student_id': self.student_id,
+            'course_id': self.course_id,
+            'teacher_id': self.teacher_id,
+            'exam_order': self.exam_order,
+            'exam_type': self.exam_type,
+            'exam_type_name': EXAM_TYPE_MAP.get(self.exam_type, '未知'),
+            'score': float(self.score) if self.score is not None else None,
+            'score_level': self.score_level,
+            'is_pass': self.is_pass,
+            'exam_date': str(self.exam_date),
+            'remark': self.remark,
+        }

model/statistics_model.py

@@ -0,0 +1,118 @@
+from sqlalchemy import *
+from database import DATETIME,Base,Column,Integer,String
+
+
+class StudentInfo(Base):
+    """学生基本信息表"""
+    __tablename__ = 'student_info'
+
+    student_id = Column(String(20), primary_key=True, comment='学生id')
+    student_name = Column(String(50), nullable=False, comment='学生姓名')
+    gender = Column(Enum('男', '女'), nullable=False, comment='性别')
+    id_card = Column(String(18), nullable=False, comment='身份证号')
+    birthday = Column(Date, nullable=False, comment='出生日期')
+    ethnicity = Column(String(20), nullable=False, comment='民族')
+    region_id = Column(String(20), ForeignKey('region.regio_id'),nullable=False, comment='籍贯编码，连接地区表')
+    phone = Column(String(11), nullable=False, comment='手机号')
+    major = Column(String(20), nullable=False, comment='就读专业')
+    class_id = Column(String(20),  nullable=False, comment='班级，连接班级表')
+    enrollment_date = Column(Date, nullable=False, comment='入学日期')
+    graduation_date = Column(Date, nullable=False, comment='毕业日期')
+    student_status = Column(Enum('在读', '休学', '退学', '毕业'), nullable=False, comment='学籍状态')
+    education_level = Column(Enum('大专', '本科', '硕士研究生', '博士研究生'), nullable=False, comment='学历')
+    is_deleted = Column(Enum('0', '1'), nullable=False, comment='逻辑删除，0=未删除, 1=已删除')
+    create_time = Column(Date, nullable=False, comment='创建时间')
+    update_time = Column(Date, nullable=False, comment='更新时间')
+
+
+class Region(Base):
+    """地区维度表"""
+    __tablename__ = 'region'
+
+    regio_id = Column(String(50), primary_key=True, comment='地区编码')
+    province = Column(String(50), nullable=False, comment='省/直辖市')
+    city = Column(String(50), nullable=False, comment='市')
+    district = Column(String(50), nullable=False, comment='区，县')
+    is_deleted = Column(Enum('0', '1'), nullable=False, comment='逻辑删除：0=未删除, 1=已删除')
+    create_time = Column(DATETIME, nullable=False, comment='创建时间')
+    update_time = Column(DATETIME, nullable=False, comment='更新时间')
+
+class EmploymentInfo(Base):
+    """学生就业信息表"""
+    __tablename__ = 'employment_info'
+
+    emp_id = Column(String(50), primary_key=True, comment='就业记录ID')
+    student_id = Column(String(20), ForeignKey('student_info.student_id'), nullable=False,
+                        comment='学号，连接学生表')
+    regio_id = Column(String(50), ForeignKey('region.regio_id'), nullable=False, comment='就业地区编码，连接地区表')
+    job_name = Column(String(50), nullable=False, comment='就业岗位名称')
+    company_name = Column(String(50), nullable=False, comment='就业公司名称')
+    salary = Column(Float, nullable=False, comment='就业月薪（元）')
+    offer_date = Column(Date, nullable=False, comment='offer下发/签约时间')
+    part_time = Column(Date, nullable=False, comment='就业时间')
+    resume_open_date = Column(Date, nullable=True, comment='开放简历时间')
+    employment_status = Column(Enum('1', '2', '3', '4'), nullable=False,
+                               comment='就业状态：1=未就业, 2,=已就业，3=升学, 4=灵活就业')
+    is_deleted = Column(Enum('0', '1'), nullable=False, comment='逻辑删除：0=未删除, 1=已删除')
+    create_time = Column(DATETIME, nullable=False, comment='创建时间')
+    update_time = Column(DATETIME, nullable=False, comment='更新时间')
+
+class StudentScore(Base):
+    """学生考核成绩表"""
+    __tablename__ = 'student_score'
+
+    score_id = Column(Integer, primary_key=True, autoincrement=True, comment='成绩记录自增主键')
+    student_id = Column(String(20), ForeignKey('student_info.student_id'),nullable=False,
+                        comment='学号，连接学生表')
+    course_id = Column(String(20),  ForeignKey('course_info.course_id'),nullable=False, comment='课程编号，连接课程表')
+    exam_type = Column(Enum('期中', '期末', '月考', '周考', '模拟考'), nullable=False,
+                       comment='考试类型：1=期中, 2=期末, 3=月考, 4=周考，5=模拟考')
+    score = Column(Float, nullable=True, comment='考试分数0~100，缺考为NULL')
+    score_level = Column(String(10), nullable=False, comment='成绩等级')
+    is_pass = Column(Enum('0', '1'), nullable=False, comment='是否及格：1=及格, 0=不及格（可由score>=60判断）')
+    exam_date = Column(Date, nullable=False, comment='考试日期')
+    exam_category = Column(Enum('首考', '重考', '补考'), nullable=False, comment='考试分类：1=首考, 2=补考, 3=重考')
+    is_deleted = Column(Enum('0', '1'), nullable=False, comment='逻辑删除：0=未删除, 1=已删除')
+    create_time = Column(DATETIME, nullable=False, comment='创建时间')
+    update_time = Column(DATETIME, nullable=False, comment='更新时间')
+
+class ClassInfo(Base):
+    """班级信息表"""
+    __tablename__ = 'class_info'
+
+    class_id = Column(String(50), primary_key=True, comment='班级业务编码')
+    class_name = Column(String(50), nullable=False, comment='班级名称')
+    grade_year = Column(Date, nullable=False, comment='入学年级')
+    status = Column(Enum('1', '2', '3'), nullable=False, comment='班级状态：1=在读 2=已毕业 3=已停用')
+    tags = Column(String(100), nullable=False, comment='存储多维标签，例如：["重点班", "科技特长", "2026届"]')
+    is_deleted = Column(Enum('0', '1'), nullable=False, comment='逻辑删除：0=未删除, 1=已删除')
+    create_time = Column(DATETIME, nullable=False, comment='创建时间')
+    update_time = Column(DATETIME, nullable=False, comment='更新时间')
+
+class TeacherInfo(Base):
+    """教师基本信息表"""
+    __tablename__ = 'teacher_info'
+
+    teacher_id = Column(String(20), primary_key=True, comment='教师编号')
+    teacher_name = Column(String(20), nullable=False, comment='教师姓名')
+    gender = Column(Enum('男', '女'), nullable=False, comment='性别')
+    birthday = Column(Date, nullable=False, comment='出生日期')
+    region_id = Column(String(50), ForeignKey('region.regio_id'), nullable=False, comment='籍贯地区编码')
+    email = Column(String(50), nullable=False, comment='邮箱')
+    phone = Column(String(11), nullable=False, comment='手机号码')
+    graduation_school = Column(String(50), nullable=False, comment='毕业学校')
+    teacher_rank = Column(Enum('1', '2', '3', '4'), nullable=False, comment='职称：1=初级, 2=中级, 3=高级, 4=特级')
+    is_deleted = Column(Enum('0', '1'), nullable=False, comment='逻辑删除：0=未删除, 1=已删除')
+    create_time = Column(DATETIME, nullable=False, comment='创建时间')
+    update_time = Column(DATETIME, nullable=False, comment='更新时间')
+
+class CourseInfo(Base):
+    """课程基础信息"""
+    __tablename__ = 'course_info'
+
+    course_id = Column(String(30), primary_key=True, comment='课程编号')
+    course_name = Column(String(30), nullable=False, comment='课程名称')
+    course_type = Column(Enum('1', '2', '3'), nullable=False, comment='课程类型：1 必修课 2 选修课 3 实训课')
+    is_deleted = Column(Enum('0', '1'), nullable=False, comment='逻辑删除：0=未删除, 1=已删除')
+    create_time = Column(DATETIME, nullable=False, comment='创建时间')
+    update_time = Column(DATETIME, nullable=False, comment='更新时间')

model/student_info_region_model.py

@@ -2,7 +2,7 @@
 from database import Base, engine as db_engine
 
 # 从 sqlalchemy 导入 Enum，并别名为 SAEnum，用于定义数据库枚举列
-from sqlalchemy import Enum as SAEnum, Integer
+from sqlalchemy import Enum as SAEnum, Integer, String
 
 # 从 sqlalchemy 导入常用组件：inspect（检查表是否存在）、Column（列）、
 # VARCHAR/BIGINT/Date/DateTime（字段类型）、func（SQL 函数，如 now()）

schema/course_request.py

@@ -0,0 +1,15 @@
+from pydantic import BaseModel, Field
+from typing import Optional
+
+
+class CourseRequest(BaseModel):
+    course_id: str = Field(..., description='课程编号')
+    course_name: str = Field(..., description='课程名称')
+    teacher_id: Optional[str] = Field(None, description='授课教师编号')
+
+
+class CourseResponse(BaseModel):
+    code: int = 200
+    detail: str = 'OK'
+    totals: int = 0
+    data: str | dict | tuple | list

schema/score_request.py

@@ -0,0 +1,29 @@
+from pydantic import BaseModel, Field
+from datetime import date
+from typing import Optional
+
+
+class ScoreRequest(BaseModel):
+    student_id: str = Field(...,min_length=1, max_length=20,description="学号")
+    course_id: str = Field(..., min_length=1, max_length=20,description="课程编号")
+    teacher_id: str = Field(..., min_length=1, max_length=20,description="教师编号")
+    exam_order: int = Field(1,ge=1, description="考核序次(第几次考核）")
+    exam_type: int = Field(...,ge=1, le=4,description="考试类型：1=期中，2=期末，3=月考，4=模拟")
+    score: Optional[float] = Field(None,ge=0, le=100,description="分数 0~100，缺考填 null")
+    exam_date: date = Field(..., description="考试日期（YYYY-MM-DD）")
+    remark: Optional[str] = Field(None, description="备注，例如：缺考、缓考")
+
+
+class ScoreUpdateRequest(BaseModel):
+    score: Optional[float] = Field(
+        None,
+        ge=0, le=100,
+        description="新分数 0~100"
+    )
+    remark: Optional[str] = Field(None, description="新备注")
+
+class ScoreResponse(BaseModel):
+    code: int = 200
+    detail: str = 'OK'
+    totals: int = 0
+    data: str | dict | tuple | list

schema/statistics_schema.py

@@ -0,0 +1,125 @@
+from pydantic import BaseModel, ConfigDict
+from datetime import date
+from typing import List, Optional
+
+
+class StudentInfo1(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    student_id: str
+    student_name: str
+    gender: str
+    id_card: str
+    birthday: Optional[date] = None
+    ethnicity: Optional[str] = None
+    region_id: Optional[str] = None
+    phone: Optional[str] = None
+    major: str
+    class_id: str
+    enrollment_date: Optional[date] = None
+    graduation_date: Optional[date] = None
+    student_status: str
+    education_level: str
+
+
+class ResponseModel(BaseModel):
+    code: int
+    message: str = 'ok'
+    total: int
+    total_pages: int
+    data: List[StudentInfo1]
+
+
+class ClassStatResponse(BaseModel):
+    code: int = 200
+    message: str = 'ok'
+    class_id: str
+    total_count: int
+    male_count: int
+    female_count: int
+
+
+class StudentInfo2(BaseModel):
+    student_id: str
+    student_name: str
+    score: Optional[float] = None
+
+
+class ResponseModel1(BaseModel):
+    code: int
+    message: str = 'ok'
+    total: int
+    total_pages: int
+    data: List[StudentInfo2]
+
+
+class StudentInfo3(BaseModel):
+    student_id: str
+    student_name: str
+    fail_count: int
+
+
+class ResponseModel2(BaseModel):
+    code: int
+    message: str = 'ok'
+    total: int
+    total_pages: int
+    data: List[StudentInfo3]
+
+
+class ClassExamAvg(BaseModel):
+    course_id: str
+    class_id: str
+    class_name: str
+    avg_score: float
+
+
+class ResponseModel3(BaseModel):
+    code: int
+    message: str = 'ok'
+    total: int
+    total_pages: int
+    data: List[ClassExamAvg]
+
+
+class SalaryTop(BaseModel):
+    student_name: str
+    class_name: str
+    company_name: str
+    offer_date: Optional[date] = None
+    salary: Optional[float] = None
+
+
+class ResponseModel4(BaseModel):
+    code: int
+    message: str = 'ok'
+    data: List[SalaryTop]
+
+
+class TimeSize(BaseModel):
+    student_id: str
+    student_name: str
+    offer_date: Optional[date] = None
+    resume_open_date: Optional[date] = None
+    time_size: Optional[int] = None
+
+
+class ResponseModel5(BaseModel):
+    code: int
+    message: str = 'ok'
+    total: int
+    total_pages: int
+    data: List[TimeSize]
+
+
+class ClassAvgTimeSize(BaseModel):
+    class_id: str
+    class_name: str
+    avg_time_size: float
+
+
+class ResponseModel6(BaseModel):
+    code: int
+    message: str = 'ok'
+    total: int
+    total_pages: int
+    data: List[ClassAvgTimeSize]