mirror of
https://github.com/kuhyx/signal-bot.git
synced 2026-11-10 19:27:20 +01:00
Krzysztof kuhy Rudnicki
3d6ae0680c
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
126 lines
4.8 KiB
Python
126 lines
4.8 KiB
Python
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
|
|
"""Telling a group that the bot has been upgraded.
|
|
|
|
A group is told about every release it has not seen, and a group that has
|
|
never been told anything has not seen any of them -- so the first
|
|
announcement starts at 0.0.0 "Petrus" rather than skipping straight to
|
|
whatever is current. Watching the ladder start at Peter is the point of
|
|
naming releases after popes at all.
|
|
|
|
That does mean a group added years from now would be handed the whole
|
|
backlog, so the message is capped at MAX_RELEASES and says how many older
|
|
versions it left out.
|
|
|
|
The announcement carries the same three things `!papiez` does: the pope's
|
|
portrait, his Latin name and his Polish Wikipedia article. Only for the pope
|
|
in the header -- ten attachments is the wall MAX_RELEASES exists to avoid.
|
|
|
|
The version is recorded *before* the message is sent. `Restart=always` with
|
|
`RestartSec=5` means a crash between the two would otherwise re-announce on
|
|
every restart, and a missed announcement is far cheaper than a looping one.
|
|
That is also why a portrait that will not download costs the photo and not
|
|
the announcement: the marker is already written, so a Commons outage at boot
|
|
would otherwise eat that group's release notes permanently.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import TYPE_CHECKING, Final
|
|
|
|
from signal_bot.changelog import since
|
|
from signal_bot.versioning import article_url, latin_name, version_for
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import Iterable
|
|
|
|
from signal_bot.changelog import Release
|
|
from signal_bot.portraits import PortraitCache
|
|
from signal_bot.store import Store
|
|
from signal_bot.transport_base import Transport
|
|
from signal_bot.versioning import Pope
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
ANNOUNCED_KEY: Final = "announced_version"
|
|
|
|
# How many releases one announcement may list before it becomes a wall.
|
|
MAX_RELEASES: Final = 10
|
|
|
|
|
|
def render(pope: Pope, releases: Iterable[Release]) -> str:
|
|
"""Return the message announcing a new version, in Polish.
|
|
|
|
One release is rendered flat, because the header already names it. Several
|
|
get a block each, so the group can see the ladder it is climbing. Pure on
|
|
purpose: whether this text travels as a caption or as a plain message is
|
|
announce_to's call, not this function's.
|
|
"""
|
|
shown = list(releases)
|
|
header = f"Wersja {version_for(pope.ordinal)} \u00ab{latin_name(pope)}\u00bb wydana"
|
|
header = f"{header}\n{article_url(pope)}"
|
|
omitted = max(0, len(shown) - MAX_RELEASES)
|
|
shown = shown[-MAX_RELEASES:]
|
|
if len(shown) == 1:
|
|
body = "\n".join(f"- {line}" for line in shown[0].lines)
|
|
return f"{header}\nChangelog:\n{body}"
|
|
blocks = [
|
|
"\n".join(
|
|
[
|
|
f"{release.version} \u00ab{latin_name(release.pope_entry)}\u00bb",
|
|
*(f"- {x}" for x in release.lines),
|
|
]
|
|
)
|
|
for release in shown
|
|
]
|
|
tail = f"\n\n(pominieto {omitted} starszych wersji)" if omitted else ""
|
|
return f"{header}\nChangelog:\n\n" + "\n\n".join(blocks) + tail
|
|
|
|
|
|
async def announce_to(
|
|
transport: Transport,
|
|
store: Store,
|
|
group_id: str,
|
|
releases: tuple[Release, ...],
|
|
pope: Pope,
|
|
portraits: PortraitCache,
|
|
) -> None:
|
|
"""Announce the current version to one group, if it has not seen it."""
|
|
version = version_for(pope.ordinal)
|
|
previous = store.meta(group_id, ANNOUNCED_KEY)
|
|
if previous == version:
|
|
return
|
|
# No record at all means the group has seen nothing, not that it should be
|
|
# skipped: that is how Petrus gets announced rather than silently assumed.
|
|
pending = releases if previous is None else since(releases, previous)
|
|
store.set_meta(group_id, ANNOUNCED_KEY, version)
|
|
if not pending:
|
|
logger.info("group %s recorded at %s with nothing to announce", group_id, version)
|
|
return
|
|
text = render(pope, pending)
|
|
try:
|
|
image = await portraits.get_base64(pope)
|
|
except Exception:
|
|
# The marker is already written, so this must never propagate: a
|
|
# Commons outage costs the portrait, not the release notes.
|
|
logger.exception("portrait for %s unavailable; announcing %s as text", version, group_id)
|
|
await transport.send_text(text, group_id)
|
|
return
|
|
await transport.send_image(image, group_id, caption=text)
|
|
|
|
|
|
async def announce_all(
|
|
transport: Transport,
|
|
store: Store,
|
|
group_ids: Iterable[str],
|
|
releases: tuple[Release, ...],
|
|
pope: Pope,
|
|
portraits: PortraitCache,
|
|
) -> None:
|
|
"""Announce to every watched group; one failure must not silence the rest."""
|
|
for group_id in group_ids:
|
|
try:
|
|
await announce_to(transport, store, group_id, releases, pope, portraits)
|
|
except Exception:
|
|
logger.exception("release announcement failed for group %s", group_id)
|