feat: CRUD for product barcode images

barcodes/images_uploader/__init__.py

@@ -0,0 +1 @@
+from .images_uploader import BarcodeImagesUploader

barcodes/images_uploader/base.py

@@ -0,0 +1,18 @@
+from abc import abstractmethod
+
+from fastapi import UploadFile
+
+
+class BaseImagesUploader:
+
+    @abstractmethod
+    def get_url(self, filename: str) -> bytes:
+        pass
+
+    @abstractmethod
+    def delete(self, filename: str):
+        pass
+
+    @abstractmethod
+    async def upload(self, upload_file: UploadFile) -> str:
+        pass

barcodes/images_uploader/images_uploader.py

@@ -0,0 +1,46 @@
+from pathlib import Path
+from uuid import uuid4
+
+from aioshutil import copyfileobj
+from fastapi import UploadFile
+
+from barcodes.images_uploader.base import BaseImagesUploader
+from barcodes.pdf import PDFGenerator
+from constants import APP_PATH, API_ROOT
+
+
+class BarcodeImagesUploader(BaseImagesUploader):
+    def __init__(self):
+        self.relative_path = Path("static/images/product_barcodes")
+        self.storage_path = APP_PATH / self.relative_path
+        if not Path.exists(self.storage_path):
+            Path.mkdir(self.storage_path)
+
+    def get_url(self, filename: str) -> str:
+        file_location = self.relative_path / filename
+        return f"{API_ROOT}/{file_location}"
+
+    def delete(self, filename: str):
+        file_location = self.storage_path / filename
+        if file_location.exists():
+            file_location.unlink()
+
+    async def upload(self, upload_file: UploadFile) -> str:
+        # Create temp file in filesystem
+        temp_filename = str(uuid4()) + '.' + upload_file.filename.split('.')[-1]
+        temp_file_location = self.storage_path / temp_filename
+        with open(temp_file_location, 'wb') as buffer:
+            await copyfileobj(upload_file.file, buffer)
+
+        # Generate PDF file and save it
+        res_filename = str(uuid4()) + '.pdf'
+        res_file_location = f"{self.storage_path}/{res_filename}"
+        temp_file_url = f"{self.relative_path}/{temp_filename}"
+
+        pdf_gen = PDFGenerator()
+        pdf_gen.generate_barcode_image(temp_file_url, res_file_location)
+
+        # Remove temp file
+        self.delete(temp_filename)
+
+        return res_filename

barcodes/pdf/generator.py

@@ -7,6 +7,7 @@ from reportlab.lib.pagesizes import mm
 from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
 from reportlab.pdfbase import pdfmetrics
 from reportlab.pdfbase.ttfonts import TTFont
+from reportlab.pdfgen import canvas
 from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, PageBreak
 
 from constants import APP_PATH
@@ -181,3 +182,10 @@ class PDFGenerator:
 
         buffer.seek(0)
         return buffer
+
+    def generate_barcode_image(self, barcode_image_url: str, path_to_save_pdf: str):
+        print(type(path_to_save_pdf))
+        print(path_to_save_pdf)
+        c = canvas.Canvas(path_to_save_pdf, pagesize=(self.page_width, self.page_height))
+        c.drawImage(barcode_image_url, 0, 0, width=self.page_width, height=self.page_height)
+        c.save()

constants.py

@@ -12,6 +12,8 @@ ENV.globals['now'] = datetime.now
 ENV.globals['encode128'] = encode128
 ENV.globals['format_number'] = lambda x: '{:,}'.format(x).replace(',', ' ')
 
+API_ROOT = "/api"
+
 APP_PATH = os.path.dirname(sys.executable) if getattr(sys, 'frozen', False) else os.path.dirname(__file__)
 
 allowed_telegram_ids = [

main.py

@@ -5,6 +5,7 @@ import platform
 from starlette.staticfiles import StaticFiles
 
 import routers
+from constants import API_ROOT
 
 origins = [
     'http://localhost:5173'
@@ -17,7 +18,7 @@ if platform.system() == 'Linux':
 
     class Worker(uvicorn.workers.UvicornWorker):
         CONFIG_KWARGS = {
-            'root_path': '/api'
+            'root_path': API_ROOT
         }
 
 app.add_middleware(

models/product.py

@@ -23,6 +23,8 @@ class Product(BaseModel):
     barcode_template_id = Column(Integer, ForeignKey('barcode_templates.id'), nullable=True)
     barcode_template = relationship('BarcodeTemplate', lazy='joined')
 
+    barcode_image = relationship('ProductBarcodeImage', back_populates='product', lazy='joined', uselist=False)
+
     # Attributes
     # TODO move to another table
     brand = Column(String, nullable=True, comment='Бренд')
@@ -61,3 +63,11 @@ class ProductBarcode(BaseModel):
     product: Mapped["Product"] = relationship(back_populates='barcodes')
 
     barcode = Column(String, nullable=False, index=True, comment='ШК товара', primary_key=True)
+
+
+class ProductBarcodeImage(BaseModel):
+    __tablename__ = 'product_barcode_images'
+    product_id = Column(Integer, ForeignKey('products.id'), primary_key=True, comment='ID товара')
+    product: Mapped["Product"] = relationship(back_populates='barcode_image')
+
+    filename = Column(String, nullable=False)

requirements.txt

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

routers/product.py

@@ -164,3 +164,40 @@ async def upload_product_image(
 ):
     file_bytes = upload_file.file.read()
     return await ProductService(session).upload_image(product_id, file_bytes)
+
+
+@product_router.post(
+    '/barcode/upload-image/{product_id}',
+    response_model=ProductUploadBarcodeImageResponse,
+    operation_id='upload_product_barcode_image'
+)
+async def upload_product_barcode_image(
+        product_id: int,
+        upload_file: UploadFile,
+        session: Annotated[AsyncSession, Depends(get_session)]
+):
+    return await ProductService(session).upload_barcode_image(product_id, upload_file)
+
+
+@product_router.post(
+    '/barcode/delete-image/{product_id}',
+    response_model=ProductDeleteBarcodeImageResponse,
+    operation_id='delete_product_barcode_image'
+)
+async def delete_product_barcode_image(
+        product_id: int,
+        session: Annotated[AsyncSession, Depends(get_session)]
+):
+    return await ProductService(session).delete_barcode_image(product_id)
+
+
+@product_router.post(
+    '/barcode/image/{product_id}',
+    response_model=ProductGetBarcodeImageResponse,
+    operation_id='get_product_barcode_image'
+)
+async def get_product_barcode_image(
+        product_id: int,
+        session: Annotated[AsyncSession, Depends(get_session)]
+):
+    return await ProductService(session).get_barcode_image(product_id)

schemas/product.py

@@ -115,4 +115,18 @@ class ProductExistsBarcodeResponse(BaseSchema):
 
 class ProductUploadImageResponse(OkMessageSchema):
     image_url: str | None = None
+
+
+class ProductUploadBarcodeImageResponse(OkMessageSchema):
+    barcode_image_url: str | None = None
+
+
+class ProductDeleteBarcodeImageResponse(OkMessageSchema):
+    pass
+
+
+class ProductGetBarcodeImageResponse(BaseSchema):
+    barcode_image_url: str | None = None
+
+
 # endregion

services/product.py

@@ -1,11 +1,14 @@
-from fastapi import HTTPException
+from typing import Optional
+
+from fastapi import HTTPException, UploadFile
 from sqlalchemy import select, func, Integer, update, or_
 from sqlalchemy.orm import selectinload, Query
 
 import utils.barcodes
 from backend import config
+from barcodes.images_uploader import BarcodeImagesUploader
 from external.s3_uploader.uploader import S3Uploader
-from models.product import Product, ProductImage
+from models.product import Product, ProductImage, ProductBarcodeImage
 from schemas.base import PaginationSchema
 from schemas.product import *
 from services.base import BaseService
@@ -107,7 +110,8 @@ class ProductService(BaseService):
             )
             .options(
                 selectinload(Product.barcodes)
-                .noload(ProductBarcode.product)
+                .noload(ProductBarcode.product),
+                selectinload(Product.barcode_image),
             )
             .where(
                 Product.client_id == client_id
@@ -257,4 +261,68 @@ class ProductService(BaseService):
             return ProductGenerateBarcodeResponse(ok=True, message='Штрих-код успешно сгенерирован', barcode=barcode)
         except Exception as e:
             return ProductGenerateBarcodeResponse(ok=False, message=str(e))
+
+    async def get_model_by_id(self, product_id: int) -> Product:
+        product: Optional[Product] = await self.session.get(Product, product_id)
+        if not product:
+            raise Exception('Не удалось найти товар с указанным ID')
+        return product
+
+    async def delete_model_barcode_image(self, uploader: BarcodeImagesUploader, product_id: int) -> None:
+        barcode_image: Optional[ProductBarcodeImage] = await self.session.get(ProductBarcodeImage, product_id)
+        if barcode_image:
+            uploader.delete(barcode_image.filename)
+            await self.session.delete(barcode_image)
+            await self.session.commit()
+
+    async def upload_barcode_image(self, product_id: int, upload_file: UploadFile) -> ProductUploadBarcodeImageResponse:
+        try:
+            await self.get_model_by_id(product_id)
+
+            uploader = BarcodeImagesUploader()
+            await self.delete_model_barcode_image(uploader, product_id)
+            filename = await uploader.upload(upload_file)
+            barcode_image_url = uploader.get_url(filename)
+
+            product_barcode_image = ProductBarcodeImage(
+                product_id=product_id,
+                filename=filename,
+            )
+            self.session.add(product_barcode_image)
+            await self.session.commit()
+            return ProductUploadBarcodeImageResponse(
+                ok=True,
+                message='Штрих-код для товара успешно загружен',
+                barcode_image_url=barcode_image_url,
+            )
+        except Exception as e:
+            print(e)
+            return ProductUploadBarcodeImageResponse(ok=False, message=str(e))
+
+    async def delete_barcode_image(self, product_id: int) -> ProductDeleteBarcodeImageResponse:
+        try:
+            await self.get_model_by_id(product_id)
+
+            uploader = BarcodeImagesUploader()
+            await self.delete_model_barcode_image(uploader, product_id)
+
+            return ProductDeleteBarcodeImageResponse(
+                ok=True,
+                message='Штрих-код для товара успешно удален',
+            )
+        except Exception as e:
+            return ProductDeleteBarcodeImageResponse(ok=False, message=str(e))
+
+    async def get_barcode_image(self, product_id: int) -> ProductGetBarcodeImageResponse:
+        product: Optional[Product] = await self.session.get(Product, product_id)
+        if not product:
+            raise HTTPException(404, 'Не удалось найти товар с указанным ID')
+        barcode_image: Optional[ProductBarcodeImage] = await self.session.get(ProductBarcodeImage, product_id)
+        if not barcode_image:
+            return ProductGetBarcodeImageResponse(barcode_image_url="")
+
+        uploader = BarcodeImagesUploader()
+        url = uploader.get_url(barcode_image.filename)
+        return ProductGetBarcodeImageResponse(barcode_image_url=url)
+
     # endregion