diff --git a/efb_telegram_master/bot_manager.py b/efb_telegram_master/bot_manager.py index 116633c7..85bc1a3f 100644 --- a/efb_telegram_master/bot_manager.py +++ b/efb_telegram_master/bot_manager.py @@ -13,6 +13,7 @@ from telegram import Update, InputFile, User, File from telegram.ext import CallbackContext, Filters, MessageHandler, Updater, Dispatcher +from . import utils from .locale_handler import LocaleHandler from .locale_mixin import LocaleMixin @@ -581,7 +582,8 @@ def graceful_stop(self): def _detect_empty_file(self, file, chat, caption, prefix, suffix): empty = True if isinstance(file, str): - empty = os.stat(file).st_size == 0 + local_path = utils.coerce_local_path(file) + empty = os.stat(local_path).st_size == 0 elif hasattr(file, "seekable"): if file.seekable(): file.seek(0, 2) diff --git a/efb_telegram_master/slave_message.py b/efb_telegram_master/slave_message.py index 8c4e9eeb..49a7bb87 100644 --- a/efb_telegram_master/slave_message.py +++ b/efb_telegram_master/slave_message.py @@ -543,7 +543,7 @@ def slave_message_animation(self, msg: Message, tg_dest: TelegramChatID, msg_tem else: assert msg.file and msg.path file = self.process_file_obj(msg.file, msg.path) - file_: Union[IO[bytes], bytes] = open(file, 'rb') if isinstance(file, str) else file + file_: Union[IO[bytes], bytes] = open(utils.coerce_local_path(file), 'rb') if isinstance(file, str) else file return self.bot.send_animation(tg_dest, InputFile(file_, filename=msg.filename), prefix=msg_template, suffix=reactions, caption=text, parse_mode="HTML", diff --git a/efb_telegram_master/utils.py b/efb_telegram_master/utils.py index f6b9cc77..6e564838 100644 --- a/efb_telegram_master/utils.py +++ b/efb_telegram_master/utils.py @@ -6,6 +6,8 @@ import os import subprocess import sys +import urllib.parse +import urllib.request from io import BytesIO from shutil import copyfileobj from tempfile import NamedTemporaryFile @@ -79,6 +81,25 @@ def b64de(s: str) -> str: return base64.urlsafe_b64decode(s).decode() +def coerce_local_path(path_or_uri: str) -> str: + parts = urllib.parse.urlsplit(path_or_uri) + if parts.scheme != "file": + return path_or_uri + + path = urllib.request.url2pathname(urllib.parse.unquote(parts.path)) + if os.name == "nt": + if parts.netloc: + unc_path = path.replace("/", "\\") + return f"\\\\{parts.netloc}{unc_path}" + if len(path) >= 3 and path[0] == "/" and path[2] == ":": + path = path[1:] + return path + + if parts.netloc: + return f"//{parts.netloc}{path}" + return path + + def message_id_to_str(chat_id: Optional[TelegramChatID] = None, message_id: Optional[TelegramMessageID] = None, update: Optional[telegram.Update] = None) -> TgChatMsgIDStr: