Add support for Shlink URL shortener

bot.py

@@ -27,6 +27,7 @@ import commands
 import db
 import irc_format as fmt
 import rss as rss_module
+import shlink as shlink_module
 import webhook_github
 import webhook_gitea
 import webhook_gitlab
@@ -84,6 +85,10 @@ class Bot:
         self._config_path = config_path
         self._database    = db.connect(config.get("database", "gitbot.db"))
         self._clients: dict[str, IRCClient] = {}
+        self._shlink = shlink_module.from_config(config.get("shlink", {}))
+
+        if self._shlink:
+            log.info("Shlink URL shortener enabled")
 
         self._load_static_webhooks()
         self._load_static_rss()
@@ -157,7 +162,8 @@ class Bot:
         except Exception as e:
             return f"Failed to reload config: {e}"
 
-        self._cfg = new_cfg
+        self._cfg    = new_cfg
+        self._shlink = shlink_module.from_config(new_cfg.get("shlink", {}))
         self._load_static_webhooks()
         self._load_static_rss()
 
@@ -276,6 +282,8 @@ class Bot:
                 continue
 
             for message, url in outputs:
+                if url and self._shlink:
+                    url = await self._shlink.shorten(url)
                 line = f"{forge_tag} ({source}) {message}"
                 if url:
                     line = f"{line} - {url}"

gitbot.toml.example

@@ -72,3 +72,14 @@ nickname = "gitbot"
 username = "gitbot"
 realname = "git webhook + RSS bot"
 channels = ["#myproject"]
+
+# ── Shlink URL shortener ──────────────────────────────────────────────────────
+# When enabled, all URLs posted to IRC will be shortened via your Shlink
+# instance.  Requires a Shlink API key with "Short URLs / Create" permission.
+#
+# [shlink]
+# url     = "https://shlink.example.com"   # base URL of your Shlink instance
+# api_key = "YOUR-API-KEY"
+# # enabled = true                          # set false to bypass shortening
+# # timeout = 5                             # seconds to wait for the API
+# # domain  = "s.example.com"              # custom domain (optional)

shlink.py

@@ -0,0 +1,84 @@
+"""Shlink short-URL client.
+
+Provides a single async helper, ``shorten(url)``, that calls the Shlink
+REST API and returns the shortened URL.  Returns the original URL unchanged
+on any error so that the bot always has *something* to post.
+
+Configuration (gitbot.toml):
+
+    [shlink]
+    url    = "https://shlink.example.com"   # base URL of your Shlink instance
+    api_key = "YOUR-API-KEY"                 # Shlink API key
+    # enabled = true                         # set false to disable (default true)
+    # timeout = 5                            # HTTP timeout in seconds (default 5)
+    # domain  = "s.example.com"             # custom domain if you have several
+"""
+
+import asyncio
+import json
+import logging
+import urllib.error
+import urllib.parse
+import urllib.request
+
+log = logging.getLogger("shlink")
+
+
+class ShlinkClient:
+    def __init__(self, base_url: str, api_key: str,
+                 timeout: int = 5, domain: str | None = None):
+        self._base    = base_url.rstrip("/")
+        self._api_key = api_key
+        self._timeout = timeout
+        self._domain  = domain
+
+    async def shorten(self, url: str) -> str:
+        """Return a shortened URL, or ``url`` unchanged on failure."""
+        try:
+            return await asyncio.get_event_loop().run_in_executor(
+                None, self._call, url)
+        except Exception as e:
+            log.warning("Shlink error for %s: %s", url, e)
+            return url
+
+    def _call(self, url: str) -> str:
+        endpoint = f"{self._base}/rest/v3/short-urls"
+        payload  = {"longUrl": url}
+        if self._domain:
+            payload["domain"] = self._domain
+        body = json.dumps(payload).encode()
+        req  = urllib.request.Request(
+            endpoint,
+            data=body,
+            headers={
+                "Content-Type": "application/json",
+                "X-Api-Key":    self._api_key,
+            },
+            method="POST",
+        )
+        with urllib.request.urlopen(req, timeout=self._timeout) as resp:
+            data = json.loads(resp.read())
+        short = data["shortUrl"]
+        log.debug("Shortened %s → %s", url, short)
+        return short
+
+
+def from_config(cfg: dict) -> "ShlinkClient | None":
+    """Build a ShlinkClient from the ``[shlink]`` config section.
+
+    Returns ``None`` when shlink is disabled or misconfigured.
+    """
+    if not cfg.get("enabled", True):
+        return None
+    base    = cfg.get("url", "").strip()
+    api_key = cfg.get("api_key", "").strip()
+    if not base or not api_key:
+        if base or api_key:
+            log.warning("[shlink] Both 'url' and 'api_key' are required — disabling")
+        return None
+    return ShlinkClient(
+        base_url=base,
+        api_key=api_key,
+        timeout=int(cfg.get("timeout", 5)),
+        domain=cfg.get("domain") or None,
+    )