Объявления: модель, API игрока и админки, очистка HTML

Объявление администрации показывается игроку окном в свой период, пока игрок
не закроет его («Понятно»). Отметка о закрытии хранится на сервере с номером
версии: правка с «показать заново» поднимает версию, и закрывшие прежнюю
увидят объявление снова — ответ помечен updated («обновлено»). Флаг
show_to_new_players=false прячет объявление от зарегистрировавшихся после
начала показа. Пересекающиеся объявления идут от старого к новому.

Текст приходит HTML-ом из редактора админки и сохраняется только после
очистки по белому списку (b, em, mark и mark.red, p, br): атрибуты
отбрасываются, script/style/svg — вместе с содержимым, текст экранируется
заново. Фронт вставляет только этот HTML.

API: GET /api/announcements/pending, POST /api/announcements/{id}/ack;
админка — список со статусом и счётчиком «закрыли N из M», создание, правка,
«снять с показа», удаление, всё в аудит. SSE-событие announcements активным
игрокам. Миграция 0015 идемпотентная.

Тесты: очистка (XSS-попытки, вложенные div), права, период и порядок,
«новые игроки», повторный показ, снятие, удаление, валидация. #84

Перенесено в main без рейтинга из a0a0e52; миграция 0015 — сразу от 0013
(как в ec0445f).

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

backend/app/services/announcement_service.py

@@ -0,0 +1,336 @@
+"""Объявления администрации (#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()