167 lines
6.2 KiB
Python
167 lines
6.2 KiB
Python
# dao/employment_dao.py
|
|
# 就业表的数据访问层（基础表 + offer 表联动 + 学生状态联动）
|
|
from datetime import date
|
|
from typing import List, Optional, Tuple
|
|
|
|
from sqlalchemy.orm import Session, joinedload
|
|
|
|
from model.employment import EmploymentBase, EmploymentOffer
|
|
from model.students import Student
|
|
from model.classes import Classinfo
|
|
|
|
|
|
class EmploymentDAO:
|
|
@staticmethod
|
|
def get_base(db: Session, stu_id: int, include_deleted: bool = False) -> Optional[EmploymentBase]:
|
|
"""获取学生就业基础记录"""
|
|
query = db.query(EmploymentBase).filter(EmploymentBase.stu_id == stu_id)
|
|
if not include_deleted:
|
|
query = query.filter(EmploymentBase.is_deleted == 0)
|
|
return query.first()
|
|
|
|
# ==================== 就业开放登记 ====================
|
|
@staticmethod
|
|
def open_employment(db: Session, stu_id: int, open_time: date) -> EmploymentBase:
|
|
"""
|
|
登记就业开放（开放简历）：
|
|
1. 插入就业基础表记录（冗余字段从 student/class 同步写入）
|
|
2. 学生表状态更新为『进入就业』（需求 2.3 联动）
|
|
"""
|
|
student = db.query(Student).filter(Student.stu_id == stu_id).one()
|
|
class_row = (
|
|
db.query(Classinfo).filter(Classinfo.class_id == student.class_id).one()
|
|
)
|
|
db_base = EmploymentBase(
|
|
stu_id=stu_id,
|
|
employment_open_time=open_time,
|
|
stu_name=student.stu_name,
|
|
class_name=class_row.class_name,
|
|
)
|
|
student.status = "进入就业"
|
|
db.add(db_base)
|
|
db.commit()
|
|
db.refresh(db_base)
|
|
return db_base
|
|
|
|
# ==================== offer 登记 ====================
|
|
@staticmethod
|
|
def add_offer(db: Session, stu_id: int, offer_time: date, company_name: str, salary: float) -> EmploymentOffer:
|
|
"""
|
|
登记offer（拿到 offer）：
|
|
1. offer 表插入记录（offer_id 同一学生内自增）
|
|
2. 就业基础表更新 offer 时间 / 公司 / 薪资
|
|
3. 学生表状态更新为『已就业』（需求 2.3 联动）
|
|
"""
|
|
max_offer_id = (
|
|
db.query(EmploymentOffer.offer_id)
|
|
.filter(EmploymentOffer.stu_id == stu_id)
|
|
.order_by(EmploymentOffer.offer_id.desc())
|
|
.first()
|
|
)
|
|
offer_id = (max_offer_id[0] + 1) if max_offer_id else 1
|
|
db_offer = EmploymentOffer(
|
|
stu_id=stu_id,
|
|
offer_id=offer_id,
|
|
offer_time=offer_time,
|
|
company_name=company_name,
|
|
salary=salary,
|
|
)
|
|
# 更新基础表（以最新 offer 为准）
|
|
db_base = EmploymentDAO.get_base(db, stu_id)
|
|
db_base.job_time = offer_time
|
|
db_base.company_name = company_name
|
|
db_base.salary = salary
|
|
# 学生状态联动
|
|
student = db.query(Student).filter(Student.stu_id == stu_id).one()
|
|
student.status = "已就业"
|
|
db.add(db_offer)
|
|
db.commit()
|
|
db.refresh(db_offer)
|
|
return db_offer
|
|
|
|
# ==================== 查询 ====================
|
|
@staticmethod
|
|
def get_by_student(db: Session, stu_id: int) -> Optional[EmploymentBase]:
|
|
"""获取学生就业信息（含 offer 列表）"""
|
|
return (
|
|
db.query(EmploymentBase)
|
|
.options(joinedload(EmploymentBase.offers))
|
|
.filter(EmploymentBase.stu_id == stu_id, EmploymentBase.is_deleted == 0)
|
|
.first()
|
|
)
|
|
|
|
@staticmethod
|
|
def get_by_class(db: Session, class_id: int, skip: int = 0, limit: int = 100) -> Tuple[int, List[EmploymentBase]]:
|
|
"""获取班级学生的就业信息（通过学生表关联班级）"""
|
|
query = (
|
|
db.query(EmploymentBase)
|
|
.join(Student, EmploymentBase.stu_id == Student.stu_id)
|
|
.filter(Student.class_id == class_id, EmploymentBase.is_deleted == 0)
|
|
)
|
|
total = query.count()
|
|
items = (
|
|
query.options(joinedload(EmploymentBase.offers))
|
|
.order_by(EmploymentBase.stu_id)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
return total, items
|
|
|
|
@staticmethod
|
|
def get_all(
|
|
db: Session,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
stu_id: Optional[int] = None,
|
|
company_name: Optional[str] = None,
|
|
salary_min: Optional[float] = None,
|
|
salary_max: Optional[float] = None,
|
|
) -> Tuple[int, List[EmploymentBase]]:
|
|
"""多条件查询就业信息：学号精确、公司名模糊、薪资范围"""
|
|
query = db.query(EmploymentBase).filter(EmploymentBase.is_deleted == 0)
|
|
if stu_id is not None:
|
|
query = query.filter(EmploymentBase.stu_id == stu_id)
|
|
if company_name:
|
|
query = query.filter(EmploymentBase.company_name.like(f"%{company_name}%"))
|
|
if salary_min is not None:
|
|
query = query.filter(EmploymentBase.salary >= salary_min)
|
|
if salary_max is not None:
|
|
query = query.filter(EmploymentBase.salary <= salary_max)
|
|
total = query.count()
|
|
items = (
|
|
query.options(joinedload(EmploymentBase.offers))
|
|
.order_by(EmploymentBase.stu_id)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
return total, items
|
|
|
|
# ==================== 修改 / 删除 ====================
|
|
@staticmethod
|
|
def update_base(db: Session, stu_id: int, update_data: dict) -> Optional[EmploymentBase]:
|
|
"""修改就业基础信息（只更新传入的非 None 字段）"""
|
|
db_base = EmploymentDAO.get_base(db, stu_id)
|
|
if db_base is None:
|
|
return None
|
|
for key, value in update_data.items():
|
|
setattr(db_base, key, value)
|
|
db.commit()
|
|
db.refresh(db_base)
|
|
return db_base
|
|
|
|
@staticmethod
|
|
def delete_light(db: Session, stu_id: int) -> bool:
|
|
"""逻辑删除学生就业基础记录"""
|
|
db_base = EmploymentDAO.get_base(db, stu_id)
|
|
if db_base is None:
|
|
return False
|
|
db_base.is_deleted = 1
|
|
# 学生状态回退为在读
|
|
student = db.query(Student).filter(Student.stu_id == stu_id).first()
|
|
if student:
|
|
student.status = "在读"
|
|
db.commit()
|
|
return True
|