# 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 os import requests import aiofiles import base64 from pyrogram import Client, enums, filters from pyrogram.types import Message, InputMediaPhoto from pyrogram.errors import MediaCaptionTooLong, MessageTooLong from utils.misc import prefix, modules_help 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: async with 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", }