上传文件至「dao」

2026-09-22 21:35:55 +08:00
parent 8d5c1749c4
commit 44c8d68ada
5 changed files with 414 additions and 0 deletions
@@ -0,0 +1,112 @@
+from model.class_model import Class
+from model.th_model import Teacher
+from sqlalchemy.orm import Session
+def add_class_dao(c,db):
+    try:
+        if db.query(Class).filter(
+                Class.class_id == c['class_id'],#如果请求体传进来的classid在表里找的到
+                Class.is_deleted==0 #且没有处于软删除状态
+        ).first():
+            return False, '班级编号已存在'
+        new_class = Class(**c)
+        db.add(new_class)
+        db.commit()
+        return True,new_class
+    except Exception as e:
+        db.rollback()
+        print('创建失败：',repr(e))
+        return False,str(e)
+def update_class_dao(id,update_data,db):
+    try:
+        rows = db.query(Class).filter(Class.id == id).update(update_data)
+    except Exception as e:
+        db.rollback()
+        print('更新失败：', repr(e))
+        return False, str(e)
+    else:
+        db.commit()
+        return rows
+'''
+def get_class_dao(id,db,page,page_size):
+        q = db.query(Class)
+        if id:
+            q = q.filter(Class.id == id,Class.is_deleted==0)
+        rows = q.offset((page-1)*page_size ).limit(page_size).all()
+        if rows:
+            return[{'class_id':i.id,'class_no':i.class_no,'class_name':i.class_name,'ht_id':i.ht_id,'t_id':i.t_id,'start_time':i.start_time}for i in rows]
+'''
+def get_class_list_dao(db: Session, page: int, page_size: int, **filters):
+    #软删除过滤
+    q = db.query(Class).filter(Class.is_deleted == 0)
+
+    # 精确匹配字段
+    if filters.get("class_no"):
+        q = q.filter(Class.class_id == filters["class_id"])
+    if filters.get("ht_id") is not None:
+        q = q.filter(Class.ht_id == filters["ht_id"])
+    if filters.get("t_id") is not None:
+        q = q.filter(Class.t_id == filters["t_id"])
+
+    # 模糊匹配字段（例如班级名称）
+    if filters.get("class_name"):
+        q = q.filter(Class.class_name.like(f"%{filters['class_name']}%"))
+
+    # 获取满足条件的总记录数（分页必备）
+    total = q.count()
+
+    # 分页查询
+    rows = q.offset((page - 1) * page_size).limit(page_size).all()
+
+    items = [ #如果匹配不到，则会返回空列表
+        {
+            'id': i.id,
+            'class_id': i.class_id,
+            'class_name': i.class_name,
+            'ht_id': i.ht_id,
+            't_id': i.t_id,
+            'start_time': i.start_time
+        }
+        for i in rows
+    ]
+    return total, items
+def get_class_by_id_dao(id: int, db: Session):
+    i = db.query(Class).filter(Class.id == id, Class.is_deleted == 0).first()
+    if not i:
+        return None
+    return {
+        'id': i.id,
+        'class_id': i.class_id,
+        'class_name': i.class_name,
+        'ht_id': i.ht_id,
+        't_id': i.t_id,
+        'start_time': i.start_time
+    }
+def delete_class_dao(id,db): #软删除
+    try:
+        rows = (db.query(Class)
+                .filter(Class.id == id,Class.is_deleted==0)#只匹配还没处于软删除状态的行
+                .update({Class.is_deleted:1},synchronize_session=False))#软删除就是把这个字段改成True
+    except Exception as e:
+        db.rollback()
+        print('删除失败：', repr(e))
+        return False, str(e)
+    else:
+        db.commit()
+        return rows
+def hard_delete_class_dao(id,db): #物理删除
+    try:
+        rows = (db.query(Class).filter(Class.id == id).delete(synchronize_session=False))
+        db.commit()
+        return rows
+    except IntegrityError:
+        db.rollback()
+        return False, '该班级被其他数据引用，无法彻底删除'
+    except Exception as e:
+        db.rollback()
+        print('物理删除失败', repr(e))
+        return False, str(e)
+def class_no_exist(class_id,db):
+    return (db.query(Class)
+            .filter(Class.class_id == class_id,Class.is_deleted==0)
+            .first() is not None
+            )
@@ -0,0 +1,65 @@
+from http.client import HTTPException
+
+from model.epy_model import Employment
+from fastapi import HTTPException,Depends
+from database import *
+
+
+def get_employment_dao(db, stu_id:str, company:str, salary:int):     # get变动
+    q = db.query(Employment).filter(Employment.is_deleted == False)
+    if stu_id:
+        q = q.filter(Employment.stu_id == stu_id)
+    if company:
+        q = q.filter(Employment.company == company)
+    if salary is not None:
+        q = q.filter(Employment.salary == salary)
+    return q.all()
+
+
+# def get_employment_dao(db):
+#     try:
+#         r = db.query(Employment).all()
+#     except Exception as e:
+#         db.rollback()
+#         raise HTTPException(status_code=500, detail=f'查询失败:str(e)')
+#     else:
+#         db.commit()
+#     return [{'stu_id':i.stu_id,'class_name':i.class_name,'company':i.company,'salary':i.salary} for i in r ]
+
+
+def wq_employment_dao(o,db):
+    try:
+        o1 = Employment(**o)
+        db.add(o1)
+    except Exception as e:
+        db.rollback()
+        raise HTTPException(status_code=500, detail=f'添加失败:str(e)')
+    else:
+        db.commit()
+        return 1
+
+def upd_employment_dao(e,db):
+    try:
+        rows = db.query(Employment).filter(Employment.stu_id == e['stu_id']).update(e)
+    except:
+        db.rollback()
+        raise HTTPException(status_code=500, detail='更新异常，请稍后再执行！')
+    else:
+        db.commit()
+        return 1
+
+def del_employment_dao(stu_id,db):
+    try:
+        row = db.query(Employment).filter(Employment.stu_id == stu_id).all()
+        if not row:
+            raise HTTPException(status_code=404,detail='记录不存在')
+        for i in row:
+            if i.is_deleted == 0:
+                db.query(Employment).filter(Employment.stu_id == i.stu_id).update({'is_deleted':1})
+    except Exception as e:
+        db.rollback()
+        raise HTTPException(status_code=500,detail=f'删除失败:{str(e)}')
+    else:
+        db.commit()
+        return 1
+
@@ -0,0 +1,47 @@
+from database import Session
+from model.sc_model import Scores
+def get_db():
+    db = Session()
+    try:
+        yield db
+    finally:
+        db.close()
+
+def a(d,db):
+    o = Scores(**d)
+    db.add(o)
+    db.commit()
+
+def a1(stu_id,db):
+    r=db.query(Scores).filter(Scores.stu_id==stu_id).delete()
+    db.commit()
+    return r
+
+def a2(stu_id,exam_seq,s,db):
+    r = db.query(Scores).filter((Scores.stu_id == stu_id) & (Scores.exam_seq == exam_seq)).update(
+    s.model_dump(exclude={'stu_id', 'exam_seq'}))
+    db.commit()
+    return r
+
+def a3(stu_id,exam_seq,db):
+    r=db.query(Scores).filter((Scores.stu_id==stu_id)&(Scores.exam_seq==exam_seq)).first()
+    return r
+
+def a4(stu_id, exam_seq, db):
+        r=db.query(Scores).filter((Scores.stu_id==stu_id)&(Scores.exam_seq==exam_seq)).update({Scores.deleted:1})
+        db.commit()
+        return r
+
+def a5(stu_id,db):
+    r = db.query(Scores).filter((Scores.stu_id == stu_id)).all()
+    s = 0
+    a = []
+    for i in r:
+        s += i.score
+        a.append(i.score)
+    return s,r,a
+
+
+
+
+
@@ -0,0 +1,121 @@
+from sqlalchemy import func, desc
+from sqlalchemy.exc import SQLAlchemyError
+from model.epy_model import Employment
+from fastapi import HTTPException
+from model.stu_model import Student
+from model.sc_model import Scores
+
+# ① 超30岁学员
+def get_stu_over_30(db):
+    try:
+        rows = db.query(Student).filter(Student.age>30,Student.is_deleted == 0).all()
+        return [{'id': i.stu_id, 'name': i.stu_name} for i in rows]
+    except SQLAlchemyError as e:
+        db.rollback()
+        raise HTTPException(status_code=400,detail=f'查询超三十岁学员信息失败:{str(e)}')
+
+# ② 各班人数 + 男女比
+def get_class_gender_count(db):
+    try:
+        rows = db.query(
+            Student.class_id,
+            Student.gender,
+            func.count(1).label('num')
+        ).group_by(Student.class_id, Student.gender).all()
+        return [{'班级': r.class_id, '性别': r.gender, '人数':r.num } for r in rows]
+    except Exception as e:
+        db.rollback()
+        raise HTTPException(status_code=400,detail=f'查询各班人数与男女比例失败:{str(e)}')
+
+# ③ 每次考试都≥80分的学生
+def get_stu_all_pass_80(db):
+    try:
+        rows = db.query(
+            Scores.stu_id,
+            Student.stu_name,
+            func.min(Scores.score).label('min_score')
+        ).join(Student, Student.stu_id == Scores.stu_id) \
+         .group_by(Scores.stu_id, Student.stu_name) \
+         .having(func.min(Scores.score) >= 80).all()
+        t = ['stu_id', 'stu_name', 'min_score']
+        l = [dict(zip(t, row)) for row in rows]
+        return  l
+    except Exception as e:
+        db.rollback()
+        raise HTTPException(status_code=400,detail=f'查询成绩大于80的学生失败:{str(e)}')
+
+# ④ 两次以上不及格的学生
+def get_stu_two_fail(db):
+    try:
+        rows = db.query(
+            Student.stu_name,
+            Student.class_id,
+        ).join(Scores, Scores.stu_id == Student.stu_id) \
+         .filter(Scores.score < 60) \
+         .group_by(Scores.stu_id,Student.stu_name,Student.class_id) \
+         .having(func.count(Scores.id) >= 2).all()
+        t = ['stu_name', 'class_id']
+        l = [dict(zip(t, row)) for row in rows]
+        return l
+    except Exception as e:
+        db.rollback()
+        raise HTTPException(status_code=400,detail=f'查询两次不及格的学生失败:{str(e)}')
+
+# ⑤ 每次考试各班平均分（降序）
+def get_class_avg_sc(db):
+    try:
+        rows = db.query(
+            Scores.exam_seq,
+            Student.class_id,
+            func.avg(Scores.score).label('avg_score')
+        ).join(Student, Student.stu_id == Scores.stu_id) \
+         .group_by(Scores.exam_seq, Student.class_id) \
+         .order_by(desc('avg_score')).all()
+        t = ['exam_seq', 'class_id','avg_score']
+        l = [dict(zip(t, row)) for row in rows]
+        return l
+    except Exception as e:
+        db.rollback()
+        raise HTTPException(status_code=400, detail=f'查询各班平均分失败:{str(e)}')
+
+# ⑥ 薪资前五名
+def get_top5_salary(db):
+    try:
+        rows = db.query(Student.stu_name, Student.class_id,Employment.work_date,Employment.company,Employment.salary).join(Student, Student.stu_id == Employment.stu_id).order_by(Employment.salary.desc()).limit(5).all()
+        t = ['stu_name','class_id','work_date','company','salary']
+        l = [dict(zip(t, row)) for row in rows]
+        print(l)
+        return {'code': 200, 'data': l}
+    except Exception as e:
+        db.rollback()
+        raise HTTPException(status_code=400,detail=f'查询薪资前五名失败失败:{str(e)}')
+
+# ⑦ 每个学生就业时长
+def get_stu_duration(db):
+    try:
+        rows = db.query(Employment.stu_id,func.datediff(Employment.offer_date,Employment.open_date).label('offer_diff')).group_by(Employment.stu_id,Employment.offer_date,Employment.open_date).all()
+        t = ['s_id', 'offer_diff']
+        l = [dict(zip(t, row)) for row in rows]
+        return {'code': 200, 'data': l}
+    except Exception as e:
+        db.rollback()
+        raise HTTPException(status_code=400,detail=f'查询每个学生的就业时长失败:{str(e)}')
+
+# ⑧ 各班平均就业时长
+def get_class_avg_duration(db):
+    try:
+        rows = db.query(
+            Student.class_id,
+          func.avg(func.datediff(Employment.offer_date, Employment.open_date)).label('avg_duration')
+        ).join(Student, Student.stu_id == Employment.stu_id).filter(
+            Employment.open_date.isnot(None)
+        ).group_by(Student.class_id).all()
+        t = ['class_id', 'avg_duration']
+        l = [dict(zip(t, row)) for row in rows]
+        return l
+    except Exception as e:
+        db.rollback()
+        raise HTTPException(status_code=400,detail=f'查询各班平均就业时长失败:{str(e)}')
+
+
+
@@ -0,0 +1,69 @@
+from sqlalchemy.orm import Session
+from model.stu_model import Student
+from sqlalchemy.exc import SQLAlchemyError
+from schema.stu_schema import StudentCreateRequest,StudentUpdateRequest
+def create_student(db: Session, student_create: StudentCreateRequest):
+
+    try:
+        db_student = Student(**student_create.model_dump())
+        db.add(db_student)
+        db.commit()
+    except SQLAlchemyError:
+        db.rollback()
+        return None
+    new_student = db.query(Student).filter(
+        Student.stu_id == student_create.stu_id,
+        Student.is_deleted == False).first()
+    return new_student
+
+
+def get_student_by_id(db: Session, stu_id: str):
+    query = db.query(Student).filter(
+        Student.stu_id == stu_id,
+        Student.is_deleted == False)
+    return query.first()
+
+
+def get_student_list(db: Session, stu_id: str | None = None,
+                     class_id: str = None, page: int = 1, page_size: int = 10, stu_name: str = None):
+    page = max(1, page)
+    page_size = max(1, min(100, page_size))
+    query = db.query(Student).filter(Student.is_deleted == False)
+    if stu_id:
+        query = query.filter(Student.stu_id == stu_id)
+    if stu_name:
+        query = query.filter(Student.stu_name.like(f"%{stu_name}%"))
+    if class_id:
+        query = query.filter(Student.class_id == class_id)
+
+    total = query.count()
+    offset = (page - 1) * page_size
+    db_student_list = query.offset(offset).limit(page_size).all()
+    return total, db_student_list
+
+
+def update_student(db: Session, stu_id: str, student_update: StudentUpdateRequest):
+    try:
+        db_student = get_student_by_id(db=db, stu_id=stu_id)
+        if not db_student:
+            return None
+        update_data = student_update.model_dump(exclude_unset=True)
+        for k, v in update_data.items():
+            setattr(db_student, k, v)
+        db.commit()
+    except SQLAlchemyError:
+        db.rollback()
+        return None
+    return db_student
+
+def delete_student_logic(db: Session, stu_id: str):
+    try:
+        db_student = get_student_by_id(db, stu_id)
+        if not db_student:
+            return None
+        db_student.is_deleted = True
+        db.commit()
+    except SQLAlchemyError:
+        db.rollback()
+        return None
+    return db_student