React in 1:1 DMs alongside servers/groups, surface the resolved friend username at startup and in the stats panel, verify reactions actually stick after adding them (with jump links for manual spot-checking), and prune stale targets from stats on each run so only the currently mutual set is shown.
70 lines
2.1 KiB
Python
70 lines
2.1 KiB
Python
import asyncio
|
|
from typing import List, Optional
|
|
|
|
from rich.console import Console
|
|
from rich.live import Live
|
|
from rich.panel import Panel
|
|
from rich.table import Table
|
|
|
|
from bot.storage import Storage
|
|
|
|
_TYPE_LABELS = {"guild": "server", "group_dm": "group", "dm": "DM"}
|
|
|
|
|
|
def _render(stats: List[dict], friend_label: Optional[str] = None) -> Panel:
|
|
table = Table(expand=True)
|
|
table.add_column("Target")
|
|
table.add_column("Type")
|
|
table.add_column("Reactions", justify="right")
|
|
table.add_column("Last reaction (UTC)")
|
|
|
|
total = 0
|
|
for row in stats:
|
|
total += row["reaction_count"]
|
|
table.add_row(
|
|
row["name"] or row["target_id"],
|
|
_TYPE_LABELS.get(row["target_type"], row["target_type"]),
|
|
str(row["reaction_count"]),
|
|
row["last_reaction_at"] or "-",
|
|
)
|
|
|
|
if not stats:
|
|
table.add_row("(no targets discovered yet)", "-", "-", "-")
|
|
|
|
who = f" for {friend_label}" if friend_label else ""
|
|
return Panel(table, title=f"Reaction stats{who} — total: {total}", border_style="cyan")
|
|
|
|
|
|
class StatsDisplay:
|
|
"""A stats panel pinned to the bottom of the terminal, with normal
|
|
log/print output scrolling above it (via rich.live.Live)."""
|
|
|
|
def __init__(self, console: Console, storage: Storage):
|
|
self.console = console
|
|
self.storage = storage
|
|
self.friend_label: Optional[str] = None
|
|
self._live = Live(
|
|
_render([]),
|
|
console=console,
|
|
refresh_per_second=4,
|
|
transient=False,
|
|
)
|
|
|
|
def __enter__(self):
|
|
self._live.__enter__()
|
|
return self
|
|
|
|
def __exit__(self, *exc_info):
|
|
return self._live.__exit__(*exc_info)
|
|
|
|
def set_friend(self, label: str) -> None:
|
|
self.friend_label = label
|
|
|
|
def refresh_sync(self) -> None:
|
|
stats = self.storage.get_stats()
|
|
self._live.update(_render(stats, self.friend_label))
|
|
|
|
async def refresh(self) -> None:
|
|
stats = await asyncio.to_thread(self.storage.get_stats)
|
|
self._live.update(_render(stats, self.friend_label))
|