From 101bc0abfed9a64c376ffe9ef06dab7feb453150 Mon Sep 17 00:00:00 2001 From: Roger Fedor Date: Tue, 25 Aug 2026 20:31:40 -0500 Subject: [PATCH] Add support for shlink and shorten_url message filter --- config.ini.example | 2 + modules/commands/path_command.py | 1 + modules/commands/test_command.py | 1 + modules/response_template.py | 11 ++++ modules/url_shortener.py | 106 +++++++++++++++++++++++-------- 5 files changed, 93 insertions(+), 28 deletions(-) diff --git a/config.ini.example b/config.ini.example index 95c759d..f1d19b9 100644 --- a/config.ini.example +++ b/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 diff --git a/modules/commands/path_command.py b/modules/commands/path_command.py index d23353f..b390e6c 100644 --- a/modules/commands/path_command.py +++ b/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: diff --git a/modules/commands/test_command.py b/modules/commands/test_command.py index 0a1529a..9f2d320 100644 --- a/modules/commands/test_command.py +++ b/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: diff --git a/modules/response_template.py b/modules/response_template.py index 66a4a07..d5824a2 100644 --- a/modules/response_template.py +++ b/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: diff --git a/modules/url_shortener.py b/modules/url_shortener.py index b815e8b..b7fe5e7 100644 --- a/modules/url_shortener.py +++ b/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(