426 lines
18 KiB
Python
426 lines
18 KiB
Python
from datetime import datetime
|
|
from db.handlers.stock import StockHandler
|
|
from db.handlers.user import UserHandler
|
|
from utils import logger, saveImage, safeFilename
|
|
from db import CRUD
|
|
from db.schemas.toolkit import Toolkit, ToolkitCompatibility
|
|
from sqlalchemy import select
|
|
from db.handlers.records import ServiceRecordsHandler
|
|
from utils.image import deleteImage
|
|
|
|
|
|
def handleToolkitImage(imageData, title: str):
|
|
import base64
|
|
|
|
title = safeFilename(title)
|
|
fileName = f"static/images/tools/{title}.png"
|
|
if imageData.startswith("data:image"):
|
|
header, encoded = imageData.split(",", 1)
|
|
else:
|
|
encoded = imageData
|
|
file_bytes = base64.b64decode(encoded)
|
|
if not saveImage(file_bytes, fileName):
|
|
return None
|
|
return fileName
|
|
|
|
|
|
class ToolkitHandler:
|
|
|
|
@staticmethod
|
|
async def add(toolkitData: dict, user_id: int = None):
|
|
title = toolkitData.get("title", None)
|
|
if not title:
|
|
logger.error("Не указано название инструмента")
|
|
return {"errorMessage": "Не указано название инструмента"}
|
|
|
|
query = select(Toolkit).where(Toolkit.title == title)
|
|
toolkit = await CRUD.read(query)
|
|
if toolkit:
|
|
logger.error("Инструмент с таким названием уже существует")
|
|
return {"errorMessage": "Инструмент с таким названием уже существует"}
|
|
|
|
try:
|
|
imageDict = {"main": "static/images/tools/default.png", "additional": []}
|
|
if "image" in toolkitData:
|
|
imageData = toolkitData.pop("image")
|
|
mainImage = imageData.get("main")
|
|
if mainImage.startswith("static/images/"):
|
|
imageDict["main"] = mainImage
|
|
else:
|
|
imageFileName = handleToolkitImage(mainImage, title)
|
|
if imageFileName:
|
|
imageDict["main"] = imageFileName
|
|
additionalImages = imageData.get("additional", [])
|
|
if len(additionalImages) > 0:
|
|
for image in additionalImages:
|
|
if image.startswith("static/images/"):
|
|
imageDict["additional"].append(image)
|
|
else:
|
|
imageFileName = handleToolkitImage(image, title)
|
|
if imageFileName:
|
|
imageDict["additional"].append(imageFileName)
|
|
toolkitData["image"] = imageDict
|
|
newToolkit = await Toolkit(**toolkitData).save()
|
|
except Exception as e:
|
|
logger.error(f"Ошибка сохранения инструмента: {str(e)}")
|
|
return {"errorMessage": f"Ошибка сохранения инструмента: {str(e)}"}
|
|
|
|
if not newToolkit:
|
|
logger.error("Инструмент не сохранен")
|
|
return {"errorMessage": "Инструмент не сохранен"}
|
|
|
|
logger.info(f"Инструмент {newToolkit.title} успешно создан")
|
|
await ServiceRecordsHandler.add(user_id, {"Добавлен инструмент": toolkitData})
|
|
return newToolkit.toDict()
|
|
|
|
@staticmethod
|
|
async def updateMovindDate(toolkitId: int):
|
|
toolkit = await CRUD.read(select(Toolkit).where(Toolkit.id == toolkitId))
|
|
if not toolkit:
|
|
logger.error("Инструмент не найден")
|
|
return False
|
|
editedToolkit = await toolkit.edit(moved_at=datetime.now())
|
|
if not editedToolkit:
|
|
logger.error("Инструмент не обновлен")
|
|
return False
|
|
return True
|
|
|
|
@staticmethod
|
|
async def updateRefillDate(toolkitId: int):
|
|
logger.info(f"Обновление даты пополнения инструмента {toolkitId}...")
|
|
toolkit = await CRUD.read(select(Toolkit).where(Toolkit.id == toolkitId))
|
|
if not toolkit:
|
|
logger.error("Инструмент не найден")
|
|
return False
|
|
editedToolkit = await toolkit.edit(refilled_at=datetime.now())
|
|
if not editedToolkit:
|
|
logger.error("Инструмент не обновлен")
|
|
return False
|
|
return True
|
|
|
|
@staticmethod
|
|
async def hideToolkit(userId: int, toolkitId: int, hidden: bool = True):
|
|
logger.info(
|
|
f"{'Скрытие' if hidden else 'Отображение'} инструмента {toolkitId}..."
|
|
)
|
|
return await ToolkitHandler.edit(userId, id=toolkitId, hidden=hidden)
|
|
|
|
@staticmethod
|
|
async def edit(user_id: int, **kwargs):
|
|
title = kwargs.get("title", None)
|
|
toolkitId = kwargs.pop("id")
|
|
if title:
|
|
query = select(Toolkit).where(Toolkit.title == title)
|
|
toolkit = await CRUD.read(query)
|
|
if toolkit:
|
|
if toolkit.id != toolkitId:
|
|
logger.error("Инструмент с таким названием уже существует")
|
|
return {
|
|
"errorMessage": "Инструмент с таким названием уже существует"
|
|
}
|
|
|
|
query = select(Toolkit).where(Toolkit.id == toolkitId)
|
|
toolkit = await CRUD.read(query)
|
|
if not toolkit:
|
|
logger.error("Инструмент не найден")
|
|
return {"errorMessage": "Инструмент не найден"}
|
|
try:
|
|
if "image" in kwargs:
|
|
title = kwargs.get("title", toolkit.title)
|
|
|
|
imageData = kwargs.pop("image")
|
|
|
|
imageDict = {"main": "", "additional": []}
|
|
|
|
if imageData.get("main").startswith("static/images/"):
|
|
imageDict["main"] = imageData.get("main")
|
|
else:
|
|
imageFileName = handleToolkitImage(imageData.get("main"), title)
|
|
if imageFileName:
|
|
imageDict["main"] = imageFileName
|
|
deleteImage(toolkit.image.get("main"))
|
|
|
|
for image in imageData.get("additional"):
|
|
if image.startswith("static/images/"):
|
|
imageDict["additional"].append(image)
|
|
else:
|
|
imageFileName = handleToolkitImage(image, title)
|
|
if imageFileName:
|
|
imageDict["additional"].append(imageFileName)
|
|
|
|
for existImage in toolkit.image.get("additional"):
|
|
if existImage not in imageDict.get("additional"):
|
|
deleteImage(existImage)
|
|
|
|
kwargs["image"] = imageDict
|
|
logger.debug(f"Обновление инструмента {toolkit.title}...")
|
|
editedToolkit = await toolkit.edit(**kwargs)
|
|
except Exception as e:
|
|
logger.error(f"Ошибка обновления инструмента: {str(e)}")
|
|
return {"errorMessage": f"Ошибка обновления инструмента: {str(e)}"}
|
|
|
|
if not editedToolkit:
|
|
logger.error("Инструмент не обновлен")
|
|
return {"errorMessage": "Инструмент не обновлен"}
|
|
|
|
logger.info(
|
|
f"Инструмент {editedToolkit.title} успешно обновлен, изменены данные: {kwargs.keys()}"
|
|
)
|
|
await ServiceRecordsHandler.add(
|
|
user_id, {f"Обновлен инструмент": editedToolkit.toDict()}
|
|
)
|
|
return editedToolkit.toDict()
|
|
|
|
@staticmethod
|
|
async def getAll():
|
|
query = select(Toolkit).order_by(Toolkit.id)
|
|
toolkits = await CRUD.read(query, True)
|
|
return [toolkit.toDict() for toolkit in toolkits] if toolkits else []
|
|
|
|
@staticmethod
|
|
async def get(toolkitId: int):
|
|
query = select(Toolkit).where(Toolkit.id == toolkitId)
|
|
toolkit = await CRUD.read(query)
|
|
if not toolkit:
|
|
logger.error("Инструмент не найден")
|
|
return {}
|
|
data = toolkit.toDict()
|
|
if toolkit.comment_user_id:
|
|
user_data = await UserHandler.get(toolkit.comment_user_id)
|
|
data["comment_user_data"] = user_data
|
|
return data
|
|
|
|
@staticmethod
|
|
async def getSeveral(toolkitIds: list[int]) -> list[dict]:
|
|
query = select(Toolkit).where(Toolkit.id.in_(toolkitIds)).order_by(Toolkit.id)
|
|
toolkits = await CRUD.read(query, True)
|
|
return [toolkit.toDict() for toolkit in toolkits] if toolkits else []
|
|
|
|
@staticmethod
|
|
async def checkCatogoryUse(category_id: int):
|
|
query = select(Toolkit).where(Toolkit.category_id == category_id)
|
|
toolkit = await CRUD.read(query)
|
|
return True if toolkit else False
|
|
|
|
@staticmethod
|
|
async def delete(toolkitId: int, user_id: int = None):
|
|
movements = await StockHandler.checkToolkitExists(toolkitId)
|
|
if movements:
|
|
logger.error("По инструменту было движение")
|
|
return {"errorMessage": "По инструменту было движение"}
|
|
query = select(Toolkit).where(Toolkit.id == toolkitId)
|
|
toolkit = await CRUD.read(query)
|
|
if not toolkit:
|
|
logger.error("Инструмент не найден")
|
|
return {"errorMessage": "Инструмент не найден"}
|
|
try:
|
|
toolkitTitle = toolkit.title
|
|
result = await CRUD.delete(toolkit)
|
|
except Exception as e:
|
|
logger.error(f"Ошибка удаления инструмента: {str(e)}")
|
|
return {"errorMessage": f"Ошибка удаления инструмента: {str(e)}"}
|
|
logger.info(
|
|
f"Инструмент {toolkitTitle} {'успешно удален' if result else 'не удален'}"
|
|
)
|
|
await ServiceRecordsHandler.add(
|
|
user_id, {"Удален инструмент": f"Название: {toolkitTitle}"}
|
|
)
|
|
return {"status": "ok"} if result else {"errorMessage": "Инструмент не удален"}
|
|
|
|
@staticmethod
|
|
async def addComment(toolkitId: int, user_id: int, comment: str):
|
|
query = select(Toolkit).where(Toolkit.id == toolkitId)
|
|
toolkit = await CRUD.read(query)
|
|
if not toolkit:
|
|
logger.error("Инструмент не найден")
|
|
return {"errorMessage": "Инструмент не найден"}
|
|
try:
|
|
await toolkit.edit(
|
|
comment_text=comment, comment_user_id=user_id, comment_at=datetime.now()
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Ошибка добавления комментария: {str(e)}")
|
|
return {"errorMessage": f"Ошибка добавления комментария: {str(e)}"}
|
|
logger.info(f"Комментарий к инструменту {toolkit.title} успешно добавлен")
|
|
return {"status": "ok"}
|
|
|
|
@staticmethod
|
|
async def addCompatibility(userId, data):
|
|
newCompatibility = await ToolkitCompatibility.add_compatibility(
|
|
int(data.get("toolkitId")), int(data.get("compatibleToolkitId"))
|
|
)
|
|
if "errorMessage" not in newCompatibility:
|
|
await ServiceRecordsHandler.add(
|
|
userId,
|
|
{
|
|
f"Добавлена совместимость": f"{data.get('toolkitId')} - {data.get('compatibleToolkitId')}"
|
|
},
|
|
)
|
|
return newCompatibility
|
|
|
|
@staticmethod
|
|
async def deleteCompatibility(userId, data):
|
|
deleteCompatibility = await ToolkitCompatibility.remove_compatibility(
|
|
int(data.get("toolkitId")), int(data.get("compatibleToolkitId"))
|
|
)
|
|
if "errorMessage" not in deleteCompatibility:
|
|
await ServiceRecordsHandler.add(
|
|
userId,
|
|
{
|
|
f"Удалена совместимость": f"{data.get('toolkitId')} - {data.get('compatibleToolkitId')}"
|
|
},
|
|
)
|
|
return deleteCompatibility
|
|
|
|
@staticmethod
|
|
async def getCompatibility(toolkitId: int):
|
|
result = await ToolkitCompatibility.get_compatibility(toolkitId)
|
|
if "errorMessage" in result:
|
|
return result
|
|
toolkitsIds = list(result.get("data").values())
|
|
toolkitsList = await ToolkitHandler.getSeveral(toolkitsIds)
|
|
toolkitsData = {toolkit["id"]: toolkit for toolkit in toolkitsList}
|
|
data = {"records": result.get("data"), "toolkits": toolkitsData}
|
|
return {"status": "ok", "data": data}
|
|
|
|
@staticmethod
|
|
async def initialize():
|
|
from .categories import CategoryHandler
|
|
|
|
logger.info("Инициализация инструментов")
|
|
categoriesList = await CategoryHandler.getAll()
|
|
categories = {category["title"]: category["id"] for category in categoriesList}
|
|
|
|
baseToolkits = [
|
|
{
|
|
"title": "Фреза №1",
|
|
"description": "Фреза такая сякая этакая #1",
|
|
"specifications": {
|
|
"Диаметр": "10",
|
|
"Длина": "20",
|
|
"Ещё что-то": "Ещё столько-то",
|
|
},
|
|
"category_id": categories["Фрезеровка"],
|
|
"quantity_min": 20,
|
|
"quantity_min_extra": 10,
|
|
"external_link": "https://nazv.ru",
|
|
},
|
|
{
|
|
"title": "Фреза №2",
|
|
"description": "Фреза такая сякая этакая #2",
|
|
"specifications": {
|
|
"Диаметр": "10",
|
|
"Длина": "20",
|
|
"Ещё что-то": "Ещё столько-то",
|
|
},
|
|
"category_id": categories["Фрезеровка"],
|
|
"quantity_min": 20,
|
|
"quantity_min_extra": 10,
|
|
"external_link": "https://nazv.ru",
|
|
},
|
|
{
|
|
"title": "Фреза №3",
|
|
"description": "Фреза такая сякая этакая #3",
|
|
"specifications": {
|
|
"Диаметр": "10",
|
|
"Длина": "20",
|
|
"Ещё что-то": "Ещё столько-то",
|
|
},
|
|
"category_id": categories["Фрезеровка"],
|
|
"quantity_min": 20,
|
|
"quantity_min_extra": 10,
|
|
"external_link": "https://nazv.ru",
|
|
},
|
|
{
|
|
"title": "Пластина №1",
|
|
"description": "Пластина такая сякая этакая #1",
|
|
"specifications": {
|
|
"Размер": "10",
|
|
"Радиус": "0.4",
|
|
"Ещё что-то": "Ещё столько-то",
|
|
},
|
|
"image": {
|
|
"main": "static/images/tools/default.png",
|
|
"additional": [
|
|
"static/images/users/default.png",
|
|
"static/images/logo.png",
|
|
],
|
|
},
|
|
"category_id": categories["Токарка"],
|
|
"quantity_min": 20,
|
|
"quantity_min_extra": 10,
|
|
"external_link": "https://nazv.ru",
|
|
},
|
|
{
|
|
"title": "Пластина №2",
|
|
"description": "Пластина такая сякая этакая #2",
|
|
"specifications": {
|
|
"Размер": "10",
|
|
"Радиус": "0.4",
|
|
"Ещё что-то": "Ещё столько-то",
|
|
},
|
|
"category_id": categories["Токарка"],
|
|
"quantity_min": 20,
|
|
"quantity_min_extra": 10,
|
|
"external_link": "https://nazv.ru",
|
|
},
|
|
{
|
|
"title": "Пластина №3",
|
|
"description": "Пластина такая сякая этакая #3",
|
|
"specifications": {
|
|
"Размер": "10",
|
|
"Радиус": "0.4",
|
|
"Ещё что-то": "Ещё столько-то",
|
|
},
|
|
"category_id": categories["Токарка"],
|
|
"quantity_min": 20,
|
|
"quantity_min_extra": 10,
|
|
"external_link": "https://nazv.ru",
|
|
},
|
|
{
|
|
"title": "Сверло №1",
|
|
"description": "Сверло такое сякое этакое #1",
|
|
"specifications": {
|
|
"Длина": "30",
|
|
"Диаметр": "5",
|
|
"Ещё что-то": "Ещё столько-то",
|
|
},
|
|
"category_id": categories["Слесарка"],
|
|
"quantity_min": 20,
|
|
"quantity_min_extra": 10,
|
|
"external_link": "https://nazv.ru",
|
|
},
|
|
{
|
|
"title": "Сверло №2",
|
|
"description": "Сверло такое сякое этакое #2",
|
|
"specifications": {
|
|
"Длина": "30",
|
|
"Диаметр": "5",
|
|
"Ещё что-то": "Ещё столько-то",
|
|
},
|
|
"category_id": categories["Слесарка"],
|
|
"quantity_min": 20,
|
|
"quantity_min_extra": 10,
|
|
"external_link": "https://nazv.ru",
|
|
},
|
|
{
|
|
"title": "Сверло №3",
|
|
"description": "Сверло такое сякое этакое #3",
|
|
"specifications": {
|
|
"Длина": "30",
|
|
"Диаметр": "5",
|
|
"Ещё что-то": "Ещё столько-то",
|
|
},
|
|
"category_id": categories["Слесарка"],
|
|
"quantity_min": 20,
|
|
"quantity_min_extra": 10,
|
|
"external_link": "https://nazv.ru",
|
|
},
|
|
]
|
|
|
|
for toolkit in baseToolkits:
|
|
await ToolkitHandler.add(toolkit)
|
|
|
|
logger.info("Базовые инструменты успешно созданы")
|
|
return
|