feat: add routes for printing

.gitignore

@@ -13,3 +13,5 @@ static
 ### Sensible ###
 /private.pem
 /public.pem
+
+/tmp

pyproject.toml

@@ -21,6 +21,7 @@ ruff = "^0.6.7"
 uvicorn = "^0.31.0"
 pycups = "^2.0.4"
 unoserver = "^3.1"
+python-multipart = "^0.0.20"
 
 [tool.poetry.group.mongo.dependencies]
 beanie = "^1.26.0"

src/api/app.py

@@ -59,6 +59,8 @@ def generate_unique_operation_id(route: APIRoute) -> str:
     allow_headers=["*"],
 )
 
+from src.modules.printing.routes import router as router_printing  # noqa: E402
 from src.modules.users.routes import router as router_users  # noqa: E402
 
 app.include_router(router_users)
+app.include_router(router_printing)

src/api/lifespan.py

@@ -1,7 +1,9 @@
 __all__ = ["lifespan"]
 
 import asyncio
+import glob
 import json
+import os
 from contextlib import asynccontextmanager
 
 from beanie import init_beanie
@@ -47,6 +49,11 @@ async def lifespan(_app: FastAPI):
     from src.modules.innohassle_accounts import innohassle_accounts  # noqa: E402
 
     await innohassle_accounts.update_key_set()
+
+    for rubbish in glob.glob("tmp/*"):
+        if "gitkeep" not in rubbish:
+            os.remove(rubbish)
+
     yield
 
     # -- Application shutdown --

src/modules/converting/repository.py

@@ -1,35 +1,14 @@
 __all__ = ["Converting"]
 
-import os
-
 import unoserver.client
 
 
 class Converting:
     def __init__(self, server="0.0.0.0", port=2003):
         self.client = unoserver.client.UnoClient(server, port)
 
-    def conversion_is_allowed(self, filename: str):
-        return filename[filename.rfind(".") + 1 :] in [
-            "doc",
-            "docx",
-            "png",
-            "txt",
-            "jpg",
-            "md",
-            "bmp",
-            "xlsx",
-            "xls",
-            "odt",
-            "ods",
-        ]
-
-    def any2pdf(self, filename: str):
-        if not self.conversion_is_allowed(filename):
-            print("The file cannot be converted.")
-            return
-        filepath = (here := os.path.dirname(__file__))[: here.index("src")] + f"files_to_be_printed/{filename}"
-        self.client.convert(inpath=filepath, outpath=f"{filepath[:filepath.rfind('.')]}.pdf")
+    def any2pdf(self, inpath: str, outpath: str):
+        self.client.convert(inpath=inpath, outpath=outpath)
 
 
 converting_repository: Converting = Converting()

src/modules/printing/repository.py

@@ -1,4 +1,4 @@
-__all__ = ["printing_repository"]
+__all__ = ["printing_repository", "PrintingOptions", "JobAttributes"]
 
 from typing import Literal
 
@@ -9,11 +9,11 @@
 class PrintingOptions(BaseModel):
     copies: str | None = None
     "Count of copies"
-    page_ranges: str | None = Field(alias="pages-ranges", default=None)
+    page_ranges: str | None = Field(None, alias="pages-ranges")
     "Which page ranges to print"
     sides: Literal["one-sided", "two-sided-long-edge"] | None = None
     "One-sided or double-sided printing"
-    number_up: Literal["1", "4", "9"] | None = Field(alias="number-up", default=None)
+    number_up: Literal["1", "4", "9"] | None = Field(None, alias="number-up")
     "Count of pages on a list"
 
 
@@ -27,8 +27,8 @@ class JobAttributes(BaseModel):
         "job-printing",
     ] = Field(alias="job-state-reasons")
     "The current state of a job from the getJobAttributes function"
-    printer_state: list[str] | None = Field(alias="job-printer-state-reasons", default=None)
-    "The current state of printer: 'cups-waiting-for-job-completed', 'media-needed-warning', 'media-empty-error', 'input-tray-missing'"
+    printer_state: list[str] | None = Field(None, alias="job-printer-state-reasons")
+    "The current state of printer: 'cups-waiting-for-job-completed', 'media-needed-warning', 'media-empty-error', 'input-tray-missing', 'media-empty-report'"
 
 
 # noinspection PyMethodMayBeStatic

src/modules/printing/routes.py

@@ -0,0 +1,76 @@
+import os
+import pathlib
+import tempfile
+
+from fastapi import APIRouter, UploadFile
+from fastapi.exceptions import HTTPException
+from starlette.responses import FileResponse
+
+from src.modules.converting.repository import converting_repository
+from src.modules.printing.repository import PrintingOptions, printing_repository
+
+router = APIRouter(prefix="/print", tags=["Print"])
+
+
+@router.get("/job_status")
+async def job_status(job_id: int) -> str:
+    """
+    Returns the status of a job
+    """
+
+    return printing_repository.get_job_status(job_id)
+
+
+tempfiles = {}
+dir_for_temp = os.getcwd() + "/tmp"
+
+
+@router.post("/prepare", responses={400: {"description": "Unsupported format"}})
+async def prepare_printing(file: UploadFile) -> str:
+    """
+    Convert a file to pdf and return the path to the converted file
+    """
+
+    ext = file.filename[file.filename.rfind(".") :]
+    if ext == ".pdf":
+        f = tempfile.NamedTemporaryFile(dir=dir_for_temp, suffix=".pdf")
+        f.write(await file.read())
+        tempfiles[f.name] = f
+        return f.name
+    elif ext in [".doc", ".docx", ".png", ".txt", ".jpg", ".md", ".bmp", ".xlsx", ".xls", ".odt", ".ods"]:
+        with (
+            tempfile.NamedTemporaryFile(dir=dir_for_temp, suffix=ext) as in_f,
+            tempfile.NamedTemporaryFile(dir=dir_for_temp, suffix=".pdf", delete=False) as out_f,
+        ):
+            in_f.write(await file.read())
+            in_f.flush()
+            converting_repository.any2pdf(in_f.name, out_f.name)
+            in_f.close()
+            tempfiles[out_f.name] = out_f
+        return out_f.name
+    else:
+        raise HTTPException(400, "Unsupported format")
+
+
+@router.post("/print", responses={404: {"description": "No such file"}})
+def actual_print(filename: str, printer_name: str, printing_options: PrintingOptions = PrintingOptions()) -> int:
+    """
+    Returns job identifier
+    """
+
+    if filename in tempfiles:
+        job_id = printing_repository.print_file(printer_name, filename, "job", printing_options)
+        os.unlink(filename)
+        del tempfiles[filename]
+        return job_id
+    else:
+        raise HTTPException(404, "No such file")
+
+
+@router.get("/get_file", responses={404: {"description": "No such file"}})
+def get_file(filename: str) -> FileResponse:
+    if filename in tempfiles:
+        short_name = pathlib.Path(filename).name
+        return FileResponse(filename, headers={"Content-Disposition": f"attachment; filename={short_name}"})
+    else:
+        raise HTTPException(404, "No such file")