111 lines
3.6 KiB
Python
111 lines
3.6 KiB
Python
import pathlib
|
|
|
|
from snek.system import security
|
|
from snek.system.service import BaseService
|
|
|
|
|
|
class UserService(BaseService):
|
|
mapper_name = "user"
|
|
|
|
async def get_by_username(self, username):
|
|
return await self.get(username=username)
|
|
|
|
async def search(self, query, **kwargs):
|
|
query = query.strip().lower()
|
|
if not query:
|
|
return []
|
|
results = []
|
|
async for result in self.find(username={"ilike": "%" + query + "%"}, **kwargs):
|
|
results.append(result)
|
|
return results
|
|
|
|
async def validate_login(self, username, password):
|
|
model = await self.get(username=username)
|
|
if not model:
|
|
return False
|
|
if not await security.verify(password, model["password"]):
|
|
return False
|
|
return True
|
|
|
|
async def save(self, user):
|
|
if not user["color"]:
|
|
user["color"] = await self.services.util.random_light_hex_color()
|
|
return await super().save(user)
|
|
|
|
def authenticate_sync(self, username, password):
|
|
user = self.get_by_username_sync(username)
|
|
|
|
if not user:
|
|
return False
|
|
if not security.verify_sync(password, user["password"]):
|
|
return False
|
|
return True
|
|
|
|
async def authenticate(self, username, password):
|
|
success = await self.validate_login(username, password)
|
|
if not success:
|
|
return None
|
|
|
|
model = await self.get(username=username, deleted_at=None)
|
|
return model
|
|
|
|
def get_admin_uids(self):
|
|
return self.mapper.get_admin_uids()
|
|
|
|
async def get_repository_path(self, user_uid):
|
|
return pathlib.Path(f"./drive/repositories/{user_uid}")
|
|
|
|
async def get_static_path(self, user_uid):
|
|
path = pathlib.Path(f"./drive/{user_uid}/snek/static")
|
|
if not path.exists():
|
|
return None
|
|
return path
|
|
|
|
async def get_template_path(self, user_uid):
|
|
path = pathlib.Path(f"./drive/{user_uid}/snek/templates")
|
|
if not path.exists():
|
|
return None
|
|
return path
|
|
|
|
def get_by_username_sync(self, username):
|
|
user = self.mapper.db["user"].find_one(username=username, deleted_at=None)
|
|
return dict(user)
|
|
|
|
def get_home_folder_by_username(self, username):
|
|
user = self.get_by_username_sync(username)
|
|
folder = pathlib.Path(f"./drive/{user['uid']}")
|
|
if not folder.exists():
|
|
try:
|
|
folder.mkdir(parents=True, exist_ok=True)
|
|
except:
|
|
pass
|
|
return folder
|
|
|
|
async def get_home_folder(self, user_uid):
|
|
folder = pathlib.Path(f"./drive/{user_uid}")
|
|
if not folder.exists():
|
|
try:
|
|
folder.mkdir(parents=True, exist_ok=True)
|
|
except:
|
|
pass
|
|
return folder
|
|
|
|
async def register(self, email, username, password):
|
|
if await self.exists(username=username):
|
|
raise Exception("User already exists.")
|
|
model = await self.new()
|
|
model["nick"] = username
|
|
model["color"] = await self.services.util.random_light_hex_color()
|
|
model.email.value = email
|
|
model.username.value = username
|
|
model.password.value = await security.hash(password)
|
|
if await self.save(model):
|
|
if model:
|
|
channel = await self.services.channel.ensure_public_channel(
|
|
model["uid"]
|
|
)
|
|
if not channel:
|
|
raise Exception("Failed to create public channel.")
|
|
return model
|
|
raise Exception(f"Failed to create user: {model.errors}.")
|