ForbiddenStarsApp/backend/app/services/announcement_service.py

"""Объявления администрации (#84): очистка текста, период показа, кому и что показать.

Объявление видно игроку, пока идёт его период и игрок не закрыл текущую версию. Закрытие
(«Понятно») пишет отметку с номером версии; правка с «показать заново» поднимает версию —
и закрывшие прежнюю увидят объявление снова, с пометкой «обновлено».

Текст хранится HTML-ом из редактора админки, но только после очистки по белому списку
(sanitize_body): фронт вставляет его без экранирования, так что это единственный барьер
между полем редактора и страницей игрока.
"""
from __future__ import annotations

import html
from datetime import datetime, timezone
from html.parser import HTMLParser

from sqlalchemy import func
from sqlmodel import Session, select

from app.core.errors import NotFoundError, ValidationError
from app.core.timeutil import utcnow
from app.models import Announcement, AnnouncementView, User

TITLE_MAX = 60
TEXT_MAX = 600  # видимых символов, без разметки

# ─── Очистка HTML ────────────────────────────────────────────────────────────

# Что оставляем и во что превращаем. Редактор (contenteditable + execCommand) в разных
# браузерах пишет то <b>, то <strong>, абзацы — <div> или <p>; приводим к одному виду.
_TAGS = {
    "b": "b",
    "strong": "b",
    "i": "em",
    "em": "em",
    "mark": "mark",
    "p": "p",
    "div": "p",
    "br": "br",
}
# Теги, которые выбрасываются вместе с содержимым: их текст не предназначен для показа.
_DROP_WITH_CONTENT = {
    "script", "style", "template", "noscript", "iframe", "object", "embed",
    "svg", "math", "head", "title", "textarea", "select",
}
_VOID = {"br"}


class _Sanitizer(HTMLParser):
    """Пересобирает HTML из разобранных токенов: теги — только из белого списка и без
    атрибутов (кроме class="red" у <mark>), весь текст экранируется заново. Всё, что
    парсер не распознал как тег из списка, становится текстом или пропадает."""

    def __init__(self) -> None:
        super().__init__(convert_charrefs=True)
        self.out: list[str] = []
        self.text: list[str] = []
        self.stack: list[str] = []
        self.drop_depth = 0

    def _pop(self) -> str:
        top = self.stack.pop()
        last = self.out[-1] if self.out else ""
        # Точное сравнение: startswith("<b") поймал бы и <br>.
        if last == f"<{top}>" or last.startswith(f"<{top} "):
            self.out.pop()  # пустая пара (<p></p> от вложенных <div>) — выбрасываем
        else:
            self.out.append(f"</{top}>")
        return top

    def _close_to(self, tag: str) -> None:
        while self.stack:
            if self._pop() == tag:
                return

    def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
        if tag in _DROP_WITH_CONTENT:
            self.drop_depth += 1
            return
        if self.drop_depth or tag not in _TAGS:
            return
        name = _TAGS[tag]
        if name in _VOID:
            self.out.append(f"<{name}>")
            return
        if name == "p" and "p" in self.stack:
            # Абзац внутри абзаца (вложенные <div> из contenteditable) — закрываем прежний.
            self._close_to("p")
        opening = f"<{name}>"
        if name == "mark":
            classes = next((v or "" for k, v in attrs if k == "class"), "").split()
            if "red" in classes:
                opening = '<mark class="red">'
        self.stack.append(name)
        self.out.append(opening)

    def handle_startendtag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
        # <br/> и прочие самозакрытые: содержимого нет, так что drop-теги тут ни при чём.
        if not self.drop_depth and _TAGS.get(tag) in _VOID:
            self.out.append(f"<{_TAGS[tag]}>")

    def handle_endtag(self, tag: str) -> None:
        if tag in _DROP_WITH_CONTENT:
            self.drop_depth = max(0, self.drop_depth - 1)
            return
        if self.drop_depth:
            return
        name = _TAGS.get(tag)
        if name and name not in _VOID and name in self.stack:
            self._close_to(name)

    def handle_data(self, data: str) -> None:
        if self.drop_depth:
            return
        self.text.append(data)
        self.out.append(html.escape(data, quote=False))

    def result(self) -> tuple[str, str]:
        self.close()
        while self.stack:
            self._pop()
        return "".join(self.out), "".join(self.text)


def sanitize_body(raw: str) -> tuple[str, str]:
    """(очищенный HTML, видимый текст). Видимый текст нужен для проверки длины."""
    parser = _Sanitizer()
    parser.feed(raw)
    return parser.result()


# ─── Проверки и время ─────────────────────────────────────────────────────────

def _now() -> datetime:
    # В SQLite моменты лежат наивными (UTC) — сравниваем с наивным же «сейчас».
    return utcnow().replace(tzinfo=None)


def _naive_utc(dt: datetime) -> datetime:
    """Момент из запроса → наивный UTC. Время без смещения считаем UTC."""
    if dt.tzinfo is not None:
        dt = dt.astimezone(timezone.utc).replace(tzinfo=None)
    return dt.replace(second=0, microsecond=0)


def _clean_fields(
    title: str, body_html: str, starts_at: datetime, ends_at: datetime
) -> tuple[str, str, datetime, datetime]:
    title = title.strip()
    if not title:
        raise ValidationError("Заголовок не может быть пустым.")
    if len(title) > TITLE_MAX:
        raise ValidationError(f"Заголовок длиннее {TITLE_MAX} символов.")
    clean_html, text = sanitize_body(body_html)
    text = text.strip()
    if not text:
        raise ValidationError("Текст объявления не может быть пустым.")
    if len(text) > TEXT_MAX:
        raise ValidationError(f"Текст длиннее {TEXT_MAX} символов ({len(text)}).")
    start, end = _naive_utc(starts_at), _naive_utc(ends_at)
    if end <= start:
        raise ValidationError("Конец показа должен быть позже начала.")
    return title, clean_html, start, end


def status_of(a: Announcement, now: datetime | None = None) -> str:
    now = now or _now()
    if now < a.starts_at:
        return "planned"
    if now >= a.ends_at:
        return "finished"
    return "live"


def get(session: Session, announcement_id: int) -> Announcement:
    a = session.get(Announcement, announcement_id)
    if a is None:
        raise NotFoundError("Объявление не найдено.")
    return a


# ─── Админка ─────────────────────────────────────────────────────────────────

def create(
    session: Session,
    *,
    title: str,
    body_html: str,
    starts_at: datetime,
    ends_at: datetime,
    show_to_new_players: bool,
    actor_id: int | None,
) -> Announcement:
    title, body_html, start, end = _clean_fields(title, body_html, starts_at, ends_at)
    if end <= _now():
        raise ValidationError("Период показа уже закончился.")
    a = Announcement(
        title=title,
        body_html=body_html,
        starts_at=start,
        ends_at=end,
        show_to_new_players=show_to_new_players,
        created_by=actor_id,
    )
    session.add(a)
    session.flush()
    return a


def update(
    session: Session,
    announcement_id: int,
    *,
    title: str,
    body_html: str,
    starts_at: datetime,
    ends_at: datetime,
    show_to_new_players: bool,
    reshow: bool,
) -> Announcement:
    a = get(session, announcement_id)
    a.title, a.body_html, a.starts_at, a.ends_at = _clean_fields(
        title, body_html, starts_at, ends_at
    )
    a.show_to_new_players = show_to_new_players
    if reshow:
        # Новая версия: отметки о закрытии прежней больше не прячут объявление.
        a.revision += 1
    a.updated_at = utcnow()
    session.add(a)
    session.flush()
    return a


def stop(session: Session, announcement_id: int) -> Announcement:
    """«Снять с показа»: период заканчивается сейчас. Только у идущего объявления —
    у запланированного конец раньше начала нарушил бы период; его просто удаляют."""
    a = get(session, announcement_id)
    now = _now()
    if status_of(a, now) != "live" or now <= a.starts_at:
        raise ValidationError("Снять с показа можно только идущее объявление.")
    a.ends_at = now
    a.updated_at = utcnow()
    session.add(a)
    session.flush()
    return a


def delete(session: Session, announcement_id: int) -> None:
    session.delete(get(session, announcement_id))
    session.flush()


def _audience_filter(a: Announcement):
    """Условия «игрок — адресат объявления» (для счётчиков и для показа)."""
    conds = [User.role == "player", User.is_active.is_(True)]  # type: ignore[union-attr]
    if not a.show_to_new_players:
        conds.append(User.created_at <= a.starts_at)
    return conds


def _admin_item(session: Session, a: Announcement, now: datetime) -> dict:
    """Объявление со статусом и счётчиком «закрыли N из M»: N — закрывшие текущую
    версию, M — сколько активных игроков сейчас в адресатах."""
    audience = _audience_filter(a)
    audience_count = session.exec(select(func.count()).select_from(User).where(*audience)).one()
    seen_count = session.exec(
        select(func.count())
        .select_from(AnnouncementView)
        .join(User, User.id == AnnouncementView.user_id)
        .where(
            AnnouncementView.announcement_id == a.id,
            AnnouncementView.revision >= a.revision,
            *audience,
        )
    ).one()
    return {"a": a, "status": status_of(a, now), "seen": seen_count, "audience": audience_count}


def admin_item(session: Session, announcement_id: int) -> dict:
    return _admin_item(session, get(session, announcement_id), _now())


def list_admin(session: Session) -> list[dict]:
    """Все объявления, новые сверху. Объявлений единицы, поэтому счётчики — по паре
    запросов на объявление."""
    now = _now()
    rows = session.exec(
        select(Announcement).order_by(Announcement.starts_at.desc(), Announcement.id.desc())  # type: ignore[union-attr]
    ).all()
    return [_admin_item(session, a, now) for a in rows]


# ─── Игрок ───────────────────────────────────────────────────────────────────

def pending_for_user(session: Session, user: User) -> list[tuple[Announcement, bool]]:
    """Что показать игроку сейчас — от старого к новому. Второй элемент — «обновлено»:
    игрок закрывал прежнюю версию этого объявления."""
    now = _now()
    registered = user.created_at.replace(tzinfo=None)  # только что созданный — aware UTC
    rows = session.exec(
        select(Announcement, AnnouncementView.revision)
        .join(
            AnnouncementView,
            (AnnouncementView.announcement_id == Announcement.id)
            & (AnnouncementView.user_id == user.id),
            isouter=True,
        )
        .where(Announcement.starts_at <= now, Announcement.ends_at > now)
        .order_by(Announcement.starts_at, Announcement.id)
    ).all()
    result = []
    for a, seen_revision in rows:
        if seen_revision is not None and seen_revision >= a.revision:
            continue
        if not a.show_to_new_players and registered > a.starts_at:
            continue
        result.append((a, seen_revision is not None))
    return result


def acknowledge(session: Session, user_id: int, announcement_id: int, revision: int) -> None:
    """Игрок закрыл объявление в версии revision. Версию берём с клиента: если админ
    выпустил новую, пока окно было открыто, игрок закрыл старую и новую ещё увидит."""
    a = get(session, announcement_id)
    revision = min(revision, a.revision)
    view = session.get(AnnouncementView, (announcement_id, user_id))
    if view is None:
        session.add(
            AnnouncementView(announcement_id=announcement_id, user_id=user_id, revision=revision)
        )
    elif revision > view.revision:
        view.revision = revision
        view.closed_at = utcnow()
        session.add(view)
    session.flush()