update 26.7.25

4600Pro/tgd/module/storage.py

@@ -0,0 +1,104 @@
+"""Stable storage naming helpers for downloaded Telegram messages."""
+
+import os
+from dataclasses import dataclass
+from typing import Optional
+
+from module.pyrogram_extension import get_extension
+from utils.format import truncate_filename, validate_title
+
+
+@dataclass
+class StoragePaths:
+    """Resolved local paths for one downloaded message."""
+
+    final_path: str
+    temp_path: str
+    stored_file_name: str
+    file_extension: str
+    file_format: Optional[str]
+    original_file_name: Optional[str]
+    chat_title: str
+    media_datetime: str
+
+
+def normalize_chat_id(chat_id) -> str:
+    """Return a filesystem-safe chat id string."""
+    return validate_title(str(chat_id))
+
+
+def get_chat_title(chat_id, message) -> str:
+    """Return a filesystem-safe chat title, falling back to chat id."""
+    title = None
+    if getattr(message, "chat", None) and getattr(message.chat, "title", None):
+        title = message.chat.title
+    return validate_title(str(title if title else chat_id))
+
+
+def get_media_datetime(message, date_format: str) -> str:
+    """Return the configured media datetime directory name."""
+    if getattr(message, "date", None):
+        return message.date.strftime(date_format)
+    return "0"
+
+
+def get_original_file_name(media_obj) -> Optional[str]:
+    """Return Telegram's original file name when the media type exposes one."""
+    return getattr(media_obj, "file_name", None)
+
+
+def get_file_format(media_type: str, media_obj) -> Optional[str]:
+    """Return the format used by existing file-format filtering."""
+    if media_type in ["audio", "document", "video", "voice", "video_note"]:
+        mime_type = getattr(media_obj, "mime_type", "")
+        if "/" in mime_type:
+            return mime_type.split("/")[-1]
+    return None
+
+
+def resolve_file_extension(media_obj, original_file_name: Optional[str]) -> str:
+    """Resolve a dotted file extension for storage."""
+    if original_file_name:
+        _, suffix = os.path.splitext(os.path.basename(original_file_name))
+        if suffix:
+            return suffix
+
+    file_id = getattr(media_obj, "file_id", "")
+    mime_type = getattr(media_obj, "mime_type", "")
+    if file_id or mime_type:
+        suffix = get_extension(file_id, mime_type)
+        if suffix:
+            return suffix if suffix.startswith(".") else f".{suffix}"
+
+    return ".unknown"
+
+
+def build_stored_file_name(chat_id, message_id: int, file_extension: str) -> str:
+    """Build `{chat_id}_{message_id}.{extension}` with safe path characters."""
+    extension = file_extension if file_extension.startswith(".") else f".{file_extension}"
+    return validate_title(f"{normalize_chat_id(chat_id)}_{message_id}{extension}")
+
+
+def build_download_paths(app, chat_id, message, media_obj, media_type: str) -> StoragePaths:
+    """Build final and temporary paths using the stable file name rule."""
+    chat_title = get_chat_title(chat_id, message)
+    media_datetime = get_media_datetime(message, app.date_format)
+    original_file_name = get_original_file_name(media_obj)
+    file_extension = resolve_file_extension(media_obj, original_file_name)
+    file_format = get_file_format(media_type, media_obj)
+    stored_file_name = build_stored_file_name(chat_id, message.id, file_extension)
+
+    final_dir = app.get_file_save_path(media_type, chat_title, media_datetime)
+    final_path = os.path.join(final_dir, stored_file_name)
+    temp_path = os.path.join(app.temp_save_path, chat_title, stored_file_name)
+
+    return StoragePaths(
+        final_path=truncate_filename(final_path),
+        temp_path=truncate_filename(temp_path),
+        stored_file_name=stored_file_name,
+        file_extension=file_extension,
+        file_format=file_format,
+        original_file_name=original_file_name,
+        chat_title=chat_title,
+        media_datetime=media_datetime,
+    )