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