| 1 | """Helpers for restart notification messages.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import os |
| 6 | import time |
| 7 | from dataclasses import dataclass |
| 8 | |
| 9 | RESTART_NOTIFY_CHANNEL_ENV = "NANOBOT_RESTART_NOTIFY_CHANNEL" |
| 10 | RESTART_NOTIFY_CHAT_ID_ENV = "NANOBOT_RESTART_NOTIFY_CHAT_ID" |
| 11 | RESTART_STARTED_AT_ENV = "NANOBOT_RESTART_STARTED_AT" |
| 12 | |
| 13 | |
| 14 | @dataclass(frozen=True) |
| 15 | class RestartNotice: |
| 16 | channel: str |
| 17 | chat_id: str |
| 18 | started_at_raw: str |
| 19 | |
| 20 | |
| 21 | def format_restart_completed_message(started_at_raw: str) -> str: |
| 22 | """Build restart completion text and include elapsed time when available.""" |
| 23 | elapsed_suffix = "" |
| 24 | if started_at_raw: |
| 25 | try: |
| 26 | elapsed_s = max(0.0, time.time() - float(started_at_raw)) |
| 27 | elapsed_suffix = f" in {elapsed_s:.1f}s" |
| 28 | except ValueError: |
| 29 | pass |
| 30 | return f"Restart completed{elapsed_suffix}." |
| 31 | |
| 32 | |
| 33 | def set_restart_notice_to_env(*, channel: str, chat_id: str) -> None: |
| 34 | """Write restart notice env values for the next process.""" |
| 35 | os.environ[RESTART_NOTIFY_CHANNEL_ENV] = channel |
| 36 | os.environ[RESTART_NOTIFY_CHAT_ID_ENV] = chat_id |
| 37 | os.environ[RESTART_STARTED_AT_ENV] = str(time.time()) |
| 38 | |
| 39 | |
| 40 | def consume_restart_notice_from_env() -> RestartNotice | None: |
| 41 | """Read and clear restart notice env values once for this process.""" |
| 42 | channel = os.environ.pop(RESTART_NOTIFY_CHANNEL_ENV, "").strip() |
| 43 | chat_id = os.environ.pop(RESTART_NOTIFY_CHAT_ID_ENV, "").strip() |
| 44 | started_at_raw = os.environ.pop(RESTART_STARTED_AT_ENV, "").strip() |
| 45 | if not (channel and chat_id): |
| 46 | return None |
| 47 | return RestartNotice(channel=channel, chat_id=chat_id, started_at_raw=started_at_raw) |
| 48 | |
| 49 | |
| 50 | def should_show_cli_restart_notice(notice: RestartNotice, session_id: str) -> bool: |
| 51 | """Return True when a restart notice should be shown in this CLI session.""" |
| 52 | if notice.channel != "cli": |
| 53 | return False |
| 54 | if ":" in session_id: |
| 55 | _, cli_chat_id = session_id.split(":", 1) |
| 56 | else: |
| 57 | cli_chat_id = session_id |
| 58 | return not notice.chat_id or notice.chat_id == cli_chat_id |
| 59 |