import asyncio import os import shlex import heroku3 from functools import wraps from pyrogram.types import Message from typing import Tuple from html_telegraph_poster import TelegraphPoster from bot import HEROKU_API_KEY, HEROKU_APP_NAME # Implement by https://github.com/jusidama18 # Setting Message def get_text(message: Message) -> [None, str]: """Extract Text From Commands""" text_to_return = message.text if message.text is None: return None if " " in text_to_return: try: return message.text.split(None, 1)[1] except IndexError: return None else: return None # Preparing For Setting Config # Implement by https://github.com/jusidama18 and Based on this https://github.com/DevsExpo/FridayUserbot/blob/master/plugins/heroku_helpers.py heroku_client = None if HEROKU_API_KEY: heroku_client = heroku3.from_key(HEROKU_API_KEY) def check_heroku(func): @wraps(func) async def heroku_cli(client, message): heroku_app = None if not heroku_client: await message.reply_text("`Please Add HEROKU_API_KEY Key For This To Function To Work!`", parse_mode="markdown") elif not HEROKU_APP_NAME: await message.reply_text("`Please Add HEROKU_APP_NAME For This To Function To Work!`", parse_mode="markdown") if HEROKU_APP_NAME and heroku_client: try: heroku_app = heroku_client.app(HEROKU_APP_NAME) except: await message.reply_text(message, "`Heroku Api Key And App Name Doesn't Match!`", parse_mode="markdown") if heroku_app: await func(client, message, heroku_app) return heroku_cli # Preparing For Update Bot # Implement by https://github.com/jusidama18 and Based on this https://github.com/DevsExpo/FridayUserbot/blob/master/plugins/updater.py def fetch_heroku_git_url(api_key, app_name): if not api_key: return None if not app_name: return None heroku = heroku3.from_key(api_key) try: heroku_applications = heroku.apps() except: return None heroku_app = None for app in heroku_applications: if app.name == app_name: heroku_app = app break if not heroku_app: return None return heroku_app.git_url.replace("https://", "https://api:" + api_key + "@") HEROKU_URL = fetch_heroku_git_url(HEROKU_API_KEY, HEROKU_APP_NAME) def post_to_telegraph(a_title: str, content: str) -> str: """ Create a Telegram Post using HTML Content """ post_client = TelegraphPoster(use_api=True) auth_name = "slam-mirrorbot" post_client.create_api_token(auth_name) post_page = post_client.post( title=a_title, author=auth_name, author_url="https://github.com/breakdowns/slam-mirrorbot", text=content, ) return post_page["url"] async def runcmd(cmd: str) -> Tuple[str, str, int, int]: """ run command in terminal """ args = shlex.split(cmd) process = await asyncio.create_subprocess_exec( *args, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) stdout, stderr = await process.communicate() return ( stdout.decode("utf-8", "replace").strip(), stderr.decode("utf-8", "replace").strip(), process.returncode, process.pid, ) # Solves ValueError: No closing quotation by removing ' or " in file name def safe_filename(path_): if path_ is None: return safename = path_.replace("'", "").replace('"', "") if safename != path_: os.rename(path_, safename) return safename