ForbiddenStarsApp/backend/app/services/user_service.py

"""Пользователи: создание из внешней личности, ник, активная группа, профиль."""
from __future__ import annotations

import re
from datetime import datetime, timezone
from pathlib import Path

from sqlmodel import Session, select

from app.auth.password import validate_new_password
from app.auth.provider import ExternalIdentity
from app.core.config import settings
from app.core.errors import (
    NicknameTakenError,
    NotFoundError,
    TelegramAlreadyLinkedError,
    TelegramTakenError,
    ValidationError,
)
from app.core.security import hash_password
from app.core.timeutil import utcnow
from app.models import AuthIdentity, Faction, 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 register_local(session: Session, nickname: str, password: str) -> User:
    """Регистрация по логину и паролю. Логин — это ник; Telegram можно привязать позже."""
    nickname = (nickname or "").strip()
    if not nickname_format_ok(nickname):
        raise ValidationError("Ник: 2–64 символа, буквы/цифры/пробел/.-_")
    if not nickname_available(session, nickname):
        raise NicknameTakenError()
    validate_new_password(password)
    user = User(
        nickname=nickname,
        role="player",
        auth_provider="local",
        password_hash=hash_password(password),
    )
    session.add(user)
    session.commit()
    session.refresh(user)
    return user


def set_password(session: Session, user: User, new_password: str) -> User:
    """Записать новый пароль. Проверку текущего делает вызывающий (игрок — да, админ — нет)."""
    validate_new_password(new_password)
    user.password_hash = hash_password(new_password)
    session.add(user)
    session.commit()
    session.refresh(user)
    return user


def link_telegram(session: Session, user: User, identity: ExternalIdentity) -> User:
    """Привязать Telegram к существующему аккаунту; ник не меняется.

    После привязки вход через Telegram попадает в этот аккаунт: find_by_identity находит
    его по той же связке provider+external_id, что создаёт регистрация через Telegram."""
    already = session.exec(
        select(AuthIdentity).where(
            AuthIdentity.user_id == user.id, AuthIdentity.provider == identity.provider
        )
    ).first()
    if user.telegram_id is not None or already is not None:
        raise TelegramAlreadyLinkedError()
    taken_by_id = session.exec(select(User).where(User.telegram_id == identity.telegram_id)).first()
    if find_by_identity(session, identity) is not None or taken_by_id is not None:
        raise TelegramTakenError()
    user.telegram_id = identity.telegram_id
    session.add(user)
    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 update_nickname(session: Session, user: User, new_nickname: str) -> User:
    new_nickname = (new_nickname or "").strip()
    if not nickname_format_ok(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:
        from app.services import group_service  # избегаем цикла импорта

        if group_service.get_membership(session, group_id, user.id) 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
    # В БД время наивное и хранится в UTC. .timestamp() у наивного значения считает
    # его локальным, и версия разъезжалась с лидербордом, где то же поле считает SQL
    # (strftime('%s') читает его как UTC) — один аватар качался браузером дважды.
    version = 0
    if updated_at is not None:
        aware = updated_at if updated_at.tzinfo else updated_at.replace(tzinfo=timezone.utc)
        version = int(aware.timestamp())
    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


_HISTORY_MODES = {"all", "best"}
_HISTORY_DETAILS = {"compact", "full"}


def update_history_prefs(
    session: Session, user: User, *, mode: str | None = None, detail: str | None = None
) -> User:
    """Витрина истории партий: что показывать и насколько подробно. None — не менять."""
    if mode is not None:
        if mode not in _HISTORY_MODES:
            raise ValidationError("Неизвестный режим истории партий.")
        user.history_mode = mode
    if detail is not None:
        if detail not in _HISTORY_DETAILS:
            raise ValidationError("Неизвестная подробность истории партий.")
        user.history_detail = detail
    session.add(user)
    session.commit()
    session.refresh(user)
    return user


def update_favorite_faction(session: Session, user: User, faction_id: int | None) -> User:
    """Любимая фракция — личный выбор игрока; None очищает выбор."""
    if faction_id is not None and session.get(Faction, faction_id) is None:
        raise ValidationError("Такой фракции не существует.")
    user.favorite_faction_id = faction_id
    session.add(user)
    session.commit()
    session.refresh(user)
    return user


def read_capped_image(file, max_bytes: int, limit_message: str) -> tuple[bytes, str]:
    """Прочитать загруженный файл с ограничением размера и убедиться, что это картинка.

    Читаем на байт больше лимита: так превышение видно, не загружая файл целиком.
    Один хелпер на все загрузки (аватар, фото партии, иконка ачивки) — иначе
    правка лимита или списка форматов расходится по четырём роутерам."""
    content = file.file.read(max_bytes + 1)
    if len(content) > max_bytes:
        raise ValidationError(limit_message)
    ext = sniff_image_ext(content)
    if ext is None:
        raise ValidationError("Поддерживаются только изображения PNG, JPEG или WebP.")
    return content, ext


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
    # Файл перезаписывается по тому же пути, поэтому при том же расширении avatar_path
    # не меняется, UPDATE не эмитится и onupdate не срабатывает. Без явного бампа
    # кэш-бастер остаётся прежним, и браузер час показывает старую картинку.
    user.updated_at = utcnow()
    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
    user.updated_at = utcnow()
    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),
    }