Commit 0b25bbc

@@ -135,6 +135,7 @@ class TOKEN:
     SPOTIFY_CLIENT_SECRET = os.getenv("SPOTIFY_CLIENT_SECRET", "")
     V2EX = os.getenv("V2EX_TOKEN", "")
     TMDB = os.getenv("TMDB_TOKEN", "")
+    SPOOME = os.getenv("SPOOME_TOKEN", "")
 
 
 class PROXY:  # format: socks5://127.0.0.1:7890

@@ -5,21 +5,23 @@ import asyncio
 import contextlib
 import json
 import re
+from datetime import timedelta
 from pathlib import Path
 from typing import Any, Literal
 from urllib.parse import parse_qs, urlparse
 
 import anyio
 from curl_cffi.requests.impersonate import BrowserTypeLiteral
+from glom import glom
 from httpx import AsyncClient, AsyncHTTPTransport, HTTPStatusError, Request, RequestError, Response
 from httpx._types import RequestContent, RequestData, RequestFiles
 from httpx_curl_cffi import AsyncCurlTransport, CurlOpt
 from loguru import logger
 
-from config import DOWNLOAD_DIR, PROXY, REQUEST_TIMEOUT, cache, semaphore
+from config import DOWNLOAD_DIR, PROXY, REQUEST_TIMEOUT, TOKEN, cache, semaphore
 from messages.progress import modify_progress
 from messages.utils import summay_media
-from utils import av2bv, bare_url, check_data, https_url, is_supported_by_ytdlp, match_urls, readable_size
+from utils import av2bv, bare_url, check_data, https_url, is_supported_by_ytdlp, match_urls, nowdt, readable_size
 
 
 # ruff: noqa: RUF001
@@ -550,9 +552,36 @@ async def flatten_rediercts(
     return texts.replace(url, rediercted_url)
 
 
+async def shorten_url(url: str, services: list[str] | None = None) -> str:
+    """Shorten URL."""
+    if not url:
+        return url
+    supported = ["spoo.me", "cleanuri.com"]
+    if services is None:
+        services = supported
+    services = [x for x in services if x.lower() in supported]
+    for service in services:
+        if service == "spoo.me":
+            expire_after = nowdt() + timedelta(days=365 * 2)
+            headers = {"Content-Type": "application/json"}
+            payload = {"long_url": url, "expire_after": round(expire_after.timestamp()), "block_bots": False}
+            if TOKEN.SPOOME:
+                headers |= {"Authorization": f"Bearer {TOKEN.SPOOME}"}
+                payload |= {"private_stats": False}
+            resp = await hx_req("https://spoo.me/api/v1/shorten", "POST", headers=headers, json_data=payload, check_kv={"status": "ACTIVE"})
+            if short_url := glom(resp, "short_url", default=""):
+                return short_url
+        if service == "cleanuri.com":
+            resp = await hx_req("https://cleanuri.com/api/v1/shorten", "POST", json_data={"url": url}, check_keys=["result_url"])
+            if short_url := glom(resp, "result_url", default=""):
+                return short_url
+    return url
+
+
 if __name__ == "__main__":
     import asyncio
 
+    asyncio.run(shorten_url("https://www.google.com"))
     check_data(json.dumps({"foo": "bar", "baz": {"qux": "quux"}, "lst": ["1", "2", "3"]}), check_keys=["baz.qux"], check_kv={"foo": "bar", "baz.qux": "quux", "lst": ["1", "2", "3"]})
     # asyncio.run(match_social_media_link("https://b23.tv/3MSgT4q/", flatten_first=True))
     # print(asyncio.run(match_social_media_link("https://mp.weixin.qq.com/s/bd_giuPEyPBu9LTOtC2VHw", flatten_first=True)))