Xtao-Labs/MTPyroger
3
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Cleanup

Browse Source
This commit is contained in:
Dan 2020-07-09 00:20:46 +02:00
parent 6b2d6ffacf
commit 4a8e6fb855
12 changed files with 216 additions and 544 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

289
pyrogram/client/client.py
View File

@ -15,7 +15,7 @@
#
# You should have received a copy of the GNU Lesser General Public License
# along with Pyrogram. If not, see <http://www.gnu.org/licenses/>.
import io
import logging
import math
import os
@ -1231,9 +1231,9 @@ class Client(Methods, BaseClient):
temp_file_path = ""
final_file_path = ""
path = [None]
try:
data, done, progress, progress_args, out, path, to_file = packet
data, directory, file_name, done, progress, progress_args, path = packet
temp_file_path = self.get_file(
media_type=data.media_type,
@ -1250,15 +1250,13 @@ class Client(Methods, BaseClient):
file_size=data.file_size,
is_big=data.is_big,
progress=progress,
progress_args=progress_args,
out=out
progress_args=progress_args
)
if to_file:
final_file_path = out.name
else:
final_file_path = ''
if to_file:
out.close()
if temp_file_path:
final_file_path = os.path.abspath(re.sub("\\\\", "/", os.path.join(directory, file_name)))
os.makedirs(directory, exist_ok=True)
shutil.move(temp_file_path, final_file_path)
except Exception as e:
log.error(e, exc_info=True)
@ -1715,7 +1713,7 @@ class Client(Methods, BaseClient):
def save_file(
self,
path: Union[str, io.IOBase],
path: str,
file_id: int = None,
file_part: int = 0,
progress: callable = None,
@ -1767,20 +1765,9 @@ class Client(Methods, BaseClient):
Raises:
RPCError: In case of a Telegram RPC error.
ValueError: if path is not str or file-like readable object
"""
part_size = 512 * 1024
if isinstance(path, str):
fp = open(path, 'rb')
filename = os.path.basename(path)
elif hasattr(path, 'write'):
fp = path
filename = fp.name
else:
raise ValueError("Invalid path passed! Pass file pointer or path to file")
fp.seek(0, os.SEEK_END)
file_size = fp.tell()
fp.seek(0)
file_size = os.path.getsize(path)
if file_size == 0:
raise ValueError("File size equals to 0 B")
@ -1798,74 +1785,67 @@ class Client(Methods, BaseClient):
session.start()
try:
fp.seek(part_size * file_part)
with open(path, "rb") as f:
f.seek(part_size * file_part)
while True:
chunk = fp.read(part_size)
while True:
chunk = f.read(part_size)
if not chunk:
if not is_big:
md5_sum = "".join([hex(i)[2:].zfill(2) for i in md5_sum.digest()])
break
for _ in range(3):
if is_big:
rpc = functions.upload.SaveBigFilePart(
file_id=file_id,
file_part=file_part,
file_total_parts=file_total_parts,
bytes=chunk
)
else:
rpc = functions.upload.SaveFilePart(
file_id=file_id,
file_part=file_part,
bytes=chunk
)
if session.send(rpc):
if not chunk:
if not is_big:
md5_sum = "".join([hex(i)[2:].zfill(2) for i in md5_sum.digest()])
break
else:
raise AssertionError("Telegram didn't accept chunk #{} of {}".format(file_part, path))
if is_missing_part:
return
for _ in range(3):
if is_big:
rpc = functions.upload.SaveBigFilePart(
file_id=file_id,
file_part=file_part,
file_total_parts=file_total_parts,
bytes=chunk
)
else:
rpc = functions.upload.SaveFilePart(
file_id=file_id,
file_part=file_part,
bytes=chunk
)
if not is_big:
md5_sum.update(chunk)
if session.send(rpc):
break
else:
raise AssertionError("Telegram didn't accept chunk #{} of {}".format(file_part, path))
file_part += 1
if is_missing_part:
return
if progress:
progress(min(file_part * part_size, file_size), file_size, *progress_args)
if not is_big:
md5_sum.update(chunk)
file_part += 1
if progress:
progress(min(file_part * part_size, file_size), file_size, *progress_args)
except Client.StopTransmission:
if isinstance(path, str):
fp.close()
raise
except Exception as e:
if isinstance(path, str):
fp.close()
log.error(e, exc_info=True)
else:
if isinstance(path, str):
fp.close()
if is_big:
return types.InputFileBig(
id=file_id,
parts=file_total_parts,
name=filename,
name=os.path.basename(path),
)
else:
return types.InputFile(
id=file_id,
parts=file_total_parts,
name=filename,
name=os.path.basename(path),
md5_checksum=md5_sum
)
finally:
if isinstance(path, str):
fp.close()
session.stop()
def get_file(
@ -1884,8 +1864,7 @@ class Client(Methods, BaseClient):
file_size: int,
is_big: bool,
progress: callable,
progress_args: tuple = (),
out: io.IOBase = None
progress_args: tuple = ()
) -> str:
with self.media_sessions_lock:
session = self.media_sessions.get(dc_id, None)
@ -1971,10 +1950,7 @@ class Client(Methods, BaseClient):
limit = 1024 * 1024
offset = 0
file_name = ""
if not out:
f = tempfile.NamedTemporaryFile("wb", delete=False)
else:
f = out
try:
r = session.send(
functions.upload.GetFile(
@ -1985,37 +1961,36 @@ class Client(Methods, BaseClient):
)
if isinstance(r, types.upload.File):
if hasattr(f, "name"):
with tempfile.NamedTemporaryFile("wb", delete=False) as f:
file_name = f.name
while True:
chunk = r.bytes
while True:
chunk = r.bytes
if not chunk:
break
if not chunk:
break
f.write(chunk)
f.write(chunk)
offset += limit
offset += limit
if progress:
progress(
if progress:
progress(
min(offset, file_size)
if file_size != 0
else offset,
file_size,
*progress_args
)
min(offset, file_size)
if file_size != 0
else offset,
file_size,
*progress_args
r = session.send(
functions.upload.GetFile(
location=location,
offset=offset,
limit=limit
)
)
r = session.send(
functions.upload.GetFile(
location=location,
offset=offset,
limit=limit
)
)
elif isinstance(r, types.upload.FileCdnRedirect):
with self.media_sessions_lock:
cdn_session = self.media_sessions.get(r.dc_id, None)
@ -2028,71 +2003,70 @@ class Client(Methods, BaseClient):
self.media_sessions[r.dc_id] = cdn_session
try:
if hasattr(f, "name"):
with tempfile.NamedTemporaryFile("wb", delete=False) as f:
file_name = f.name
while True:
r2 = cdn_session.send(
functions.upload.GetCdnFile(
file_token=r.file_token,
offset=offset,
limit=limit
)
)
if isinstance(r2, types.upload.CdnFileReuploadNeeded):
try:
session.send(
functions.upload.ReuploadCdnFile(
file_token=r.file_token,
request_token=r2.request_token
)
while True:
r2 = cdn_session.send(
functions.upload.GetCdnFile(
file_token=r.file_token,
offset=offset,
limit=limit
)
except VolumeLocNotFound:
)
if isinstance(r2, types.upload.CdnFileReuploadNeeded):
try:
session.send(
functions.upload.ReuploadCdnFile(
file_token=r.file_token,
request_token=r2.request_token
)
)
except VolumeLocNotFound:
break
else:
continue
chunk = r2.bytes
# https://core.telegram.org/cdn#decrypting-files
decrypted_chunk = AES.ctr256_decrypt(
chunk,
r.encryption_key,
bytearray(
r.encryption_iv[:-4]
+ (offset // 16).to_bytes(4, "big")
)
)
hashes = session.send(
functions.upload.GetCdnFileHashes(
file_token=r.file_token,
offset=offset
)
)
# https://core.telegram.org/cdn#verifying-files
for i, h in enumerate(hashes):
cdn_chunk = decrypted_chunk[h.limit * i: h.limit * (i + 1)]
assert h.hash == sha256(cdn_chunk).digest(), "Invalid CDN hash part {}".format(i)
f.write(decrypted_chunk)
offset += limit
if progress:
progress(
min(offset, file_size)
if file_size != 0
else offset,
file_size,
*progress_args
)
if len(chunk) < limit:
break
else:
continue
chunk = r2.bytes
# https://core.telegram.org/cdn#decrypting-files
decrypted_chunk = AES.ctr256_decrypt(
chunk,
r.encryption_key,
bytearray(
r.encryption_iv[:-4]
+ (offset // 16).to_bytes(4, "big")
)
)
hashes = session.send(
functions.upload.GetCdnFileHashes(
file_token=r.file_token,
offset=offset
)
)
# https://core.telegram.org/cdn#verifying-files
for i, h in enumerate(hashes):
cdn_chunk = decrypted_chunk[h.limit * i: h.limit * (i + 1)]
assert h.hash == sha256(cdn_chunk).digest(), "Invalid CDN hash part {}".format(i)
f.write(decrypted_chunk)
offset += limit
if progress:
progress(
min(offset, file_size)
if file_size != 0
else offset,
file_size,
*progress_args
)
if len(chunk) < limit:
break
except Exception as e:
raise e
except Exception as e:
@ -2100,8 +2074,7 @@ class Client(Methods, BaseClient):
log.error(e, exc_info=True)
try:
if out:
os.remove(file_name)
os.remove(file_name)
except OSError:
pass

13
pyrogram/client/methods/messages/download_media.py
View File

@ -17,9 +17,7 @@
# along with Pyrogram. If not, see <http://www.gnu.org/licenses/>.
import binascii
import io
import os
import re
import struct
import time
from datetime import datetime
@ -39,7 +37,6 @@ class DownloadMedia(BaseClient):
message: Union["pyrogram.Message", str],
file_ref: str = None,
file_name: str = DEFAULT_DOWNLOAD_DIR,
out: io.IOBase = None,
block: bool = True,
progress: callable = None,
progress_args: tuple = ()
@ -61,9 +58,6 @@ class DownloadMedia(BaseClient):
You can also specify a path for downloading files in a custom location: paths that end with "/"
are considered directories. All non-existent folders will be created automatically.
out (``io.IOBase``, *optional*):
A custom *file-like object* to be used when downloading file. Overrides file_name
block (``bool``, *optional*):
Blocks the code execution until the file has been downloaded.
Defaults to True.
@ -244,13 +238,6 @@ class DownloadMedia(BaseClient):
extension
)
if not out:
out = open(os.path.abspath(re.sub("\\\\", "/", os.path.join(directory, file_name))), 'wb')
os.makedirs(directory, exist_ok=True)
to_file = True
else:
to_file = False
self.download_queue.put((data, done, progress, progress_args, out, path, to_file))
# Cast to string because Path objects aren't supported by Python 3.5
self.download_queue.put((data, str(directory), str(file_name), done, progress, progress_args, path))

153
pyrogram/client/methods/messages/send_animated_sticker.py
View File

@ -1,153 +0,0 @@
# Pyrogram - Telegram MTProto API Client Library for Python
# Copyright (C) 2017-2019 Dan Tès <https://github.com/delivrance>
#
# This file is part of Pyrogram.
#
# Pyrogram is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Pyrogram is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Pyrogram. If not, see <http://www.gnu.org/licenses/>.
import io
import os
from typing import Union
import pyrogram
from pyrogram.api import functions, types
from pyrogram.client.ext import BaseClient, utils
from pyrogram.errors import FilePartMissing
class SendAnimatedSticker(BaseClient):
def send_animated_sticker(
self,
chat_id: Union[int, str],
animated_sticker: Union[str, io.IOBase],
disable_notification: bool = None,
reply_to_message_id: int = None,
reply_markup: Union[
"pyrogram.InlineKeyboardMarkup",
"pyrogram.ReplyKeyboardMarkup",
"pyrogram.ReplyKeyboardRemove",
"pyrogram.ForceReply"
] = None,
progress: callable = None,
progress_args: tuple = ()
) -> Union["pyrogram.Message", None]:
"""Send .tgs animated stickers.
Parameters:
chat_id (``int`` | ``str``):
Unique identifier (int) or username (str) of the target chat.
For your personal cloud (Saved Messages) you can simply use "me" or "self".
For a contact that exists in your Telegram address book you can use his phone number (str).
animated_sticker (``str`` | file-like object):
Animated sticker to send.
Pass a file_id as string to send a animated sticker that exists on the Telegram servers,
pass an HTTP URL as a string for Telegram to get a .webp animated sticker file from the Internet, or
pass a file path as string to upload a new animated sticker that exists on your local machine.
pass a readable file-like object with .name
disable_notification (``bool``, *optional*):
Sends the message silently.
Users will receive a notification with no sound.
reply_to_message_id (``int``, *optional*):
If the message is a reply, ID of the original message.
reply_markup (:obj:`InlineKeyboardMarkup` | :obj:`ReplyKeyboardMarkup` | :obj:`ReplyKeyboardRemove` | :obj:`ForceReply`, *optional*):
Additional interface options. An object for an inline keyboard, custom reply keyboard,
instructions to remove reply keyboard or to force a reply from the user.
progress (``callable``, *optional*):
Pass a callback function to view the upload progress.
The function must take *(client, current, total, \*args)* as positional arguments (look at the section
below for a detailed description).
progress_args (``tuple``, *optional*):
Extra custom arguments for the progress callback function. Useful, for example, if you want to pass
a chat_id and a message_id in order to edit a message with the updated progress.
Other Parameters:
client (:obj:`Client`):
The Client itself, useful when you want to call other API methods inside the callback function.
current (``int``):
The amount of bytes uploaded so far.
total (``int``):
The size of the file.
*args (``tuple``, *optional*):
Extra custom arguments as defined in the *progress_args* parameter.
You can either keep *\*args* or add every single extra argument in your function signature.
Returns:
:obj:`Message` | ``None``: On success, the sent animated sticker message is returned, otherwise, in case the
upload is deliberately stopped with :meth:`~Client.stop_transmission`, None is returned.
Raises:
RPCError: In case of a Telegram RPC error.
"""
file = None
try:
if isinstance(animated_sticker, str):
if os.path.exists(animated_sticker):
file = self.save_file(animated_sticker, progress=progress, progress_args=progress_args)
media = types.InputMediaUploadedDocument(
mime_type=self.guess_mime_type(animated_sticker) or "application/x-tgsticker",
file=file,
attributes=[
types.DocumentAttributeFilename(file_name=os.path.basename(animated_sticker))
]
)
elif animated_sticker.startswith("http"):
media = types.InputMediaDocumentExternal(
url=animated_sticker
)
else:
media = utils.get_input_media_from_file_id(animated_sticker, 5)
elif hasattr(animated_sticker, "read"):
file = self.save_file(animated_sticker, progress=progress, progress_args=progress_args)
media = types.InputMediaUploadedDocument(
mime_type=self.guess_mime_type(animated_sticker.name) or "application/x-tgsticker",
file=file,
attributes=[
types.DocumentAttributeFilename(file_name=animated_sticker.name)
]
)
while True:
try:
r = self.send(
functions.messages.SendMedia(
peer=self.resolve_peer(chat_id),
media=media,
silent=disable_notification or None,
reply_to_msg_id=reply_to_message_id,
random_id=self.rnd_id(),
reply_markup=reply_markup.write() if reply_markup else None,
message=""
)
)
except FilePartMissing as e:
self.save_file(animated_sticker, file_id=file.id, file_part=e.x)
else:
for i in r.updates:
if isinstance(i, (types.UpdateNewMessage, types.UpdateNewChannelMessage)):
return pyrogram.Message._parse(
self, i.message,
{i.id: i for i in r.users},
{i.id: i for i in r.chats}
)
except BaseClient.StopTransmission:
return None

45
pyrogram/client/methods/messages/send_animation.py
View File

@ -15,7 +15,7 @@
#
# You should have received a copy of the GNU Lesser General Public License
# along with Pyrogram. If not, see <http://www.gnu.org/licenses/>.
import io
import os
from typing import Union
@ -29,7 +29,6 @@ class SendAnimation(BaseClient):
def send_animation(
self,
chat_id: Union[int, str],
animation: Union[str, io.IOBase],
animation: str,
file_ref: str = None,
caption: str = "",
@ -60,13 +59,11 @@ class SendAnimation(BaseClient):
For your personal cloud (Saved Messages) you can simply use "me" or "self".
For a contact that exists in your Telegram address book you can use his phone number (str).
animation (``str``| file-like object):
animation (``str``):
Animation to send.
Pass a file_id as string to send an animation that exists on the Telegram servers,
pass an HTTP URL as a string for Telegram to get an animation from the Internet, or
pass a file path as string to upload a new animation that exists on your local machine.
pass a readable file-like object with .name
file_ref (``str``, *optional*):
A valid file reference obtained by a recently fetched media message.
@ -166,36 +163,11 @@ class SendAnimation(BaseClient):
file = None
try:
if isinstance(animation, str):
if os.path.exists(animation):
thumb = None if thumb is None else self.save_file(thumb)
file = self.save_file(animation, progress=progress, progress_args=progress_args)
media = types.InputMediaUploadedDocument(
mime_type=self.guess_mime_type(animation) or "video/mp4",
file=file,
thumb=thumb,
attributes=[
types.DocumentAttributeVideo(
supports_streaming=True,
duration=duration,
w=width,
h=height
),
types.DocumentAttributeFilename(file_name=file_name or os.path.basename(animation)),
types.DocumentAttributeAnimated()
]
)
elif animation.startswith("http"):
media = types.InputMediaDocumentExternal(
url=animation
)
else:
media = utils.get_input_media_from_file_id(animation, file_ref, 10)
elif hasattr(animation, "read"):
if os.path.exists(animation):
thumb = None if thumb is None else self.save_file(thumb)
file = self.save_file(animation, progress=progress, progress_args=progress_args)
media = types.InputMediaUploadedDocument(
mime_type=self.guess_mime_type(animation.name) or "video/mp4",
mime_type=self.guess_mime_type(animation) or "video/mp4",
file=file,
thumb=thumb,
attributes=[
@ -205,10 +177,17 @@ class SendAnimation(BaseClient):
w=width,
h=height
),
types.DocumentAttributeFilename(file_name=animation.name),
types.DocumentAttributeFilename(file_name=file_name or os.path.basename(animation)),
types.DocumentAttributeAnimated()
]
)
elif animation.startswith("http"):
media = types.InputMediaDocumentExternal(
url=animation
)
else:
media = utils.get_input_media_from_file_id(animation, file_ref, 10)
while True:
try:
r = self.send(

41
pyrogram/client/methods/messages/send_audio.py
View File

@ -15,7 +15,7 @@
#
# You should have received a copy of the GNU Lesser General Public License
# along with Pyrogram. If not, see <http://www.gnu.org/licenses/>.
import io
import os
from typing import Union
@ -29,7 +29,6 @@ class SendAudio(BaseClient):
def send_audio(
self,
chat_id: Union[int, str],
audio: Union[str, io.IOBase],
audio: str,
file_ref: str = None,
caption: str = "",
@ -61,12 +60,11 @@ class SendAudio(BaseClient):
For your personal cloud (Saved Messages) you can simply use "me" or "self".
For a contact that exists in your Telegram address book you can use his phone number (str).
audio (``str``, file-like object):
audio (``str``):
Audio file to send.
Pass a file_id as string to send an audio file that exists on the Telegram servers,
pass an HTTP URL as a string for Telegram to get an audio file from the Internet, or
pass a file path as string to upload a new audio file that exists on your local machine.
pass a readable file-like object with .name
file_ref (``str``, *optional*):
A valid file reference obtained by a recently fetched media message.
@ -165,34 +163,11 @@ class SendAudio(BaseClient):
file = None
try:
if isinstance(audio, str):
if os.path.exists(audio):
thumb = None if thumb is None else self.save_file(thumb)
file = self.save_file(audio, progress=progress, progress_args=progress_args)
media = types.InputMediaUploadedDocument(
mime_type=self.guess_mime_type(audio) or "audio/mpeg",
file=file,
thumb=thumb,
attributes=[
types.DocumentAttributeAudio(
duration=duration,
performer=performer,
title=title
),
types.DocumentAttributeFilename(file_name=file_name or os.path.basename(audio))
]
)
elif audio.startswith("http"):
media = types.InputMediaDocumentExternal(
url=audio
)
else:
media = utils.get_input_media_from_file_id(audio, file_ref, 9)
elif hasattr(audio, "read"):
if os.path.exists(audio):
thumb = None if thumb is None else self.save_file(thumb)
file = self.save_file(audio, progress=progress, progress_args=progress_args)
media = types.InputMediaUploadedDocument(
mime_type=self.guess_mime_type(audio.name) or "audio/mpeg",
mime_type=self.guess_mime_type(audio) or "audio/mpeg",
file=file,
thumb=thumb,
attributes=[
@ -201,9 +176,15 @@ class SendAudio(BaseClient):
performer=performer,
title=title
),
types.DocumentAttributeFilename(file_name=os.path.basename(audio.name))
types.DocumentAttributeFilename(file_name=file_name or os.path.basename(audio))
]
)
elif audio.startswith("http"):
media = types.InputMediaDocumentExternal(
url=audio
)
else:
media = utils.get_input_media_from_file_id(audio, file_ref, 9)
while True:
try:

37
pyrogram/client/methods/messages/send_document.py
View File

@ -15,7 +15,7 @@
#
# You should have received a copy of the GNU Lesser General Public License
# along with Pyrogram. If not, see <http://www.gnu.org/licenses/>.
import io
import os
from typing import Union
@ -29,7 +29,6 @@ class SendDocument(BaseClient):
def send_document(
self,
chat_id: Union[int, str],
document: Union[str, io.IOBase],
document: str,
file_ref: str = None,
thumb: str = None,
@ -56,13 +55,11 @@ class SendDocument(BaseClient):
For your personal cloud (Saved Messages) you can simply use "me" or "self".
For a contact that exists in your Telegram address book you can use his phone number (str).
document (``str`` | file-like object):
document (``str``):
File to send.
Pass a file_id as string to send a file that exists on the Telegram servers,
pass an HTTP URL as a string for Telegram to get a file from the Internet, or
pass a file path as string to upload a new file that exists on your local machine.
pass a readable file-like object with .name
file_ref (``str``, *optional*):
A valid file reference obtained by a recently fetched media message.
@ -146,35 +143,23 @@ class SendDocument(BaseClient):
file = None
try:
if isinstance(document, str):
if os.path.exists(document):
thumb = None if thumb is None else self.save_file(thumb)
file = self.save_file(document, progress=progress, progress_args=progress_args)
media = types.InputMediaUploadedDocument(
mime_type=self.guess_mime_type(document) or "application/zip",
file=file,
thumb=thumb,
attributes=[
types.DocumentAttributeFilename(file_name=file_name or os.path.basename(document))
]
)
elif document.startswith("http"):
media = types.InputMediaDocumentExternal(
url=document
)
else:
media = utils.get_input_media_from_file_id(document, file_ref, 5)
else:
if os.path.exists(document):
thumb = None if thumb is None else self.save_file(thumb)
file = self.save_file(document, progress=progress, progress_args=progress_args)
media = types.InputMediaUploadedDocument(
mime_type=self.guess_mime_type(document.name) or "application/zip",
mime_type=self.guess_mime_type(document) or "application/zip",
file=file,
thumb=thumb,
attributes=[
types.DocumentAttributeFilename(file_name=document.name)
types.DocumentAttributeFilename(file_name=file_name or os.path.basename(document))
]
)
elif document.startswith("http"):
media = types.InputMediaDocumentExternal(
url=document
)
else:
media = utils.get_input_media_from_file_id(document, file_ref, 5)
while True:
try:

29
pyrogram/client/methods/messages/send_photo.py
View File

@ -15,7 +15,7 @@
#
# You should have received a copy of the GNU Lesser General Public License
# along with Pyrogram. If not, see <http://www.gnu.org/licenses/>.
import io
import os
from typing import Union
@ -29,7 +29,6 @@ class SendPhoto(BaseClient):
def send_photo(
self,
chat_id: Union[int, str],
photo: Union[str, io.IOBase],
photo: str,
file_ref: str = None,
caption: str = "",
@ -55,12 +54,11 @@ class SendPhoto(BaseClient):
For your personal cloud (Saved Messages) you can simply use "me" or "self".
For a contact that exists in your Telegram address book you can use his phone number (str).
photo (``str`` | file-like object):
photo (``str``):
Photo to send.
Pass a file_id as string to send a photo that exists on the Telegram servers,
pass an HTTP URL as a string for Telegram to get a photo from the Internet, or
pass a file path as string to upload a new photo that exists on your local machine.
pass a readable file-like object with .name
file_ref (``str``, *optional*):
A valid file reference obtained by a recently fetched media message.
@ -139,26 +137,19 @@ class SendPhoto(BaseClient):
file = None
try:
if isinstance(photo, str):
if os.path.exists(photo):
file = self.save_file(photo, progress=progress, progress_args=progress_args)
media = types.InputMediaUploadedPhoto(
file=file,
ttl_seconds=ttl_seconds
)
elif photo.startswith("http"):
media = types.InputMediaPhotoExternal(
url=photo,
ttl_seconds=ttl_seconds
)
else:
media = utils.get_input_media_from_file_id(photo, file_ref, 2)
elif hasattr(photo, "read"):
if os.path.exists(photo):
file = self.save_file(photo, progress=progress, progress_args=progress_args)
media = types.InputMediaUploadedPhoto(
file=file,
ttl_seconds=ttl_seconds
)
elif photo.startswith("http"):
media = types.InputMediaPhotoExternal(
url=photo,
ttl_seconds=ttl_seconds
)
else:
media = utils.get_input_media_from_file_id(photo, file_ref, 2)
while True:
try:

34
pyrogram/client/methods/messages/send_sticker.py
View File

@ -15,7 +15,7 @@
#
# You should have received a copy of the GNU Lesser General Public License
# along with Pyrogram. If not, see <http://www.gnu.org/licenses/>.
import io
import os
from typing import Union
@ -29,7 +29,6 @@ class SendSticker(BaseClient):
def send_sticker(
self,
chat_id: Union[int, str],
sticker: Union[str, io.IOBase],
sticker: str,
file_ref: str = None,
disable_notification: bool = None,
@ -52,12 +51,11 @@ class SendSticker(BaseClient):
For your personal cloud (Saved Messages) you can simply use "me" or "self".
For a contact that exists in your Telegram address book you can use his phone number (str).
sticker (``str`` | file-like object):
sticker (``str``):
Sticker to send.
Pass a file_id as string to send a sticker that exists on the Telegram servers,
pass an HTTP URL as a string for Telegram to get a .webp sticker file from the Internet, or
pass a file path as string to upload a new sticker that exists on your local machine.
pass a readable file-like object with .name
file_ref (``str``, *optional*):
A valid file reference obtained by a recently fetched media message.
@ -115,31 +113,21 @@ class SendSticker(BaseClient):
file = None
try:
if isinstance(sticker, str):
if os.path.exists(sticker):
file = self.save_file(sticker, progress=progress, progress_args=progress_args)
media = types.InputMediaUploadedDocument(
mime_type=self.guess_mime_type(sticker) or "image/webp",
file=file,
attributes=[
types.DocumentAttributeFilename(file_name=os.path.basename(sticker))
]
)
elif sticker.startswith("http"):
media = types.InputMediaDocumentExternal(
url=sticker
)
else:
media = utils.get_input_media_from_file_id(sticker, file_ref, 8)
elif hasattr(sticker, "read"):
if os.path.exists(sticker):
file = self.save_file(sticker, progress=progress, progress_args=progress_args)
media = types.InputMediaUploadedDocument(
mime_type=self.guess_mime_type(sticker.name) or "image/webp",
mime_type=self.guess_mime_type(sticker) or "image/webp",
file=file,
attributes=[
types.DocumentAttributeFilename(file_name=sticker.name)
types.DocumentAttributeFilename(file_name=os.path.basename(sticker))
]
)
elif sticker.startswith("http"):
media = types.InputMediaDocumentExternal(
url=sticker
)
else:
media = utils.get_input_media_from_file_id(sticker, file_ref, 8)
while True:
try:

42
pyrogram/client/methods/messages/send_video.py
View File

@ -15,7 +15,7 @@
#
# You should have received a copy of the GNU Lesser General Public License
# along with Pyrogram. If not, see <http://www.gnu.org/licenses/>.
import io
import os
from typing import Union
@ -29,7 +29,6 @@ class SendVideo(BaseClient):
def send_video(
self,
chat_id: Union[int, str],
video: Union[str, io.IOBase],
video: str,
file_ref: str = None,
caption: str = "",
@ -60,12 +59,11 @@ class SendVideo(BaseClient):
For your personal cloud (Saved Messages) you can simply use "me" or "self".
For a contact that exists in your Telegram address book you can use his phone number (str).
video (``str`` | file-like object):
video (``str``):
Video to send.
Pass a file_id as string to send a video that exists on the Telegram servers,
pass an HTTP URL as a string for Telegram to get a video from the Internet, or
pass a file path as string to upload a new video that exists on your local machine.
pass a readable file-like object with .name
file_ref (``str``, *optional*):
A valid file reference obtained by a recently fetched media message.
@ -162,35 +160,11 @@ class SendVideo(BaseClient):
file = None
try:
if isinstance(video, str):
if os.path.exists(video):
thumb = None if thumb is None else self.save_file(thumb)
file = self.save_file(video, progress=progress, progress_args=progress_args)
media = types.InputMediaUploadedDocument(
mime_type=self.guess_mime_type(video) or "video/mp4",
file=file,
thumb=thumb,
attributes=[
types.DocumentAttributeVideo(
supports_streaming=supports_streaming or None,
duration=duration,
w=width,
h=height
),
types.DocumentAttributeFilename(file_name=file_name or os.path.basename(video))
]
)
elif video.startswith("http"):
media = types.InputMediaDocumentExternal(
url=video
)
else:
media = utils.get_input_media_from_file_id(video, file_ref, 4)
elif hasattr(video, "read"):
if os.path.exists(video):
thumb = None if thumb is None else self.save_file(thumb)
file = self.save_file(video, progress=progress, progress_args=progress_args)
media = types.InputMediaUploadedDocument(
mime_type=self.guess_mime_type(video.name) or "video/mp4",
mime_type=self.guess_mime_type(video) or "video/mp4",
file=file,
thumb=thumb,
attributes=[
@ -200,9 +174,15 @@ class SendVideo(BaseClient):
w=width,
h=height
),
types.DocumentAttributeFilename(file_name=video.name)
types.DocumentAttributeFilename(file_name=file_name or os.path.basename(video))
]
)
elif video.startswith("http"):
media = types.InputMediaDocumentExternal(
url=video
)
else:
media = utils.get_input_media_from_file_id(video, file_ref, 4)
while True:
try:

31
pyrogram/client/methods/messages/send_video_note.py
View File

@ -15,7 +15,7 @@
#
# You should have received a copy of the GNU Lesser General Public License
# along with Pyrogram. If not, see <http://www.gnu.org/licenses/>.
import io
import os
from typing import Union
@ -29,7 +29,6 @@ class SendVideoNote(BaseClient):
def send_video_note(
self,
chat_id: Union[int, str],
video_note: Union[str, io.IOBase],
video_note: str,
file_ref: str = None,
duration: int = 0,
@ -55,12 +54,11 @@ class SendVideoNote(BaseClient):
For your personal cloud (Saved Messages) you can simply use "me" or "self".
For a contact that exists in your Telegram address book you can use his phone number (str).
video_note (``str``, file-like object):
video_note (``str``):
Video note to send.
Pass a file_id as string to send a video note that exists on the Telegram servers, or
pass a file path as string to upload a new video note that exists on your local machine.
Sending video notes by a URL is currently unsupported.
pass a readable file-like object with .name
file_ref (``str``, *optional*):
A valid file reference obtained by a recently fetched media message.
@ -130,30 +128,11 @@ class SendVideoNote(BaseClient):
file = None
try:
if isinstance(video_note, str):
if os.path.exists(video_note):
thumb = None if thumb is None else self.save_file(thumb)
file = self.save_file(video_note, progress=progress, progress_args=progress_args)
media = types.InputMediaUploadedDocument(
mime_type=self.guess_mime_type(video_note) or "video/mp4",
file=file,
thumb=thumb,
attributes=[
types.DocumentAttributeVideo(
round_message=True,
duration=duration,
w=length,
h=length
)
]
)
else:
media = utils.get_input_media_from_file_id(video_note, file_ref, 13)
elif hasattr(video_note, "read"):
if os.path.exists(video_note):
thumb = None if thumb is None else self.save_file(thumb)
file = self.save_file(video_note, progress=progress, progress_args=progress_args)
media = types.InputMediaUploadedDocument(
mime_type=self.guess_mime_type(video_note.name) or "video/mp4",
mime_type=self.guess_mime_type(video_note) or "video/mp4",
file=file,
thumb=thumb,
attributes=[
@ -165,6 +144,8 @@ class SendVideoNote(BaseClient):
)
]
)
else:
media = utils.get_input_media_from_file_id(video_note, file_ref, 13)
while True:
try:

35
pyrogram/client/methods/messages/send_voice.py
View File

@ -15,7 +15,7 @@
#
# You should have received a copy of the GNU Lesser General Public License
# along with Pyrogram. If not, see <http://www.gnu.org/licenses/>.
import io
import os
from typing import Union
@ -29,7 +29,6 @@ class SendVoice(BaseClient):
def send_voice(
self,
chat_id: Union[int, str],
voice: Union[str, io.IOBase],
voice: str,
file_ref=None,
caption: str = "",
@ -55,12 +54,11 @@ class SendVoice(BaseClient):
For your personal cloud (Saved Messages) you can simply use "me" or "self".
For a contact that exists in your Telegram address book you can use his phone number (str).
voice (``str``, file-like object):
voice (``str``):
Audio file to send.
Pass a file_id as string to send an audio that exists on the Telegram servers,
pass an HTTP URL as a string for Telegram to get an audio from the Internet, or
pass a file path as string to upload a new audio that exists on your local machine.
pass a readable file-like object with .name
file_ref (``str``, *optional*):
A valid file reference obtained by a recently fetched media message.
@ -134,29 +132,10 @@ class SendVoice(BaseClient):
file = None
try:
if isinstance(voice, str):
if os.path.exists(voice):
file = self.save_file(voice, progress=progress, progress_args=progress_args)
media = types.InputMediaUploadedDocument(
mime_type=self.guess_mime_type(voice) or "audio/mpeg",
file=file,
attributes=[
types.DocumentAttributeAudio(
voice=True,
duration=duration
)
]
)
elif voice.startswith("http"):
media = types.InputMediaDocumentExternal(
url=voice
)
else:
media = utils.get_input_media_from_file_id(voice, file_ref, 3)
elif hasattr(voice, "read"):
if os.path.exists(voice):
file = self.save_file(voice, progress=progress, progress_args=progress_args)
media = types.InputMediaUploadedDocument(
mime_type=self.guess_mime_type(voice.name) or "audio/mpeg",
mime_type=self.guess_mime_type(voice) or "audio/mpeg",
file=file,
attributes=[
types.DocumentAttributeAudio(
@ -165,6 +144,12 @@ class SendVoice(BaseClient):
)
]
)
elif voice.startswith("http"):
media = types.InputMediaDocumentExternal(
url=voice
)
else:
media = utils.get_input_media_from_file_id(voice, file_ref, 3)
while True:
try:

7
pyrogram/client/types/messages_and_media/message.py
View File

@ -15,7 +15,7 @@
#
# You should have received a copy of the GNU Lesser General Public License
# along with Pyrogram. If not, see <http://www.gnu.org/licenses/>.
import io
from functools import partial
from typing import List, Match, Union
@ -2985,7 +2985,6 @@ class Message(Object, Update):
def download(
self,
file_name: str = "",
out: io.IOBase = None,
block: bool = True,
progress: callable = None,
progress_args: tuple = ()
@ -3010,9 +3009,6 @@ class Message(Object, Update):
You can also specify a path for downloading files in a custom location: paths that end with "/"
are considered directories. All non-existent folders will be created automatically.
out (``io.IOBase``, *optional*):
A custom *file-like object* to be used when downloading file. Overrides file_name
block (``bool``, *optional*):
Blocks the code execution until the file has been downloaded.
Defaults to True.
@ -3049,7 +3045,6 @@ class Message(Object, Update):
return self._client.download_media(
message=self,
file_name=file_name,
out=out,
block=block,
progress=progress,
progress_args=progress_args,