mirror of
https://github.com/TehPeGaSuS/GitBot.git
synced 2026-06-30 16:45:45 +02:00
TehPeGaSuS
98ced2fbec
Refactor payload and headers creation for clarity. Simplify the from_config function by combining multiple lines into single lines.
46 lines
1.6 KiB
Python
46 lines
1.6 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.
|
|
|
|
Requires: httpx
|
|
"""
|
|
|
|
import logging
|
|
import httpx
|
|
|
|
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
|
|
self._endpoint = f"{self._base}/rest/v3/short-urls"
|
|
|
|
async def shorten(self, url: str) -> str:
|
|
"""Return a shortened URL, or ``url`` unchanged on failure."""
|
|
payload = {"longUrl": url, "findIfExists": True}
|
|
if self._domain:
|
|
payload["domain"] = self._domain
|
|
|
|
headers = {"X-Api-Key": self._api_key, "Content-Type": "application/json"}
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=self._timeout) as client:
|
|
resp = await client.post(self._endpoint, json=payload, headers=headers)
|
|
resp.raise_for_status()
|
|
return resp.json()["shortUrl"]
|
|
except Exception as e:
|
|
log.warning("Shlink error for %s: %s", url, e)
|
|
return url
|
|
|
|
def from_config(cfg: dict) -> "ShlinkClient | None":
|
|
if not cfg.get("enabled", True): return None
|
|
base, api_key = cfg.get("url", "").strip(), cfg.get("api_key", "").strip()
|
|
if not base or not api_key: return None
|
|
return ShlinkClient(base, api_key, int(cfg.get("timeout", 5)), cfg.get("domain"))
|