""" Shlink short-URL client. Provides a single async helper, ``shorten(url)``, that calls the Shlink REST API via httpx. Returns the original URL unchanged on any error. """ 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() data = resp.json() short = data["shortUrl"] log.debug("Shortened %s → %s", url, short) return short except Exception as e: log.warning("Shlink error for %s: %s", url, e) return url def from_config(cfg: dict) -> "ShlinkClient | None": """Build a ShlinkClient from the [shlink] config section.""" 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, )