mirror of
https://github.com/TehPeGaSuS/GitBot.git
synced 2026-06-28 17:05:47 +02:00
ThePeGaSuS
85 lines
2.8 KiB
Python
85 lines
2.8 KiB
Python
"""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,
|
|
)
|