62 lines
1.7 KiB
Python
62 lines
1.7 KiB
Python
import secrets
|
|
from typing import Any
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.auth.tokens import hash_token
|
|
from app.db.models import CharacterCard, User
|
|
from app.character.card import DEFAULT_CARD, normalize_card
|
|
import json
|
|
|
|
|
|
def find_user_by_token(db: Session, token: str) -> User | None:
|
|
token_hash = hash_token(token.strip())
|
|
return db.scalar(
|
|
select(User).where(User.api_token_hash == token_hash, User.is_active.is_(True))
|
|
)
|
|
|
|
|
|
def user_to_dict(user: User) -> dict[str, Any]:
|
|
return {
|
|
"id": user.id,
|
|
"username": user.username,
|
|
"display_name": user.display_name or user.username,
|
|
}
|
|
|
|
|
|
def create_user(
|
|
db: Session,
|
|
*,
|
|
username: str,
|
|
display_name: str = "",
|
|
api_token: str | None = None,
|
|
) -> tuple[User, str]:
|
|
clean = username.strip().lower()
|
|
if not clean:
|
|
raise ValueError("username не может быть пустым")
|
|
existing = db.scalar(select(User).where(User.username == clean))
|
|
if existing:
|
|
raise ValueError(f"Пользователь «{clean}» уже существует")
|
|
|
|
plain_token = (api_token or "").strip() or secrets.token_urlsafe(32)
|
|
user = User(
|
|
username=clean,
|
|
display_name=(display_name or clean).strip(),
|
|
api_token_hash=hash_token(plain_token),
|
|
is_active=True,
|
|
)
|
|
db.add(user)
|
|
db.flush()
|
|
|
|
card = normalize_card(DEFAULT_CARD)
|
|
db.add(
|
|
CharacterCard(
|
|
user_id=user.id,
|
|
card_json=json.dumps(card, ensure_ascii=False),
|
|
)
|
|
)
|
|
db.commit()
|
|
db.refresh(user)
|
|
return user, plain_token
|