from __future__ import annotations import html from datetime import datetime, timezone from typing import Iterable, List from .boinc import BoincSnapshot, Task def progress_bar(percent: float, width: int = 16) -> str: clamped = max(0, min(percent, 100)) filled = int(round((clamped / 100.0) * width)) return "[" + "#" * filled + "." * (width - filled) + "]" def humanize_seconds(seconds: float) -> str: if seconds <= 0: return "0s" minutes, sec = divmod(int(seconds), 60) hours, minutes = divmod(minutes, 60) days, hours = divmod(hours, 24) parts = [] if days: parts.append(f"{days}d") if hours: parts.append(f"{hours}h") if minutes: parts.append(f"{minutes}m") if sec and len(parts) < 2: parts.append(f"{sec}s") return " ".join(parts) if parts else f"{sec}s" def format_timestamp(dt: datetime) -> str: if not dt: return "-" if not dt.tzinfo: dt = dt.replace(tzinfo=timezone.utc) return dt.astimezone(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") def format_task(task: Task) -> str: safe_name = html.escape(task.name) safe_project = html.escape(task.project) eta = humanize_seconds(task.remaining_seconds or 0) deadline = format_timestamp(task.deadline) if task.deadline else "-" bar = progress_bar(task.progress_percent) return ( f"{safe_name} ({safe_project})\n" f"{bar} {task.progress_percent}% | ETA {eta} | Deadline {deadline}" ) def format_tasks_section(title: str, tasks: Iterable[Task]) -> str: items = list(tasks) if not items: return f"{html.escape(title)}: ничего нет" lines: List[str] = [f"{html.escape(title)}:"] for task in items: lines.append(format_task(task)) return "\n".join(lines) def format_dashboard(snapshot: BoincSnapshot) -> str: lines = [ "BOINC / grcpool монитор", f"Обновлено: {format_timestamp(snapshot.fetched_at)}", "", f"Активные: {len(snapshot.active)} | В очереди: {len(snapshot.queued)} | Завершены: {len(snapshot.completed)}", f"Средний прогресс: {snapshot.average_progress:.1f}% | Оставшееся время: {humanize_seconds(snapshot.total_remaining_seconds)}", "", format_tasks_section("Активные задачи", snapshot.active[:5]), ] if snapshot.queued: lines.extend(["", format_tasks_section("Ожидают запуска", snapshot.queued[:5])]) if snapshot.completed: lines.extend(["", format_tasks_section("Готовы к отправке", snapshot.completed[:5])]) return "\n".join(lines)