feat: work shifts by QR codes

generators/work_shifts_qr_code_generator/__init__.py

@@ -0,0 +1 @@
+from .generator import WorkShiftsQRCodeGenerator

generators/work_shifts_qr_code_generator/generator.py

@@ -0,0 +1,68 @@
+import os
+from io import BytesIO
+
+from reportlab.pdfgen.canvas import Canvas
+from reportlab_qrcode import QRCodeImage
+from reportlab.lib.styles import ParagraphStyle, getSampleStyleSheet
+from reportlab.lib.units import mm
+from reportlab.pdfbase import pdfmetrics
+from reportlab.pdfbase.ttfonts import TTFont
+from reportlab.platypus import SimpleDocTemplate, Paragraph
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from constants import APP_PATH
+from models import User
+from services.user import UserService
+
+
+class WorkShiftsQRCodeGenerator:
+    def __init__(self, session: AsyncSession):
+        self._session = session
+        assets_folder = os.path.join(APP_PATH, 'assets')
+        fonts_folder = os.path.join(assets_folder, 'fonts')
+        font_file_path = os.path.join(fonts_folder, 'DejaVuSans.ttf')
+        self.page_width = 58 * mm
+        self.page_height = 40 * mm
+        pdfmetrics.registerFont(TTFont('DejaVuSans', font_file_path))
+
+        styles = getSampleStyleSheet()
+        self.small_style = ParagraphStyle(
+            'Small',
+            parent=styles['Normal'],
+            alignment=1,
+            fontName='DejaVuSans',  # Specify the new font
+            fontSize=9,
+            leading=12,
+            spaceAfter=2,
+            rightIndent=2,
+        )
+
+    def _create_doc(self, buffer):
+        return SimpleDocTemplate(
+            buffer,
+            pagesize=(self.page_width, self.page_height),
+            rightMargin=1,
+            leftMargin=1,
+            topMargin=1,
+            bottomMargin=1
+        )
+
+    async def generate(self, user_id: int) -> BytesIO:
+        buffer = BytesIO()
+        doc = self._create_doc(buffer)
+
+        def on_first_page(canvas: Canvas, doc):
+            qr = QRCodeImage(str(user_id), size=33 * mm)
+            qr.drawOn(canvas, 34, 0)
+
+        user: User = await UserService(self._session).get_by_id(user_id)
+        position = user.position.name if user.position else ""
+        user_info = Paragraph(
+            f"{user.first_name} {user.second_name}\n{position}",
+            self.small_style
+        )
+
+        doc.build([user_info], on_first_page)
+
+        buffer.seek(0)
+        return buffer

main.py

@@ -45,6 +45,7 @@ routers_list = [
     routers.time_tracking_router,
     routers.billing_router,
     routers.task_router,
+    routers.work_shifts_router,
 ]
 for router in routers_list:
     app.include_router(router)

models/auth.py

@@ -5,6 +5,7 @@ from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from enums.user import UserRole
 from models.base import BaseModel
+from models.work_shifts import WorkShift
 
 if TYPE_CHECKING:
     from models.payroll import PayRate, PaymentRecord
@@ -91,6 +92,12 @@ class User(BaseModel):
         uselist=True,
         foreign_keys="PaymentRecord.user_id"
     )
+    work_shifts: Mapped[list["WorkShift"]] = relationship(
+        "WorkShift",
+        back_populates="user",
+        uselist=True,
+        foreign_keys="WorkShift.user_id"
+    )
 
 
 class Position(BaseModel):

models/work_shifts.py

@@ -0,0 +1,30 @@
+from datetime import datetime
+from typing import TYPE_CHECKING
+
+from sqlalchemy import ForeignKey
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from models.base import BaseModel
+
+if TYPE_CHECKING:
+    from models import User
+
+
+class WorkShift(BaseModel):
+    __tablename__ = "work_shifts"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    started_at: Mapped[datetime] = mapped_column(
+        nullable=False,
+    )
+    finished_at: Mapped[datetime] = mapped_column(
+        nullable=True,
+    )
+
+    user_id: Mapped[int] = mapped_column(
+        ForeignKey("users.id"),
+    )
+    user: Mapped["User"] = relationship(
+        "User",
+        back_populates="work_shifts",
+    )

requirements.txt

@@ -26,6 +26,7 @@ lexorank-py
 celery[redis]
 celery
 aioshutil
+reportlab-qrcode
 # PDF
 reportlab
 weasyprint

routers/__init__.py

@@ -13,3 +13,4 @@ from .payroll import payroll_router
 from .time_tracking import time_tracking_router
 from .billing import billing_router
 from .task import task_router
+from .work_shifts import work_shifts_router

routers/work_shifts.py

@@ -0,0 +1,87 @@
+from io import BytesIO
+
+from fastapi import APIRouter, Response
+
+from backend.dependecies import SessionDependency, CurrentUserDependency
+from generators.work_shifts_qr_code_generator import WorkShiftsQRCodeGenerator
+from schemas.work_shifts import StartShiftResponse, FinishShiftResponse, ActiveWorkShiftsResponse, DeleteShiftResponse, \
+    FinishShiftByIdResponse
+from services.work_shifts import WorkShiftsService
+
+work_shifts_router = APIRouter(
+    prefix="/work-shifts",
+    tags=["work-shifts"],
+)
+
+
+@work_shifts_router.get(
+    "/generate-qr-code/{user_id}",
+    operation_id="generate_qr_code",
+)
+async def generate_qr_code(
+        session: SessionDependency,
+        user_id: int,
+):
+    pdf_file: BytesIO = await WorkShiftsQRCodeGenerator(session).generate(user_id)
+    return Response(pdf_file.getvalue(), media_type="application/pdf")
+
+
+@work_shifts_router.post(
+    "/start-shift/{user_id}",
+    response_model=StartShiftResponse,
+    operation_id="start_shift",
+)
+async def start_shift(
+        session: SessionDependency,
+        user_id: int,
+):
+    return await WorkShiftsService(session).start_shift(user_id)
+
+
+@work_shifts_router.post(
+    "/finish-shift/{user_id}",
+    response_model=FinishShiftResponse,
+    operation_id="finish_shift",
+)
+async def finish_shift(
+        session: SessionDependency,
+        user_id: int,
+        user: CurrentUserDependency,
+):
+    return await WorkShiftsService(session).finish_shift(user, user_id)
+
+
+@work_shifts_router.post(
+    "/finish-shift-by-id/{shift_id}",
+    response_model=FinishShiftByIdResponse,
+    operation_id="finish_work_shift_by_id",
+)
+async def finish_work_shift_by_id(
+        session: SessionDependency,
+        user: CurrentUserDependency,
+        shift_id: int,
+):
+    return await WorkShiftsService(session).finish_shift_by_id(user, shift_id)
+
+
+@work_shifts_router.get(
+    "/get-active-shifts",
+    response_model=ActiveWorkShiftsResponse,
+    operation_id="get_active_shifts",
+)
+async def get_active_shifts(
+        session: SessionDependency,
+):
+    return await WorkShiftsService(session).get_active_shifts()
+
+
+@work_shifts_router.delete(
+    "/delete-shift/{shift_id}",
+    response_model=DeleteShiftResponse,
+    operation_id="delete_work_shift",
+)
+async def delete_work_shift(
+        session: SessionDependency,
+        shift_id: int,
+):
+    return await WorkShiftsService(session).delete_shift(shift_id)

schemas/work_shifts.py

@@ -0,0 +1,39 @@
+from datetime import datetime
+from typing import List
+
+from schemas.base import OkMessageSchema, BaseSchema
+from schemas.user import UserSchema
+
+
+# region Entities
+
+class ActiveWorkShiftSchema(BaseSchema):
+    id: int
+    started_at: datetime
+    user: UserSchema
+
+
+# endregion
+
+# region Responses
+
+class StartShiftResponse(OkMessageSchema):
+    pass
+
+
+class FinishShiftResponse(OkMessageSchema):
+    pass
+
+
+class FinishShiftByIdResponse(OkMessageSchema):
+    pass
+
+
+class ActiveWorkShiftsResponse(BaseSchema):
+    shifts: List[ActiveWorkShiftSchema]
+
+
+class DeleteShiftResponse(OkMessageSchema):
+    pass
+
+# endregion

services/work_shifts.py

@@ -0,0 +1,114 @@
+from datetime import datetime, date, timedelta
+
+from sqlalchemy import select, delete
+from sqlalchemy.orm import joinedload
+
+from models import WorkShift, User
+from schemas.time_tracking import UpdateTimeTrackingRecordRequest
+from schemas.user import *
+from schemas.work_shifts import StartShiftResponse, FinishShiftResponse, ActiveWorkShiftsResponse, DeleteShiftResponse, \
+    FinishShiftByIdResponse
+from services.base import BaseService
+from services.time_tracking import TimeTrackingService
+
+
+class WorkShiftsService(BaseService):
+    async def _get_last_work_shift_for_today(self, user_id: int) -> Optional[WorkShift]:
+        stmt = (
+            select(WorkShift)
+            .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 work_shift and work_shift.started_at.date() == date.today():
+            return work_shift
+        return None
+
+    async def start_shift(self, user_id: int) -> StartShiftResponse:
+        employee = await self.session.get(User, user_id)
+        if not employee:
+            return StartShiftResponse(ok=False, message=f"Пользователь с ID {user_id} не найден")
+
+        work_shift = await self._get_last_work_shift_for_today(user_id)
+        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(self, user: User, user_id: int) -> FinishShiftResponse:
+        employee = await self.session.get(User, user_id)
+        if not employee:
+            return FinishShiftResponse(ok=False, message=f"Пользователь с ID {user_id} не найден")
+
+        work_shift = await self._get_last_work_shift_for_today(user_id)
+
+        if not work_shift or work_shift.finished_at:
+            return FinishShiftResponse(ok=False, message="Смена для сотрудника еще не начата")
+
+        work_shift.finished_at = datetime.now()
+        await self.session.commit()
+
+        diff: timedelta = work_shift.finished_at - work_shift.started_at
+        hours = int(diff.total_seconds() // 3600)
+
+        data = UpdateTimeTrackingRecordRequest(
+            user_id=user_id,
+            date=work_shift.started_at.date(),
+            hours=hours,
+        )
+        await TimeTrackingService(self.session).update_record(user, data)
+
+        return FinishShiftResponse(ok=True, message=f"Смена закончена. Отработано {hours} ч.")
+
+    async def finish_shift_by_id(self, user: User, shift_id: int) -> FinishShiftByIdResponse:
+        work_shift = await self.session.get(WorkShift, shift_id)
+
+        if not work_shift or work_shift.finished_at:
+            return FinishShiftByIdResponse(ok=False, message="Смена для сотрудника еще не начата")
+
+        work_shift.finished_at = datetime.now()
+        await self.session.commit()
+
+        diff: timedelta = work_shift.finished_at - work_shift.started_at
+        hours = int(diff.total_seconds() // 3600)
+
+        data = UpdateTimeTrackingRecordRequest(
+            user_id=work_shift.user_id,
+            date=work_shift.started_at.date(),
+            hours=hours,
+        )
+        await TimeTrackingService(self.session).update_record(user, data)
+
+        return FinishShiftByIdResponse(ok=True, message=f"Смена закончена. Отработано {hours} ч.")
+
+    async def get_active_shifts(self) -> ActiveWorkShiftsResponse:
+        stmt = (
+            select(WorkShift)
+            .options(joinedload(WorkShift.user))
+            .where(WorkShift.finished_at == None)
+            .order_by(WorkShift.started_at.desc())
+        )
+
+        shifts = await self.session.execute(stmt)
+        shifts = shifts.scalars().all()
+        response = ActiveWorkShiftsResponse(shifts=shifts)
+        return response
+
+    async def delete_shift(self, shift_id: int) -> DeleteShiftResponse:
+        stmt = (
+            delete(WorkShift)
+            .where(WorkShift.id == shift_id)
+        )
+        await self.session.execute(stmt)
+        await self.session.commit()
+        return DeleteShiftResponse(ok=True, message="Запись о смене успешно удалена")