NotBigGhostandClaude Fable 5
e5b7420bae
До 10 подсказок по подстроке ника среди активных игроков (role=player). Регистронезависимо и для кириллицы: SQLite LIKE/lower сворачивают регистр только у ASCII, поэтому фильтрация/ранжирование в Python (casefold): префиксные совпадения раньше подстрочных, внутри — по алфавиту. Пустой q — пустой список. Тесты + регенерированный schema.d.ts. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
257 lines
10 KiB
Python
257 lines
10 KiB
Python
"""Пользователи: создание из внешней личности, ник, активная группа, профиль."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from sqlmodel import Session, select
|
|
|
|
from app.auth.provider import ExternalIdentity
|
|
from app.core.config import settings
|
|
from app.core.errors import NicknameTakenError, NotFoundError, ValidationError
|
|
from app.models import AuthIdentity, GroupMember, User
|
|
|
|
_NICK_RE = re.compile(r"^[\w .\-]{2,64}$", re.UNICODE)
|
|
_BIO_MAX = 500
|
|
|
|
# Аватары: файл на диске под settings.upload_dir/avatars, путь относительный.
|
|
_AVATAR_SUBDIR = "avatars"
|
|
_IMAGE_MEDIA = {"png": "image/png", "jpg": "image/jpeg", "webp": "image/webp"}
|
|
|
|
|
|
def get_user(session: Session, user_id: int) -> User:
|
|
user = session.get(User, user_id)
|
|
if user is None:
|
|
raise NotFoundError("Пользователь не найден.")
|
|
return user
|
|
|
|
|
|
def nickname_available(session: Session, nickname: str, exclude_user_id: int | None = None) -> bool:
|
|
stmt = select(User).where(User.nickname == nickname)
|
|
existing = session.exec(stmt).first()
|
|
return existing is None or existing.id == exclude_user_id
|
|
|
|
|
|
def search_users(session: Session, query: str, limit: int = 3) -> list[User]:
|
|
"""Подсказки по нику: активные игроки (role='player'), без учёта регистра.
|
|
|
|
SQLite LIKE/lower() сворачивают регистр только для ASCII, а ники бывают
|
|
кириллицей — поэтому выбираем активных игроков и фильтруем/ранжируем в
|
|
Python (casefold). Пользователей десятки, полный проход дёшев; при росте
|
|
до тысяч — заменить на индексированную колонку nickname_lower.
|
|
Сначала префиксные совпадения, затем подстрочные, внутри — по алфавиту.
|
|
"""
|
|
q = (query or "").strip().casefold()
|
|
if not q:
|
|
return []
|
|
users = session.exec(
|
|
select(User).where(User.role == "player", User.is_active == True) # noqa: E712
|
|
).all()
|
|
matched = [u for u in users if q in u.nickname.casefold()]
|
|
matched.sort(
|
|
key=lambda u: (0 if u.nickname.casefold().startswith(q) else 1, u.nickname.casefold())
|
|
)
|
|
return matched[:limit]
|
|
|
|
|
|
def nickname_format_ok(nickname: str) -> bool:
|
|
"""Ник подходит по формату (2–64, буквы/цифры/пробел/.-_)."""
|
|
return bool(_NICK_RE.match((nickname or "").strip()))
|
|
|
|
|
|
def _unique_nickname(session: Session, base: str) -> str:
|
|
base = (base or "Игрок").strip()[:60] or "Игрок"
|
|
candidate = base
|
|
suffix = 1
|
|
while session.exec(select(User).where(User.nickname == candidate)).first() is not None:
|
|
suffix += 1
|
|
candidate = f"{base} {suffix}"
|
|
return candidate
|
|
|
|
|
|
def find_by_identity(session: Session, identity: ExternalIdentity) -> User | None:
|
|
"""Существующий пользователь по внешней личности (provider+external_id).
|
|
|
|
Для Telegram external_id — это НЕИЗМЕНЯЕМЫЙ числовой id аккаунта (не @тег), поэтому
|
|
вход с любого устройства/после смены тега ведёт к одному и тому же пользователю.
|
|
"""
|
|
link = session.exec(
|
|
select(AuthIdentity).where(
|
|
AuthIdentity.provider == identity.provider,
|
|
AuthIdentity.external_id == identity.external_id,
|
|
)
|
|
).first()
|
|
return get_user(session, link.user_id) if link is not None else None
|
|
|
|
|
|
def _create_from_identity(session: Session, identity: ExternalIdentity, nickname: str) -> User:
|
|
user = User(
|
|
nickname=nickname,
|
|
role="player",
|
|
auth_provider=identity.provider,
|
|
telegram_id=identity.telegram_id,
|
|
)
|
|
session.add(user)
|
|
session.flush()
|
|
session.add(
|
|
AuthIdentity(
|
|
user_id=user.id, # type: ignore[arg-type]
|
|
provider=identity.provider,
|
|
external_id=identity.external_id,
|
|
)
|
|
)
|
|
session.commit()
|
|
session.refresh(user)
|
|
return user
|
|
|
|
|
|
def get_or_create_from_identity(session: Session, identity: ExternalIdentity) -> User:
|
|
"""Найти по личности или создать, авто-подбирая уникальный ник (dev-вход по нику)."""
|
|
user = find_by_identity(session, identity)
|
|
if user is not None:
|
|
return user
|
|
nickname = _unique_nickname(session, identity.suggested_nickname or identity.external_id)
|
|
return _create_from_identity(session, identity, nickname)
|
|
|
|
|
|
def register_from_identity(
|
|
session: Session, identity: ExternalIdentity, nickname: str
|
|
) -> User:
|
|
"""Создать пользователя с ЯВНО выбранным ником (регистрация через Telegram).
|
|
|
|
В отличие от get_or_create — ник не подбирается автоматически: занятый/неподходящий
|
|
ник вызывает ошибку, чтобы пользователь выбрал другой.
|
|
"""
|
|
nickname = (nickname or "").strip()
|
|
if not nickname_format_ok(nickname):
|
|
raise ValidationError("Ник: 2–64 символа, буквы/цифры/пробел/.-_")
|
|
if not nickname_available(session, nickname):
|
|
raise NicknameTakenError()
|
|
return _create_from_identity(session, identity, nickname)
|
|
|
|
|
|
def update_nickname(session: Session, user: User, new_nickname: str) -> User:
|
|
new_nickname = (new_nickname or "").strip()
|
|
if not _NICK_RE.match(new_nickname):
|
|
raise ValidationError("Ник: 2–64 символа, буквы/цифры/пробел/.-_")
|
|
if not nickname_available(session, new_nickname, exclude_user_id=user.id):
|
|
raise NicknameTakenError()
|
|
user.nickname = new_nickname
|
|
session.add(user)
|
|
session.commit()
|
|
session.refresh(user)
|
|
return user
|
|
|
|
|
|
def set_active_group(session: Session, user: User, group_id: int | None) -> User:
|
|
if group_id is not None:
|
|
member = session.exec(
|
|
select(GroupMember).where(
|
|
GroupMember.group_id == group_id, GroupMember.user_id == user.id
|
|
)
|
|
).first()
|
|
if member is None:
|
|
raise ValidationError("Нельзя сделать активной группу, в которой вы не состоите.")
|
|
user.active_group_id = group_id
|
|
session.add(user)
|
|
session.commit()
|
|
session.refresh(user)
|
|
return user
|
|
|
|
|
|
# ─── Профиль: «о себе», аватар ────────────────────────────────────────────────
|
|
|
|
def avatar_url_for(user_id: int, avatar_path: str | None, updated_at: datetime | None) -> str | None:
|
|
"""URL аватара с кэш-бастингом по updated_at. None, если аватара нет.
|
|
|
|
Один источник для build_me, публичного профиля и лидерборда — чтобы клиент
|
|
подтягивал новую картинку после смены (файл перезаписывается по тому же пути)."""
|
|
if not avatar_path:
|
|
return None
|
|
version = int(updated_at.timestamp()) if updated_at else 0
|
|
return f"/api/users/{user_id}/avatar?v={version}"
|
|
|
|
|
|
def update_bio(session: Session, user: User, bio: str | None) -> User:
|
|
bio = (bio or "").strip()
|
|
if len(bio) > _BIO_MAX:
|
|
raise ValidationError(f"«О себе»: не более {_BIO_MAX} символов.")
|
|
user.bio = bio or None
|
|
session.add(user)
|
|
session.commit()
|
|
session.refresh(user)
|
|
return user
|
|
|
|
|
|
def sniff_image_ext(content: bytes) -> str | None:
|
|
"""Расширение по магическим байтам (PNG/JPEG/WebP), без Pillow. None — не картинка."""
|
|
if content.startswith(b"\x89PNG\r\n\x1a\n"):
|
|
return "png"
|
|
if content.startswith(b"\xff\xd8\xff"):
|
|
return "jpg"
|
|
if len(content) >= 12 and content[0:4] == b"RIFF" and content[8:12] == b"WEBP":
|
|
return "webp"
|
|
return None
|
|
|
|
|
|
def avatar_media_type(ext: str) -> str:
|
|
return _IMAGE_MEDIA.get(ext, "application/octet-stream")
|
|
|
|
|
|
def avatar_abs_path(avatar_path: str | None) -> Path | None:
|
|
"""Абсолютный путь файла аватара (или None), для отдачи через FileResponse."""
|
|
if not avatar_path:
|
|
return None
|
|
return Path(settings.upload_dir) / avatar_path
|
|
|
|
|
|
def set_avatar(session: Session, user: User, content: bytes, ext: str) -> User:
|
|
avatars_dir = Path(settings.upload_dir) / _AVATAR_SUBDIR
|
|
avatars_dir.mkdir(parents=True, exist_ok=True)
|
|
# Удаляем прежний файл (мог быть с другим расширением), затем пишем новый.
|
|
old = avatar_abs_path(user.avatar_path)
|
|
if old and old.exists():
|
|
try:
|
|
old.unlink()
|
|
except OSError:
|
|
pass
|
|
rel = f"{_AVATAR_SUBDIR}/{user.id}.{ext}"
|
|
(Path(settings.upload_dir) / rel).write_bytes(content)
|
|
user.avatar_path = rel
|
|
session.add(user)
|
|
session.commit()
|
|
session.refresh(user)
|
|
return user
|
|
|
|
|
|
def clear_avatar(session: Session, user: User) -> User:
|
|
path = avatar_abs_path(user.avatar_path)
|
|
if path and path.exists():
|
|
try:
|
|
path.unlink()
|
|
except OSError:
|
|
pass
|
|
user.avatar_path = None
|
|
session.add(user)
|
|
session.commit()
|
|
session.refresh(user)
|
|
return user
|
|
|
|
|
|
def public_profile(session: Session, user_id: int) -> dict:
|
|
"""Публичный профиль игрока: ник/«о себе»/аватар + глобальная статистика."""
|
|
from app.services import stats_service # ленивый импорт (избегаем цикла)
|
|
|
|
user = session.get(User, user_id)
|
|
if user is None:
|
|
raise NotFoundError("Пользователь не найден.")
|
|
return {
|
|
"user_id": user.id,
|
|
"nickname": user.nickname,
|
|
"bio": user.bio,
|
|
"avatar_url": avatar_url_for(user.id, user.avatar_path, user.updated_at), # type: ignore[arg-type]
|
|
"stats": stats_service.profile_stats(session, user_id, group_id=None),
|
|
}
|