+
+
+
+
+
+ 
+
+
+
+git clone https://github.com/EvamariaTG/evamaria +# Install Packages +pip3 install -U -r requirements.txt +Edit info.py with variables as given below then run bot +python3 bot.py ++ +
add a filter in chat
+• /filters - list all the filters of a chat
+• /del - delete a specific filter in chat
+• /delall - delete the whole filters in a chat (chat owner only)"""
+ BUTTON_TXT = """Help: Buttons
+
+- Eva Maria Supports both url and alert inline buttons.
+
+NOTE:
+1. Telegram will not allows you to send buttons without any content, so content is mandatory.
+2. Eva Maria supports buttons with any telegram media type.
+3. Buttons should be properly parsed as markdown format
+
+URL buttons:
+[Button Text](buttonurl:https://t.me/EvaMariaBot)
+
+Alert buttons:
+[Button Text](buttonalert:This is an alert message)"""
+ AUTOFILTER_TXT = """Help: Auto Filter
+
+NOTE:
+1. Make me the admin of your channel if it's private.
+2. make sure that your channel does not contains camrips, porn and fake files.
+3. Forward the last message to me with quotes.
+ I'll add all the files in that channel to my db."""
+ CONNECTION_TXT = """Help: Connections
+
+- Used to connect bot to PM for managing filters
+- it helps to avoid spamming in groups.
+
+NOTE:
+1. Only admins can add a connection.
+2. Send /connect for connecting me to ur PM
+
+Commands and Usage:
+• /connect - connect a particular chat to your PM
+• /disconnect - disconnect from a chat
+• /connections - list all your connections"""
+ EXTRAMOD_TXT = """Help: Extra Modules
+
+NOTE:
+these are the extra features of Eva Maria
+
+Commands and Usage:
+• /id - get id of a specified user.
+• /info - get information about a user.
+• /imdb - get the film information from IMDb source.
+• /search - get the film information from various sources."""
+ ADMIN_TXT = """Help: Admin mods
+
+NOTE:
+This module only works for my admins
+
+Commands and Usage:
+• /logs - to get the rescent errors
+• /stats - to get status of files in db.
+• /delete - to delete a specific file from db.
+• /users - to get list of my users and ids.
+• /chats - to get list of the my chats and ids
+• /leave - to leave from a chat.
+• /disable - do disable a chat.
+• /ban - to ban a user.
+• /unban - to unban a user.
+• /channel - to get list of total connected channels
+• /broadcast - to broadcast a message to all users"""
+ STATUS_TXT = """★ 𝚃𝙾𝚃𝙰𝙻 𝙵𝙸𝙻𝙴𝚂: {}
+★ 𝚃𝙾𝚃𝙰𝙻 𝚄𝚂𝙴𝚁𝚂: {}
+★ 𝚃𝙾𝚃𝙰𝙻 𝙲𝙷𝙰𝚃𝚂: {}
+★ 𝚄𝚂𝙴𝙳 𝚂𝚃𝙾𝚁𝙰𝙶𝙴: {} 𝙼𝚒𝙱
+★ 𝙵𝚁𝙴𝙴 𝚂𝚃𝙾𝚁𝙰𝙶𝙴: {} 𝙼𝚒𝙱"""
+ LOG_TEXT_G = """#NewGroup
+Group = {}({})
+Total Members = {}
+Added By - {}
+"""
+ LOG_TEXT_P = """#NewUser
+ID - {}
+Name - {}
+"""
diff --git a/app.json b/app.json
new file mode 100644
index 0000000..a11da70
--- /dev/null
+++ b/app.json
@@ -0,0 +1,81 @@
+{
+ "name": "EvaMariaBot",
+ "description": "When you going to send file on telegram channel this bot will save that in database, So you can search that easily in inline mode",
+ "stack": "container",
+ "keywords": [
+ "telegram",
+ "auto-filter",
+ "filter",
+ "best",
+ "indian",
+ "pyrogram",
+ "media",
+ "search",
+ "channel",
+ "index",
+ "inline"
+ ],
+ "website": "https://github.com/EvamariaTG/EvaMaria",
+ "repository": "https://github.com/EvamariaTG/EvaMaria",
+ "env": {
+ "BOT_TOKEN": {
+ "description": "Your bot token.",
+ "required": true
+ },
+ "API_ID": {
+ "description": "Get this value from https://my.telegram.org",
+ "required": true
+ },
+ "API_HASH": {
+ "description": "Get this value from https://my.telegram.org",
+ "required": true
+ },
+ "CHANNELS": {
+ "description": "Username or ID of channel or group. Separate multiple IDs by space.",
+ "required": false
+ },
+ "ADMINS": {
+ "description": "Username or ID of Admin. Separate multiple Admins by space.",
+ "required": true
+ },
+ "PICS": {
+ "description": "Add some telegraph link of pictures .",
+ "required": false
+ },
+ "LOG_CHANNEL": {
+ "description": "Bot Logs,Give a channel id with -100xxxxxxx",
+ "required": true
+ },
+ "AUTH_USERS": {
+ "description": "Username or ID of users to give access of inline search. Separate multiple users by space.\nLeave it empty if you don't want to restrict bot usage.",
+ "required": false
+ },
+ "AUTH_CHANNEL": {
+ "description": "ID of channel.Make sure bot is admin in this channel. Without subscribing this channel users cannot use bot.",
+ "required": false
+ },
+ "DATABASE_URI": {
+ "description": "mongoDB URI. Get this value from https://www.mongodb.com. For more help watch this video - https://youtu.be/dsuTn4qV2GA",
+ "required": true
+ },
+ "DATABASE_NAME": {
+ "description": "Name of the database in mongoDB. For more help watch this video - https://youtu.be/dsuTn4qV2GA",
+ "required": false
+ },
+ "COLLECTION_NAME": {
+ "description": "Name of the collections. Defaults to Telegram_files. If you are using the same database, then use different collection name for each bot",
+ "value": "Telegram_files",
+ "required": false
+ }
+ },
+ "addons": [],
+ "buildpacks": [{
+ "url": "heroku/python"
+ }],
+ "formation": {
+ "worker": {
+ "quantity": 1,
+ "size": "free"
+ }
+ }
+}
diff --git a/bot.py b/bot.py
new file mode 100644
index 0000000..8bb7a23
--- /dev/null
+++ b/bot.py
@@ -0,0 +1,91 @@
+import logging
+import logging.config
+
+# Get logging configurations
+logging.config.fileConfig('logging.conf')
+logging.getLogger().setLevel(logging.INFO)
+logging.getLogger("pyrogram").setLevel(logging.ERROR)
+logging.getLogger("imdbpy").setLevel(logging.ERROR)
+
+from pyrogram import Client, __version__
+from pyrogram.raw.all import layer
+from database.ia_filterdb import Media
+from database.users_chats_db import db
+from info import SESSION, API_ID, API_HASH, BOT_TOKEN, LOG_STR
+from utils import temp
+from typing import Union, Optional, AsyncGenerator
+from pyrogram import types
+
+class Bot(Client):
+
+ def __init__(self):
+ super().__init__(
+ name=SESSION,
+ api_id=API_ID,
+ api_hash=API_HASH,
+ bot_token=BOT_TOKEN,
+ workers=50,
+ plugins={"root": "plugins"},
+ sleep_threshold=5,
+ )
+
+ async def start(self):
+ b_users, b_chats = await db.get_banned()
+ temp.BANNED_USERS = b_users
+ temp.BANNED_CHATS = b_chats
+ await super().start()
+ await Media.ensure_indexes()
+ me = await self.get_me()
+ temp.ME = me.id
+ temp.U_NAME = me.username
+ temp.B_NAME = me.first_name
+ self.username = '@' + me.username
+ logging.info(f"{me.first_name} with for Pyrogram v{__version__} (Layer {layer}) started on {me.username}.")
+ logging.info(LOG_STR)
+
+ async def stop(self, *args):
+ await super().stop()
+ logging.info("Bot stopped. Bye.")
+
+ async def iter_messages(
+ self,
+ chat_id: Union[int, str],
+ limit: int,
+ offset: int = 0,
+ ) -> Optional[AsyncGenerator["types.Message", None]]:
+ """Iterate through a chat sequentially.
+ This convenience method does the same as repeatedly calling :meth:`~pyrogram.Client.get_messages` in a loop, thus saving
+ you from the hassle of setting up boilerplate code. It is useful for getting the whole chat messages with a
+ single call.
+ 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).
+
+ limit (``int``):
+ Identifier of the last message to be returned.
+
+ offset (``int``, *optional*):
+ Identifier of the first message to be returned.
+ Defaults to 0.
+ Returns:
+ ``Generator``: A generator yielding :obj:`~pyrogram.types.Message` objects.
+ Example:
+ .. code-block:: python
+ for message in app.iter_messages("pyrogram", 1, 15000):
+ print(message.text)
+ """
+ current = offset
+ while True:
+ new_diff = min(200, limit - current)
+ if new_diff <= 0:
+ return
+ messages = await self.get_messages(chat_id, list(range(current, current+new_diff+1)))
+ for message in messages:
+ yield message
+ current += 1
+
+
+app = Bot()
+app.run()
diff --git a/heroku.yml b/heroku.yml
new file mode 100644
index 0000000..375fc05
--- /dev/null
+++ b/heroku.yml
@@ -0,0 +1,3 @@
+build:
+ docker:
+ worker: Dockerfile
\ No newline at end of file
diff --git a/info.py b/info.py
new file mode 100644
index 0000000..2f61bc4
--- /dev/null
+++ b/info.py
@@ -0,0 +1,65 @@
+import re
+from os import environ
+
+id_pattern = re.compile(r'^.\d+$')
+def is_enabled(value, default):
+ if value.lower() in ["true", "yes", "1", "enable", "y"]:
+ return True
+ elif value.lower() in ["false", "no", "0", "disable", "n"]:
+ return False
+ else:
+ return default
+
+# Bot information
+SESSION = environ.get('SESSION', 'Media_search')
+API_ID = int(environ['API_ID'])
+API_HASH = environ['API_HASH']
+BOT_TOKEN = environ['BOT_TOKEN']
+
+# Bot settings
+CACHE_TIME = int(environ.get('CACHE_TIME', 300))
+USE_CAPTION_FILTER = bool(environ.get('USE_CAPTION_FILTER', False))
+PICS = (environ.get('PICS', 'https://telegra.ph/file/7e56d907542396289fee4.jpg https://telegra.ph/file/9aa8dd372f4739fe02d85.jpg https://telegra.ph/file/adffc5ce502f5578e2806.jpg https://telegra.ph/file/6937b60bc2617597b92fd.jpg https://telegra.ph/file/09a7abaab340143f9c7e7.jpg https://telegra.ph/file/5a82c4a59bd04d415af1c.jpg https://telegra.ph/file/323986d3bd9c4c1b3cb26.jpg https://telegra.ph/file/b8a82dcb89fb296f92ca0.jpg https://telegra.ph/file/31adab039a85ed88e22b0.jpg https://telegra.ph/file/c0e0f4c3ed53ac8438f34.jpg https://telegra.ph/file/eede835fb3c37e07c9cee.jpg https://telegra.ph/file/e17d2d068f71a9867d554.jpg https://telegra.ph/file/8fb1ae7d995e8735a7c25.jpg https://telegra.ph/file/8fed19586b4aa019ec215.jpg https://telegra.ph/file/8e6c923abd6139083e1de.jpg https://telegra.ph/file/0049d801d29e83d68b001.jpg')).split()
+
+# Admins, Channels & Users
+ADMINS = [int(admin) if id_pattern.search(admin) else admin for admin in environ.get('ADMINS', '').split()]
+CHANNELS = [int(ch) if id_pattern.search(ch) else ch for ch in environ.get('CHANNELS', '0').split()]
+auth_users = [int(user) if id_pattern.search(user) else user for user in environ.get('AUTH_USERS', '').split()]
+AUTH_USERS = (auth_users + ADMINS) if auth_users else []
+auth_channel = environ.get('AUTH_CHANNEL')
+auth_grp = environ.get('AUTH_GROUP')
+AUTH_CHANNEL = int(auth_channel) if auth_channel and id_pattern.search(auth_channel) else None
+AUTH_GROUPS = [int(ch) for ch in auth_grp.split()] if auth_grp else None
+
+# MongoDB information
+DATABASE_URI = environ.get('DATABASE_URI', "")
+DATABASE_NAME = environ.get('DATABASE_NAME', "Rajappan")
+COLLECTION_NAME = environ.get('COLLECTION_NAME', 'Telegram_files')
+
+# Others
+LOG_CHANNEL = int(environ.get('LOG_CHANNEL', 0))
+SUPPORT_CHAT = environ.get('SUPPORT_CHAT', 'TeamEvamaria')
+P_TTI_SHOW_OFF = is_enabled((environ.get('P_TTI_SHOW_OFF', "False")), False)
+IMDB = is_enabled((environ.get('IMDB', "True")), True)
+SINGLE_BUTTON = is_enabled((environ.get('SINGLE_BUTTON', "False")), False)
+CUSTOM_FILE_CAPTION = environ.get("CUSTOM_FILE_CAPTION", None)
+BATCH_FILE_CAPTION = environ.get("BATCH_FILE_CAPTION", CUSTOM_FILE_CAPTION)
+IMDB_TEMPLATE = environ.get("IMDB_TEMPLATE", "Query: {query} \nIMDb Data:\n\n🏷 Title: {title}\n🎭 Genres: {genres}\n📆 Year: {year}\n🌟 Rating: {rating} / 10")
+LONG_IMDB_DESCRIPTION = is_enabled(environ.get("LONG_IMDB_DESCRIPTION", "False"), False)
+SPELL_CHECK_REPLY = is_enabled(environ.get("SPELL_CHECK_REPLY", "True"), True)
+MAX_LIST_ELM = environ.get("MAX_LIST_ELM", None)
+INDEX_REQ_CHANNEL = int(environ.get('INDEX_REQ_CHANNEL', LOG_CHANNEL))
+FILE_STORE_CHANNEL = [int(ch) for ch in (environ.get('FILE_STORE_CHANNEL', '')).split()]
+MELCOW_NEW_USERS = is_enabled((environ.get('MELCOW_NEW_USERS', "True")), True)
+PROTECT_CONTENT = is_enabled((environ.get('PROTECT_CONTENT', "False")), False)
+PUBLIC_FILE_STORE = is_enabled((environ.get('PUBLIC_FILE_STORE', "True")), True)
+
+LOG_STR = "Current Cusomized Configurations are:-\n"
+LOG_STR += ("IMDB Results are enabled, Bot will be showing imdb details for you queries.\n" if IMDB else "IMBD Results are disabled.\n")
+LOG_STR += ("P_TTI_SHOW_OFF found , Users will be redirected to send /start to Bot PM instead of sending file file directly\n" if P_TTI_SHOW_OFF else "P_TTI_SHOW_OFF is disabled files will be send in PM, instead of sending start.\n")
+LOG_STR += ("SINGLE_BUTTON is Found, filename and files size will be shown in a single button instead of two separate buttons\n" if SINGLE_BUTTON else "SINGLE_BUTTON is disabled , filename and file_sixe will be shown as different buttons\n")
+LOG_STR += (f"CUSTOM_FILE_CAPTION enabled with value {CUSTOM_FILE_CAPTION}, your files will be send along with this customized caption.\n" if CUSTOM_FILE_CAPTION else "No CUSTOM_FILE_CAPTION Found, Default captions of file will be used.\n")
+LOG_STR += ("Long IMDB storyline enabled." if LONG_IMDB_DESCRIPTION else "LONG_IMDB_DESCRIPTION is disabled , Plot will be shorter.\n")
+LOG_STR += ("Spell Check Mode Is Enabled, bot will be suggesting related movies if movie not found\n" if SPELL_CHECK_REPLY else "SPELL_CHECK_REPLY Mode disabled\n")
+LOG_STR += (f"MAX_LIST_ELM Found, long list will be shortened to first {MAX_LIST_ELM} elements\n" if MAX_LIST_ELM else "Full List of casts and crew will be shown in imdb template, restrict them by adding a value to MAX_LIST_ELM\n")
+LOG_STR += f"Your current IMDB template is {IMDB_TEMPLATE}"
diff --git a/logging.conf b/logging.conf
new file mode 100644
index 0000000..371455a
--- /dev/null
+++ b/logging.conf
@@ -0,0 +1,32 @@
+[loggers]
+keys=root
+
+[handlers]
+keys=consoleHandler,fileHandler
+
+[formatters]
+keys=consoleFormatter,fileFormatter
+
+[logger_root]
+level=DEBUG
+handlers=consoleHandler,fileHandler
+
+[handler_consoleHandler]
+class=StreamHandler
+level=INFO
+formatter=consoleFormatter
+args=(sys.stdout,)
+
+[handler_fileHandler]
+class=FileHandler
+level=ERROR
+formatter=fileFormatter
+args=('TelegramBot.log','w',)
+
+[formatter_consoleFormatter]
+format=%(asctime)s - %(lineno)d - %(name)s - %(module)s - %(levelname)s - %(message)s
+datefmt=%I:%M:%S %p
+
+[formatter_fileFormatter]
+format=[%(asctime)s:%(name)s:%(lineno)d:%(levelname)s] %(message)s
+datefmt=%m/%d/%Y %I:%M:%S %p
\ No newline at end of file
diff --git a/requirements.txt b/requirements.txt
new file mode 100644
index 0000000..368e55c
--- /dev/null
+++ b/requirements.txt
@@ -0,0 +1,9 @@
+pyrogram>=2.0.30
+tgcrypto
+pymongo[srv]==3.12.3
+motor==2.5.1
+marshmallow==3.14.1
+umongo==3.0.1
+requests
+bs4
+imdbpy==2021.4.18
diff --git a/runtime.txt b/runtime.txt
new file mode 100644
index 0000000..3e4835c
--- /dev/null
+++ b/runtime.txt
@@ -0,0 +1 @@
+python-3.8.7
diff --git a/sample_info.py b/sample_info.py
new file mode 100644
index 0000000..273c541
--- /dev/null
+++ b/sample_info.py
@@ -0,0 +1,24 @@
+# Bot information
+SESSION = 'Media_search'
+USER_SESSION = 'User_Bot'
+API_ID = 12345
+API_HASH = '0123456789abcdef0123456789abcdef'
+BOT_TOKEN = '123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11'
+USERBOT_STRING_SESSION = ''
+
+# Bot settings
+CACHE_TIME = 300
+USE_CAPTION_FILTER = False
+
+# Admins, Channels & Users
+ADMINS = [12345789, 'admin123', 98765432]
+CHANNELS = [-10012345678, -100987654321, 'channelusername']
+AUTH_USERS = []
+AUTH_CHANNEL = None
+
+# MongoDB information
+DATABASE_URI = "mongodb://[username:password@]host1[:port1][,...hostN[:portN]][/[defaultauthdb]?retryWrites=true&w=majority"
+DATABASE_NAME = 'Telegram'
+COLLECTION_NAME = 'channel_files' # If you are using the same database, then use different collection name for each bot
+
+
diff --git a/start.sh b/start.sh
new file mode 100644
index 0000000..fb1333b
--- /dev/null
+++ b/start.sh
@@ -0,0 +1,12 @@
+if [ -z $UPSTREAM_REPO ]
+then
+ echo "Cloning main Repository"
+ git clone https://github.com/EvamariaTG/EvaMaria.git /EvaMaria
+else
+ echo "Cloning Custom Repo from $UPSTREAM_REPO "
+ git clone $UPSTREAM_REPO /EvaMaria
+fi
+cd /EvaMaria
+pip3 install -U -r requirements.txt
+echo "Starting Bot...."
+python3 bot.py
\ No newline at end of file
diff --git a/utils.py b/utils.py
new file mode 100644
index 0000000..496cf43
--- /dev/null
+++ b/utils.py
@@ -0,0 +1,377 @@
+import logging
+from pyrogram.errors import InputUserDeactivated, UserNotParticipant, FloodWait, UserIsBlocked, PeerIdInvalid
+from info import AUTH_CHANNEL, LONG_IMDB_DESCRIPTION, MAX_LIST_ELM
+from imdb import IMDb
+import asyncio
+from pyrogram.types import Message, InlineKeyboardButton
+from pyrogram import enums
+from typing import Union
+import re
+import os
+from datetime import datetime
+from typing import List
+from database.users_chats_db import db
+from bs4 import BeautifulSoup
+import requests
+
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.INFO)
+
+BTN_URL_REGEX = re.compile(
+ r"(\[([^\[]+?)\]\((buttonurl|buttonalert):(?:/{0,2})(.+?)(:same)?\))"
+)
+
+imdb = IMDb()
+
+BANNED = {}
+SMART_OPEN = '“'
+SMART_CLOSE = '”'
+START_CHAR = ('\'', '"', SMART_OPEN)
+
+# temp db for banned
+class temp(object):
+ BANNED_USERS = []
+ BANNED_CHATS = []
+ ME = None
+ CURRENT=int(os.environ.get("SKIP", 2))
+ CANCEL = False
+ MELCOW = {}
+ U_NAME = None
+ B_NAME = None
+ SETTINGS = {}
+
+async def is_subscribed(bot, query):
+ try:
+ user = await bot.get_chat_member(AUTH_CHANNEL, query.from_user.id)
+ except UserNotParticipant:
+ pass
+ except Exception as e:
+ logger.exception(e)
+ else:
+ if user.status != 'kicked':
+ return True
+
+ return False
+
+async def get_poster(query, bulk=False, id=False, file=None):
+ if not id:
+ # https://t.me/GetTGLink/4183
+ query = (query.strip()).lower()
+ title = query
+ year = re.findall(r'[1-2]\d{3}$', query, re.IGNORECASE)
+ if year:
+ year = list_to_str(year[:1])
+ title = (query.replace(year, "")).strip()
+ elif file is not None:
+ year = re.findall(r'[1-2]\d{3}', file, re.IGNORECASE)
+ if year:
+ year = list_to_str(year[:1])
+ else:
+ year = None
+ movieid = imdb.search_movie(title.lower(), results=10)
+ if not movieid:
+ return None
+ if year:
+ filtered=list(filter(lambda k: str(k.get('year')) == str(year), movieid))
+ if not filtered:
+ filtered = movieid
+ else:
+ filtered = movieid
+ movieid=list(filter(lambda k: k.get('kind') in ['movie', 'tv series'], filtered))
+ if not movieid:
+ movieid = filtered
+ if bulk:
+ return movieid
+ movieid = movieid[0].movieID
+ else:
+ movieid = query
+ movie = imdb.get_movie(movieid)
+ if movie.get("original air date"):
+ date = movie["original air date"]
+ elif movie.get("year"):
+ date = movie.get("year")
+ else:
+ date = "N/A"
+ plot = ""
+ if not LONG_IMDB_DESCRIPTION:
+ plot = movie.get('plot')
+ if plot and len(plot) > 0:
+ plot = plot[0]
+ else:
+ plot = movie.get('plot outline')
+ if plot and len(plot) > 800:
+ plot = plot[0:800] + "..."
+
+ return {
+ 'title': movie.get('title'),
+ 'votes': movie.get('votes'),
+ "aka": list_to_str(movie.get("akas")),
+ "seasons": movie.get("number of seasons"),
+ "box_office": movie.get('box office'),
+ 'localized_title': movie.get('localized title'),
+ 'kind': movie.get("kind"),
+ "imdb_id": f"tt{movie.get('imdbID')}",
+ "cast": list_to_str(movie.get("cast")),
+ "runtime": list_to_str(movie.get("runtimes")),
+ "countries": list_to_str(movie.get("countries")),
+ "certificates": list_to_str(movie.get("certificates")),
+ "languages": list_to_str(movie.get("languages")),
+ "director": list_to_str(movie.get("director")),
+ "writer":list_to_str(movie.get("writer")),
+ "producer":list_to_str(movie.get("producer")),
+ "composer":list_to_str(movie.get("composer")) ,
+ "cinematographer":list_to_str(movie.get("cinematographer")),
+ "music_team": list_to_str(movie.get("music department")),
+ "distributors": list_to_str(movie.get("distributors")),
+ 'release_date': date,
+ 'year': movie.get('year'),
+ 'genres': list_to_str(movie.get("genres")),
+ 'poster': movie.get('full-size cover url'),
+ 'plot': plot,
+ 'rating': str(movie.get("rating")),
+ 'url':f'https://www.imdb.com/title/tt{movieid}'
+ }
+# https://github.com/odysseusmax/animated-lamp/blob/2ef4730eb2b5f0596ed6d03e7b05243d93e3415b/bot/utils/broadcast.py#L37
+
+async def broadcast_messages(user_id, message):
+ try:
+ await message.copy(chat_id=user_id)
+ return True, "Success"
+ except FloodWait as e:
+ await asyncio.sleep(e.x)
+ return await broadcast_messages(user_id, message)
+ except InputUserDeactivated:
+ await db.delete_user(int(user_id))
+ logging.info(f"{user_id}-Removed from Database, since deleted account.")
+ return False, "Deleted"
+ except UserIsBlocked:
+ logging.info(f"{user_id} -Blocked the bot.")
+ return False, "Blocked"
+ except PeerIdInvalid:
+ await db.delete_user(int(user_id))
+ logging.info(f"{user_id} - PeerIdInvalid")
+ return False, "Error"
+ except Exception as e:
+ return False, "Error"
+
+async def search_gagala(text):
+ usr_agent = {
+ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) '
+ 'Chrome/61.0.3163.100 Safari/537.36'
+ }
+ text = text.replace(" ", '+')
+ url = f'https://www.google.com/search?q={text}'
+ response = requests.get(url, headers=usr_agent)
+ response.raise_for_status()
+ soup = BeautifulSoup(response.text, 'html.parser')
+ titles = soup.find_all( 'h3' )
+ return [title.getText() for title in titles]
+
+
+async def get_settings(group_id):
+ settings = temp.SETTINGS.get(group_id)
+ if not settings:
+ settings = await db.get_settings(group_id)
+ temp.SETTINGS[group_id] = settings
+ return settings
+
+async def save_group_settings(group_id, key, value):
+ current = await get_settings(group_id)
+ current[key] = value
+ temp.SETTINGS[group_id] = current
+ await db.update_settings(group_id, current)
+
+def get_size(size):
+ """Get size in readable format"""
+
+ units = ["Bytes", "KB", "MB", "GB", "TB", "PB", "EB"]
+ size = float(size)
+ i = 0
+ while size >= 1024.0 and i < len(units):
+ i += 1
+ size /= 1024.0
+ return "%.2f %s" % (size, units[i])
+
+def split_list(l, n):
+ for i in range(0, len(l), n):
+ yield l[i:i + n]
+
+def get_file_id(msg: Message):
+ if msg.media:
+ for message_type in (
+ "photo",
+ "animation",
+ "audio",
+ "document",
+ "video",
+ "video_note",
+ "voice",
+ "sticker"
+ ):
+ obj = getattr(msg, message_type)
+ if obj:
+ setattr(obj, "message_type", message_type)
+ return obj
+
+def extract_user(message: Message) -> Union[int, str]:
+ """extracts the user from a message"""
+ # https://github.com/SpEcHiDe/PyroGramBot/blob/f30e2cca12002121bad1982f68cd0ff9814ce027/pyrobot/helper_functions/extract_user.py#L7
+ user_id = None
+ user_first_name = None
+ if message.reply_to_message:
+ user_id = message.reply_to_message.from_user.id
+ user_first_name = message.reply_to_message.from_user.first_name
+
+ elif len(message.command) > 1:
+ if (
+ len(message.entities) > 1 and
+ message.entities[1].type == enums.MessageEntityType.TEXT_MENTION
+ ):
+
+ required_entity = message.entities[1]
+ user_id = required_entity.user.id
+ user_first_name = required_entity.user.first_name
+ else:
+ user_id = message.command[1]
+ # don't want to make a request -_-
+ user_first_name = user_id
+ try:
+ user_id = int(user_id)
+ except ValueError:
+ pass
+ else:
+ user_id = message.from_user.id
+ user_first_name = message.from_user.first_name
+ return (user_id, user_first_name)
+
+def list_to_str(k):
+ if not k:
+ return "N/A"
+ elif len(k) == 1:
+ return str(k[0])
+ elif MAX_LIST_ELM:
+ k = k[:int(MAX_LIST_ELM)]
+ return ' '.join(f'{elem}, ' for elem in k)
+ else:
+ return ' '.join(f'{elem}, ' for elem in k)
+
+def last_online(from_user):
+ time = ""
+ if from_user.is_bot:
+ time += "🤖 Bot :("
+ elif from_user.status == enums.UserStatus.RECENTLY:
+ time += "Recently"
+ elif from_user.status == enums.UserStatus.LAST_WEEK:
+ time += "Within the last week"
+ elif from_user.status == enums.UserStatus.LAST_MONTH:
+ time += "Within the last month"
+ elif from_user.status == enums.UserStatus.LONG_AGO:
+ time += "A long time ago :("
+ elif from_user.status == enums.UserStatus.ONLINE:
+ time += "Currently Online"
+ elif from_user.status == enums.UserStatus.OFFLINE:
+ time += from_user.last_online_date.strftime("%a, %d %b %Y, %H:%M:%S")
+ return time
+
+
+def split_quotes(text: str) -> List:
+ if not any(text.startswith(char) for char in START_CHAR):
+ return text.split(None, 1)
+ counter = 1 # ignore first char -> is some kind of quote
+ while counter < len(text):
+ if text[counter] == "\\":
+ counter += 1
+ elif text[counter] == text[0] or (text[0] == SMART_OPEN and text[counter] == SMART_CLOSE):
+ break
+ counter += 1
+ else:
+ return text.split(None, 1)
+
+ # 1 to avoid starting quote, and counter is exclusive so avoids ending
+ key = remove_escapes(text[1:counter].strip())
+ # index will be in range, or `else` would have been executed and returned
+ rest = text[counter + 1:].strip()
+ if not key:
+ key = text[0] + text[0]
+ return list(filter(None, [key, rest]))
+
+def parser(text, keyword):
+ if "buttonalert" in text:
+ text = (text.replace("\n", "\\n").replace("\t", "\\t"))
+ buttons = []
+ note_data = ""
+ prev = 0
+ i = 0
+ alerts = []
+ for match in BTN_URL_REGEX.finditer(text):
+ # Check if btnurl is escaped
+ n_escapes = 0
+ to_check = match.start(1) - 1
+ while to_check > 0 and text[to_check] == "\\":
+ n_escapes += 1
+ to_check -= 1
+
+ # if even, not escaped -> create button
+ if n_escapes % 2 == 0:
+ note_data += text[prev:match.start(1)]
+ prev = match.end(1)
+ if match.group(3) == "buttonalert":
+ # create a thruple with button label, url, and newline status
+ if bool(match.group(5)) and buttons:
+ buttons[-1].append(InlineKeyboardButton(
+ text=match.group(2),
+ callback_data=f"alertmessage:{i}:{keyword}"
+ ))
+ else:
+ buttons.append([InlineKeyboardButton(
+ text=match.group(2),
+ callback_data=f"alertmessage:{i}:{keyword}"
+ )])
+ i += 1
+ alerts.append(match.group(4))
+ elif bool(match.group(5)) and buttons:
+ buttons[-1].append(InlineKeyboardButton(
+ text=match.group(2),
+ url=match.group(4).replace(" ", "")
+ ))
+ else:
+ buttons.append([InlineKeyboardButton(
+ text=match.group(2),
+ url=match.group(4).replace(" ", "")
+ )])
+
+ else:
+ note_data += text[prev:to_check]
+ prev = match.start(1) - 1
+ else:
+ note_data += text[prev:]
+
+ try:
+ return note_data, buttons, alerts
+ except:
+ return note_data, buttons, None
+
+def remove_escapes(text: str) -> str:
+ res = ""
+ is_escaped = False
+ for counter in range(len(text)):
+ if is_escaped:
+ res += text[counter]
+ is_escaped = False
+ elif text[counter] == "\\":
+ is_escaped = True
+ else:
+ res += text[counter]
+ return res
+
+
+def humanbytes(size):
+ if not size:
+ return ""
+ power = 2**10
+ n = 0
+ Dic_powerN = {0: ' ', 1: 'Ki', 2: 'Mi', 3: 'Gi', 4: 'Ti'}
+ while size > power:
+ size /= power
+ n += 1
+ return str(round(size, 2)) + " " + Dic_powerN[n] + 'B'