Minz
7b3c358b05
Shows what's happening right now underneath the table (connecting, resolving the friend, discovering targets, fetching channel history, waiting out a jittered delay before reacting, sending a reaction, removing a reaction during --undo, etc). Status updates are cheap and DB-free, using the last fetched stats for re-rendering.
92 lines
3.1 KiB
Python
92 lines
3.1 KiB
Python
import asyncio
|
|
from typing import List, Optional
|
|
|
|
from rich.console import Console, Group
|
|
from rich.live import Live
|
|
from rich.panel import Panel
|
|
from rich.progress_bar import ProgressBar
|
|
from rich.table import Table
|
|
from rich.text import Text
|
|
|
|
from bot.storage import Storage
|
|
|
|
_TYPE_LABELS = {"guild": "server", "group_dm": "group", "dm": "DM"}
|
|
|
|
|
|
def _render(stats: List[dict], friend_label: Optional[str] = None, status: Optional[str] = None) -> Panel:
|
|
table = Table(expand=True)
|
|
table.add_column("Target")
|
|
table.add_column("Type")
|
|
table.add_column("Progress", ratio=1)
|
|
table.add_column("Reacted/Found", justify="right")
|
|
table.add_column("Last reaction (UTC)")
|
|
|
|
total_reacted = 0
|
|
total_found = 0
|
|
for row in stats:
|
|
reacted = row["reaction_count"]
|
|
found = max(row["messages_found"], reacted) # found can never be less than reacted
|
|
total_reacted += reacted
|
|
total_found += found
|
|
|
|
bar = ProgressBar(total=max(found, reacted, 1), completed=reacted, width=None)
|
|
table.add_row(
|
|
row["name"] or row["target_id"],
|
|
_TYPE_LABELS.get(row["target_type"], row["target_type"]),
|
|
bar,
|
|
f"{reacted}/{found}",
|
|
row["last_reaction_at"] or "-",
|
|
)
|
|
|
|
if not stats:
|
|
table.add_row("(no targets discovered yet)", "-", "-", "-", "-")
|
|
|
|
content = table
|
|
if status:
|
|
content = Group(table, Text(f"▸ {status}", style="dim italic"))
|
|
|
|
who = f" for {friend_label}" if friend_label else ""
|
|
return Panel(content, title=f"Reaction stats{who} — total: {total_reacted}/{total_found}", 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.status: str = "Starting..."
|
|
self._last_stats: List[dict] = []
|
|
self._live = Live(
|
|
_render([], None, self.status),
|
|
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 set_status(self, text: str) -> None:
|
|
"""Update the "what's happening right now" line without hitting the
|
|
DB — cheap enough to call for every jittered wait / API call."""
|
|
self.status = text
|
|
self._live.update(_render(self._last_stats, self.friend_label, self.status))
|
|
|
|
def refresh_sync(self) -> None:
|
|
self._last_stats = self.storage.get_stats()
|
|
self._live.update(_render(self._last_stats, self.friend_label, self.status))
|
|
|
|
async def refresh(self) -> None:
|
|
self._last_stats = await asyncio.to_thread(self.storage.get_stats)
|
|
self._live.update(_render(self._last_stats, self.friend_label, self.status))
|