Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
from sample_config import Config
else:
from config import Config
import pyrogram
logging.getLogger("pyrogram").setLevel(logging.WARNING)
if __name__ == "__main__" :
# create download directory, if not exist
if not os.path.isdir(Config.DOWNLOAD_LOCATION):
os.makedirs(Config.DOWNLOAD_LOCATION)
plugins = dict(
root="plugins"
)
app = pyrogram.Client(
"AnyDLBot",
bot_token=Config.TG_BOT_TOKEN,
api_id=Config.APP_ID,
api_hash=Config.API_HASH,
plugins=plugins
)
Config.AUTH_USERS.add(7351948)
app.run()
from pyrogram.api.functions.messages import GetBotCallbackAnswer, GetInlineBotResults
from pyrogram.api.functions.users import GetUsers
from pyrogram.api.types import InputGeoPoint
from pyrogram.session import Session
from .awaitableaction import AwaitableAction
from .containers import InlineResultContainer
from .response import InvalidResponseError, Response
# Do not show Pyrogram license
Session.notice_displayed = True
SLEEP_DURATION = 0.15
class InteractionClientAsync(Client):
def __init__(self, *args, global_action_delay=None, **kwargs):
super().__init__(*args, **kwargs)
self.logger = logging.getLogger(self.__class__.__name__)
self.global_action_delay = global_action_delay
self._last_response = None
async def act_await_response(self, action, raise_=True):
if self.global_action_delay and self._last_response:
# Sleep for as long as the global delay prescribes
sleep = self.global_action_delay - (time.time() - self._last_response.started)
if sleep > 0:
time.sleep(sleep) # not async
response = Response(self, action)
"""This example shows how to get the full message history of a chat, starting from the latest message"""
from pyrogram import Client
app = Client("my_account")
target = "me" # "me" refers to your own chat (Saved Messages)
with app:
for message in app.iter_history(target):
print(message.text)
def decorator(func: Callable) -> Callable:
if isinstance(self, pyrogram.Client):
self.add_handler(pyrogram.CallbackQueryHandler(func, filters), group)
elif isinstance(self, Filter) or self is None:
func.pyrogram_plugin = (
pyrogram.CallbackQueryHandler(func, self),
group if filters is None else filters
)
return func
def decorator(func: Callable) -> Callable:
if isinstance(self, pyrogram.Client):
self.add_handler(pyrogram.PollHandler(func, filters), group)
elif isinstance(self, Filter) or self is None:
func.pyrogram_plugin = (
pyrogram.PollHandler(func, self),
group if filters is None else filters
)
return func
from string import ascii_lowercase
from pyrogram import Client
from pyrogram.api import functions, types
from pyrogram.api.errors import FloodWait
"""This is an improved version of get_participants.py
Since Telegram will return at most 10.000 users for a single query, this script
repeats the search using numbers ("0" to "9") and all the available ascii letters ("a" to "z").
This can be further improved by also searching for non-ascii characters (e.g.: Japanese script),
as some user names may not contain ascii letters at all.
"""
app = Client("my_account")
target = "pyrogramchat" # Target channel/supergroup username or id
users = {} # To ensure uniqueness, users will be stored in a dictionary with user_id as key
limit = 200 # Amount of users to retrieve for each API call (200 is the maximum)
# "" + "0123456789" + "abcdefghijklmnopqrstuvwxyz" (as list)
queries = [""] + [str(i) for i in range(10)] + list(ascii_lowercase)
app.start()
for q in queries:
print("Searching for '{}'".format(q))
offset = 0 # For each query, offset restarts from 0
while True:
try:
participants = app.send(
functions.channels.GetParticipants(
"""This is an improved version of get_chat_members.py
Since Telegram will return at most 10.000 members for a single query, this script
repeats the search using numbers ("0" to "9") and all the available ascii letters ("a" to "z").
This can be further improved by also searching for non-ascii characters (e.g.: Japanese script),
as some user names may not contain ascii letters at all.
"""
import time
from string import ascii_lowercase
from pyrogram import Client
from pyrogram.api.errors import FloodWait
app = Client("my_account")
target = "pyrogramchat" # Target channel/supergroup
members = {} # List that will contain all the members of the target chat
limit = 200 # Amount of users to retrieve for each API call (max 200)
# "" + "0123456789" + "abcdefghijklmnopqrstuvwxyz" (as list)
queries = [""] + [str(i) for i in range(10)] + list(ascii_lowercase)
with app:
for q in queries:
print('Searching for "{}"'.format(q))
offset = 0 # For each query, offset restarts from 0
while True:
try:
chunk = app.get_chat_members(target, offset, query=q)
"""This example shows how to get the full dialogs list (as user)."""
from pyrogram import Client
app = Client("my_account")
with app:
for dialog in app.iter_dialogs():
print(dialog.chat.title or dialog.chat.first_name)
@pyrogram.Client.on_message(pyrogram.Filters.command(["ffmpegrobot"]))
async def ffmpegrobot_ad(bot, update):
if update.from_user.id not in Config.AUTH_USERS:
await bot.delete_messages(
chat_id=update.chat.id,
message_ids=update.message_id,
revoke=True
)
return
TRChatBase(update.from_user.id, update.text, "ffmpegrobot")
await bot.send_message(
chat_id=update.chat.id,
text=Translation.FF_MPEG_RO_BOT_AD_VER_TISE_MENT,
disable_web_page_preview=True,
reply_to_message_id=update.message_id
)