avnazv/toolbox
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

final

Browse Source
This commit is contained in:
Macbook
2025-12-14 18:48:06 +03:00
parent ccec507033
commit 1a51349eed
17 changed files with 1081 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files
db/handlers/__pycache__/access.cpython-313.pyc
BIN
View File
Binary file not shown.
db/handlers/__pycache__/records.cpython-313.pyc
BIN
View File
Binary file not shown.
db/handlers/__pycache__/user.cpython-313.pyc
BIN
View File
Binary file not shown.
db/handlers/access.py
+11 -13
View File
@@ -6,21 +6,20 @@ from db.handlers.records import ServiceRecordsHandler
class AccessLevelHandler:
async def add(newData):
async def add(newData, userId: int = None):
title = newData.get("title", None)
if not title:
logger.error("Не указано название уровня доступа")
return {}
return {"error": "Не указано название уровня доступа"}
exists = await CRUD.read(select(AccessLevel).where(AccessLevel.title == title))
if exists:
logger.error("Уровень доступа с таким названием уже существует")
return {}
return {"error": "Уровень доступа с таким названием уже существует"}
try:
logger.info(f"Создание уровня доступа {title}")
user_id = newData.pop("user_id", None)
accessData = await AccessLevel(**newData).save()
await ServiceRecordsHandler.add(
user_id, {"Добавлен уровень доступа": accessData.toDict()}
userId, {"Добавлен уровень доступа": accessData.toDict()}
)
except Exception as e:
logger.error(f"Ошибка создания уровня доступа: {str(e)}")
@@ -36,23 +35,22 @@ class AccessLevelHandler:
return {}
return accessData.toDict()
async def edit(accessId: int, **kwargs):
query = select(AccessLevel).where(AccessLevel.id == accessId)
async def edit(levelData, userId: int = None):
query = select(AccessLevel).where(AccessLevel.id == levelData.pop("id"))
accessData = await CRUD.read(query)
if not accessData:
logger.error("Уровень доступа не найден")
return {}
return {"error": "Уровень доступа не найден"}
try:
user_id = kwargs.pop("user_id", None)
editedAccessData = await accessData.edit(**kwargs)
editedAccessData = await accessData.edit(**levelData)
await ServiceRecordsHandler.add(
user_id, {"Обновлен уровень доступа": editedAccessData.toDict()}
userId, {"Обновлен уровень доступа": editedAccessData.toDict()}
)
except Exception as e:
logger.error(f"Ошибка обновления уровня доступа: {str(e)}")
return {}
return {"error": "Ошибка обновления уровня доступа"}
logger.info(
f"Уровень доступа {editedAccessData.title} успешно обновлен, изменены данные: {kwargs.keys()}"
f"Уровень доступа {editedAccessData.title} успешно обновлен, изменены данные: {levelData.keys()}"
)
return editedAccessData.toDict()
db/handlers/records.py
+16 -1
View File
@@ -1,6 +1,6 @@
from datetime import date, datetime, time, timedelta
from sqlalchemy import select
from sqlalchemy import func, select
from db.handlers.stock import StockHandler
from db.schemas.records import StocksRecords, ServicesRecords
@@ -175,6 +175,21 @@ class StocksRecordsHandler:
logger.exception("Ошибка получения записей")
return []
async def getUserRecords(user_id: int) -> int:
from db import CRUD
try:
query = select(func.count(StocksRecords.id)).where(
StocksRecords.init_user_id == user_id,
)
logger.debug(f"Получение всех записей пользователя {user_id}")
records = await CRUD.read(query)
logger.debug(f"{records} записей пользователя {user_id} успешно получены")
return records
except Exception as e:
logger.error(f"Ошибка получения записей: {str(e)}")
return 0
async def get(user_id: int, manager: bool):
from db import CRUD
db/handlers/user.py
+77 -48
View File
@@ -4,13 +4,20 @@ from db.handlers.access import AccessLevelHandler
from db.handlers.toolbox import ToolboxHandler
from utils import logger, pwd_hash, saveImage, safeFilename, deleteImage, pwd_verify
from db.schemas.user import User
from db.handlers.records import ServiceRecordsHandler
from db.handlers.records import ServiceRecordsHandler, StocksRecordsHandler
def handleUserPhoto(imageData, login: str):
import base64
login = safeFilename(login)
fileName = f"users/{login}.png"
if not saveImage(imageData, fileName):
fileName = f"static/images/users/{login}.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
@@ -39,9 +46,9 @@ class UserHandler:
userData["hashed_password"] = pwd_hash(userData.pop("password"))
if "photo" in userData:
imageData = userData.pop("photo")
photoFile = handleUserPhoto(imageData, login)
if photoFile:
userData["photo"] = photoFile
imageFileName = handleUserPhoto(imageData, login)
if imageFileName:
userData["photo"] = imageFileName
try:
newUser = await User(**userData).save()
except Exception as e:
@@ -67,10 +74,9 @@ class UserHandler:
logger.info(
f"Тулбокс {newToolbox['title']} успешно создан и закреплен за пользователем {newUser.username}"
)
await ServiceRecordsHandler.add(
user_id, {"Добавлен пользователь": newUser.toDict()}
)
newUserData = newUser.toDict()
newUserData.pop("hashed_password")
await ServiceRecordsHandler.add(user_id, {"Добавлен пользователь": newUserData})
newUserData["access_level_data"] = userAccessLevel
return newUserData
@@ -78,50 +84,66 @@ class UserHandler:
id = userData.get("id", None)
if not id:
logger.error("Не указан id пользователя")
return {}
return {"error": "Не указан id пользователя"}
query = select(User).where(User.id == id)
user = await CRUD.read(query)
if not user:
logger.error("Пользователь с таким id не найден")
return {}
changedUserData = userData.get("changedUserData", {})
if len(changedUserData.keys()) == 0:
return {"error": "Пользователь не найден"}
if len(userData.keys()) == 0:
logger.error("Не указаны изменяемые данные")
return {}
if "password" in changedUserData:
userData["hashed_password"] = pwd_hash(changedUserData.pop("password"))
if "photo" in changedUserData:
imageData = changedUserData.pop("photo")
photoFile = handleUserPhoto(imageData, user.login)
if photoFile:
changedUserData["photo"] = photoFile
return {"error": "Не указаны изменяемые данные"}
if "password" in userData:
userData["hashed_password"] = pwd_hash(userData.pop("password"))
if "photo" in userData:
imageData = userData.pop("photo")
if imageData != "":
login = user.login if "login" not in userData else userData["login"]
photoFile = handleUserPhoto(imageData, login)
if photoFile:
userData["photo"] = photoFile
deleteImage(user.photo)
else:
logger.error("Ошибка обновления фото пользователя")
return {"error": "Ошибка обновления фото пользователя"}
else:
userData["photo"] = "static/images/users/default.png"
deleteImage(user.photo)
try:
editedUser = await user.edit(**changedUserData)
userData.pop("id")
editedUser = await user.edit(**userData)
except Exception as e:
logger.error(f"Ошибка обновления пользователя: {str(e)}")
return {}
return {"error": "Ошибка обновления пользователя"}
if not editedUser:
logger.error("Ошибка обновления пользователя")
return {}
return {"error": "Ошибка обновления пользователя"}
logger.info(
f"Пользователь {editedUser.username} успешно обновлен, изменены данные: {changedUserData.keys()}"
f"Пользователь {editedUser.username} успешно обновлен, изменены данные: {userData.keys()}"
)
if not user.available_own_toolbox:
if editedUser.available_own_toolbox:
newToolboxData = {
"title": editedUser.username,
"description": f"Оборудование, полученное сотрудником '{editedUser.username}' под личную материальную ответственность",
"owner_id": editedUser.id,
}
newToolbox = await ToolboxHandler.addNewToolbox(newToolboxData)
logger.info(
f"Тулбокс {newToolbox['title']} успешно создан и закреплен за пользователем {editedUser.username}"
)
await ServiceRecordsHandler.add(
user_id, {"Изменен пользователь": editedUser.toDict()}
)
return editedUser.toDict()
if user.access_level_id != editedUser.access_level_id:
userAccessLevel = await AccessLevelHandler.get(user.access_level_id)
userAccessLevelNew = await AccessLevelHandler.get(
editedUser.access_level_id
)
if not userAccessLevel or not userAccessLevelNew:
logger.error("Уровень доступа не найден")
return {"error": "Уровень доступа не найден"}
if not userAccessLevel.get("available_own_toolbox"):
if userAccessLevelNew.get("available_own_toolbox"):
newToolboxData = {
"title": editedUser.username,
"description": f"Оборудование, полученное сотрудником '{editedUser.username}' под личную материальную ответственность",
"owner_id": editedUser.id,
}
newToolbox = await ToolboxHandler.addNewToolbox(newToolboxData)
logger.info(
f"Тулбокс {newToolbox['title']} успешно создан и закреплен за пользователем {editedUser.username}"
)
editUserData = editedUser.toDict()
editUserData.pop("hashed_password")
await ServiceRecordsHandler.add(user_id, {"Изменен пользователь": editUserData})
return editUserData
async def getAll() -> list[dict]:
query = select(User)
@@ -136,23 +158,30 @@ class UserHandler:
return {}
return user.toDict()
async def delete(id: int, user_id: int = None) -> bool:
async def delete(id: int, user_id: int = None) -> dict:
userRecordsCount = await StocksRecordsHandler.getUserRecords(id)
if userRecordsCount > 0:
logger.error(f"У пользователя {id} есть записи: {userRecordsCount}")
return {"error": "У пользователя есть записи"}
query = select(User).where(User.id == id)
user = await CRUD.read(query)
if not user:
logger.error("Пользователь с таким id не найден")
return False
return {"error": "Пользователь не найден"}
try:
userName = user.username
photoFile = user.photo
result = await CRUD.delete(user)
except Exception as e:
logger.error(f"Ошибка удаления пользователя: {str(e)}")
return False
return {"error": "Ошибка удаления пользователя"}
if result:
deleteImage(photoFile)
logger.info(
f"Пользователь {userName} {'успешно удален' if result else 'не удален'}"
)
await ServiceRecordsHandler.add(user_id, {"Удален пользователь": userName})
return result
return {"error": "Ошибка удаления пользователя"} if not result else {}
async def deletePhoto(id: int, user_id: int = None) -> bool:
query = select(User).where(User.id == id)
@@ -205,25 +234,25 @@ class UserHandler:
baseUsers = {
"admin": {
"login": "admin",
"username": "Администратор - Demo",
"username": "Администратор",
"password": password,
"access_level_id": acessLevels["Администратор"],
},
"manager": {
"login": "manager",
"username": "Менеджер - Demo",
"username": "Менеджер",
"password": password,
"access_level_id": acessLevels["Менеджер"],
},
"storekeeper": {
"login": "storekeeper",
"username": "Кладовщик - Demo",
"username": "Кладовщик",
"password": password,
"access_level_id": acessLevels["Кладовщик"],
},
"employee": {
"login": "employee",
"username": "Сотрудник - Demo",
"username": "Сотрудник",
"password": password,
"access_level_id": acessLevels["Сотрудник"],
},