137 lines
5.3 KiB
Python
137 lines
5.3 KiB
Python
|
|
from typing import Any, Dict, Union
|
|
|
|
import os
|
|
import magic
|
|
|
|
from matemat.webserver import pagelet, RequestArguments, PageletResponse, RedirectResponse, TemplateResponse
|
|
from matemat.db import MatematDatabase
|
|
from matemat.db.primitives import User
|
|
from matemat.exceptions import DatabaseConsistencyError, HttpException
|
|
|
|
|
|
@pagelet('/admin')
|
|
def admin(method: str,
|
|
path: str,
|
|
args: RequestArguments,
|
|
session_vars: Dict[str, Any],
|
|
headers: Dict[str, str],
|
|
config: Dict[str, str]) \
|
|
-> Union[str, bytes, PageletResponse]:
|
|
if 'authentication_level' not in session_vars or 'authenticated_user' not in session_vars:
|
|
return RedirectResponse('/login')
|
|
authlevel: int = session_vars['authentication_level']
|
|
uid: int = session_vars['authenticated_user']
|
|
if authlevel < 2:
|
|
raise HttpException(403)
|
|
|
|
with MatematDatabase(config['DatabaseFile']) as db:
|
|
user = db.get_user(uid)
|
|
if method == 'POST' and 'change' in args:
|
|
handle_change(args, user, db, config)
|
|
elif method == 'POST' and 'adminchange' in args and user.is_admin:
|
|
handle_admin_change(args, db, config)
|
|
|
|
users = db.list_users()
|
|
products = db.list_products()
|
|
return TemplateResponse('admin.html',
|
|
authuser=user, authlevel=authlevel, users=users, products=products,
|
|
setupname=config['InstanceName'])
|
|
|
|
|
|
def handle_change(args: RequestArguments, user: User, db: MatematDatabase, config: Dict[str, str]) -> None:
|
|
try:
|
|
change = str(args.change)
|
|
|
|
if change == 'account':
|
|
if 'username' not in args or 'email' not in args:
|
|
return
|
|
username = str(args.username)
|
|
email = str(args.email)
|
|
if len(email) == 0:
|
|
email = None
|
|
try:
|
|
db.change_user(user, name=username, email=email)
|
|
except DatabaseConsistencyError:
|
|
pass
|
|
|
|
elif change == 'password':
|
|
if 'oldpass' not in args or 'newpass' not in args or 'newpass2' not in args:
|
|
return
|
|
oldpass = str(args.oldpass)
|
|
newpass = str(args.newpass)
|
|
newpass2 = str(args.newpass2)
|
|
if newpass != newpass2:
|
|
raise ValueError('New passwords don\'t match')
|
|
db.change_password(user, oldpass, newpass)
|
|
|
|
elif change == 'touchkey':
|
|
if 'touchkey' not in args:
|
|
return
|
|
touchkey = str(args.touchkey)
|
|
if len(touchkey) == 0:
|
|
touchkey = None
|
|
db.change_touchkey(user, '', touchkey, verify_password=False)
|
|
|
|
elif change == 'avatar':
|
|
if 'avatar' not in args:
|
|
return
|
|
avatar = bytes(args.avatar)
|
|
filemagic: magic.FileMagic = magic.detect_from_content(avatar)
|
|
if filemagic.mime_type != 'image/png':
|
|
# TODO: Optionally convert to png
|
|
return
|
|
abspath: str = os.path.join(os.path.abspath(config['UploadDir']), 'thumbnails/users/')
|
|
os.makedirs(abspath, exist_ok=True)
|
|
with open(os.path.join(abspath, f'{user.id}.png'), 'wb') as f:
|
|
f.write(avatar)
|
|
|
|
except UnicodeDecodeError:
|
|
raise ValueError('an argument not a string')
|
|
|
|
|
|
def handle_admin_change(args: RequestArguments, db: MatematDatabase, config: Dict[str, str]):
|
|
try:
|
|
change = str(args.adminchange)
|
|
|
|
if change == 'newuser':
|
|
if 'username' not in args or 'email' not in args or 'password' not in args:
|
|
return
|
|
username = str(args.username)
|
|
email = str(args.email)
|
|
if len(email) == 0:
|
|
email = None
|
|
password = str(args.password)
|
|
is_member = 'ismember' in args
|
|
is_admin = 'isadmin' in args
|
|
db.create_user(username, password, email, member=is_member, admin=is_admin)
|
|
|
|
elif change == 'newproduct':
|
|
if 'name' not in args or 'pricemember' not in args or 'pricenonmember' not in args:
|
|
return
|
|
name = str(args.name)
|
|
price_member = int(str(args.pricemember))
|
|
price_non_member = int(str(args.pricenonmember))
|
|
newproduct = db.create_product(name, price_member, price_non_member)
|
|
if 'image' in args:
|
|
image = bytes(args.image)
|
|
filemagic: magic.FileMagic = magic.detect_from_content(image)
|
|
if filemagic.mime_type != 'image/png':
|
|
# TODO: Optionally convert to png
|
|
return
|
|
if len(image) > 0:
|
|
abspath: str = os.path.join(os.path.abspath(config['UploadDir']), 'thumbnails/products/')
|
|
os.makedirs(abspath, exist_ok=True)
|
|
with open(os.path.join(abspath, f'{newproduct.id}.png'), 'wb') as f:
|
|
f.write(image)
|
|
|
|
elif change == 'restock':
|
|
if 'productid' not in args or 'amount' not in args:
|
|
return
|
|
productid = int(str(args.productid))
|
|
amount = int(str(args.amount))
|
|
product = db.get_product(productid)
|
|
db.restock(product, amount)
|
|
|
|
except UnicodeDecodeError:
|
|
raise ValueError('an argument not a string')
|