forked from pyrogram/pyrogram
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathutils.py
More file actions
137 lines (108 loc) 路 4.04 KB
/
Copy pathutils.py
File metadata and controls
137 lines (108 loc) 路 4.04 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
# 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 struct
from base64 import b64decode, b64encode
from typing import Union
from . import BaseClient
from ...api import types
def decode(s: str) -> bytes:
s = b64decode(s + "=" * (-len(s) % 4), "-_")
r = b""
assert s[-1] == 2
i = 0
while i < len(s) - 1:
if s[i] != 0:
r += bytes([s[i]])
else:
r += b"\x00" * s[i + 1]
i += 1
i += 1
return r
def encode(s: bytes) -> str:
r = b""
n = 0
for i in s + bytes([2]):
if i == 0:
n += 1
else:
if n:
r += b"\x00" + bytes([n])
n = 0
r += bytes([i])
return b64encode(r, b"-_").decode().rstrip("=")
def get_peer_id(input_peer) -> int:
return (
input_peer.user_id if isinstance(input_peer, types.InputPeerUser)
else -input_peer.chat_id if isinstance(input_peer, types.InputPeerChat)
else int("-100" + str(input_peer.channel_id))
)
def get_input_peer(peer_id: int, access_hash: int):
return (
types.InputPeerUser(user_id=peer_id, access_hash=access_hash) if peer_id > 0
else types.InputPeerChannel(channel_id=int(str(peer_id)[4:]), access_hash=access_hash)
if (str(peer_id).startswith("-100") and access_hash)
else types.InputPeerChat(chat_id=-peer_id)
)
def get_offset_date(dialogs):
for m in reversed(dialogs.messages):
if isinstance(m, types.MessageEmpty):
continue
else:
return m.date
else:
return 0
def get_input_media_from_file_id(
file_id_str: str,
expected_media_type: int = None
) -> Union[types.InputMediaPhoto, types.InputMediaDocument]:
try:
decoded = decode(file_id_str)
except Exception:
raise ValueError("Failed to decode file_id: {}".format(file_id_str))
else:
media_type = decoded[0]
if expected_media_type is not None:
if media_type != expected_media_type:
media_type_str = BaseClient.MEDIA_TYPE_ID.get(media_type, None)
expected_media_type_str = BaseClient.MEDIA_TYPE_ID.get(expected_media_type, None)
raise ValueError(
'Expected: "{}", got "{}" file_id instead'.format(expected_media_type_str, media_type_str)
)
if media_type in (0, 1, 14):
raise ValueError("This file_id can only be used for download: {}".format(file_id_str))
if media_type == 2:
unpacked = struct.unpack("<iiqqc", decoded)
dc_id, file_id, access_hash, thumb_size = unpacked[1:]
return types.InputMediaPhoto(
id=types.InputPhoto(
id=file_id,
access_hash=access_hash,
file_reference=b""
)
)
if media_type in (3, 4, 5, 8, 9, 10, 13):
unpacked = struct.unpack("<iiqq", decoded)
dc_id, file_id, access_hash = unpacked[1:]
return types.InputMediaDocument(
id=types.InputDocument(
id=file_id,
access_hash=access_hash,
file_reference=b""
)
)
raise ValueError("Unknown media type: {}".format(file_id_str))