"""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, )