import math from datetime import date, timedelta, time from fastapi import HTTPException, status from sqlalchemy import select, func, extract, literal, label, Select from sqlalchemy.orm import joinedload, selectinload from logger import logger_builder from models import WorkShift, User from models.work_shifts import WorkShiftPause from schemas.base import PaginationSchema from schemas.time_tracking import UpdateTimeTrackingRecordRequest from schemas.work_shifts import * from services.base import BaseService from services.time_tracking import TimeTrackingService from utils.dependecies import is_valid_pagination from utils.work_time import hours_to_hours_and_minutes class WorkShiftsService(BaseService): async def _get_last_work_shift(self, user_id: int, for_today: bool = False) -> Optional[WorkShift]: stmt = ( select(WorkShift) .options(selectinload(WorkShift.pauses)) .where(WorkShift.user_id == user_id) .order_by(WorkShift.started_at.desc()) .limit(1) ) work_shift = await self.session.execute(stmt) work_shift = work_shift.one_or_none() work_shift = work_shift[0] if work_shift else None if for_today: return work_shift if work_shift and work_shift.started_at.date() == date.today() else None return work_shift async def start_shift(self, user_id: int) -> StartShiftResponse: employee = await self.session.get(User, user_id) if not employee or employee.is_deleted: return StartShiftResponse(ok=False, message=f"Пользователь с ID {user_id} не найден") work_shift = await self._get_last_work_shift(user_id, for_today=True) if work_shift: if not work_shift.finished_at: return StartShiftResponse(ok=False, message="Предыдущая смена еще не закончена") return StartShiftResponse(ok=False, message="Смена для сотрудника на сегодня уже закончена") work_shift = WorkShift(user_id=user_id, started_at=datetime.now()) self.session.add(work_shift) await self.session.commit() return StartShiftResponse(ok=True, message="Смена начата") async def finish_shift_by_user_id(self, user: User, user_id: int) -> FinishShiftResponse: employee = await self.session.get(User, user_id) if not employee or employee.is_deleted: return FinishShiftResponse(ok=False, message=f"Пользователь с ID {user_id} не найден") work_shift = await self._get_last_work_shift(user_id) ok, message = await self._finish_shift_common(user, work_shift) return FinishShiftResponse(ok=ok, message=message) async def finish_shift_by_id(self, user: User, shift_id: int) -> FinishShiftByIdResponse: stmt = ( select(WorkShift) .options(selectinload(WorkShift.pauses)) .where(WorkShift.id == shift_id) ) work_shift = await self.session.execute(stmt) work_shift = work_shift.scalars().one_or_none() ok, message = await self._finish_shift_common(user, work_shift) return FinishShiftByIdResponse(ok=ok, message=message) async def _finish_shift_common(self, user: User, work_shift: Optional[WorkShift]) -> tuple[bool, str]: logger = logger_builder.get_logger() try: if not work_shift or work_shift.finished_at: return False, "Смена для сотрудника еще не начата" # End active pause if work_shift.is_paused: await self.finish_pause_by_shift_id(work_shift.id) work_shift.finished_at = datetime.now() # Collect pauses pauses = [ (pause.started_at, pause.finished_at) for pause in work_shift.pauses ] # Build raw work intervals # Start with one interval: whole shift raw_intervals = [(work_shift.started_at, work_shift.finished_at)] # Subtract pauses from work intervals work_intervals = self.subtract_pauses(raw_intervals, pauses) # Split intervals by days daily_hours = self.split_intervals_by_days(work_intervals) # Create work records per day tts = TimeTrackingService(self.session) for day, hours in daily_hours.items(): if hours < (1 / 60): # ignore <1 minute continue data = UpdateTimeTrackingRecordRequest( user_id=work_shift.user_id, date=day, hours=hours, ) ok, msg = await tts.update_work_record(user, data, False) if not ok: raise Exception(msg) await self.session.commit() # Build human-readable result message total_work_seconds = sum(hours * 3600 for hours in daily_hours.values()) total_td = timedelta(seconds=total_work_seconds) h, m = hours_to_hours_and_minutes(total_td) logger.info(f"Успешное завершение смены. userID: {work_shift.user_id}. Отработано суммарно: {h} ч. {m} мин.") return True, f"Смена закончена. Отработано {h} ч. {m} мин." except Exception as e: logger.error(f"Ошибка завершения смены. userID: {work_shift.user_id}. Ошибка: {str(e)}") await self.session.rollback() return False, str(e) def subtract_pauses( self, work_intervals: list[tuple[datetime, datetime]], pauses: list[tuple[datetime, datetime]] ) -> list[tuple[datetime, datetime]]: result = [] for w_start, w_end in work_intervals: temp = [(w_start, w_end)] for p_start, p_end in pauses: new_temp = [] for s, e in temp: # pause outside interval → keep original if p_end <= s or p_start >= e: new_temp.append((s, e)) else: # pause cuts interval if p_start > s: new_temp.append((s, p_start)) if p_end < e: new_temp.append((p_end, e)) temp = new_temp result.extend(temp) return result def split_intervals_by_days(self, intervals: list[tuple[datetime, datetime]]) -> dict[date, float]: from collections import defaultdict res = defaultdict(float) for start, end in intervals: cur = start while cur.date() < end.date(): # end of current day day_end = datetime.combine(cur.date(), time.max) seconds = (day_end - cur).total_seconds() res[cur.date()] += seconds / 3600 cur = day_end + timedelta(seconds=1) # last segment (same day) seconds = (end - cur).total_seconds() res[cur.date()] += seconds / 3600 return res @staticmethod def get_work_shifts_history_stmt() -> Select: sub_hours = ( select( WorkShift.id, label( "total_hours", extract('epoch', WorkShift.finished_at) - extract('epoch', WorkShift.started_at) ), func.sum( extract('epoch', WorkShiftPause.finished_at) - extract('epoch', WorkShiftPause.started_at) ).label("pause_hours"), ) .join(WorkShiftPause, isouter=True) .where(WorkShift.finished_at.is_not(None)) .group_by(WorkShift.id) .subquery() ) return ( select( WorkShift, sub_hours.c.total_hours, sub_hours.c.pause_hours, ) .join(sub_hours, sub_hours.c.id == WorkShift.id) .options(joinedload(WorkShift.user)) .order_by(WorkShift.started_at.desc()) ) @staticmethod def get_active_work_shifts_stmt() -> Select: return ( select( WorkShift, literal(0), literal(0), ) .options(joinedload(WorkShift.user)) .where(WorkShift.finished_at.is_(None)) .order_by(WorkShift.started_at.desc()) ) @staticmethod def validate_pagination(pagination: PaginationSchema): if not is_valid_pagination(pagination): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid pagination") async def get_total_shifts_count(self, is_active: bool) -> int: count_query = ( select(func.count()) .select_from(WorkShift) .where(WorkShift.finished_at.is_(None) if is_active else WorkShift.finished_at.is_not(None)) ) return await self.session.scalar(count_query) @staticmethod def empty_shifts_response() -> GetWorkShiftsResponse: return GetWorkShiftsResponse( shifts=[], pagination_info=PaginationInfoSchema( total_pages=0, total_items=0 ) ) async def get_shifts(self, is_active: bool, pagination: PaginationSchema) -> GetWorkShiftsResponse: self.validate_pagination(pagination) page = max(0, pagination.page - 1) total_shifts = await self.get_total_shifts_count(is_active) if not total_shifts: return self.empty_shifts_response() total_pages = math.ceil(total_shifts / pagination.items_per_page) if is_active: stmt = self.get_active_work_shifts_stmt() else: stmt = self.get_work_shifts_history_stmt() stmt_with_pagination = ( stmt .offset(page * pagination.items_per_page) .limit(pagination.items_per_page) ) shifts_rows = await self.session.execute(stmt_with_pagination) shifts = [] for shift, total_hours, pause_hours in shifts_rows: shift = WorkShiftRowSchema( work_shift=shift, total_hours=total_hours, pause_hours=pause_hours, ) shifts.append(shift) return GetWorkShiftsResponse( shifts=shifts, pagination_info=PaginationInfoSchema( total_pages=total_pages, total_items=total_shifts, ) ) async def delete_shift(self, shift_id: int) -> DeleteShiftResponse: work_shift = await self.session.get(WorkShift, shift_id) if work_shift: await self.session.delete(work_shift) await self.session.commit() return DeleteShiftResponse(ok=True, message="Запись о смене успешно удалена") async def _get_last_work_shift_pause(self, work_shift_id: int) -> Optional[WorkShiftPause]: stmt = ( select(WorkShiftPause) .where(WorkShiftPause.work_shift_id == work_shift_id) .order_by(WorkShiftPause.started_at.desc()) .limit(1) ) work_shift_pause = await self.session.execute(stmt) work_shift_pause = work_shift_pause.one_or_none() work_shift_pause = work_shift_pause[0] if work_shift_pause else None return work_shift_pause if work_shift_pause and work_shift_pause.started_at.date() == date.today() else None async def start_pause(self, work_shift: Optional[WorkShift]) -> tuple[bool, str]: if not work_shift: return False, "Смена не найдена" if work_shift.finished_at: return False, "Смена уже завершена" work_shift_pause = await self._get_last_work_shift_pause(work_shift.id) if work_shift_pause and not work_shift_pause.finished_at: return False, "Перерыв для смены уже начат" work_shift.is_paused = True work_shift_pause = WorkShiftPause(work_shift_id=work_shift.id, started_at=datetime.now()) self.session.add(work_shift_pause) await self.session.commit() return True, "Перерыв начат" async def start_pause_by_user_id(self, user_id: int) -> StartPauseByUserIdResponse: last_shift = await self._get_last_work_shift(user_id) ok, message = await self.start_pause(last_shift) return StartPauseByUserIdResponse(ok=ok, message=message) async def start_pause_by_shift_id(self, shift_id: int) -> StartPauseByShiftIdResponse: work_shift = await self.session.get(WorkShift, shift_id) ok, message = await self.start_pause(work_shift) return StartPauseByShiftIdResponse(ok=ok, message=message) async def finish_pause(self, work_shift: Optional[WorkShift]) -> tuple[bool, str]: if not work_shift: return False, "Смена не найдена" if work_shift.finished_at: return False, "Смена уже завершена" work_shift_pause = await self._get_last_work_shift_pause(work_shift.id) if not work_shift_pause or work_shift_pause.finished_at: return False, "Перерыв еще не начат" work_shift.is_paused = False work_shift_pause.finished_at = datetime.now() await self.session.commit() diff: timedelta = work_shift_pause.finished_at - work_shift_pause.started_at hours, minutes = hours_to_hours_and_minutes(diff) return True, f"Перерыв закончен: {hours} ч. {minutes} мин." async def finish_pause_by_user_id(self, user_id: int) -> FinishPauseByUserIdResponse: last_shift = await self._get_last_work_shift(user_id) ok, message = await self.finish_pause(last_shift) return FinishPauseByUserIdResponse(ok=ok, message=message) async def finish_pause_by_shift_id(self, shift_id: int) -> FinishPauseByShiftIdResponse: work_shift = await self.session.get(WorkShift, shift_id) ok, message = await self.finish_pause(work_shift) return FinishPauseByShiftIdResponse(ok=ok, message=message)