"""Uptime / ping / resource numbers for /status and /stats. Inside a container psutil reports the *host's* CPU and RAM, so the cgroup files are read first and psutil is only the fallback. """ from __future__ import annotations import asyncio import os import platform import shutil import time from typing import Any, List, Optional, Tuple import psutil from . import __version__ from .fasttransfer import BIG_FILE_THRESHOLD from .utils import clip, esc, human_size, human_uptime START_TIME = time.time() CGROUP_ROOT = "/sys/fs/cgroup" _NET_BASE = psutil.net_io_counters() def _read(path: str) -> Optional[str]: try: with open(path, "r", encoding="utf-8") as fh: return fh.read().strip() except OSError: return None def parse_cpu_max(text: Optional[str]) -> Optional[float]: """cgroup v2 ``cpu.max`` -> number of cores, ``None`` when unlimited.""" if not text: return None parts = text.split() if len(parts) != 2 or parts[0] == "max": return None try: quota, period = int(parts[0]), int(parts[1]) except ValueError: return None return quota / period if quota > 0 and period > 0 else None def parse_mem_limit(text: Optional[str]) -> Optional[int]: """cgroup ``memory.max`` / v1 ``limit_in_bytes`` -> bytes, ``None`` when unlimited.""" if not text or text.strip() == "max": return None try: value = int(text.strip()) except ValueError: return None return value if value < 2 ** 60 else None def parse_stat_value(text: Optional[str], key: str) -> Optional[int]: for line in (text or "").splitlines(): parts = line.split() if len(parts) == 2 and parts[0] == key: try: return int(parts[1]) except ValueError: return None return None def cpu_cores() -> float: quota = parse_cpu_max(_read(f"{CGROUP_ROOT}/cpu.max")) return quota if quota else float(psutil.cpu_count() or 1) def _cgroup_cpu_usec() -> Optional[int]: return parse_stat_value(_read(f"{CGROUP_ROOT}/cpu.stat"), "usage_usec") async def cpu_percent(interval: float = 0.5) -> float: """CPU use of this container as a percentage of its own limit.""" first = _cgroup_cpu_usec() if first is None: return psutil.cpu_percent(interval=None) t0 = time.monotonic() await asyncio.sleep(interval) second = _cgroup_cpu_usec() elapsed = time.monotonic() - t0 if second is None or elapsed <= 0: return psutil.cpu_percent(interval=None) return max(0.0, min(100.0, (second - first) / (elapsed * 1_000_000) / cpu_cores() * 100)) def memory_usage() -> Tuple[int, int]: """(used, total) bytes for this container.""" current = _read(f"{CGROUP_ROOT}/memory.current") limit = parse_mem_limit(_read(f"{CGROUP_ROOT}/memory.max")) if current is None: # cgroup v1 current = _read(f"{CGROUP_ROOT}/memory/memory.usage_in_bytes") limit = parse_mem_limit(_read(f"{CGROUP_ROOT}/memory/memory.limit_in_bytes")) host_total = psutil.virtual_memory().total if current is None or not current.isdigit(): vm = psutil.virtual_memory() return vm.used, vm.total used = int(current) inactive = parse_stat_value(_read(f"{CGROUP_ROOT}/memory.stat"), "inactive_file") or 0 used = max(0, used - inactive) # do not count reclaimable page cache return used, min(limit, host_total) if limit else host_total def disk_usage(path: str) -> Tuple[int, int, int]: os.makedirs(path, exist_ok=True) du = shutil.disk_usage(path) return du.used, du.total, du.free def net_since_start() -> Tuple[int, int]: now = psutil.net_io_counters() return max(0, now.bytes_recv - _NET_BASE.bytes_recv), max(0, now.bytes_sent - _NET_BASE.bytes_sent) def uptime() -> float: return time.time() - START_TIME def crypto_backend() -> str: """What Telethon encrypts and decrypts with: ``cryptg`` or ``libssl`` (both native code, fast), else ``python`` (slow).""" from telethon.crypto import aes if aes.cryptg is not None: return "cryptg" return "libssl" if aes.libssl.encrypt_ige and aes.libssl.decrypt_ige else "python" def loop_backend() -> str: """``uvloop`` when the running event loop is uvloop's, ``asyncio`` for the standard one, ``none`` outside a loop.""" try: loop = asyncio.get_running_loop() except RuntimeError: return "none" return "uvloop" if type(loop).__module__.split(".")[0] == "uvloop" else "asyncio" def speed_checks(parallel: bool, connections: int, stream: Optional[bool] = None) -> List[Tuple[str, str, bool]]: """``(short name, detail, fine)`` for the things transfer speed depends on; ``stream`` (when given) adds the streaming rename.""" crypto, loop = crypto_backend(), loop_backend() if crypto == "cryptg": encryption = ("cryptg", "cryptg", True) elif crypto == "libssl": encryption = ("libssl", "libssl (cryptg is not installed)", True) else: encryption = ("slow Python crypto", "pure Python, slow: cryptg is missing", False) event_loop = ("uvloop", "uvloop", True) if loop == "uvloop" else ("no uvloop", f"{loop} loop: uvloop is not running", False) transfers = (f"parallel ×{connections}", f"on, {connections} connections", True) if parallel else ("parallel off", "off (FAST_TRANSFER=0)", False) checks = [encryption, event_loop, transfers] if stream is not None: if not parallel: checks.append(("no streaming", "off: it needs the parallel transfer", False)) elif stream: checks.append(("streaming", "on: big files are uploaded while they download", True)) else: checks.append(("no streaming", "off (STREAM_RENAME=0)", False)) return checks def speed_summary(parallel: bool, connections: int, stream: Optional[bool] = None) -> str: """One line for the start-up notice: ``cryptg ✅ · uvloop ✅ · parallel ×8 ✅ · streaming ✅``.""" return " · ".join(f"{short} {'✅' if fine else '⚠️'}" for short, _detail, fine in speed_checks(parallel, connections, stream)) def format_speed(parallel: bool, connections: int, stats: Any, stream: Optional[bool] = None) -> str: """The Speed section of /status: what is switched on, and how the transfers since the start went.""" labels = ("🔐 Encryption", "🔁 Event loop", "🚀 Parallel transfer", "⬇⬆ Streaming rename") lines = ["⚡ Speed"] for label, (_short, detail, fine) in zip(labels, speed_checks(parallel, connections, stream)): lines.append(f"{label}: {esc(detail)} {'✅' if fine else '⚠️'}") small = f"{BIG_FILE_THRESHOLD // 2 ** 20} MB" lines.append(f"⬇ Downloads: {stats.fast_down} parallel · {stats.standard_down} standard") lines.append(f"⬆ Uploads: {stats.fast_up} parallel · {stats.small_up} up to {small} · {stats.standard_up} standard") if stream is not None: lines.append(f"⬇⬆ Streamed renames: {stats.streamed}") if stats.fallbacks: lines.append( f"⚠️ The parallel transfer failed {stats.fallbacks}× and the standard, slower one took over. " f"Last: {esc(clip(stats.last_fallback, 160))}" ) if stats.stream_fallbacks: lines.append( f"⚠️ Streaming failed {stats.stream_fallbacks}× and those files were saved first and then uploaded. " f"Last: {esc(clip(stats.last_stream_fallback, 160))}" ) return "\n".join(lines) def format_status(ping_ms: Optional[float], dc_id: Optional[int], telethon_version: str, speed: str = "") -> str: ping = f"{ping_ms:.0f} ms" if ping_ms is not None else "n/a" dc = f" (DC {dc_id})" if dc_id else "" return ( "🟢 Bot status\n\n" f"⏱ Uptime: {human_uptime(uptime())}\n" f"🏓 Ping: {ping}{dc}\n" f"🤖 Version: v{__version__}\n" f"🐍 Python {platform.python_version()} · Telethon {telethon_version}" + (f"\n\n{speed}" if speed else "") ) async def format_stats(users: int, files: int, total_bytes: int, running: int, waiting: int, temp_dir: str) -> str: cpu = await cpu_percent() used, total = memory_usage() d_used, d_total, d_free = disk_usage(temp_dir) rx, tx = net_since_start() rss = psutil.Process().memory_info().rss ram_pct = used / total * 100 if total else 0.0 disk_pct = d_used / d_total * 100 if d_total else 0.0 return ( "📊 System stats\n\n" f"👥 Users: {users}\n" f"📁 Files renamed: {files} ({human_size(total_bytes)})\n" f"🧵 Queue: {running} running · {waiting} waiting\n\n" f"🖥 CPU: {cpu:.1f}% of {cpu_cores():g} core(s)\n" f"🧠 RAM: {human_size(used)} / {human_size(total)} ({ram_pct:.1f}%) · bot {human_size(rss)}\n" f"💾 Disk: {human_size(d_used)} / {human_size(d_total)} ({disk_pct:.1f}%) · {human_size(d_free)} free\n" f"🌐 Network since start: ↓ {human_size(rx)} · ↑ {human_size(tx)}" )