Print-pages limit.

.github/workflows/build_and_publish.yml

@@ -108,6 +108,7 @@ jobs:
             --env ALLOW_STUDENT_NUMBER=true \
             --env STATIC_FOLDER=/app/static \
             --env STORAGE_TIME=30 \
+            --env MAX_PAGE_COUNT=20 \
             --env AUTH_URL=https://auth.api.test.profcomff.com/ \
             --name ${{ env.CONTAITER_NAME }} \
             ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:test
@@ -174,6 +175,7 @@ jobs:
             --env ALLOW_STUDENT_NUMBER=true \
             --env STATIC_FOLDER=/app/static \
             --env STORAGE_TIME=168 \
+            --env MAX_PAGE_COUNT=20 \
             --env AUTH_URL=https://auth.api.profcomff.com/ \
             --name ${{ env.CONTAITER_NAME }} \
             ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:latest

migrations/versions/d63e9f7661dd_page_count.py

@@ -0,0 +1,24 @@
+"""page_count
+
+Revision ID: d63e9f7661dd
+Revises: f6fb6304fb74
+Create Date: 2023-05-15 18:38:40.964981
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+
+# revision identifiers, used by Alembic.
+revision = 'd63e9f7661dd'
+down_revision = 'f6fb6304fb74'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column('print_fact', sa.Column('sheets_used', sa.Integer(), nullable=True))
+
+
+def downgrade():
+    op.drop_column('print_fact', 'sheets_used')

print_service/base.py

@@ -0,0 +1,17 @@
+from pydantic import BaseModel
+
+
+class Base(BaseModel):
+    def __repr__(self) -> str:
+        attrs = []
+        for k, v in self.__class__.schema().items():
+            attrs.append(f"{k}={v}")
+        return "{}({})".format(self.__class__.__name__, ', '.join(attrs))
+
+    class Config:
+        orm_mode = True
+
+
+class StatusResponseModel(Base):
+    status: str
+    message: str

print_service/exceptions.py

@@ -0,0 +1,92 @@
+from print_service.settings import get_settings
+
+
+settings = get_settings()
+
+
+class ObjectNotFound(Exception):
+    pass
+
+
+class TerminalTokenNotFound(ObjectNotFound):
+    pass
+
+
+class TerminalQRNotFound(ObjectNotFound):
+    pass
+
+
+class PINNotFound(ObjectNotFound):
+    def __init__(self, pin: str):
+        self.pin = pin
+
+
+class UserNotFound(ObjectNotFound):
+    pass
+
+
+class FileNotFound(ObjectNotFound):
+    def __init__(self, count: int):
+        self.count = count
+
+
+class TooManyPages(Exception):
+    def __init__(self):
+        super().__init__(f'Content too large, count of page: {settings.MAX_PAGE_COUNT} is allowed')
+
+
+class TooLargeSize(Exception):
+    def __init__(self):
+        super().__init__(f'Content too large, {settings.MAX_SIZE} bytes allowed')
+
+
+class InvalidPageRequest(Exception):
+    def __init__(self):
+        super().__init__(f'Invalid format')
+
+
+class UnionStudentDuplicate(Exception):
+    def __init__(self):
+        super().__init__('Duplicates by union_numbers or student_numbers')
+
+
+class NotInUnion(Exception):
+    def __init__(self):
+        super().__init__(f'User is not found in trade union list')
+
+
+class PINGenerateError(Exception):
+    def __init__(self):
+        super().__init__(f'Can not generate PIN. Too many users?')
+
+
+class FileIsNotReceived(Exception):
+    def __init__(self):
+        super().__init__(f'No file was recieved')
+
+
+class InvalidType(Exception):
+    def __init__(self, content_type: str):
+        super().__init__(
+            f'Only {", ".join(settings.CONTENT_TYPES)} files allowed, but {content_type} was recieved'
+        )
+
+
+class AlreadyUploaded(Exception):
+    def __init__(self):
+        super().__init__(f'File has been already uploaded')
+
+
+class IsCorrupted(Exception):
+    def __init__(self):
+        super().__init__(f'File is corrupted')
+
+
+class IsNotUploaded(Exception):
+    def __init__(self):
+        super().__init__(f'File has not been uploaded yet')
+
+
+class UnprocessableFileInstance(Exception):
+    def __init__(self):
+        super().__init__(f'Unprocessable file instance')

print_service/models/__init__.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import math
 from datetime import datetime
 
 from sqlalchemy import Column, DateTime, Integer, String
@@ -45,6 +46,37 @@ class File(Model):
     owner: Mapped[UnionMember] = relationship('UnionMember', back_populates='files')
     print_facts: Mapped[list[PrintFact]] = relationship('PrintFact', back_populates='file')
 
+    @property
+    def flatten_pages(self) -> list[int] | None:
+        '''Возвращает расширенный список из элементов списков внутренних целочисленных точек переданного множества отрезков
+        "1-5, 3, 2" --> [1, 2, 3, 4, 5, 3, 2]'''
+        if self.number_of_pages is None:
+            return None
+        result = list()
+        if self.option_pages == '':
+            return result
+        for part in self.option_pages.split(','):
+            x = part.split('-')
+            result.extend(range(int(x[0]), int(x[-1]) + 1))
+        return result
+
+    @property
+    def sheets_count(self) -> int | None:
+        '''Возвращает количество элементов списков внутренних целочисленных точек переданного множества отрезков
+        "1-5, 3, 2" --> 7
+        P.S. 1, 2, 3, 4, 5, 3, 2 -- 7 чисел'''
+        if self.number_of_pages is None:
+            return None
+        if not self.flatten_pages:
+            return (
+                math.ceil(self.number_of_pages - (self.option_two_sided * self.number_of_pages / 2))
+                * self.option_copies
+            )
+        if self.option_two_sided:
+            return math.ceil(len(self.flatten_pages) / 2) * self.option_copies
+        else:
+            return len(self.flatten_pages) * self.option_copies
+
 
 class PrintFact(Model):
     __tablename__ = 'print_fact'
@@ -56,3 +88,5 @@ class PrintFact(Model):
 
     owner: Mapped[UnionMember] = relationship('UnionMember', back_populates='print_facts')
     file: Mapped[File] = relationship('File', back_populates='print_facts')
+
+    sheets_used: Mapped[int] = Column(Integer)

print_service/routes/__init__.py

@@ -1 +1,5 @@
+from . import exc_handlers
 from .base import app
+
+
+__all__ = ["app", "exc_handlers"]

print_service/routes/admin.py

@@ -5,6 +5,7 @@
 from fastapi import APIRouter, Depends, HTTPException
 from redis import Redis
 
+from print_service.exceptions import TerminalTokenNotFound
 from print_service.schema import BaseModel
 from print_service.settings import Settings, get_settings
 
@@ -52,7 +53,7 @@ async def manual_update_terminal(
         sender.redis.close()
         return {'status': 'ok'}
     sender.redis.close()
-    raise HTTPException(400, 'Terminal not found by token')
+    raise TerminalTokenNotFound()
 
 
 @router.post("/reboot")
@@ -65,4 +66,4 @@ async def reboot_terminal(
         sender.redis.close()
         return {'status': 'ok'}
     sender.redis.close()
-    raise HTTPException(400, 'Terminal not found by token')
+    raise TerminalTokenNotFound()

print_service/routes/exc_handlers.py

@@ -0,0 +1,147 @@
+import starlette.requests
+from starlette.responses import JSONResponse
+
+from print_service.base import StatusResponseModel
+from print_service.exceptions import (
+    AlreadyUploaded,
+    FileIsNotReceived,
+    FileNotFound,
+    InvalidPageRequest,
+    InvalidType,
+    IsCorrupted,
+    IsNotUploaded,
+    NotInUnion,
+    PINGenerateError,
+    PINNotFound,
+    TerminalQRNotFound,
+    TerminalTokenNotFound,
+    TooLargeSize,
+    TooManyPages,
+    UnionStudentDuplicate,
+    UnprocessableFileInstance,
+    UserNotFound,
+)
+from print_service.routes.base import app
+
+
+@app.exception_handler(TooLargeSize)
+async def too_large_size(req: starlette.requests.Request, exc: TooLargeSize):
+    return JSONResponse(
+        content=StatusResponseModel(status="Error", message=f"{exc}").dict(), status_code=413
+    )
+
+
+@app.exception_handler(TooManyPages)
+async def too_many_pages(req: starlette.requests.Request, exc: TooManyPages):
+    return JSONResponse(
+        content=StatusResponseModel(status="Error", message=f"{exc}").dict(), status_code=413
+    )
+
+
+@app.exception_handler(InvalidPageRequest)
+async def invalid_format(req: starlette.requests.Request, exc: TooManyPages):
+    return JSONResponse(
+        content=StatusResponseModel(status="Error", message=f"{exc}").dict(), status_code=416
+    )
+
+
+@app.exception_handler(TerminalQRNotFound)
+async def terminal_not_found_by_qr(req: starlette.requests.Request, exc: TerminalQRNotFound):
+    return JSONResponse(
+        content=StatusResponseModel(status="Error", message=f"Terminal not found by QR").dict(),
+        status_code=400,
+    )
+
+
+@app.exception_handler(TerminalTokenNotFound)
+async def terminal_not_found_by_token(req: starlette.requests.Request, exc: TerminalTokenNotFound):
+    return JSONResponse(
+        content=StatusResponseModel(status="Error", message=f"Terminal not found by token").dict(),
+        status_code=400,
+    )
+
+
+@app.exception_handler(UserNotFound)
+async def user_not_found(req: starlette.requests.Request, exc: UserNotFound):
+    return JSONResponse(
+        content=StatusResponseModel(status="Error", message=f"User not found").dict(), status_code=404
+    )
+
+
+@app.exception_handler(UnionStudentDuplicate)
+async def student_duplicate(req: starlette.requests.Request, exc: UnionStudentDuplicate):
+    return JSONResponse(
+        content=StatusResponseModel(status="Error", message=f"{exc}").dict(), status_code=400
+    )
+
+
+@app.exception_handler(NotInUnion)
+async def not_in_union(req: starlette.requests.Request, exc: NotInUnion):
+    return JSONResponse(
+        content=StatusResponseModel(status="Error", message=f"{exc}").dict(), status_code=403
+    )
+
+
+@app.exception_handler(PINGenerateError)
+async def generate_error(req: starlette.requests.Request, exc: PINGenerateError):
+    return JSONResponse(
+        content=StatusResponseModel(status="Error", message=f"{exc}").dict(), status_code=500
+    )
+
+
+@app.exception_handler(FileIsNotReceived)
+async def file_not_received(req: starlette.requests.Request, exc: FileIsNotReceived):
+    return JSONResponse(
+        content=StatusResponseModel(status="Error", message=f"{exc}").dict(), status_code=400
+    )
+
+
+@app.exception_handler(PINNotFound)
+async def pin_not_found(req: starlette.requests.Request, exc: PINNotFound):
+    return JSONResponse(
+        content=StatusResponseModel(status="Error", message=f"Pin {exc.pin} not found").dict(),
+        status_code=404,
+    )
+
+
+@app.exception_handler(InvalidType)
+async def invalid_type(req: starlette.requests.Request, exc: InvalidType):
+    return JSONResponse(
+        content=StatusResponseModel(status="Error", message=f"{exc}").dict(), status_code=415
+    )
+
+
+@app.exception_handler(AlreadyUploaded)
+async def already_upload(req: starlette.requests.Request, exc: AlreadyUploaded):
+    return JSONResponse(
+        content=StatusResponseModel(status="Error", message=f"{exc}").dict(), status_code=415
+    )
+
+
+@app.exception_handler(IsCorrupted)
+async def is_corrupted(req: starlette.requests.Request, exc: IsCorrupted):
+    return JSONResponse(
+        content=StatusResponseModel(status="Error", message=f"{exc}").dict(), status_code=415
+    )
+
+
+@app.exception_handler(UnprocessableFileInstance)
+async def unprocessable_file_instance(req: starlette.requests.Request, exc: UnprocessableFileInstance):
+    return JSONResponse(
+        content=StatusResponseModel(status="Error", message=f"{exc}").dict(), status_code=422
+    )
+
+
+@app.exception_handler(FileNotFound)
+async def file_not_found(req: starlette.requests.Request, exc: FileNotFound):
+    return JSONResponse(
+        content=StatusResponseModel(status="Error", message=f"{exc.count} file(s) not found").dict(),
+        status_code=404,
+    )
+
+
+@app.exception_handler(IsNotUploaded)
+async def not_uploaded(req: starlette.requests.Request, exc: IsNotUploaded):
+    return JSONResponse(
+        content=StatusResponseModel(status="Error", message=f"{exc}").dict(), status_code=415
+    )