diff options
Diffstat (limited to 'server/szurubooru/func/users.py')
| -rw-r--r-- | server/szurubooru/func/users.py | 185 |
1 files changed, 102 insertions, 83 deletions
diff --git a/server/szurubooru/func/users.py b/server/szurubooru/func/users.py index e5946dc..5cbe3cc 100644 --- a/server/szurubooru/func/users.py +++ b/server/szurubooru/func/users.py @@ -1,9 +1,11 @@ -from datetime import datetime -from typing import Any, Optional, Union, List, Dict, Callable import re +from datetime import datetime +from typing import Any, Callable, Dict, List, Optional, Union + import sqlalchemy as sa -from szurubooru import config, db, model, errors, rest -from szurubooru.func import auth, util, serialization, files, images + +from szurubooru import config, db, errors, model, rest +from szurubooru.func import auth, files, images, serialization, util class UserNotFoundError(errors.NotFoundError): @@ -35,36 +37,41 @@ class InvalidAvatarError(errors.ValidationError): def get_avatar_path(user_name: str) -> str: - return 'avatars/' + user_name.lower() + '.png' + return "avatars/" + user_name.lower() + ".png" def get_avatar_url(user: model.User) -> str: assert user if user.avatar_style == user.AVATAR_GRAVATAR: assert user.email or user.name - return 'https://gravatar.com/avatar/%s?d=retro&s=%d' % ( + return "https://gravatar.com/avatar/%s?d=retro&s=%d" % ( util.get_md5((user.email or user.name).lower()), - config.config['thumbnails']['avatar_width']) + config.config["thumbnails"]["avatar_width"], + ) assert user.name - return '%s/avatars/%s.png' % ( - config.config['data_url'].rstrip('/'), user.name.lower()) + return "%s/avatars/%s.png" % ( + config.config["data_url"].rstrip("/"), + user.name.lower(), + ) def get_email( - user: model.User, - auth_user: model.User, - force_show_email: bool) -> Union[bool, str]: + user: model.User, auth_user: model.User, force_show_email: bool +) -> Union[bool, str]: assert user assert auth_user - if not force_show_email \ - and auth_user.user_id != user.user_id \ - and not auth.has_privilege(auth_user, 'users:edit:any:email'): + if ( + not force_show_email + and auth_user.user_id != user.user_id + and not auth.has_privilege(auth_user, "users:edit:any:email") + ): return False return user.email def get_liked_post_count( - user: model.User, auth_user: model.User) -> Union[bool, int]: + user: model.User, auth_user: model.User +) -> Union[bool, int]: assert user assert auth_user if auth_user.user_id != user.user_id: @@ -73,7 +80,8 @@ def get_liked_post_count( def get_disliked_post_count( - user: model.User, auth_user: model.User) -> Union[bool, int]: + user: model.User, auth_user: model.User +) -> Union[bool, int]: assert user assert auth_user if auth_user.user_id != user.user_id: @@ -83,29 +91,30 @@ def get_disliked_post_count( class UserSerializer(serialization.BaseSerializer): def __init__( - self, - user: model.User, - auth_user: model.User, - force_show_email: bool = False) -> None: + self, + user: model.User, + auth_user: model.User, + force_show_email: bool = False, + ) -> None: self.user = user self.auth_user = auth_user self.force_show_email = force_show_email def _serializers(self) -> Dict[str, Callable[[], Any]]: return { - 'name': self.serialize_name, - 'creationTime': self.serialize_creation_time, - 'lastLoginTime': self.serialize_last_login_time, - 'version': self.serialize_version, - 'rank': self.serialize_rank, - 'avatarStyle': self.serialize_avatar_style, - 'avatarUrl': self.serialize_avatar_url, - 'commentCount': self.serialize_comment_count, - 'uploadedPostCount': self.serialize_uploaded_post_count, - 'favoritePostCount': self.serialize_favorite_post_count, - 'likedPostCount': self.serialize_liked_post_count, - 'dislikedPostCount': self.serialize_disliked_post_count, - 'email': self.serialize_email, + "name": self.serialize_name, + "creationTime": self.serialize_creation_time, + "lastLoginTime": self.serialize_last_login_time, + "version": self.serialize_version, + "rank": self.serialize_rank, + "avatarStyle": self.serialize_avatar_style, + "avatarUrl": self.serialize_avatar_url, + "commentCount": self.serialize_comment_count, + "uploadedPostCount": self.serialize_uploaded_post_count, + "favoritePostCount": self.serialize_favorite_post_count, + "likedPostCount": self.serialize_liked_post_count, + "dislikedPostCount": self.serialize_disliked_post_count, + "email": self.serialize_email, } def serialize_name(self) -> Any: @@ -149,20 +158,22 @@ class UserSerializer(serialization.BaseSerializer): def serialize_user( - user: Optional[model.User], - auth_user: model.User, - options: List[str] = [], - force_show_email: bool = False) -> Optional[rest.Response]: + user: Optional[model.User], + auth_user: model.User, + options: List[str] = [], + force_show_email: bool = False, +) -> Optional[rest.Response]: if not user: return None return UserSerializer(user, auth_user, force_show_email).serialize(options) def serialize_micro_user( - user: Optional[model.User], - auth_user: model.User) -> Optional[rest.Response]: + user: Optional[model.User], auth_user: model.User +) -> Optional[rest.Response]: return serialize_user( - user, auth_user=auth_user, options=['name', 'avatarUrl']) + user, auth_user=auth_user, options=["name", "avatarUrl"] + ) def get_user_count() -> int: @@ -171,33 +182,34 @@ def get_user_count() -> int: def try_get_user_by_name(name: str) -> Optional[model.User]: return ( - db.session - .query(model.User) + db.session.query(model.User) .filter(sa.func.lower(model.User.name) == sa.func.lower(name)) - .one_or_none()) + .one_or_none() + ) def get_user_by_name(name: str) -> model.User: user = try_get_user_by_name(name) if not user: - raise UserNotFoundError('User %r not found.' % name) + raise UserNotFoundError("User %r not found." % name) return user def try_get_user_by_name_or_email(name_or_email: str) -> Optional[model.User]: return ( - db.session - .query(model.User) + db.session.query(model.User) .filter( - (sa.func.lower(model.User.name) == sa.func.lower(name_or_email)) | - (sa.func.lower(model.User.email) == sa.func.lower(name_or_email))) - .one_or_none()) + (sa.func.lower(model.User.name) == sa.func.lower(name_or_email)) + | (sa.func.lower(model.User.email) == sa.func.lower(name_or_email)) + ) + .one_or_none() + ) def get_user_by_name_or_email(name_or_email: str) -> model.User: user = try_get_user_by_name_or_email(name_or_email) if not user: - raise UserNotFoundError('User %r not found.' % name_or_email) + raise UserNotFoundError("User %r not found." % name_or_email) return user @@ -207,7 +219,7 @@ def create_user(name: str, password: str, email: str) -> model.User: update_user_password(user, password) update_user_email(user, email) if get_user_count() > 0: - user.rank = util.flip(auth.RANK_MAP)[config.config['default_rank']] + user.rank = util.flip(auth.RANK_MAP)[config.config["default_rank"]] else: user.rank = model.User.RANK_ADMINISTRATOR user.creation_time = datetime.utcnow() @@ -218,17 +230,18 @@ def create_user(name: str, password: str, email: str) -> model.User: def update_user_name(user: model.User, name: str) -> None: assert user if not name: - raise InvalidUserNameError('Name cannot be empty.') + raise InvalidUserNameError("Name cannot be empty.") if util.value_exceeds_column_size(name, model.User.name): - raise InvalidUserNameError('User name is too long.') + raise InvalidUserNameError("User name is too long.") name = name.strip() - name_regex = config.config['user_name_regex'] + name_regex = config.config["user_name_regex"] if not re.match(name_regex, name): raise InvalidUserNameError( - 'User name %r must satisfy regex %r.' % (name, name_regex)) + "User name %r must satisfy regex %r." % (name, name_regex) + ) other_user = try_get_user_by_name(name) if other_user and other_user.user_id != user.user_id: - raise UserAlreadyExistsError('User %r already exists.' % name) + raise UserAlreadyExistsError("User %r already exists." % name) if user.name and files.has(get_avatar_path(user.name)): files.move(get_avatar_path(user.name), get_avatar_path(name)) user.name = name @@ -237,14 +250,16 @@ def update_user_name(user: model.User, name: str) -> None: def update_user_password(user: model.User, password: str) -> None: assert user if not password: - raise InvalidPasswordError('Password cannot be empty.') - password_regex = config.config['password_regex'] + raise InvalidPasswordError("Password cannot be empty.") + password_regex = config.config["password_regex"] if not re.match(password_regex, password): raise InvalidPasswordError( - 'Password must satisfy regex %r.' % password_regex) + "Password must satisfy regex %r." % password_regex + ) user.password_salt = auth.create_password() password_hash, revision = auth.get_password_hash( - user.password_salt, password) + user.password_salt, password + ) user.password_hash = password_hash user.password_revision = revision @@ -253,53 +268,56 @@ def update_user_email(user: model.User, email: str) -> None: assert user email = email.strip() if util.value_exceeds_column_size(email, model.User.email): - raise InvalidEmailError('Email is too long.') + raise InvalidEmailError("Email is too long.") if not util.is_valid_email(email): - raise InvalidEmailError('E-mail is invalid.') + raise InvalidEmailError("E-mail is invalid.") user.email = email or None def update_user_rank( - user: model.User, rank: str, auth_user: model.User) -> None: + user: model.User, rank: str, auth_user: model.User +) -> None: assert user if not rank: - raise InvalidRankError('Rank cannot be empty.') + raise InvalidRankError("Rank cannot be empty.") rank = util.flip(auth.RANK_MAP).get(rank.strip(), None) all_ranks = list(auth.RANK_MAP.values()) if not rank: - raise InvalidRankError( - 'Rank can be either of %r.' % all_ranks) + raise InvalidRankError("Rank can be either of %r." % all_ranks) if rank in (model.User.RANK_ANONYMOUS, model.User.RANK_NOBODY): - raise InvalidRankError('Rank %r cannot be used.' % auth.RANK_MAP[rank]) - if all_ranks.index(auth_user.rank) \ - < all_ranks.index(rank) and get_user_count() > 0: - raise errors.AuthError('Trying to set higher rank than your own.') + raise InvalidRankError("Rank %r cannot be used." % auth.RANK_MAP[rank]) + if ( + all_ranks.index(auth_user.rank) < all_ranks.index(rank) + and get_user_count() > 0 + ): + raise errors.AuthError("Trying to set higher rank than your own.") user.rank = rank def update_user_avatar( - user: model.User, - avatar_style: str, - avatar_content: Optional[bytes] = None) -> None: + user: model.User, avatar_style: str, avatar_content: Optional[bytes] = None +) -> None: assert user - if avatar_style == 'gravatar': + if avatar_style == "gravatar": user.avatar_style = user.AVATAR_GRAVATAR - elif avatar_style == 'manual': + elif avatar_style == "manual": user.avatar_style = user.AVATAR_MANUAL - avatar_path = 'avatars/' + user.name.lower() + '.png' + avatar_path = "avatars/" + user.name.lower() + ".png" if not avatar_content: if files.has(avatar_path): return - raise InvalidAvatarError('Avatar content missing.') + raise InvalidAvatarError("Avatar content missing.") image = images.Image(avatar_content) image.resize_fill( - int(config.config['thumbnails']['avatar_width']), - int(config.config['thumbnails']['avatar_height'])) + int(config.config["thumbnails"]["avatar_width"]), + int(config.config["thumbnails"]["avatar_height"]), + ) files.save(avatar_path, image.to_png()) else: raise InvalidAvatarError( - 'Avatar style %r is invalid. Valid avatar styles: %r.' % ( - avatar_style, ['gravatar', 'manual'])) + "Avatar style %r is invalid. Valid avatar styles: %r." + % (avatar_style, ["gravatar", "manual"]) + ) def bump_user_login_time(user: model.User) -> None: @@ -312,7 +330,8 @@ def reset_user_password(user: model.User) -> str: password = auth.create_password() user.password_salt = auth.create_password() password_hash, revision = auth.get_password_hash( - user.password_salt, password) + user.password_salt, password + ) user.password_hash = password_hash user.password_revision = revision return password |