Add support for shlink and shorten_url message filter

config.ini.example

@@ -644,6 +644,8 @@ log_max_bytes = 5242880
 # log_backup_count: number of rotated backup files to keep (e.g. meshcore_bot.log.1 … .3)
 log_backup_count = 3
 [External_Data]
+# URL shortener Service (gd,shlink). Default: gd
+short_url_website_service = gd
 # URL shortener API base (v.gd / is.gd-compatible create.php). Default: https://v.gd
 # See https://v.gd/apishorteningreference.php
 short_url_website = https://v.gd

modules/commands/path_command.py

@@ -425,6 +425,7 @@ class PathCommand(BaseCommand):
             {k: str(v) for k, v in fields.items()},
             message=message,
             logger=self.logger,
+            config=self.bot.config,
             prefix_hex_chars=getattr(self.bot, 'prefix_hex_chars', 2),
         ).rstrip()
         if not formatted:

modules/commands/test_command.py

@@ -674,6 +674,7 @@ class TestCommand(BaseCommand):
                 fields,
                 message=message,
                 logger=self.logger,
+                config=self.bot.config,
                 prefix_hex_chars=getattr(self.bot, 'prefix_hex_chars', 2),
             )
         except (KeyError, ValueError) as e:

modules/response_template.py

@@ -11,6 +11,7 @@ import re
 from typing import Any, Callable
 
 from .utils import message_hop_count, message_path_bytes_per_hop
+from .url_shortener import shorten_url
 
 FilterFn = Callable[[str, dict[str, Any], str], str]
 
@@ -66,12 +67,20 @@ def _filter_prefix_if_nonempty(value: str, ctx: dict[str, Any], args: str) -> st
         return ''
     return args + value
 
+def _filter_shorten_url(value: str, ctx: dict[str, Any], args: str) -> str:
+    """Shorten *value* URL using configured URL shortener (v.gd / is.gd compatible)."""
+    config = ctx.get('config')
+    if config is None:
+        return value
+    return shorten_url(value, config)
+
 
 RESPONSE_TEMPLATE_FILTERS: dict[str, FilterFn] = {
     'pathbytes_min': _filter_pathbytes_min,
     'pathbytes': _filter_pathbytes_min,
     'hops_min': _filter_hops_min,
     'prefix_if_nonempty': _filter_prefix_if_nonempty,
+    'shorten_url': _filter_shorten_url,
 }
 
 
@@ -116,6 +125,7 @@ def format_piped_template(
     *,
     message: Any = None,
     logger: Any = None,
+    config: Any = None,
     prefix_hex_chars: int = 2,
 ) -> str:
     """Replace ``{field}`` and ``{field|filter:arg|...}`` using *fields* and optional *message*.
@@ -136,6 +146,7 @@ def format_piped_template(
         'message': message,
         'logger': logger,
         'prefix_hex_chars': prefix_hex_chars,
+        'config': config,
     }
 
     def replace_placeholder(match: re.Match[str]) -> str:

modules/url_shortener.py

@@ -82,7 +82,7 @@ def _parse_simple_response(body: str) -> str | None:
     return None
 
 
-def _build_create_url(long_url: str, base: str, api_key: str) -> str:
+def _build_create_gd_url(long_url: str, base: str, api_key: str) -> str:
     from urllib.parse import urlparse, urlunparse
 
     encoded = quote(long_url, safe="")
@@ -104,6 +104,76 @@ def _build_create_url(long_url: str, base: str, api_key: str) -> str:
     )
     return rebuilt
 
+def _build_create_shlink_url(long_url: str, base: str, api_key: str) -> str:
+    from urllib.parse import urlparse, urlunparse
+
+    encoded = quote(long_url, safe="")
+    root = _normalize_base(base)
+    if "://" not in root:
+        root = f"https://{root}"
+    parsed = urlparse(root)
+    netloc = parsed.netloc
+    if not netloc and parsed.path:
+        netloc = parsed.path.split("/")[0]
+    path = (parsed.path or "").rstrip("/") + "/api/v3/short-urls"
+    if not path.startswith("/"):
+        path = "/" + path
+    query = ""
+    rebuilt = urlunparse(
+        (parsed.scheme or "https", netloc, path, "", query, "")
+    )
+    return rebuilt
+
+def _shorten_url_with_shlink(long_url: str, base: str, api_key: str, session: requests.Session | None = None, timeout: float = 5.0) -> str:
+    """Shorten a URL using Shlink API."""
+    import json
+
+    shortener_url = _build_create_shlink_url(long_url, base, api_key)
+    headers = {
+        "Content-Type": "application/json",
+        "X-Api-Key": api_key,
+    }
+    payload = json.dumps({"longUrl": long_url})
+    get = session.post if session is not None else requests.post
+    try:
+        response = get(shortener_url, headers=headers, data=payload, timeout=timeout)
+    except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e:
+        return ""
+    except Exception as e:
+        return ""
+
+    if not response.ok:
+        return ""
+
+    try:
+        data = response.json()
+        short_url = data.get("shortUrl") or data.get("shortUrlSlug")
+        if short_url:
+            return short_url
+    except Exception:
+        return ""
+
+    return ""
+
+def _shorten_url_with_gd(long_url: str, base: str, api_key: str, session: requests.Session | None = None, timeout: float = 5.0) -> str:
+    """Shorten a URL using v.gd / is.gd API."""
+    shortener_url = _build_create_gd_url(long_url, base, api_key)
+    get = session.get if session is not None else requests.get
+    try:
+        response = get(shortener_url, timeout=timeout)
+    except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e:
+        return ""
+    except Exception as e:
+        return ""
+
+    if not response.ok:
+        return ""
+
+    short = _parse_simple_response(response.text)
+    if short:
+        return short
+
+    return ""
 
 def shorten_url_sync(
     url: Any,
@@ -123,38 +193,18 @@ def shorten_url_sync(
             return ""
 
         base = _safe_config_get(config, "External_Data", "short_url_website", "")
+        service = safe_config_get(config, "External_Data", "short_url_website_service", "gd").strip().lower()
         api_key = (_safe_config_get(config, "External_Data", "short_url_website_api_key", "") or "").strip()
         base = _normalize_base(base)
 
-        shortener_url = _build_create_url(url_str, base, api_key)
-        get = session.get if session is not None else requests.get
-
-        try:
-            response = get(shortener_url, timeout=timeout)
-        except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e:
-            if logger:
-                logger.debug("Error shortening URL: %s", e)
-            return ""
-        except Exception as e:
-            if logger:
-                logger.debug("Unexpected error shortening URL: %s", e)
-            return ""
-
-        if not response.ok:
-            if logger:
-                logger.debug("Error shortening URL: HTTP %s", response.status_code)
-            return ""
-
-        short = _parse_simple_response(response.text)
-        if short:
-            return short
-        if logger:
-            logger.debug("URL shortener returned error: %s", response.text.strip()[:200])
-        return ""
+        if service == "shlink":
+            return _shorten_url_with_shlink(url_str, base, api_key, session=session, timeout=timeout)
+        else:
+            return _shorten_url_with_gd(url_str, base, api_key, session=session, timeout=timeout)
     except Exception as e:
         if logger:
-            logger.debug("shorten_url_sync failed: %s", e)
-        return ""
+            logger.debug("Unexpected error shortening URL: %s", e)
+        return ""   
 
 
 async def shorten_url(