__version__ = (1, 0, 0)
import asyncio
#
# @@@@@@ @@@@@@ @@@@@@@ @@@@@@@ @@@@@@ @@@@@@@@@@ @@@@@@ @@@@@@@ @@@ @@@ @@@ @@@@@@@@ @@@@@@
# @@@@@@@@ @@@@@@@ @@@@@@@ @@@@@@@@ @@@@@@@@ @@@@@@@@@@@ @@@@@@@@ @@@@@@@@ @@@ @@@ @@@ @@@@@@@@ @@@@@@@
# @@! @@@ !@@ @@! @@! @@@ @@! @@@ @@! @@! @@! @@! @@@ @@! @@@ @@! @@@ @@! @@! !@@
# !@! @!@ !@! !@! !@! @!@ !@! @!@ !@! !@! !@! !@! @!@ !@! @!@ !@! @!@ !@! !@! !@!
# @!@!@!@! !!@@!! @!! @!@!!@! @!@ !@! @!! !!@ @!@ @!@ !@! @!@ !@! @!@ !@! @!! @!!!:! !!@@!!
# !!!@!!!! !!@!!! !!! !!@!@! !@! !!! !@! ! !@! !@! !!! !@! !!! !@! !!! !!! !!!!!: !!@!!!
# !!: !!! !:! !!: !!: :!! !!: !!! !!: !!: !!: !!! !!: !!! !!: !!! !!: !!: !:!
# :!: !:! !:! :!: :!: !:! :!: !:! :!: :!: :!: !:! :!: !:! :!: !:! :!: :!: !:!
# :: ::: :::: :: :: :: ::: ::::: :: ::: :: ::::: :: :::: :: ::::: :: :: :::: :: :::: :::: ::
# : : : :: : : : : : : : : : : : : : : :: : : : : : : :: : : : :: :: :: : :
#
# © Copyright 2024
#
# https://t.me/Den4ikSuperOstryyPer4ik
# and
# https://t.me/ToXicUse
#
# 🔒 Licensed under the GNU AGPLv3
# https://www.gnu.org/licenses/agpl-3.0.html
#
# meta banner: https://raw.githubusercontent.com/Den4ikSuperOstryyPer4ik/Astro-modules/main/Banners/Convertio.jpg
# meta developer: @AstroModules
import base64
import os
import requests
from .. import loader, utils
async def download_media(message):
media = None
msg = None
if message.media:
media, msg = message.media, message
elif (reply := await message.get_reply_message()) and reply.media:
media, msg = reply.media, reply
if not (media and msg):
return False
return await msg.download_media()
def generate_api_key():
return requests.post("https://den4iksop.org/convertio/getApiKey").json()
class ConvertioClient:
# Original from https://github.com/PetitPotiron/python-convertio/blob/main/convertio/client.py#L14
# This is edited version
def __init__(self, token: str) -> None:
self.token = token
def upload(self, fp: str, output_format: str):
"""Converts the file found in the path provided.
Args:
fp (str): The file's path
output_format (str): The file format you want the file to be converted to
"""
r1 = requests.post(
"http://api.convertio.co/convert",
json={
"apikey": self.token,
"input": "upload",
"outputformat": output_format,
}
).json()
if r1.get("error", None):
raise ValueError(r1["error"])
id = r1['data']['id']
with open(fp, 'rb') as file:
content = file.read()
r2 = requests.put(f'http://api.convertio.co/convert/{id}/{fp.split("/")[-1]}', data=content).json()
if r2.get("error", None):
raise ValueError(r2["error"])
return id
def check_conversion(self, id: str):
"""Checks the step of a conversion
Args:
id (str) : The id of the conversion
Returns:
Conversion: The status of the conversion
"""
r = requests.get(f"https://api.convertio.co/convert/{id}/status").json()
if r.get("error", None):
raise ValueError(r["error"])
return {
"code": r['code'],
"status": r['status'],
"id": r['data']['id'],
"step": r['data']['step'],
"step_percent": r['data']['step_percent'],
"minutes": r['data']['minutes'],
}
def download(self, id: str, fp: str) -> None:
"""Writes the file content to a path."""
r = requests.get(f"https://api.convertio.co/convert/{id}/dl").json()
if r.get("error", None):
raise ValueError(r["error"])
if r['status'] == 'error':
raise ValueError(r['error'])
content = base64.b64decode(r["data"]["content"])
with open(fp, "wb") as file:
file.write(content)
return fp
class ConvertioMod(loader.Module):
"""Convert file with api from https://convertio.co"""
strings = {
"name": "Convertio",
"converting": "☺️ Wait, converting...",
"getting_api_key": "#️⃣ Wait, getting random API key...",
"error": "💔 Something went wrong...\n💭 Contact @AstroModsChat",
"no_file": "⁉️ Where is the file?",
"no_args": "❓ Where is the file format you want to convert the file to?",
"renewed": "🍀 New API-key generated and saved!",
"downloading_file": "⏳ Downloading your file...",
"converted": "⭐️ File successfully converted from {} to {}",
"api_error": "🚨 API: {}",
"uploading": "☺️ File converted. Uploading to Telegram...",
}
strings_ru = {
"_cls_doc": "Конвертирует файл с помощью https://convertio.co",
"converting": "☺️ Подождите, идёт конвертация...",
"getting_api_key": "#️⃣ Подождите, получаю рандомный API KEY...",
"error": "💔 Что-то пошло не так...\n💭 Обратитесь в @AstroModsChat",
"no_file": "⁉️ Где файл?",
"no_args": "❓ Где формат файла, в который вы хотите преобразовать файл?",
"renewed": "🍀 Новый API-ключ сохранен!",
"downloading_file": "⏳ Загружаю ваш файл...",
"converted": "⭐️ Файл успешно конвертирован с {} в {}",
"uploading": "☺️ Файл конвертирован. Загружаю в Telegram..."
}
def __init__(self):
self.config = loader.ModuleConfig(
loader.ConfigValue(
"API_KEY",
None,
lambda: "API key from https://developers.convertio.co (Getting automatically)",
validator=loader.validators.String()
)
)
async def convert_file(self, message, m) -> str:
args: str = utils.get_args_raw(message)
if not args:
raise ValueError(self.strings("no_args"))
m = await utils.answer(m, self.strings("downloading_file"))
file_name = await download_media(message)
if not file_name:
raise ValueError(self.strings("no_file"))
m = await utils.answer(m, self.strings("converting"))
client = ConvertioClient(token=self.config["API_KEY"])
conversion_id = await utils.run_sync(client.upload, file_name, args)
os.remove(file_name)
while (await utils.run_sync(client.check_conversion, conversion_id))["step"] != 'finish':
await asyncio.sleep(1)
m = await utils.answer(m, self.strings("uploading"))
output_file_path = os.path.abspath("".join([*file_name.split(".")[:-1], ".", args]))
await utils.run_sync(client.download, conversion_id, output_file_path)
return output_file_path, file_name.split(".")[-1], output_file_path.split(".")[-1], m
@loader.command(alias="renewconv")
async def renewconvertio(self, message):
"""Renew convertio api key"""
await utils.answer(message, self.strings("getting_api_key"))
response = await utils.run_sync(generate_api_key)
if response["status"] == "Failed":
await utils.answer(self.strings("error"))
return
self.config["API_KEY"] = response["api_key"]
await utils.answer(message, self.strings("renewed"))
@loader.command(
ru_doc="<выходной формат> | Пример: png",
alias="conv",
)
async def convert(self, message):
"""