gitpull polished

cogs/gitpull.py

@@ -1,5 +1,7 @@
 import asyncio
 import os
+from datetime import datetime, timezone
+
 import discord
 from discord.ext import commands
 
@@ -9,64 +11,85 @@ SUDO_PASSWORD = os.getenv("SUDO_PASSWORD", "")
 BOT_DIR       = os.path.expanduser("~/discordbot")
 SERVICE_NAME  = "yuuka"
 
-GIT_TIMEOUT    = 30
+GIT_TIMEOUT     = 30
 SYSTEMD_TIMEOUT = 15
 
-# ── colors (kept local so this cog has no dependency on music.py) ───────────
+# ── colors ───────────────────────────────────────────────────────────────────
 C_OK   = 0x50FA7B
 C_ERR  = 0xFF5555
 C_WARN = 0xFFB86C
+C_INFO = 0x8BE9FD
 
 
 # ── subprocess helpers ────────────────────────────────────────────────────────
-async def run_git_pull() -> tuple[int, str]:
+async def _run(*args, cwd=None, input_data=None, timeout=30) -> tuple[int, str]:
     proc = await asyncio.create_subprocess_exec(
-        "git", "pull",
+        *args,
-        cwd=BOT_DIR,
+        cwd=cwd,
-        stdout=asyncio.subprocess.PIPE,
+        stdin=asyncio.subprocess.PIPE if input_data else None,
-        stderr=asyncio.subprocess.PIPE,
-    )
-    try:
-        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=GIT_TIMEOUT)
-    except asyncio.TimeoutError:
-        proc.kill()
-        await proc.communicate()
-        return -1, f"git pull timed out after {GIT_TIMEOUT}s"
-
-    raw = (stdout.decode() + stderr.decode()).strip()
-    output = "\n".join(
-        line for line in raw.splitlines() if not line.strip().startswith("[sudo]")
-    )
-    return proc.returncode, output or "No output."
-
-
-async def run_sudo_systemctl(action: str) -> tuple[int, str]:
-    proc = await asyncio.create_subprocess_exec(
-        "sudo", "-S", "systemctl", action, SERVICE_NAME,
-        stdin=asyncio.subprocess.PIPE,
         stdout=asyncio.subprocess.PIPE,
         stderr=asyncio.subprocess.PIPE,
     )
     try:
         stdout, stderr = await asyncio.wait_for(
-            proc.communicate(input=f"{SUDO_PASSWORD}\n".encode()),
+            proc.communicate(input=input_data),
-            timeout=SYSTEMD_TIMEOUT,
+            timeout=timeout,
         )
     except asyncio.TimeoutError:
         proc.kill()
         await proc.communicate()
-        return -1, f"systemctl {action} timed out after {SYSTEMD_TIMEOUT}s"
+        return -1, f"timed out after {timeout}s"
 
-    output = (stdout.decode() + stderr.decode()).strip()
+    raw = (stdout.decode() + stderr.decode()).strip()
-    return proc.returncode, output or "No output."
+    output = "\n".join(
+        line for line in raw.splitlines() if not line.strip().startswith("[sudo]")
+    )
+    return proc.returncode, output or "no output"
+
+
+async def run_git_pull() -> tuple[int, str]:
+    return await _run("git", "pull", cwd=BOT_DIR, timeout=GIT_TIMEOUT)
+
+
+async def run_git_log(n: int = 5) -> tuple[int, str]:
+    return await _run(
+        "git", "log", f"--oneline", f"-{n}",
+        cwd=BOT_DIR, timeout=GIT_TIMEOUT,
+    )
+
+
+async def run_sudo_systemctl(action: str) -> tuple[int, str]:
+    return await _run(
+        "sudo", "-S", "systemctl", action, SERVICE_NAME,
+        input_data=f"{SUDO_PASSWORD}\n".encode() if SUDO_PASSWORD else None,
+        timeout=SYSTEMD_TIMEOUT,
+    )
+
+
+async def run_journalctl(n: int = 50) -> tuple[int, str]:
+    return await _run(
+        "sudo", "-S", "journalctl", "-u", SERVICE_NAME,
+        "-n", str(n), "--no-pager",
+        input_data=f"{SUDO_PASSWORD}\n".encode() if SUDO_PASSWORD else None,
+        timeout=SYSTEMD_TIMEOUT,
+    )
 
 
 def truncate(text: str, limit: int = 1900) -> str:
     if len(text) > limit:
-        return text[:limit] + "\n... (truncated)"
+        return text[:limit] + "\n… (truncated)"
     return text
 
 
+def _embed(title: str, description: str, color: int) -> discord.Embed:
+    return discord.Embed(
+        title=title,
+        description=description,
+        color=color,
+        timestamp=datetime.now(timezone.utc),
+    )
+
+
 # ── cog ──────────────────────────────────────────────────────────────────────
 class GitPull(commands.Cog):
     def __init__(self, bot: commands.Bot):
@@ -76,55 +99,21 @@ class GitPull(commands.Cog):
         return ctx.author.id == OWNER_ID
 
     async def _do_restart(self, ctx: commands.Context):
-        await ctx.reply(embed=discord.Embed(
+        await ctx.reply(embed=_embed(
-            description="Restarting service...", color=C_WARN))
+            "⟳  Restarting",
+            f"Restarting `{SERVICE_NAME}` — see you on the other side.",
+            C_WARN,
+        ))
         try:
             await run_sudo_systemctl("restart")
         except Exception:
-            # process is expected to die mid-call once the service restarts,
+            pass  # process dies mid-call, that's expected
-            # so failures here are not reported back
+
-            pass
+    # ── !gitpull ──────────────────────────────────────────────────────────────
 
     @commands.command(name="gitpull")
     async def gitpull(self, ctx: commands.Context):
-        if not self._is_owner(ctx):
+        """Pull latest changes from git without restarting."""
-            return
-
-        async with ctx.typing():
-            code, output = await run_git_pull()
-
-        color = C_OK if code == 0 else C_ERR
-        embed = discord.Embed(
-            title="Git Pull",
-            description=f"```\n{truncate(output)}\n```",
-            color=color,
-        )
-        await ctx.reply(embed=embed)
-
-    @commands.command(name="status")
-    async def status(self, ctx: commands.Context):
-        if not self._is_owner(ctx):
-            return
-
-        async with ctx.typing():
-            code, output = await run_sudo_systemctl("status")
-
-        color = C_OK if code == 0 else C_ERR
-        embed = discord.Embed(
-            title=f"Service Status: {SERVICE_NAME}",
-            description=f"```\n{truncate(output)}\n```",
-            color=color,
-        )
-        await ctx.reply(embed=embed)
-
-    @commands.command(name="restart")
-    async def restart(self, ctx: commands.Context):
-        if not self._is_owner(ctx):
-            return
-        await self._do_restart(ctx)
-
-    @commands.command(name="update")
-    async def update(self, ctx: commands.Context):
         if not self._is_owner(ctx):
             return
 
@@ -132,27 +121,120 @@ class GitPull(commands.Cog):
             code, output = await run_git_pull()
 
         if code != 0:
-            embed = discord.Embed(
+            return await ctx.reply(embed=_embed(
-                title="Update Failed",
+                "✗  Pull Failed",
-                description=f"git pull failed, restart skipped.\n```\n{truncate(output)}\n```",
+                f"```\n{truncate(output)}\n```",
-                color=C_ERR,
+                C_ERR,
-            )
+            ))
-            return await ctx.reply(embed=embed)
 
         if "Already up to date" in output:
-            embed = discord.Embed(
+            return await ctx.reply(embed=_embed(
-                title="Update",
+                "✓  Already Up to Date",
-                description="No changes to pull. Restart skipped.",
+                "Nothing new to pull.",
-                color=C_WARN,
+                C_WARN,
-            )
+            ))
-            return await ctx.reply(embed=embed)
 
-        embed = discord.Embed(
+        await ctx.reply(embed=_embed(
-            title="Update",
+            "✓  Pulled",
-            description=f"Pulled changes:\n```\n{truncate(output)}\n```",
+            f"```\n{truncate(output)}\n```",
-            color=C_OK,
+            C_OK,
-        )
+        ))
-        await ctx.reply(embed=embed)
+
+    # ── !gitlog ───────────────────────────────────────────────────────────────
+
+    @commands.command(name="gitlog")
+    async def gitlog(self, ctx: commands.Context, n: int = 5):
+        """Show the last N commits. Default: 5."""
+        if not self._is_owner(ctx):
+            return
+        n = max(1, min(n, 20))
+
+        async with ctx.typing():
+            code, output = await run_git_log(n)
+
+        color = C_OK if code == 0 else C_ERR
+        await ctx.reply(embed=_embed(
+            f"📋  Last {n} Commits",
+            f"```\n{truncate(output)}\n```",
+            color,
+        ))
+
+    # ── !logs ─────────────────────────────────────────────────────────────────
+
+    @commands.command(name="logs")
+    async def logs(self, ctx: commands.Context, n: int = 40):
+        """Show the last N lines of the service journal. Default: 40."""
+        if not self._is_owner(ctx):
+            return
+        n = max(1, min(n, 100))
+
+        async with ctx.typing():
+            code, output = await run_journalctl(n)
+
+        color = C_OK if code == 0 else C_ERR
+        await ctx.reply(embed=_embed(
+            f"📄  Logs — {SERVICE_NAME} (last {n} lines)",
+            f"```\n{truncate(output)}\n```",
+            color,
+        ))
+
+    # ── !status ───────────────────────────────────────────────────────────────
+
+    @commands.command(name="status")
+    async def status(self, ctx: commands.Context):
+        """Show systemd service status."""
+        if not self._is_owner(ctx):
+            return
+
+        async with ctx.typing():
+            code, output = await run_sudo_systemctl("status")
+
+        color = C_OK if code == 0 else C_ERR
+        await ctx.reply(embed=_embed(
+            f"{'✓' if code == 0 else '✗'}  Status — {SERVICE_NAME}",
+            f"```\n{truncate(output)}\n```",
+            color,
+        ))
+
+    # ── !restart ──────────────────────────────────────────────────────────────
+
+    @commands.command(name="restart")
+    async def restart(self, ctx: commands.Context):
+        """Restart the bot service."""
+        if not self._is_owner(ctx):
+            return
+        await self._do_restart(ctx)
+
+    # ── !update ───────────────────────────────────────────────────────────────
+
+    @commands.command(name="update")
+    async def update(self, ctx: commands.Context):
+        """Pull latest changes and restart if anything changed."""
+        if not self._is_owner(ctx):
+            return
+
+        async with ctx.typing():
+            code, output = await run_git_pull()
+
+        if code != 0:
+            return await ctx.reply(embed=_embed(
+                "✗  Update Failed",
+                f"Pull failed — restart skipped.\n```\n{truncate(output)}\n```",
+                C_ERR,
+            ))
+
+        if "Already up to date" in output:
+            return await ctx.reply(embed=_embed(
+                "✓  Already Up to Date",
+                "Nothing new to pull. Restart skipped.",
+                C_WARN,
+            ))
+
+        await ctx.reply(embed=_embed(
+            "✓  Updated",
+            f"```\n{truncate(output)}\n```",
+            C_OK,
+        ))
         await self._do_restart(ctx)