diff --git a/bot.py b/bot.py index 2e294a5..5049f4a 100644 --- a/bot.py +++ b/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}" diff --git a/gitbot.toml.example b/gitbot.toml.example index b8bcda8..f719e01 100644 --- a/gitbot.toml.example +++ b/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) diff --git a/shlink.py b/shlink.py new file mode 100644 index 0000000..4f50621 --- /dev/null +++ b/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, + )