fakz9/Fulfillment-Backend
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

crappy

Browse Source
This commit is contained in:
fakz9
2024-04-12 07:34:21 +03:00
parent 5c81af05d5
commit be623a3555
12 changed files with 513 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
models/client.py
View File

@@ -25,7 +25,7 @@ class ClientDetails(BaseModel):
address = Column(String)
phone_number = Column(String)
inn = Column(BigInteger)
inn = Column(String)
email = Column(String)
last_modified_at = Column(DateTime, nullable=False)

1
models/deal.py
View File

@@ -35,6 +35,7 @@ class Deal(BaseModel):
services = relationship('DealService', back_populates='deal')
products = relationship('DealProduct', back_populates='deal')
class DealStatusHistory(BaseModel):
__tablename__ = 'deals_status_history'
id = Column(Integer, autoincrement=True, primary_key=True, index=True)

100
routers/deal.py
View File

@@ -52,6 +52,44 @@ async def get_summary(
return await DealService(session).get_summary()
@deal_router.get(
'/get-all',
response_model=DealGetAllResponse,
operation_id='getAllDeals'
)
async def get_all(
session: Annotated[AsyncSession, Depends(get_session)]
):
return await DealService(session).get_all()
# endpoint to get deal by id
@deal_router.get(
'/get/{deal_id}',
response_model=DealSchema,
operation_id='getDealById'
)
async def get_deal_by_id(
deal_id: int,
session: Annotated[AsyncSession, Depends(get_session)]
):
return await DealService(session).get_by_id(deal_id)
@deal_router.post(
'/update-general-info',
response_model=DealUpdateGeneralInfoResponse,
operation_id='updateDealGeneralInfo'
)
async def update_general_info(
request: DealUpdateGeneralInfoRequest,
session: Annotated[AsyncSession, Depends(get_session)]
):
return await DealService(session).update_general_info(request)
# region Deal services
@deal_router.post(
'/services/add/multiple',
response_model=DealAddServicesResponse,
@@ -87,6 +125,7 @@ async def services_update(
):
return await DealService(session).update_service_quantity(request)
@deal_router.post(
'/services/delete',
response_model=DealDeleteServiceResponse,
@@ -98,6 +137,7 @@ async def services_delete(
):
return await DealService(session).delete_service(request)
@deal_router.post(
'/services/delete/multiple',
response_model=DealDeleteServicesResponse,
@@ -110,26 +150,46 @@ async def services_delete(
return await DealService(session).delete_services(request)
@deal_router.get(
'/get-all',
response_model=DealGetAllResponse,
operation_id='getAllDeals'
)
async def get_all(
session: Annotated[AsyncSession, Depends(get_session)]
):
return await DealService(session).get_all()
# endregion
# region Deal products
@deal_router.post(
'/products/update-quantity',
response_model=DealUpdateProductQuantityResponse,
operation_id='updateDealProductQuantity')
async def products_update(
request: DealUpdateProductQuantityRequest,
session: Annotated[AsyncSession, Depends(get_session)]):
return await DealService(session).update_product_quantity(request)
# endpoint to get deal by id
@deal_router.get(
'/get/{deal_id}',
response_model=DealSchema,
operation_id='getDealById'
)
async def get_deal_by_id(
deal_id: int,
session: Annotated[AsyncSession, Depends(get_session)]
):
return await DealService(session).get_by_id(deal_id)
@deal_router.post(
'/products/add',
response_model=DealAddProductResponse,
operation_id='addDealProduct')
async def products_add(
request: DealAddProductRequest,
session: Annotated[AsyncSession, Depends(get_session)]):
return await DealService(session).add_product(request)
@deal_router.post(
'/products/delete',
response_model=DealDeleteProductResponse,
operation_id='deleteDealProduct')
async def products_delete(
request: DealDeleteProductRequest,
session: Annotated[AsyncSession, Depends(get_session)]):
return await DealService(session).delete_product(request)
@deal_router.post(
'/products/delete/multiple',
response_model=DealDeleteProductsResponse,
operation_id='deleteMultipleDealProducts')
async def products_delete(
request: DealDeleteProductsRequest,
session: Annotated[AsyncSession, Depends(get_session)]):
return await DealService(session).delete_products(request)
# endregion

46
routers/product.py
View File

@@ -62,3 +62,49 @@ async def get_product(
session: Annotated[AsyncSession, Depends(get_session)]
):
return await ProductService(session).get_by_client_id(client_id, pagination)
@product_router.get('/get-by-id',
response_model=ProductSchema,
operation_id='get_product_by_id')
async def get_product_by_id(
product_id: int,
session: Annotated[AsyncSession, Depends(get_session)]
):
return await ProductService(session).get_by_id(product_id)
@product_router.post(
'/barcode/add',
response_model=ProductAddBarcodeResponse,
operation_id='add_product_barcode'
)
async def add_product_barcode(
request: ProductAddBarcodeRequest,
session: Annotated[AsyncSession, Depends(get_session)]
):
return await ProductService(session).add_barcode(request)
@product_router.get(
'/barcode/exists',
response_model=ProductExistsBarcodeResponse,
operation_id='exists_product_barcode'
)
async def exists_product_barcode(
product_id: int,
barcode: str,
session: Annotated[AsyncSession, Depends(get_session)]
):
return await ProductService(session).exists_barcode(product_id, barcode)
@product_router.post(
'/barcode/generate',
response_model=ProductGenerateBarcodeResponse,
operation_id='generate_product_barcode'
)
async def generate_product_barcode(
request: ProductGenerateBarcodeRequest,
session: Annotated[AsyncSession, Depends(get_session)]
):
return await ProductService(session).generate_barcode(request)

8
schemas/client.py
View File

@@ -1,5 +1,7 @@
from typing import List
from pydantic import validator, field_validator
from schemas.base import CustomModelCamel, OkMessageSchema
@@ -7,9 +9,13 @@ from schemas.base import CustomModelCamel, OkMessageSchema
class ClientDetailsSchema(CustomModelCamel):
address: str | None = None
phone_number: str | None = None
inn: int | None = None
inn: str | None = None
email: str | None = None
@field_validator("phone_number", "inn", "email", "address", mode="before")
def empty_string_to_none(cls, v):
return '' if v is None else v
class ClientSchema(CustomModelCamel):
id: int

82
schemas/deal.py
View File

@@ -5,6 +5,7 @@ from schemas.base import CustomModelCamel, OkMessageSchema
from schemas.client import ClientSchema
from schemas.product import ProductSchema
from schemas.service import ServiceSchema
from schemas.user import UserSchema
# region Entities
@@ -34,6 +35,14 @@ class DealProductSchema(CustomModelCamel):
quantity: int
class DealStatusHistorySchema(CustomModelCamel):
user: UserSchema
changed_at: datetime.datetime
from_status: int
to_status: int
next_status_deadline: datetime.datetime
class DealSchema(CustomModelCamel):
id: int
name: str
@@ -42,7 +51,16 @@ class DealSchema(CustomModelCamel):
current_status: int
services: List[DealServiceSchema]
products: List[DealProductSchema]
# total_price: int
status_history: List[DealStatusHistorySchema]
is_deleted: bool
is_completed: bool
client: ClientSchema
class DealGeneralInfoSchema(CustomModelCamel):
name: str
is_deleted: bool
is_completed: bool
# endregion Entities
@@ -74,10 +92,6 @@ class DealAddServicesRequest(CustomModelCamel):
services: list[DealServiceSchema]
class DealGetAllResponse(CustomModelCamel):
deals: List[DealSchema]
class DealUpdateServiceQuantityRequest(CustomModelCamel):
deal_id: int
service_id: int
@@ -95,18 +109,52 @@ class DealDeleteServiceRequest(CustomModelCamel):
service_id: int
class DealDeleteServicesResponse(OkMessageSchema):
pass
class DealDeleteServicesRequest(CustomModelCamel):
deal_id: int
service_ids: List[int]
class DealUpdateProductQuantityRequest(CustomModelCamel):
deal_id: int
product_id: int
quantity: int
class DealAddProductRequest(CustomModelCamel):
deal_id: int
product_id: int
quantity: int
class DealDeleteProductRequest(CustomModelCamel):
deal_id: int
product_id: int
class DealDeleteProductsRequest(CustomModelCamel):
deal_id: int
product_ids: List[int]
class DealUpdateGeneralInfoRequest(CustomModelCamel):
deal_id: int
data: DealGeneralInfoSchema
# endregion Requests
# region Responses
class DealUpdateProductQuantityResponse(OkMessageSchema):
pass
class DealDeleteServicesResponse(OkMessageSchema):
pass
class DealGetAllResponse(CustomModelCamel):
deals: List[DealSchema]
class DealChangeStatusResponse(CustomModelCamel):
ok: bool
@@ -140,4 +188,20 @@ class DealAddServiceResponse(OkMessageSchema):
class DealDeleteServiceResponse(OkMessageSchema):
pass
class DealDeleteProductResponse(OkMessageSchema):
pass
class DealDeleteProductsResponse(OkMessageSchema):
pass
class DealAddProductResponse(OkMessageSchema):
pass
class DealUpdateGeneralInfoResponse(OkMessageSchema):
pass
# endregion Responses

44
schemas/product.py
View File

@@ -1,19 +1,25 @@
from typing import List
from pydantic import validator, field_validator
from models import ProductBarcode
from schemas.base import CustomModelCamel, PaginationInfoSchema, OkMessageSchema
# region Entities
class ProductBarcodeSchema(CustomModelCamel):
barcode: str
class ProductSchema(CustomModelCamel):
id: int
name: str
article: str
client_id: int
barcodes: list[ProductBarcodeSchema]
barcodes: list[str]
@field_validator('barcodes', mode="before")
def barcodes_to_list(cls, v):
if isinstance(v, list) and all([type(barcode) is ProductBarcode for barcode in v]):
return [barcode.barcode for barcode in v]
return v
# endregion
@@ -34,6 +40,20 @@ class ProductUpdateRequest(CustomModelCamel):
product: ProductSchema
class ProductAddBarcodeRequest(CustomModelCamel):
product_id: int
barcode: str
class ProductDeleteBarcodeRequest(CustomModelCamel):
product_id: int
barcode: str
class ProductGenerateBarcodeRequest(CustomModelCamel):
product_id: int
# endregion
# region Responses
@@ -52,4 +72,20 @@ class ProductDeleteResponse(OkMessageSchema):
class ProductUpdateResponse(OkMessageSchema):
pass
class ProductAddBarcodeResponse(OkMessageSchema):
pass
class ProductDeleteBarcodeResponse(OkMessageSchema):
pass
class ProductGenerateBarcodeResponse(OkMessageSchema):
barcode: str
class ProductExistsBarcodeResponse(CustomModelCamel):
exists: bool
# endregion

8
schemas/user.py Normal file
View File

@@ -0,0 +1,8 @@
from schemas.base import CustomModelCamel
class UserSchema(CustomModelCamel):
id: int
telegram_id: int
phone_number: str | None = None
is_admin: bool

10
services/client.py
View File

@@ -75,7 +75,7 @@ class ClientService(BaseService):
try:
client = await self.get_by_name(request.data.name)
if client:
return ClientCreateResponse(ok=False, message='Client already exists')
return ClientCreateResponse(ok=False, message='Клиент с таким именем уже существует')
await self.create_client_raw(user, request.data.name, request.data.details)
await self.session.commit()
return ClientCreateResponse(ok=True, message='Client created')
@@ -86,11 +86,11 @@ class ClientService(BaseService):
try:
client = await self.get_by_id(request.data.id)
if not client:
return ClientUpdateResponse(ok=False, message='Client not found')
return ClientUpdateResponse(ok=False, message='Клиент не найден')
await self.session.execute(update(Client).where(Client.id == client.id).values(name=request.data.name))
await self.update_details(user, client, request.data.details)
await self.session.commit()
return ClientUpdateResponse(ok=True, message='Client updated')
return ClientUpdateResponse(ok=True, message='Клиент обновлен')
except Exception as e:
return ClientUpdateResponse(ok=False, message=str(e))
@@ -98,9 +98,9 @@ class ClientService(BaseService):
try:
client = await self.get_by_id(request.client_id)
if not client:
return ClientDeleteResponse(ok=False, message='Client not found')
return ClientDeleteResponse(ok=False, message='Клиент не найден')
await self.session.delete(client)
await self.session.commit()
return ClientDeleteResponse(ok=True, message='Client deleted')
return ClientDeleteResponse(ok=True, message='Клиент удален')
except Exception as e:
return ClientDeleteResponse(ok=False, message=str(e))

165
services/deal.py
View File

@@ -5,7 +5,7 @@ from fastapi import HTTPException
from sqlalchemy import select, func
from sqlalchemy.orm import joinedload, selectinload
from models import User, Service
from models import User, Service, Client
from models.deal import *
from schemas.client import ClientDetailsSchema
from schemas.deal import *
@@ -15,6 +15,7 @@ from services.client import ClientService
class DealService(BaseService):
# region Deal
async def _get_deal_by_id(self, deal_id) -> Union[Deal, None]:
return await self.session.get(Deal, deal_id)
@@ -116,6 +117,56 @@ class DealService(BaseService):
)
return DealSummaryResponse(summaries=summaries)
async def get_all(self) -> DealGetAllResponse:
deals_query = await self.session.scalars(select(Deal).options(joinedload(Deal.client)))
deals = deals_query.all()
result = []
for deal in deals:
result.append(DealSchema.model_validate(deal))
return DealGetAllResponse(deals=result)
async def get_by_id(self, deal_id: int) -> DealSchema:
deal = await self.session.scalar(
select(Deal)
.options(
joinedload(Deal.client).joinedload(Client.details),
selectinload(Deal.services)
.joinedload(models.secondary.DealService.service)
.joinedload(Service.category),
selectinload(Deal.products)
.joinedload(models.secondary.DealProduct.product)
.joinedload(models.Product.client),
selectinload(Deal.products)
.joinedload(models.secondary.DealProduct.product)
.joinedload(models.Product.barcodes),
selectinload(Deal.status_history)
.joinedload(DealStatusHistory.user),
selectinload(Deal.status_history)
.noload(DealStatusHistory.deal)
)
.where(Deal.id == deal_id)
)
if not deal:
raise HTTPException(status_code=404, detail="Сделка не найдена")
return DealSchema.model_validate(deal)
async def update_general_info(self, request: DealUpdateGeneralInfoRequest) -> DealUpdateGeneralInfoResponse:
try:
deal = await self.session.scalar(select(Deal).where(Deal.id == request.deal_id))
if not deal:
raise HTTPException(status_code=404, detail="Сделка не найдена")
deal.name = request.data.name
deal.is_deleted = request.data.is_deleted
deal.is_completed = request.data.is_completed
await self.session.commit()
return DealUpdateGeneralInfoResponse(ok=True, message='Данные о сделке успешно обновлены')
except Exception as e:
await self.session.rollback()
return DealUpdateGeneralInfoResponse(ok=False, message=str(e))
# endregion
# region Deal services
async def add_services(self, request: DealAddServicesRequest):
# TODO refactor
deal: Deal = await self.session.scalar(
@@ -149,41 +200,14 @@ class DealService(BaseService):
quantity = request_services_dict[service.id]
deal.services.append(
models.secondary.DealService(
service=service,
deal=deal,
service_id=service.id,
deal_id=deal.id,
quantity=quantity
)
)
await self.session.commit()
return DealAddServicesResponse(ok=True, message='Услуги успешно добавлены')
async def get_all(self) -> DealGetAllResponse:
deals_query = await self.session.scalars(select(Deal).options(joinedload(Deal.client)))
deals = deals_query.all()
result = []
for deal in deals:
result.append(DealSchema.model_validate(deal))
return DealGetAllResponse(deals=result)
async def get_by_id(self, deal_id: int) -> DealSchema:
deal = await self.session.scalar(
select(Deal)
.options(
joinedload(Deal.client),
selectinload(Deal.services)
.joinedload(models.secondary.DealService.service)
.joinedload(Service.category),
selectinload(Deal.products)
.joinedload(models.secondary.DealProduct.product)
.joinedload(models.Product.client)
.selectinload(models.Product.barcodes)
)
.where(Deal.id == deal_id)
)
if not deal:
raise HTTPException(status_code=404, detail="Сделка не найдена")
return DealSchema.model_validate(deal)
async def update_service_quantity(self,
request: DealUpdateServiceQuantityRequest) -> DealUpdateServiceQuantityResponse:
try:
@@ -259,3 +283,84 @@ class DealService(BaseService):
except Exception as e:
await self.session.rollback()
return DealDeleteServicesResponse(ok=False, message=str(e))
# endregion
# region Deal products
async def update_product_quantity(self,
request: DealUpdateProductQuantityRequest) -> DealUpdateProductQuantityResponse:
try:
# check if there is no deal or no product with different exceptions
deal_product = await self.session.scalar(
select(models.secondary.DealProduct)
.where(models.secondary.DealProduct.deal_id == request.deal_id,
models.secondary.DealProduct.product_id == request.product_id)
)
if not deal_product:
raise HTTPException(status_code=404, detail="Сделка или товар не найдена")
deal_product.quantity = request.quantity
await self.session.commit()
return DealUpdateProductQuantityResponse(ok=True, message='Количество успешно обновлено')
except Exception as e:
await self.session.rollback()
return DealUpdateProductQuantityResponse(ok=False, message=str(e))
async def add_product(self, request: DealAddProductRequest) -> DealAddProductResponse:
try:
deal = await self.session.scalar(select(Deal).where(Deal.id == request.deal_id))
if not deal:
raise HTTPException(status_code=404, detail="Сделка не найдена")
product = await self.session.scalar(select(models.Product).where(models.Product.id == request.product_id))
if not product:
raise HTTPException(status_code=404, detail="Товар не найден")
# Preventing duplicates
deal_product = await self.session.scalar(
select(models.secondary.DealProduct)
.where(models.secondary.DealProduct.deal_id == request.deal_id,
models.secondary.DealProduct.product_id == request.product_id)
)
if deal_product:
raise HTTPException(status_code=400, detail="Товар уже добавлен")
deal_product = models.secondary.DealProduct(
deal_id=request.deal_id,
product_id=request.product_id,
quantity=request.quantity
)
self.session.add(deal_product)
await self.session.commit()
return DealAddProductResponse(ok=True, message='Товар успешно добавлен')
except Exception as e:
await self.session.rollback()
return DealAddProductResponse(ok=False, message=str(e))
async def delete_product(self, request: DealDeleteProductRequest) -> DealDeleteProductResponse:
try:
deal_product = await self.session.scalar(
select(models.secondary.DealProduct)
.where(models.secondary.DealProduct.deal_id == request.deal_id,
models.secondary.DealProduct.product_id == request.product_id)
)
if not deal_product:
raise HTTPException(status_code=404, detail="Сделка не найдена")
await self.session.delete(deal_product)
await self.session.commit()
return DealDeleteProductResponse(ok=True, message='Товар успешно удален')
except Exception as e:
await self.session.rollback()
return DealDeleteProductResponse(ok=False, message=str(e))
async def delete_products(self, request: DealDeleteProductsRequest) -> DealDeleteProductsResponse:
try:
deal_products = await self.session.scalars(
select(models.secondary.DealProduct)
.where(models.secondary.DealProduct.deal_id == request.deal_id,
models.secondary.DealProduct.product_id.in_(request.product_ids))
)
for deal_product in deal_products:
await self.session.delete(deal_product)
await self.session.commit()
return DealDeleteProductsResponse(ok=True, message='Товары успешно удалены')
except Exception as e:
await self.session.rollback()
return DealDeleteProductsResponse(ok=False, message=str(e))
# endregion

115
services/product.py
View File

@@ -2,6 +2,7 @@ from fastapi import HTTPException
from sqlalchemy import select, func, Integer, update
from sqlalchemy.orm import selectinload
import utils.barcodes
from models.product import Product, ProductBarcode
from schemas.base import PaginationSchema
from services.base import BaseService
@@ -85,6 +86,9 @@ class ProductService(BaseService):
return ProductUpdateResponse(ok=True, message='Товар успешно обновлен')
async def get_by_client_id(self, client_id: int, pagination: PaginationSchema) -> ProductGetResponse:
is_pagination_valid = is_valid_pagination(pagination)
total_pages = 0
total_items = 0
stmt = (
select(Product)
.options(selectinload(Product.barcodes)
@@ -92,7 +96,7 @@ class ProductService(BaseService):
.where(Product.client_id == client_id)
.order_by(Product.id)
)
if is_valid_pagination(pagination):
if is_pagination_valid:
total_products_query = await self.session.execute(
select(
func.cast(func.ceil(func.count() / pagination.items_per_page), Integer),
@@ -101,36 +105,101 @@ class ProductService(BaseService):
.select_from(stmt.subquery())
)
total_pages, total_items = total_products_query.first()
else:
total_items_query = await self.session.execute(
select(func.count())
.select_from(stmt.subquery())
)
total_items = total_items_query.scalar()
total_pages = 1
pagination_info = PaginationInfoSchema(total_pages=total_pages, total_items=total_items)
if is_valid_pagination(pagination):
stmt = (
stmt
.offset(pagination.page * pagination.items_per_page)
.limit(pagination.items_per_page)
)
query = await self.session.execute(
stmt
.order_by(Product.id)
)
product_orm = query.scalars().all()
if not is_pagination_valid:
total_pages = 1
total_items = len(product_orm)
pagination_info = PaginationInfoSchema(total_pages=total_pages, total_items=total_items)
products: list[ProductSchema] = []
for product in query.scalars().all():
product: Product
barcodes = []
for barcode_obj in product.barcodes:
barcode_obj: ProductBarcode
barcodes.append(barcode_obj.barcode)
products.append(
ProductSchema.from_sql_model(product, {'barcodes': barcodes})
)
for product in product_orm:
products.append(ProductSchema.model_validate(product))
return ProductGetResponse(products=products, pagination_info=pagination_info)
async def get_by_id(self, product_id: int) -> ProductSchema:
stmt = (
select(Product)
.options(selectinload(Product.barcodes)
.noload(ProductBarcode.product))
.where(Product.id == product_id)
)
query = await self.session.execute(stmt)
product = query.scalar()
if not product:
raise HTTPException(status_code=404, detail='Товар не найден')
return ProductSchema.model_validate(product)
# region Barcodes
async def add_barcode(self, request: ProductAddBarcodeRequest):
try:
product = await self.session.get(Product, request.product_id)
if not product:
raise HTTPException(status_code=404, detail='Товар не найден')
existing_barcode_query = await self.session.execute(
select(ProductBarcode)
.where(ProductBarcode.product_id == request.product_id,
ProductBarcode.barcode == request.barcode)
)
existing_barcode = existing_barcode_query.first()
if existing_barcode:
return ProductAddBarcodeResponse(ok=False, message='Штрих-код уже существует у товара')
product_barcode = ProductBarcode(product_id=product.id,
barcode=request.barcode)
self.session.add(product_barcode)
await self.session.commit()
return ProductAddBarcodeResponse(ok=True, message='Штрих-код успешно добавлен')
except Exception as e:
await self.session.rollback()
return ProductAddBarcodeResponse(ok=False, message=str(e))
async def delete_barcode(self, request: ProductDeleteBarcodeRequest):
try:
product_barcode = await self.session.get(ProductBarcode, (request.product_id, request.barcode))
if not product_barcode:
return ProductDeleteBarcodeResponse(ok=False, message='Штрих-код не найден')
await self.session.delete(product_barcode)
await self.session.commit()
return ProductDeleteBarcodeResponse(ok=True, message='Штрих-код успешно удален')
except Exception as e:
await self.session.rollback()
return ProductDeleteBarcodeResponse(ok=False, message=str(e))
async def exists_barcode(self, product_id: int, barcode: str) -> ProductExistsBarcodeResponse:
product_barcode_query = await self.session.execute(
select(ProductBarcode)
.where(ProductBarcode.product_id == product_id,
ProductBarcode.barcode == barcode)
)
product_barcode = product_barcode_query.first()
return ProductExistsBarcodeResponse(exists=bool(product_barcode))
async def generate_barcode(self, request: ProductGenerateBarcodeRequest) -> ProductGenerateBarcodeResponse:
try:
product = await self.session.get(Product, request.product_id)
if not product:
raise HTTPException(status_code=404, detail='Товар не найден')
barcode = utils.barcodes.generate_barcode(product.id)
barcode_exists_query = await self.session.execute(
select(ProductBarcode)
.where(ProductBarcode.barcode == barcode)
)
barcode_exists = barcode_exists_query.first()
if barcode_exists:
raise Exception('Штрих-код уже существует')
product_barcode = ProductBarcode(product_id=product.id,
barcode=barcode)
self.session.add(product_barcode)
await self.session.commit()
return ProductGenerateBarcodeResponse(ok=True, message='Штрих-код успешно сгенерирован', barcode=barcode)
except Exception as e:
return ProductGenerateBarcodeResponse(ok=False, message=str(e))
# endregion

25
utils/barcodes.py Normal file
View File

@@ -0,0 +1,25 @@
def generate_ean13_checksum(ean12):
if len(ean12) != 12:
return None
digits = [int(digit) for digit in ean12]
odd_sum = sum(digits[::2])
even_sum = sum(digits[1::2])
total = odd_sum + 3 * even_sum
checksum = 10 - (total % 10)
if checksum == 10:
checksum = 0
return checksum
def generate_barcode(product_id: int):
product_id = str(product_id)
# 4 - D
# 6 - E
# 5 - N
# 2 - C
# 4652 - DENCO_C (C - CRM)
supplier_part = 4652_5
article_part = product_id.ljust(7, '0')
barcode = f'{supplier_part}{article_part}'
checksum_code = generate_ean13_checksum(barcode)
return f'{barcode}{checksum_code}'