moved to a better sysinfo method

This commit is contained in:
esef committed 2026-07-12 19:23:29 +03:00
1 parent b7a2338de8
commit 973330d21a
3 files changed
+62 -40

No files matched your search

+3 -3
View File
@@ -53,7 +53,7 @@ async def _fetch_portrait(session: aiohttp.ClientSession, student: dict) -> Opti
# 1. Try the face icon # 1. Try the face icon
icon_url = db.cdn_icon(student) icon_url = db.cdn_icon(student)
try: try:
async with session.get(icon_url, timeout=aiohttp.ClientTimeout(total=10)) as resp: async with session.get(icon_url, timeout=aiohttp.ClientTimeout(total=5)) as resp:
if resp.status == 200: if resp.status == 200:
data = await resp.read() data = await resp.read()
return Image.open(io.BytesIO(data)).convert("RGBA") return Image.open(io.BytesIO(data)).convert("RGBA")
@@ -63,7 +63,7 @@ async def _fetch_portrait(session: aiohttp.ClientSession, student: dict) -> Opti
# 2. Fallback: skill portrait (full-body character art) # 2. Fallback: skill portrait (full-body character art)
sp_url = db.cdn_skill_portrait(student) sp_url = db.cdn_skill_portrait(student)
try: try:
async with session.get(sp_url, timeout=aiohttp.ClientTimeout(total=10)) as resp: async with session.get(sp_url, timeout=aiohttp.ClientTimeout(total=5)) as resp:
if resp.status == 200: if resp.status == 200:
data = await resp.read() data = await resp.read()
img = Image.open(io.BytesIO(data)).convert("RGBA") img = Image.open(io.BytesIO(data)).convert("RGBA")
@@ -149,7 +149,7 @@ async def render_pull(
_draw_footer(draw, spark_count) _draw_footer(draw, spark_count)
# ── Fetch all portraits ───────────────────────────────────────────── # ── Fetch all portraits ─────────────────────────────────────────────
timeout = aiohttp.ClientTimeout(total=60, connect=10) timeout = aiohttp.ClientTimeout(total=30, connect=5)
async with aiohttp.ClientSession(timeout=timeout) as session: async with aiohttp.ClientSession(timeout=timeout) as session:
portrait_tasks = [_fetch_portrait(session, p) for p in pulls] portrait_tasks = [_fetch_portrait(session, p) for p in pulls]
portraits = await asyncio.gather(*portrait_tasks, return_exceptions=True) portraits = await asyncio.gather(*portrait_tasks, return_exceptions=True)
+51 -30
View File
@@ -1,50 +1,75 @@
import subprocess import asyncio
from discord.ext import commands import logging
import re import re
from discord.ext import commands
import psutil import psutil
import aiohttp import aiohttp
log = logging.getLogger(__name__)
class SysInfo(commands.Cog): class SysInfo(commands.Cog):
def __init__(self, bot): def __init__(self, bot):
self.bot = bot self.bot = bot
def strip_ansi(self, text): @staticmethod
def _strip_ansi(text: str) -> str:
return re.sub(r'\x1b\[[0-9;]*[a-zA-Z]|\[[0-9]+[A-Z]', '', text) return re.sub(r'\x1b\[[0-9;]*[a-zA-Z]|\[[0-9]+[A-Z]', '', text)
#this command lists all the sysinfo commands, not an actualy sysinfo command## @staticmethod
async def _run_command(*args: str, timeout: float = 8) -> str:
"""Run a shell command asynchronously, return stdout or '' on failure."""
try:
proc = await asyncio.create_subprocess_exec(
*args,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.DEVNULL,
)
stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=timeout)
return stdout.decode().strip()
except Exception:
log.debug("Command failed: %s", args, exc_info=True)
return ""
# ── commands ────────────────────────────────────────────────────────
@commands.command() @commands.command()
async def sysinfo(self, ctx): async def sysinfo(self, ctx):
sys_commands = [f'!{cmd.name}' for cmd in self.get_commands()] """List available sysinfo commands."""
await ctx.reply('\n'.join(sys_commands)) cmds = [f'!{cmd.name}' for cmd in self.get_commands()]
###################################################################### await ctx.reply('\n'.join(cmds))
@commands.command(name="fetch")
async def fetch_cmd(self, ctx):
async def fetch(self, ctx): """Show system info via fastfetch."""
result = subprocess.run(['fastfetch', '--pipe', '--logo', 'none'], capture_output=True, text=True) out = await self._run_command("fastfetch", "--pipe", "--logo", "none")
clean = self.strip_ansi(result.stdout) if not out:
return await ctx.reply("fastfetch not available")
clean = self._strip_ansi(out)
await ctx.reply(f'```\n{clean}\n```') await ctx.reply(f'```\n{clean}\n```')
@commands.command() @commands.command()
async def uptime(self, ctx): async def uptime(self, ctx):
result = subprocess.run(['uptime', '-p'], capture_output=True, text=True) out = await self._run_command("uptime", "-p")
await ctx.reply(f'⏱️ {result.stdout.strip()}') await ctx.reply(f'⏱️ {out}' if out else "uptime not available")
@commands.command() @commands.command()
async def temps(self, ctx): async def temps(self, ctx):
result = subprocess.run(['sensors'], capture_output=True, text=True) out = await self._run_command("sensors")
await ctx.reply(f'```\n{result.stdout}\n```') await ctx.reply(f'```\n{out}\n```' if out else "sensors not available")
@commands.command() @commands.command()
async def gpu(self, ctx): async def gpu(self, ctx):
result = subprocess.run(['envycontrol', '--query'], capture_output=True, text=True) out = await self._run_command("envycontrol", "--query")
await ctx.reply(f'I am on {result.stdout.strip()} mode esef!') await ctx.reply(f'I am on {out} mode esef!' if out else "envycontrol not available")
@commands.command() @commands.command()
async def top(self, ctx): async def top(self, ctx):
result = subprocess.run(['ps', 'aux', '--sort=-%cpu'], capture_output=True, text=True) out = await self._run_command("ps", "aux", "--sort=-%cpu")
lines = result.stdout.splitlines() if not out:
top5 = '\n'.join(lines[:6]) # header + 5 processes return await ctx.reply("ps not available")
lines = out.splitlines()
top5 = '\n'.join(lines[:6])
await ctx.reply(f'```\n{top5}\n```') await ctx.reply(f'```\n{top5}\n```')
@commands.command() @commands.command()
@@ -72,12 +97,12 @@ class SysInfo(commands.Cog):
battery = psutil.sensors_battery() battery = psutil.sensors_battery()
if battery: if battery:
await ctx.reply(f'Battery: {battery.percent}% {"Charging" if battery.power_plugged else "Not Charging"}') await ctx.reply(f'Battery: {battery.percent}% {"Charging" if battery.power_plugged else "Not Charging"}')
@commands.command() @commands.command()
async def processes(self, ctx): async def processes(self, ctx):
processes = psutil.pids() processes = psutil.pids()
await ctx.reply(f'Processes: {len(processes)} running processes ') await ctx.reply(f'Processes: {len(processes)} running processes')
@commands.command() @commands.command()
async def stats(self, ctx): async def stats(self, ctx):
cpu = psutil.cpu_percent(interval=1) cpu = psutil.cpu_percent(interval=1)
@@ -85,11 +110,7 @@ class SysInfo(commands.Cog):
disk = psutil.disk_usage('/').percent disk = psutil.disk_usage('/').percent
reply = f'CPU: {cpu}% | RAM: {ram}% | Disk: {disk}%' reply = f'CPU: {cpu}% | RAM: {ram}% | Disk: {disk}%'
await ctx.reply(reply) await ctx.reply(reply)
#this command fetches the current player count for Strinova from the Steam API
#it's a game i play and want to show off to my friends, not an actual sysinfo command##
#i know this is a bit of a weird place for it but i dont want to make a whole new cog just for one command
#holy shit vscode autocomplete knew exactly what i was gonna say wtf;?
@commands.command() @commands.command()
async def strinova(self, ctx): async def strinova(self, ctx):
async with aiohttp.ClientSession() as session: async with aiohttp.ClientSession() as session:
+8 -7
View File
@@ -3,7 +3,6 @@
import asyncio import asyncio
import logging import logging
import re import re
import subprocess
import time import time
import aiohttp import aiohttp
@@ -101,7 +100,7 @@ class SysInfoSync:
"netSent": {"stringValue": _fmt_bytes(net.bytes_sent)}, "netSent": {"stringValue": _fmt_bytes(net.bytes_sent)},
"netRecv": {"stringValue": _fmt_bytes(net.bytes_recv)}, "netRecv": {"stringValue": _fmt_bytes(net.bytes_recv)},
"processes": {"integerValue": len(psutil.pids())}, "processes": {"integerValue": len(psutil.pids())},
"fetch": {"stringValue": self._get_fastfetch()}, "fetch": {"stringValue": await self._get_fastfetch()},
} }
} }
@@ -111,14 +110,16 @@ class SysInfoSync:
log.debug("Firestore PATCH %d: %s", resp.status, body[:200]) log.debug("Firestore PATCH %d: %s", resp.status, body[:200])
@staticmethod @staticmethod
def _get_fastfetch() -> str: async def _get_fastfetch() -> str:
"""Run fastfetch --pipe --logo none, strip ANSI, return first 15 lines.""" """Run fastfetch --pipe --logo none, strip ANSI, return first 15 lines."""
try: try:
result = subprocess.run( proc = await asyncio.create_subprocess_exec(
["fastfetch", "--pipe", "--logo", "none"], "fastfetch", "--pipe", "--logo", "none",
capture_output=True, text=True, timeout=5, stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.DEVNULL,
) )
clean = re.sub(r"\x1b\[[0-9;]*[a-zA-Z]|\[[0-9]+[A-Z]", "", result.stdout) stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=5)
clean = re.sub(r"\x1b\[[0-9;]*[a-zA-Z]|\[[0-9]+[A-Z]", "", stdout.decode())
lines = [l for l in clean.strip().splitlines() if l.strip()] lines = [l for l in clean.strip().splitlines() if l.strip()]
return "\n".join(lines[:15]) return "\n".join(lines[:15])
except Exception: except Exception: