JhHardwareWRS_BackPoint/app/services/work_schedule.py

from dataclasses import dataclass
import re

from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import Session

from app.models import AttendancePoint, WorkSchedule


TIME_PATTERN = re.compile(r"^\d{1,2}:\d{2}$")


@dataclass(frozen=True)
class WorkScheduleConfig:
    day_start: str
    day_end: str
    lunch_start: str
    lunch_end: str
    dinner_start: str
    dinner_end: str
    overtime_start: str
    overtime_end: str
    night_start: str
    night_end: str


DEFAULT_WORK_SCHEDULE = {
    "day_start": "08:00",
    "day_end": "17:20",
    "lunch_start": "11:40",
    "lunch_end": "12:40",
    "dinner_start": "17:20",
    "dinner_end": "18:00",
    "overtime_start": "18:00",
    "overtime_end": "20:00",
    "night_start": "20:00",
    "night_end": "06:00",
}

DEFAULT_WORK_SCHEDULE_CONFIG = WorkScheduleConfig(**DEFAULT_WORK_SCHEDULE)
DEFAULT_ATTENDANCE_RADIUS_METERS = 500
DEFAULT_AUTO_SUBMIT_HOURS = 15


def normalize_time_text(value: str) -> str:
    text = str(value or "").strip()
    if not TIME_PATTERN.match(text):
        raise ValueError("时间格式必须为 HH:MM")
    hour_text, minute_text = text.split(":", 1)
    hour = int(hour_text)
    minute = int(minute_text)
    if hour < 0 or hour > 23 or minute < 0 or minute > 59:
        raise ValueError("时间必须在 00:00-23:59 范围内")
    return f"{hour:02d}:{minute:02d}"


def schedule_to_config(schedule: WorkSchedule | None) -> WorkScheduleConfig:
    if schedule is None:
        return DEFAULT_WORK_SCHEDULE_CONFIG
    return WorkScheduleConfig(
        day_start=schedule.day_start,
        day_end=schedule.day_end,
        lunch_start=schedule.lunch_start,
        lunch_end=schedule.lunch_end,
        dinner_start=schedule.dinner_start,
        dinner_end=schedule.dinner_end,
        overtime_start=schedule.overtime_start,
        overtime_end=schedule.overtime_end,
        night_start=schedule.night_start,
        night_end=schedule.night_end,
    )


def attendance_point_to_config(point: AttendancePoint | None) -> WorkScheduleConfig | None:
    if point is None:
        return None
    return WorkScheduleConfig(
        day_start=point.day_start or DEFAULT_WORK_SCHEDULE["day_start"],
        day_end=point.day_end or DEFAULT_WORK_SCHEDULE["day_end"],
        lunch_start=point.lunch_start or DEFAULT_WORK_SCHEDULE["lunch_start"],
        lunch_end=point.lunch_end or DEFAULT_WORK_SCHEDULE["lunch_end"],
        dinner_start=point.dinner_start or DEFAULT_WORK_SCHEDULE["dinner_start"],
        dinner_end=point.dinner_end or DEFAULT_WORK_SCHEDULE["dinner_end"],
        overtime_start=point.overtime_start or DEFAULT_WORK_SCHEDULE["overtime_start"],
        overtime_end=point.overtime_end or DEFAULT_WORK_SCHEDULE["overtime_end"],
        night_start=point.night_start or DEFAULT_WORK_SCHEDULE["night_start"],
        night_end=point.night_end or DEFAULT_WORK_SCHEDULE["night_end"],
    )


def get_work_schedule(db: Session) -> WorkSchedule:
    schedule = db.get(WorkSchedule, 1)
    if schedule is not None:
        return schedule
    schedule = WorkSchedule(id=1, **DEFAULT_WORK_SCHEDULE)
    db.add(schedule)
    db.commit()
    db.refresh(schedule)
    return schedule


def get_work_schedule_config(db: Session, attendance_point_name: str | None = None) -> WorkScheduleConfig:
    try:
        point_name = str(attendance_point_name or "").strip()
        if point_name:
            point_config = attendance_point_to_config(db.get(AttendancePoint, point_name))
            if point_config is not None:
                return point_config
        return schedule_to_config(get_work_schedule(db))
    except SQLAlchemyError:
        db.rollback()
        return DEFAULT_WORK_SCHEDULE_CONFIG


def normalize_schedule_payload(payload) -> dict[str, str]:
    return {
        field: normalize_time_text(getattr(payload, field, DEFAULT_WORK_SCHEDULE[field]))
        for field in DEFAULT_WORK_SCHEDULE
    }


def update_work_schedule(db: Session, payload, updated_by: str, *, commit: bool = True) -> WorkSchedule:
    values = normalize_schedule_payload(payload)
    attendance_latitude = getattr(payload, "attendance_latitude", None)
    attendance_longitude = getattr(payload, "attendance_longitude", None)
    attendance_radius_meters = getattr(payload, "attendance_radius_meters", None)
    if (attendance_latitude is None) != (attendance_longitude is None):
        raise ValueError("考勤经纬度必须同时配置")
    if attendance_latitude is not None and not (-90 <= attendance_latitude <= 90):
        raise ValueError("考勤纬度必须在 -90 到 90 之间")
    if attendance_longitude is not None and not (-180 <= attendance_longitude <= 180):
        raise ValueError("考勤经度必须在 -180 到 180 之间")
    if attendance_radius_meters is None:
        attendance_radius_meters = DEFAULT_ATTENDANCE_RADIUS_METERS
    if int(attendance_radius_meters) <= 0:
        raise ValueError("考勤范围半径必须大于0米")
    auto_submit_hours = getattr(payload, "auto_submit_hours", DEFAULT_AUTO_SUBMIT_HOURS)
    try:
        auto_submit_hours = float(auto_submit_hours)
    except (TypeError, ValueError) as exc:
        raise ValueError("系统自动提交时长必须是数字") from exc
    if auto_submit_hours < 1 or auto_submit_hours > 168:
        raise ValueError("系统自动提交时长必须在1到168小时之间")
    schedule = get_work_schedule(db)
    for field, value in values.items():
        setattr(schedule, field, value)
    schedule.attendance_latitude = attendance_latitude
    schedule.attendance_longitude = attendance_longitude
    schedule.attendance_radius_meters = int(attendance_radius_meters)
    schedule.auto_submit_hours = auto_submit_hours
    schedule.updated_by = updated_by
    db.add(schedule)
    if commit:
        db.commit()
        db.refresh(schedule)
    else:
        db.flush()
    return schedule