shanxia2.0/backend/app/crud/leave.py

"""请假申请 CRUD"""
from __future__ import annotations

from datetime import datetime
from typing import Any, Dict, List, Optional, Tuple

from sqlalchemy import and_, or_
from sqlalchemy.orm import Session

from app.core.delete_guard import assert_row_deletable_by_creator
from app.core.utils import generate_uuid
from app.models.models import BizLeaveRequest


def total_pages(total: int, page_size: int) -> int:
    return (total + page_size - 1) // page_size if page_size else 0


def _row_to_dict(row: BizLeaveRequest) -> Dict[str, Any]:
    return {
        "id": row.id,
        "leaveType": row.leave_type,
        "startDate": row.start_date,
        "endDate": row.end_date,
        "days": row.days,
        "reason": row.reason or "",
        "wfInstanceId": row.wf_instance_id or "",
        "status": row.status,
        "tenantNo": row.tenant_no,
        "orderNum": row.order_num if row.order_num is not None else 0,
        "remark": row.remark or "",
        "createBy": row.create_by or "",
        "createTime": row.create_time.strftime("%Y-%m-%d %H:%M:%S") if row.create_time else "",
        "updateBy": row.update_by or "",
        "updateTime": row.update_time.strftime("%Y-%m-%d %H:%M:%S") if row.update_time else "",
    }


def apply_leave_list_filters(
    q,
    db: Session,
    *,
    keyword: Optional[str] = None,
    status: Optional[str] = None,
    reason_in: Optional[str] = None,
    tenant_no: Optional[str] = None,
    create_by: Optional[str] = None,
    user_info=None,
):
    """列表和列头候选值共用本人、数据权限及搜索条件。"""
    q = q.filter(BizLeaveRequest.del_flag == 0)
    if tenant_no:
        q = q.filter(BizLeaveRequest.tenant_no == tenant_no)
    if status:
        q = q.filter(BizLeaveRequest.status == status)
    if create_by:
        q = q.filter(BizLeaveRequest.create_by == create_by)
    kw = (keyword or "").strip()
    if kw:
        like = f"%{kw}%"
        q = q.filter(or_(BizLeaveRequest.reason.like(like), BizLeaveRequest.leave_type.like(like)))
    reasons_in = [s.strip() for s in str(reason_in or "").split(",") if s.strip()]
    if reasons_in:
        q = q.filter(BizLeaveRequest.reason.in_(reasons_in))
    if user_info:
        from app.core.data_scope import apply_data_scope
        q = apply_data_scope(q, BizLeaveRequest, user_info, db, tenant_no)
    return q


def list_leaves(
    db: Session,
    *,
    page: int = 1,
    page_size: int = 10,
    keyword: Optional[str] = None,
    status: Optional[str] = None,
    reason_in: Optional[str] = None,
    tenant_no: Optional[str] = None,
    create_by: Optional[str] = None,
    ids: Optional[List[str]] = None,
    max_page_size: Optional[int] = None,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    q = apply_leave_list_filters(
        db.query(BizLeaveRequest), db,
        keyword=keyword, status=status, reason_in=reason_in,
        tenant_no=tenant_no, create_by=create_by, user_info=user_info,
    )
    if ids:
        q = q.filter(BizLeaveRequest.id.in_(ids))
    total = q.count()
    page = max(1, page)
    cap = int(max_page_size) if max_page_size else 500
    page_size = min(max(1, page_size), cap)
    rows = (
        q.order_by(BizLeaveRequest.create_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    return [_row_to_dict(r) for r in rows], total


def get_leave(db: Session, leave_id: str) -> Optional[Dict[str, Any]]:
    row = db.query(BizLeaveRequest).filter(BizLeaveRequest.id == leave_id, BizLeaveRequest.del_flag == 0).first()
    return _row_to_dict(row) if row else None


def create_leave(
    db: Session,
    *,
    leave_type: str,
    start_date: str,
    end_date: str,
    days: int,
    reason: str = "",
    tenant_no: str = "shanxia2.0",
    user_id: str = "",
    dept_id: str = "",
    remark: str = "",
) -> BizLeaveRequest:
    row = BizLeaveRequest(
        id=generate_uuid(),
        leave_type=leave_type,
        start_date=start_date,
        end_date=end_date,
        days=days,
        reason=reason,
        status="draft",
        tenant_no=tenant_no,
        create_by=user_id,
        create_dept_id=dept_id,
        remark=remark,
    )
    db.add(row)
    db.flush()
    return row


def update_leave(db: Session, leave_id: str, *, user_id: str = "", **fields) -> None:
    row = db.query(BizLeaveRequest).filter(BizLeaveRequest.id == leave_id, BizLeaveRequest.del_flag == 0).first()
    if not row:
        return
    mapping = {
        "leaveType": "leave_type",
        "startDate": "start_date",
        "endDate": "end_date",
        "days": "days",
        "reason": "reason",
        "remark": "remark",
    }
    for camel, snake in mapping.items():
        if camel in fields and fields[camel] is not None:
            setattr(row, snake, fields[camel])
    row.update_by = user_id
    row.update_time = datetime.now()
    db.flush()


def update_leave_status(db: Session, leave_id: str, status: str, wf_instance_id: str = "", user_id: str = "") -> None:
    row = db.query(BizLeaveRequest).filter(BizLeaveRequest.id == leave_id, BizLeaveRequest.del_flag == 0).first()
    if not row:
        return
    row.status = status
    if wf_instance_id:
        row.wf_instance_id = wf_instance_id
    row.update_by = user_id
    row.update_time = datetime.now()
    db.flush()


def soft_delete_leave(db: Session, leave_id: str, user_id: str = "", *, user_info: dict = None) -> None:
    row = db.query(BizLeaveRequest).filter(BizLeaveRequest.id == leave_id, BizLeaveRequest.del_flag == 0).first()
    if not row:
        return
    assert_row_deletable_by_creator(row, user_info)
    row.del_flag = 1
    row.update_by = user_id
    row.update_time = datetime.now()
    db.flush()