63 lines
1.7 KiB
Python
63 lines
1.7 KiB
Python
import asyncio
|
|
from typing import List
|
|
|
|
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
|
|
|
|
|
|
def _render(stats: List[dict]) -> 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"],
|
|
"server" if row["target_type"] == "guild" else "group",
|
|
str(row["reaction_count"]),
|
|
row["last_reaction_at"] or "-",
|
|
)
|
|
|
|
if not stats:
|
|
table.add_row("(no targets discovered yet)", "-", "-", "-")
|
|
|
|
return Panel(table, title=f"Reaction stats — 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._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 refresh_sync(self) -> None:
|
|
stats = self.storage.get_stats()
|
|
self._live.update(_render(stats))
|
|
|
|
async def refresh(self) -> None:
|
|
stats = await asyncio.to_thread(self.storage.get_stats)
|
|
self._live.update(_render(stats))
|