105 lines
3.7 KiB
Python
Executable File
105 lines
3.7 KiB
Python
Executable File
"""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,
|
|
)
|