# 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 . import base64 import os import aiofiles import requests 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 = '
'.join(content.split('\n')) formatted_content = '

' + formatted_content + '

' 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) 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) 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).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'Title: {title}\nRating: {rating}\nIsFree: {IsFree}\nPrice: {price}\nPackage Name: {package_name}\nGenres: {genre}\nDeveloper: {developer}\nDescription: {description}\nLink: {link}', ) ], ) except MediaCaptionTooLong: description = description[:850] await message.delete() await client.send_media_group( chat_id, [ InputMediaPhoto( 'coverImage.jpg', caption=f'Title: {title}\nRating: {rating}\nIsFree: {IsFree}\nPrice: {price}\nPackage Name: {package_name}\nGenres: {genre}\nDeveloper: {developer}\nDescription: {description}\nLink: {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) 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'Title: {titles}\nCategory: {categorys}\nLanguage: {languages}\nSize: {sizes}\nGenres: {genres}\nDescription: {descriptions}\nMagnet Link: {links}
' results.append(r) all_results_content = '
'.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).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"Title: {title}\nCategory: {category}\nLanguage: {language}\nSize: {size}\nGenres: {genre}\nDescription: {description}\nMagnet Link: Click Here\nMore Results: Click Here", ) ], ) else: await message.edit_text( f"Title: {title}\nCategory: {category}\nLanguage: {language}\nSize: {size}\nGenres: {genre}\nDescription: {description}\nMagnet Link: Click Here\nMore Results: Click Here", 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"Title: {title}\nCategory: {category}\nLanguage: {language}\nSize: {size}\nGenres: {genre}\nDescription: {description}\nMagnet Link: Click Here", ) ], ) except MessageTooLong: description = description[:150] await message.edit_text( f"Title: {title}\nCategory: {category}\nLanguage: {language}\nSize: {size}\nGenres: {genre}\nDescription: {description}\nMagnet Link: Click Here", 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('Please Wait...') try: if len(message.command) > 2: character, prompt = message.text.split(maxsplit=2)[1:] if character not in characters: await message.edit_text( f'Usage: {prefix}tts [character]* [text/reply to text]*\n Available Characters:
{characters}
' ) 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'Usage: {prefix}tts [character]* [text/reply to text]*\n Available Characters:
{characters}
' ) return else: await message.edit_text( f'Usage: {prefix}tts [character]* [text/reply to text]*\n Available Characters:
{characters}
' ) return data = {'text': prompt, 'character': character} response = requests.post(url=f'{url}/speech', headers=headers, json=data) 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'Characters: {character}\nPrompt: {prompt}', ) 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'Usage: {prefix}tts [character]* [text/reply to text]*\n Available Characters:
{characters}
' ) 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'Text: {text}', 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'Text: {cap}', 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) 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: {bins}\nTotal: {len(cards)}\nCards: \n{cards_str}' ) @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'Text: {text}', 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'Text: {cap}', 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', }