forust
3c383db9a7
Deploy to Server / deploy (push) Has been cancelled
userbot now lives at ssh://git@gitssh.forust.xyz:2221/forust/userbot.git and is included in homelab as a submodule at userbot/
520 lines
19 KiB
Python
520 lines
19 KiB
Python
# Moon-Userbot - telegram userbot
|
|
# Copyright (C) 2020-present Moon Userbot Organization
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU General Public License as published by
|
|
# the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU General Public License for more details.
|
|
|
|
# You should have received a copy of the GNU General Public License
|
|
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
|
|
|
import base64
|
|
import os
|
|
from io import BytesIO
|
|
|
|
import aiofiles
|
|
import aiohttp
|
|
import requests
|
|
from PIL import Image, ImageEnhance
|
|
from pyrogram import Client, enums, filters
|
|
from pyrogram.errors import MediaCaptionTooLong, MessageTooLong
|
|
from pyrogram.types import InputMediaPhoto, Message
|
|
from utils.misc import modules_help, prefix
|
|
from utils.scripts import format_exc
|
|
|
|
url = 'https://api.safone.co'
|
|
|
|
headers = {
|
|
'Accept-Language': 'en-US,en;q=0.9',
|
|
'Connection': 'keep-alive',
|
|
'DNT': '1',
|
|
'Referer': 'https://api.safone.dev/docs',
|
|
'Sec-Fetch-Dest': 'empty',
|
|
'Sec-Fetch-Mode': 'cors',
|
|
'Sec-Fetch-Site': 'same-origin',
|
|
'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36',
|
|
'accept': 'application/json',
|
|
'sec-ch-ua': '"Chromium";v="124", "Google Chrome";v="124", "Not-A.Brand";v="99"',
|
|
'sec-ch-ua-mobile': '?0',
|
|
'sec-ch-ua-platform': '"Linux"',
|
|
}
|
|
|
|
|
|
async def make_carbon(code):
|
|
url = 'https://carbonara.solopov.dev/api/cook'
|
|
|
|
async with aiohttp.ClientSession() as session, session.post(url, json={'code': code}) as resp:
|
|
image_data = await resp.read()
|
|
|
|
carbon_image = Image.open(BytesIO(image_data))
|
|
|
|
enhancer = ImageEnhance.Brightness(carbon_image)
|
|
bright_image = enhancer.enhance(1.0)
|
|
|
|
output_image = BytesIO()
|
|
bright_image.save(output_image, format='PNG', quality=95)
|
|
output_image.name = 'carbon.png'
|
|
|
|
return output_image
|
|
|
|
|
|
async def telegraph(title, user_name, content):
|
|
formatted_content = '<br>'.join(content.split('\n'))
|
|
formatted_content = '<p>' + formatted_content + '</p>'
|
|
|
|
data = {'title': title, 'content': formatted_content, 'author_name': user_name}
|
|
|
|
response = requests.post(url=f'{url}/telegraph/text', headers=headers, json=data, timeout=5)
|
|
|
|
result = response.json()
|
|
|
|
return result['url']
|
|
|
|
|
|
async def voice_characters():
|
|
response = requests.get(url=f'{url}/speech/characters', headers=headers, timeout=5)
|
|
|
|
result = response.json()
|
|
|
|
return ', '.join(result['characters'])
|
|
|
|
|
|
async def make_rayso(code: str, title: str, theme: str):
|
|
data = {
|
|
'code': code,
|
|
'title': title,
|
|
'theme': theme,
|
|
'padding': 64,
|
|
'language': 'auto',
|
|
'darkMode': False,
|
|
}
|
|
response = requests.post(f'{url}/rayso', data=data, headers=headers, timeout=10)
|
|
if response.status_code != 200:
|
|
return None
|
|
result = response.json()
|
|
try:
|
|
if result['error'] is not None:
|
|
return None
|
|
except KeyError:
|
|
pass
|
|
image_data = result['image']
|
|
file_name = 'rayso.png'
|
|
with open(file_name, 'wb') as f:
|
|
f.write(base64.b64decode(image_data))
|
|
return file_name
|
|
|
|
|
|
@Client.on_message(filters.command('asq', prefix) & filters.me)
|
|
async def asq(_, message: Message):
|
|
if len(message.command) > 1:
|
|
query = message.text.split(maxsplit=1)[1]
|
|
else:
|
|
await message.edit_text('Query not provided!')
|
|
return
|
|
await message.edit_text('Processing...')
|
|
response = requests.get(url=f'{url}/asq?query={query}', headers=headers, timeout=5)
|
|
if response.status_code != 200:
|
|
await message.edit_text('Something went wrong!')
|
|
return
|
|
|
|
result = response.json()
|
|
|
|
ans = result['answer']
|
|
await message.edit_text(f'Q. {query}\n A. {ans}', parse_mode=enums.ParseMode.MARKDOWN)
|
|
|
|
|
|
@Client.on_message(filters.command('sgemini', prefix) & filters.me)
|
|
async def sgemini(_, message: Message):
|
|
if len(message.command) > 1:
|
|
prompt = message.text.split(maxsplit=1)[1]
|
|
else:
|
|
await message.edit_text('prompt not provided!')
|
|
return
|
|
await message.edit_text('Processing...')
|
|
response = requests.get(url=f'{url}/bard?query={prompt}', headers=headers, timeout=10)
|
|
if response.status_code != 200:
|
|
await message.edit_text('Something went wrong!')
|
|
return
|
|
|
|
result = response.json()
|
|
|
|
ans = result['message']
|
|
await message.edit_text(f'Prompt: {prompt}\n Ans: {ans}', parse_mode=enums.ParseMode.MARKDOWN)
|
|
|
|
|
|
@Client.on_message(filters.command('app', prefix) & filters.me)
|
|
async def app(client: Client, message: Message):
|
|
try:
|
|
chat_id = message.chat.id
|
|
await message.edit_text('Processing...')
|
|
if len(message.command) > 1:
|
|
query = message.text.split(maxsplit=1)[1]
|
|
else:
|
|
message.edit_text("What should i search? You didn't provided me with any value to search")
|
|
|
|
response = requests.get(url=f'{url}/apps?query={query}&limit=1', headers=headers, timeout=5)
|
|
if response.status_code != 200:
|
|
await message.edit_text('Something went wrong')
|
|
return
|
|
|
|
result = response.json()
|
|
|
|
try:
|
|
coverImage_url = result['results'][0]['icon']
|
|
coverImage = requests.get(url=coverImage_url, timeout=10).content
|
|
async with aiofiles.open('coverImage.jpg', mode='wb') as f:
|
|
await f.write(coverImage)
|
|
|
|
except Exception:
|
|
coverImage = None
|
|
|
|
description = result['results'][0]['description']
|
|
developer = result['results'][0]['developer']
|
|
IsFree = result['results'][0]['free']
|
|
genre = result['results'][0]['genre']
|
|
package_name = result['results'][0]['id']
|
|
title = result['results'][0]['title']
|
|
price = result['results'][0]['price']
|
|
link = result['results'][0]['link']
|
|
rating = result['results'][0]['rating']
|
|
|
|
await message.delete()
|
|
await client.send_media_group(
|
|
chat_id,
|
|
[
|
|
InputMediaPhoto(
|
|
'coverImage.jpg',
|
|
caption=f'<b>Title:</b> <code>{title}</code>\n<b>Rating:</b> <code>{rating}</code>\n<b>IsFree:</b> <code>{IsFree}</code>\n<b>Price:</b> <code>{price}</code>\n<b>Package Name:</b> <code>{package_name}</code>\n<b>Genres:</b> <code>{genre}</code>\n<b>Developer:</b> <code>{developer}\n<b>Description:</b> {description}\n<b>Link:</b> {link}',
|
|
)
|
|
],
|
|
)
|
|
|
|
except MediaCaptionTooLong:
|
|
description = description[:850]
|
|
await message.delete()
|
|
await client.send_media_group(
|
|
chat_id,
|
|
[
|
|
InputMediaPhoto(
|
|
'coverImage.jpg',
|
|
caption=f'<b>Title:</b> <code>{title}</code>\n<b>Rating:</b> <code>{rating}</code>\n<b>IsFree:</b> <code>{IsFree}</code>\n<b>Price:</b> <code>{price}</code>\n<b>Package Name:</b> <code>{package_name}</code>\n<b>Genres:</b> <code>{genre}</code>\n<b>Developer:</b> <code>{developer}\n<b>Description:</b> {description}\n<b>Link:</b> {link}',
|
|
)
|
|
],
|
|
)
|
|
except Exception as e:
|
|
await message.edit_text(format_exc(e))
|
|
finally:
|
|
if os.path.exists('coverImage.jpg'):
|
|
os.remove('coverImage.jpg')
|
|
|
|
|
|
@Client.on_message(filters.command('tsearch', prefix) & filters.me)
|
|
async def tsearch(client: Client, message: Message):
|
|
try:
|
|
chat_id = message.chat.id
|
|
limit = 10
|
|
await message.edit_text('Processing...')
|
|
if len(message.command) > 1:
|
|
query = message.text.split(maxsplit=1)[1]
|
|
else:
|
|
message.edit_text("What should i search? You didn't provided me with any value to search")
|
|
|
|
response = requests.get(url=f'{url}/torrent?query={query}&limit={limit}', headers=headers, timeout=10)
|
|
if response.status_code != 200:
|
|
await message.edit_text('Something went wrong')
|
|
return
|
|
|
|
result = response.json()
|
|
|
|
coverImage_url = result['results'][0]['thumbnail']
|
|
description = result['results'][0]['description']
|
|
genre = result['results'][0]['genre']
|
|
category = result['results'][0]['category']
|
|
title = result['results'][0]['name']
|
|
link = result['results'][0]['magnetLink']
|
|
link_result = await telegraph(title=title, user_name=message.from_user.first_name, content=link)
|
|
language = result['results'][0]['language']
|
|
size = result['results'][0]['size']
|
|
|
|
results = []
|
|
|
|
for i in range(min(limit, len(result['results']))):
|
|
descriptions = result['results'][i]['description']
|
|
genres = result['results'][i]['genre']
|
|
categorys = result['results'][i]['category']
|
|
titles = result['results'][i]['name']
|
|
links = result['results'][i]['magnetLink']
|
|
languages = result['results'][i]['language']
|
|
sizes = result['results'][i]['size']
|
|
|
|
r = f'<b>Title:</b> <code>{titles}</code>\n<b>Category:</b> <code>{categorys}</code>\n<b>Language:</b> <code>{languages}</code>\n<b>Size:</b> <code>{sizes}</code>\n<b>Genres:</b> <code>{genres}</code>\n<b>Description:</b> {descriptions}\n<b>Magnet Link:</b> <code>{links}</code><br>'
|
|
results.append(r)
|
|
|
|
all_results_content = '<br>'.join(results)
|
|
|
|
link_results = await telegraph(
|
|
title='Search Results',
|
|
user_name=message.from_user.first_name,
|
|
content=all_results_content,
|
|
)
|
|
|
|
if coverImage_url is not None:
|
|
coverImage = requests.get(url=coverImage_url, timeout=10).content
|
|
async with aiofiles.open('coverImage.jpg', mode='wb') as f:
|
|
await f.write(coverImage)
|
|
|
|
await message.delete()
|
|
await client.send_media_group(
|
|
chat_id,
|
|
[
|
|
InputMediaPhoto(
|
|
'coverImage.jpg',
|
|
caption=f"<b>Title:</b> <code>{title}</code>\n<b>Category:</b> <code>{category}</code>\n<b>Language:</b> <code>{language}</code>\n<b>Size:</b> <code>{size}</code>\n<b>Genres:</b> <code>{genre}</code>\n<b>Description:</b> {description}\n<b>Magnet Link:</b> <a href='{link_result}'>Click Here</a>\n<b>More Results:</b> <a href='{link_results}'>Click Here</a>",
|
|
)
|
|
],
|
|
)
|
|
else:
|
|
await message.edit_text(
|
|
f"<b>Title:</b> <code>{title}</code>\n<b>Category:</b> <code>{category}</code>\n<b>Language:</b> <code>{language}</code>\n<b>Size:</b> <code>{size}</code>\n<b>Genres:</b> <code>{genre}</code>\n<b>Description:</b> {description}\n<b>Magnet Link:</b> <a href='{link_result}'>Click Here</a>\n<b>More Results:</b> <a href='{link_results}'>Click Here</a>",
|
|
disable_web_page_preview=True,
|
|
)
|
|
|
|
except MediaCaptionTooLong:
|
|
description = description[:850]
|
|
await message.delete()
|
|
await client.send_media_group(
|
|
chat_id,
|
|
[
|
|
InputMediaPhoto(
|
|
'coverImage.jpg',
|
|
caption=f"<b>Title:</b> <code>{title}</code>\n<b>Category:</b> <code>{category}</code>\n<b>Language:</b> <code>{language}</code>\n<b>Size:</b> <code>{size}</code>\n<b>Genres:</b> <code>{genre}</code>\n<b>Description:</b> {description}\n<b>Magnet Link:</b> <a href='{link_result}'>Click Here</a>",
|
|
)
|
|
],
|
|
)
|
|
|
|
except MessageTooLong:
|
|
description = description[:150]
|
|
await message.edit_text(
|
|
f"<b>Title:</b> <code>{title}</code>\n<b>Category:</b> <code>{category}</code>\n<b>Language:</b> <code>{language}</code>\n<b>Size:</b> <code>{size}</code>\n<b>Genres:</b> <code>{genre}</code>\n<b>Description:</b> {description}\n<b>Magnet Link:</b> <a href='{link_result}'>Click Here</a>",
|
|
disable_web_page_preview=True,
|
|
)
|
|
|
|
except Exception as e:
|
|
await message.edit_text(format_exc(e))
|
|
finally:
|
|
if os.path.exists('coverImage.jpg'):
|
|
os.remove('coverImage.jpg')
|
|
|
|
|
|
@Client.on_message(filters.command('stts', prefix) & filters.me)
|
|
async def tts(client: Client, message: Message):
|
|
characters = await voice_characters()
|
|
await message.edit_text('<code>Please Wait...</code>')
|
|
try:
|
|
if len(message.command) > 2:
|
|
character, prompt = message.text.split(maxsplit=2)[1:]
|
|
if character not in characters:
|
|
await message.edit_text(
|
|
f'<b>Usage: </b><code>{prefix}tts [character]* [text/reply to text]*</code>\n <b>Available Characters:</b> <blockquote>{characters}</blockquote>'
|
|
)
|
|
return
|
|
|
|
elif message.reply_to_message and len(message.command) > 1:
|
|
character = message.text.split(maxsplit=1)[1]
|
|
if character in characters:
|
|
prompt = message.reply_to_message.text
|
|
else:
|
|
await message.edit_text(
|
|
f'<b>Usage: </b><code>{prefix}tts [character]* [text/reply to text]*</code>\n <b>Available Characters:</b> <blockquote>{characters}</blockquote>'
|
|
)
|
|
return
|
|
|
|
else:
|
|
await message.edit_text(
|
|
f'<b>Usage: </b><code>{prefix}tts [character]* [text/reply to text]*</code>\n <b>Available Characters:</b> <blockquote>{characters}</blockquote>'
|
|
)
|
|
return
|
|
|
|
data = {'text': prompt, 'character': character}
|
|
response = requests.post(url=f'{url}/speech', headers=headers, json=data, timeout=10)
|
|
if response.status_code != 200:
|
|
await message.edit_text('Something went wrong')
|
|
return
|
|
|
|
result = response.json()
|
|
audio_data = result['audio']
|
|
audio_data = base64.b64decode(audio_data)
|
|
async with aiofiles.open(f'{prompt}.mp3', mode='wb') as f:
|
|
await f.write(audio_data)
|
|
|
|
await message.delete()
|
|
await client.send_audio(
|
|
chat_id=message.chat.id,
|
|
audio=f'{prompt}.mp3',
|
|
caption=f'<b>Characters:</b> <code>{character}</code>\n<b>Prompt:</b> <code>{prompt}</code>',
|
|
)
|
|
if os.path.exists(f'{prompt}.mp3'):
|
|
os.remove(f'{prompt}.mp3')
|
|
|
|
except KeyError:
|
|
try:
|
|
error = result['error']
|
|
await message.edit_text(error)
|
|
except KeyError:
|
|
await message.edit_text(
|
|
f'<b>Usage: </b><code>{prefix}tts [character]* [text/reply to text]*</code>\n <b>Available Characters:</b> <blockquote>{characters}</blockquote>'
|
|
)
|
|
except Exception as e:
|
|
await message.edit_text(format_exc(e))
|
|
|
|
|
|
@Client.on_message(filters.command(['carbonnowsh', 'carboon', 'carbon', 'cboon'], prefix) & filters.me)
|
|
async def carbon(client: Client, message: Message):
|
|
if message.reply_to_message:
|
|
text = message.reply_to_message.text
|
|
message_id = message.reply_to_message.id
|
|
elif len(message.command) > 1:
|
|
message_id = None
|
|
text = message.text.split(maxsplit=1)[1]
|
|
else:
|
|
await message.edit_text('Query not provided!')
|
|
return
|
|
await message.edit_text('Processing...')
|
|
|
|
image_file = await make_carbon(text)
|
|
|
|
await message.delete()
|
|
try:
|
|
await client.send_photo(
|
|
chat_id=message.chat.id,
|
|
photo=image_file,
|
|
caption=f'<b>Text:</b> <code>{text}</code>',
|
|
reply_to_message_id=message_id,
|
|
)
|
|
except MediaCaptionTooLong:
|
|
cap = text[:850]
|
|
await client.send_photo(
|
|
chat_id=message.chat.id,
|
|
photo=image_file,
|
|
caption=f'<b>Text:</b> <code>{cap}</code>',
|
|
reply_to_message_id=message_id,
|
|
)
|
|
except Exception as e:
|
|
await message.edit_text(format_exc(e))
|
|
if os.path.exists('carbon.png'):
|
|
os.remove('carbon.png')
|
|
|
|
|
|
@Client.on_message(filters.command('ccgen', prefix) & filters.me)
|
|
async def ccgen(_, message: Message):
|
|
if len(message.command) > 1:
|
|
bins = message.text.split(maxsplit=1)[1]
|
|
else:
|
|
await message.edit_text('Code not provided!')
|
|
return
|
|
await message.edit_text('Processing...')
|
|
response = requests.get(url=f'{url}/ccgen?bins={bins}', headers=headers, timeout=10)
|
|
if response.status_code != 200:
|
|
await message.edit_text('Something went wrong')
|
|
return
|
|
|
|
result = response.json()
|
|
|
|
cards = result['results'][0]['cards']
|
|
cards_str = '\n'.join([f'"{card}"' for card in cards])
|
|
bins = result['results'][0]['bin']
|
|
await message.edit_text(
|
|
f'Bins: <code>{bins}</code>\nTotal: <code>{len(cards)}</code>\nCards: \n<code>{cards_str}</code>'
|
|
)
|
|
|
|
|
|
@Client.on_message(filters.command('rayso', prefix) & filters.me)
|
|
async def rayso(client: Client, message: Message):
|
|
title = 'Untitled'
|
|
themes = [
|
|
'vercel',
|
|
'supabase',
|
|
'tailwind',
|
|
'clerk',
|
|
'mintlify',
|
|
'prisma',
|
|
'bitmap',
|
|
'noir',
|
|
'ice',
|
|
'sand',
|
|
'forest',
|
|
'mono',
|
|
'breeze',
|
|
'candy',
|
|
'crimson',
|
|
'falcon',
|
|
'meadow',
|
|
'midnight',
|
|
'raindrop',
|
|
'sunset',
|
|
]
|
|
if message.reply_to_message:
|
|
text = message.reply_to_message.text
|
|
message_id = message.reply_to_message.id
|
|
if 2 <= len(message.command) <= 3:
|
|
title = message.text.split(maxsplit=2)[1]
|
|
theme = message.text.split(maxsplit=2)[2].lower()
|
|
if theme not in themes:
|
|
theme = 'breeze'
|
|
elif len(message.command) > 1:
|
|
message_id = message.id
|
|
title = message.text.split(maxsplit=3)[1]
|
|
theme = message.text.split(maxsplit=3)[2]
|
|
if theme not in themes:
|
|
theme = 'breeze'
|
|
text = message.text.split(maxsplit=3)[3]
|
|
else:
|
|
await message.edit_text('Query not provided!')
|
|
return
|
|
await message.edit_text('Processing...')
|
|
|
|
image_file = await make_rayso(text, title, theme)
|
|
|
|
if image_file is None:
|
|
await message.edit_text('Something went wrong')
|
|
return
|
|
try:
|
|
await client.send_photo(
|
|
chat_id=message.chat.id,
|
|
photo=image_file,
|
|
caption=f'<b>Text:</b> <code>{text}</code>',
|
|
reply_to_message_id=message_id,
|
|
)
|
|
await message.delete()
|
|
except MediaCaptionTooLong:
|
|
cap = text[:850]
|
|
await client.send_photo(
|
|
chat_id=message.chat.id,
|
|
photo=image_file,
|
|
caption=f'<b>Text:</b> <code>{cap}</code>',
|
|
reply_to_message_id=message_id,
|
|
)
|
|
await message.delete()
|
|
except Exception as e:
|
|
await message.edit_text(format_exc(e))
|
|
if os.path.exists(image_file):
|
|
os.remove(image_file)
|
|
|
|
|
|
modules_help['safone'] = {
|
|
'asq [query]*': 'Asq',
|
|
'app [query]*': 'Search for an app on Play Store',
|
|
'tsearch [query]*': 'Search Torrent',
|
|
'tts [character]* [text/reply to text]*': 'Convert Text to Speech',
|
|
'sgemini [prompt]*': 'Gemini Model through safone api',
|
|
'carbon [code/file/reply]': 'Create beautiful image with your code',
|
|
'ccgen [bins]*': 'Generate credit cards',
|
|
'rayso [title]* [theme]* [text/reply to text]*': 'Create beautiful image with your text',
|
|
}
|