feat(userbot,api,frontend): file links, media downloads, story hold-pause, drop scheduled dupes

backend/src/utils/files.py

@@ -0,0 +1,134 @@
+import re
+from datetime import UTC, datetime, timedelta
+from urllib.parse import quote
+
+_MIME_EXTENSIONS = {
+    "application/pdf": ".pdf",
+    "application/x-tgsticker": ".tgs",
+    "application/zip": ".zip",
+    "audio/mpeg": ".mp3",
+    "audio/ogg": ".ogg",
+    "image/gif": ".gif",
+    "image/jpeg": ".jpg",
+    "image/png": ".png",
+    "image/webp": ".webp",
+    "text/plain": ".txt",
+    "video/mp4": ".mp4",
+    "video/quicktime": ".mov",
+    "video/webm": ".webm",
+}
+
+_KIND_EXTENSIONS = {
+    "animation": ".mp4",
+    "audio": ".mp3",
+    "gif": ".mp4",
+    "photo": ".jpg",
+    "sticker": ".webp",
+    "video": ".mp4",
+    "video_note": ".mp4",
+    "voice": ".ogg",
+}
+
+_INLINE_MIME_PREFIXES = ("image/", "video/", "audio/", "text/")
+_INLINE_MIMES = {"application/pdf", "application/json"}
+
+_PREVIEW_AGENTS = (
+    "telegrambot",
+    "twitterbot",
+    "facebookexternalhit",
+    "whatsapp",
+    "discordbot",
+    "slackbot",
+    "skypeuripreview",
+    "vkshare",
+    "redditbot",
+    "linkedinbot",
+    "embedly",
+    "quora link preview",
+    "googlebot",
+    "bingbot",
+    "applebot",
+    "yandexbot",
+    "duckduckbot",
+    "petalbot",
+    "ahrefsbot",
+    "semrushbot",
+    "headlesschrome",
+    "python-requests",
+)
+
+_UNSAFE_CHARS = re.compile(r'[\\/:*?"<>|\x00-\x1f]+')
+_SPACES = re.compile(r"\s+")
+_NAME_LIMIT = 120
+
+
+def extension_for(kind: str | None, mime: str | None) -> str:
+    if mime and mime in _MIME_EXTENSIONS:
+        return _MIME_EXTENSIONS[mime]
+    if kind and kind in _KIND_EXTENSIONS:
+        return _KIND_EXTENSIONS[kind]
+    if mime and "/" in mime:
+        tail = mime.rsplit("/", 1)[1].split(";")[0].strip()
+        if tail.isalnum():
+            return f".{tail}"
+    return ".bin"
+
+
+def sanitize_name(name: str) -> str:
+    cleaned = _SPACES.sub(" ", _UNSAFE_CHARS.sub("_", name)).strip(" .")
+    if len(cleaned) > _NAME_LIMIT:
+        head, dot, tail = cleaned.rpartition(".")
+        cleaned = (
+            f"{head[: _NAME_LIMIT - len(tail) - 1]}{dot}{tail}"
+            if dot and len(tail) < 12  # noqa: PLR2004
+            else cleaned[:_NAME_LIMIT]
+        )
+    return cleaned or "file"
+
+
+def media_file_name(
+    kind: str | None, mime: str | None, message_id: int, original: str | None = None
+) -> str:
+    if original:
+        name = sanitize_name(original)
+        return name if "." in name else f"{name}{extension_for(kind, mime)}"
+    return f"{kind or 'media'}_{message_id}{extension_for(kind, mime)}"
+
+
+def story_file_name(peer_id: int, story_id: int, media_kind: str | None) -> str:
+    return f"story_{peer_id}_{story_id}{extension_for(media_kind or 'photo', None)}"
+
+
+def is_inline_mime(mime: str | None) -> bool:
+    if not mime:
+        return False
+    return mime in _INLINE_MIMES or mime.startswith(_INLINE_MIME_PREFIXES)
+
+
+def content_disposition(file_name: str, *, attachment: bool) -> str:
+    kind = "attachment" if attachment else "inline"
+    ascii_name = file_name.encode("ascii", "replace").decode("ascii").replace('"', "_")
+    return f"{kind}; filename=\"{ascii_name}\"; filename*=UTF-8''{quote(file_name)}"
+
+
+def is_preview_agent(user_agent: str | None) -> bool:
+    if not user_agent:
+        return False
+    lowered = user_agent.lower()
+    return any(marker in lowered for marker in _PREVIEW_AGENTS)
+
+
+def counts_as_download(
+    method: str, user_agent: str | None, range_header: str | None
+) -> bool:
+    if method.upper() != "GET" or is_preview_agent(user_agent):
+        return False
+    if range_header is None:
+        return True
+    return range_header.replace(" ", "").startswith("bytes=0-")
+
+
+def expiry_from_seconds(seconds: int | None) -> datetime | None:
+    if seconds is None or seconds <= 0:
+        return None
+    return datetime.now(UTC) + timedelta(seconds=seconds)