114 lines
3.7 KiB
Python
114 lines
3.7 KiB
Python
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
|