Upload 6 files

Akeno/utils/convert.py

@@ -0,0 +1,78 @@
+import os
+import time
+
+from PIL import Image
+from pyrogram.types import Message
+
+from Akeno.utils.tools import runcmd
+
+
+async def convert_to_gif(file: str, is_video: bool = False) -> str:
+    resultFileName = f"gif_{round(time.time())}.mp4"
+    if is_video:
+        cmd = f"ffmpeg -i '{file}' -c copy '{resultFileName}'"
+    else:
+        cmd = f"lottie_convert.py '{file}' '{resultFileName}'"
+    await runcmd(cmd)
+    return resultFileName
+
+async def tgs_to_png(file: str) -> str:
+    resultFileName = f"png_{round(time.time())}.png"
+    cmd = f"lottie_convert.py '{file}' '{resultFileName}'"
+    await runcmd(cmd)
+    return resultFileName
+
+async def image_to_sticker(file: str, max_size: tuple = (512, 512)) -> tuple[bool, str]:
+    try:
+        with Image.open(file) as img:
+            original_width, original_height = img.size
+
+            new_width = min(original_width, max_size[0])
+            new_height = min(original_height, max_size[1])
+
+            if original_width > max_size[0] or original_height > max_size[1]:
+                img = img.resize((new_width, new_height), Image.LANCZOS)
+
+            file_name = f"sticker_{int(time.time())}.png"
+            img.save(file_name, "PNG")
+
+        return True, file_name
+
+    except Exception as e:
+        return False, str(e)
+
+async def video_to_png(
+    file: str, duration: float, output: str = None
+) -> tuple[str, bool]:
+    resultFileName = output or f"{os.path.basename(file)}.png"
+    cut_at = duration // 2
+    cmd = f"ffmpeg -ss {cut_at} -i '{file}' -vframes 1 '{resultFileName}'"
+    _, err, _, _ = await runcmd(cmd)
+    if err:
+        return err, False
+    return resultFileName, True
+
+async def video_to_sticker(file: Message) -> tuple[str, bool]:
+    try:
+        if file.animation:
+            width, height = file.animation.width, file.animation.height
+        elif file.video:
+            width, height = file.video.width, file.video.height
+        else:
+            return "Unsupported media type.", False
+        file_path = await file.download("downloads")
+        output_path = os.path.join("downloads", "videoSticker.webm")
+        if height > width:
+            scale_params = f"scale=-1:512"
+        else:
+            scale_params = f"scale=512:-1"
+        cmd = (
+            f"ffmpeg -i {file_path} "
+            f"-vf fps=30,{scale_params} -t 3 -c:v libvpx-vp9 -b:v 256k -an -pix_fmt yuv420p -auto-alt-ref 0 -loop 0 "
+            f"-f webm {output_path}"
+        )
+        await runcmd(cmd)
+        os.remove(file_path)
+        return output_path, True
+    except Exception as e:
+        return f"Error during conversion: {e}", False

Akeno/utils/driver.py

@@ -0,0 +1,459 @@
+import datetime
+import json
+import os
+import random
+import re
+import time
+import urllib.parse
+from urllib.parse import quote_plus
+
+import httpx
+import requests
+from pytz import country_names, country_timezones, timezone
+from selenium import webdriver
+from selenium.webdriver.chrome.options import Options
+from selenium.webdriver.chrome.service import Service
+from selenium.webdriver.common.by import By
+from selenium.webdriver.support.expected_conditions import presence_of_element_located
+from selenium.webdriver.support.wait import WebDriverWait
+
+from Akeno.utils.database import db
+from Akeno.utils.formatter import format_text
+from config import *
+
+
+class ChromeDriver:
+    def __init__(self) -> None:
+        self.carbon_theme = [
+            "3024-night",
+            "a11y-dark",
+            "blackboard",
+            "base16-dark",
+            "base16-light",
+            "cobalt",
+            "duotone-dark",
+            "hopscotch",
+            "lucario",
+            "material",
+            "monokai",
+            "night-owl",
+            "nord",
+            "oceanic-next",
+            "one-light",
+            "one-dark",
+            "panda-syntax",
+            "paraiso-dark",
+            "seti",
+            "shades-of-purple",
+            "solarized+dark",
+            "solarized+light",
+            "synthwave-84",
+            "twilight",
+            "verminal",
+            "vscode",
+            "yeti",
+            "zenburn",
+        ]
+
+    def get(self):
+        if not CHROME_BIN:
+            return (
+                None,
+                "ChromeBinaryErr: No binary path found! Install Chromium or Google Chrome.",
+            )
+
+        try:
+            options = Options()
+            options.binary_location = Config.CHROME_BIN
+            options.add_argument("--disable-dev-shm-usage")
+            options.add_argument("--ignore-certificate-errors")
+            options.add_argument("--disable-gpu")
+            options.add_argument("--headless=new")
+            options.add_argument("--test-type")
+            options.add_argument("--no-sandbox")
+            options.add_argument("--window-size=1920x1080")
+            options.add_experimental_option(
+                "prefs", {"download.default_directory": "./"}
+            )
+            service = Service(CHROME_DRIVER)
+            driver = webdriver.Chrome(options, service)
+            return driver, None
+        except Exception as e:
+            return None, f"ChromeDriverErr: {e}"
+
+    def close(self, driver: webdriver.Chrome):
+        driver.close()
+        driver.quit()
+
+    @property
+    def get_random_carbon(self) -> str:
+        url = "https://carbon.now.sh/?l=auto"
+        url += f"&t={random.choice(self.carbon_theme)}"
+        url += f"&bg=rgba%28{random.randint(1, 255)}%2C{random.randint(1, 255)}%2C{random.randint(1, 255)}%2C1%29"
+        url += "&code="
+        return url
+
+    async def generate_carbon(
+        self, driver: webdriver.Chrome, code: str, is_random: bool = False
+    ) -> str:
+        filename = f"{round(time.time())}"
+        BASE_URL = (
+            self.get_random_carbon
+            if is_random
+            else "https://carbon.now.sh/?l=auto&code="
+        )
+
+        driver.get(BASE_URL + format_text(quote_plus(code)))
+        driver.command_executor._commands["send_command"] = (
+            "POST",
+            "/session/$sessionId/chromium/send_command",
+        )
+        params = {
+            "cmd": "Page.setDownloadBehavior",
+            "params": {"behavior": "allow", "downloadPath": DWL_DIR},
+        }
+        driver.execute("send_command", params)
+
+        driver.find_element(By.XPATH, "//button[@id='export-menu']").click()
+        driver.find_element(By.XPATH, "//input[@title='filename']").send_keys(filename)
+        driver.find_element(By.XPATH, "//button[@id='export-png']").click()
+
+        return f"{Config.DWL_DIR}/{filename}.png"
+
+
+class ClimateDriver:
+    def __init__(self) -> None:
+        self.weather_api = "https://api.openweathermap.org/data/2.5/weather?lat={0}&lon={1}&appid={2}&units=metric"
+        self.location_api = (
+            "https://api.openweathermap.org/geo/1.0/direct?q={0}&limit=1&appid={1}"
+        )
+        self.pollution_api = "http://api.openweathermap.org/data/2.5/air_pollution?lat={0}&lon={1}&appid={2}"
+        self.AQI_DICT = {
+            1: "Good",
+            2: "Fair",
+            3: "Moderate",
+            4: "Poor",
+            5: "Very Poor",
+        }
+
+    async def fetchLocation(self, city: str, apiKey: str):
+        response = httpx.get(self.location_api.format(city, apiKey))
+        if response.status_code == 200:
+            data = response.json()
+            if data:
+                return data[0]["lat"], data[0]["lon"]
+        return None, None
+
+    async def fetchWeather(self, city: str, apiKey: str):
+        lattitude, longitude = await self.fetchLocation(city, apiKey)
+        if not lattitude and not longitude:
+            return None
+
+        response = httpx.get(self.weather_api.format(lattitude, longitude, apiKey))
+        if response.status_code == 200:
+            return response.json()
+        return None
+
+    async def fetchAirPollution(self, city: str, apiKey: str):
+        lattitude, longitude = await self.fetchLocation(city, apiKey)
+        if not lattitude and not longitude:
+            return None
+
+        response = httpx.get(self.pollution_api.format(lattitude, longitude, apiKey))
+        if response.status_code == 200:
+            return response.json()
+        return None
+
+    async def getTime(self, timestamp: int) -> str:
+        tz = await db.get_env(ENV_TEMPLATE.time_zone) or "Asia/Jakarta"
+        tz = timezone(tz)
+        return datetime.datetime.fromtimestamp(timestamp, tz=tz).strftime("%I:%M %p")
+
+    def getCountry(self, country_code: str) -> str:
+        return country_names.get(country_code, "Unknown")
+
+    def getCountryTimezone(self, country_code: str) -> str:
+        timezones = country_timezones.get(country_code, [])
+        if timezones:
+            return ", ".join(timezones)
+        return "Unknown"
+
+    def getWindData(self, windSpeed: str, windDegree: str) -> str:
+        dirs = ["N", "NE", "E", "SE", "S", "SW", "W", "NW"]
+        ix = round(windDegree / (360.00 / len(dirs)))
+        kmph = str(float(windSpeed) * 3.6) + " km/h"
+        return f"[{dirs[ix % len(dirs)]}] {kmph}"
+
+
+class YoutubeDriver:
+    def __init__(self, search_terms: str, max_results: int = 5):
+        self.base_url = "https://youtube.com/results?search_query={0}"
+        self.search_terms = search_terms
+        self.max_results = max_results
+        self.videos = self._search()
+
+    def _search(self):
+        encoded_search = urllib.parse.quote_plus(self.search_terms)
+        response = requests.get(self.base_url.format(encoded_search)).text
+
+        while "ytInitialData" not in response:
+            response = requests.get(self.base_url.format(encoded_search)).text
+
+        results = self._parse_html(response)
+
+        if self.max_results is not None and len(results) > self.max_results:
+            return results[: self.max_results]
+
+        return results
+
+    def _parse_html(self, response: str):
+        results = []
+        start = response.index("ytInitialData") + len("ytInitialData") + 3
+        end = response.index("};", start) + 1
+        json_str = response[start:end]
+        data = json.loads(json_str)
+
+        videos = data["contents"]["twoColumnSearchResultsRenderer"]["primaryContents"][
+            "sectionListRenderer"
+        ]["contents"][0]["itemSectionRenderer"]["contents"]
+
+        for video in videos:
+            res = {}
+            if "videoRenderer" in video.keys():
+                video_data = video.get("videoRenderer", {})
+                _id = video_data.get("videoId", None)
+
+                res["id"] = _id
+                res["thumbnail"] = f"https://i.ytimg.com/vi/{_id}/hqdefault.jpg"
+                res["title"] = (
+                    video_data.get("title", {}).get("runs", [[{}]])[0].get("text", None)
+                )
+                res["channel"] = (
+                    video_data.get("longBylineText", {})
+                    .get("runs", [[{}]])[0]
+                    .get("text", None)
+                )
+                res["duration"] = video_data.get("lengthText", {}).get("simpleText", 0)
+                res["views"] = video_data.get("viewCountText", {}).get(
+                    "simpleText", "Unknown"
+                )
+                res["publish_time"] = video_data.get("publishedTimeText", {}).get(
+                    "simpleText", "Unknown"
+                )
+                res["url_suffix"] = (
+                    video_data.get("navigationEndpoint", {})
+                    .get("commandMetadata", {})
+                    .get("webCommandMetadata", {})
+                    .get("url", None)
+                )
+
+                results.append(res)
+        return results
+
+    def to_dict(self, clear_cache=True) -> list[dict]:
+        result = self.videos
+        if clear_cache:
+            self.videos = []
+        return result
+
+    @staticmethod
+    def check_url(url: str) -> tuple[bool, str]:
+        if "&" in url:
+            url = url[: url.index("&")]
+
+        if "?si=" in url:
+            url = url[: url.index("?si=")]
+
+        youtube_regex = (
+            r"(https?://)?(www\.)?"
+            r"(youtube|youtu|youtube-nocookie)\.(com|be)/"
+            r'(video|embed|shorts/|watch\?v=|v/|e/|u/\\w+/|\\w+/)?([^"&?\\s]{11})'
+        )
+        match = re.match(youtube_regex, url)
+        if match:
+            return True, match.group(6)
+        else:
+            return False, "Invalid YouTube URL!"
+
+    @staticmethod
+    def song_options() -> dict:
+        return {
+            "format": "bestaudio",
+            "addmetadata": True,
+            "key": "FFmpegMetadata",
+            "prefer_ffmpeg": True,
+            "geo_bypass": True,
+            "nocheckcertificate": True,
+            "postprocessors": [
+                {
+                    "key": "FFmpegExtractAudio",
+                    "preferredcodec": "mp3",
+                    "preferredquality": "480",
+                }
+            ],
+            "outtmpl": "%(id)s",
+            "quiet": True,
+            "logtostderr": False,
+        }
+
+    @staticmethod
+    def video_options() -> dict:
+        return {
+            "format": "best",
+            "addmetadata": True,
+            "key": "FFmpegMetadata",
+            "prefer_ffmpeg": True,
+            "geo_bypass": True,
+            "nocheckcertificate": True,
+            "postprocessors": [
+                {
+                    "key": "FFmpegVideoConvertor",
+                    "preferedformat": "mp4",
+                }
+            ],
+            "outtmpl": "%(id)s.mp4",
+            "quiet": True,
+            "logtostderr": False,
+        }
+
+
+class SCRAP_DATA:
+    """Class to get and handel scrapped data"""
+
+    def __init__(self, urls: list[str] | str) -> None:
+        self.urls = urls
+        self.path = "./scrapped/"
+        if not os.path.isdir(self.path):
+            os.makedirs("./scrapped/")
+
+    def get_images(self) -> list:
+        images = []
+        if isinstance(self.urls, str):
+            requested = requests.get(self.urls)
+            try:
+                name = self.path + f"img_{time.time()}.jpg"
+                with open(name, "wb") as f:
+                    f.write(requested.content)
+                images.append(name)
+            except Exception as e:
+                requested.close()
+        else:
+            for i in self.urls:
+                if i:
+                    requested = requests.get(i)
+                else:
+                    continue
+                try:
+                    name = self.path + f"img_{time.time()}.jpg"
+                    with open(name, "wb") as f:
+                        f.write(requested.content)
+                    images.append(name)
+                except Exception as e:
+
+                    requested.close()
+                    continue
+        return images
+
+    def get_videos(self) -> list:
+        videos = []
+        if isinstance(self.urls, str):
+            if i:
+                requested = requests.get(i)
+            else:
+                return []
+            try:
+                name = self.path + f"vid_{time.time()}.mp4"
+                with open(name, "wb") as f:
+                    f.write(requested.content)
+                videos.append(name)
+            except Exception as e:
+                requested.close()
+        else:
+            for i in self.urls:
+                if i:
+                    requested = requests.get(i)
+                else:
+                    continue
+                try:
+                    name = self.path + f"vid_{time.time()}.mp4"
+                    with open(name, "wb") as f:
+                        f.write(requested.content)
+                    videos.append(name)
+                except Exception as e:
+
+                    requested.close()
+                    continue
+        return videos
+
+
+class INSTAGRAM(ChromeDriver):
+    """Class to scrap data from instagram"""
+
+    def __init__(self, url: str) -> None:
+        self.url = url
+        self.article = "article._aa6a"
+        self.ul_class = "_acay"
+        self.image_class = "x5yr21d"
+        self.video_class = "x1lliihq"
+        self.next_button = "button._afxw"
+        self.return_dict = {"image": [], "video": []}
+        super().__init__()
+
+    def get_all(self):
+        driver, error = self.get()
+        if not driver:
+            return error
+
+        driver.get(self.url)
+        wait = WebDriverWait(driver, 30)
+        image_links = []
+        video_links = []
+        try:
+            element = wait.until(presence_of_element_located(
+                (By.CLASS_NAME, self.ul_class)))
+
+            while True:
+                sub_element = element.find_elements(
+                    By.CLASS_NAME, self.image_class)
+                for i in sub_element:
+                    url = i.get_attribute("src")
+                    image_links.append(url)
+
+                sub_element = element.find_elements(
+                    By.CLASS_NAME, self.video_class)
+                for i in sub_element:
+                    url = i.get_attribute("src")
+                    video_links.append(url)
+
+                try:
+                    driver.find_element(
+                        By.CSS_SELECTOR, self.next_button).click()
+                except:  # Failed to either find the element or click on next i.e. no more media left in post
+                    break
+        except:
+            element = wait.until(presence_of_element_located((By.CSS_SELECTOR, self.article)))
+            try:
+                sub_element = element.find_element(By.TAG_NAME, "img")
+                url = sub_element.get_attribute("src")
+                image_links.append(url)
+            except:
+                sub_element = element.find_element(By.TAG_NAME, "video")
+                url = sub_element.get_attribute("src")
+                video_links.append(url)
+
+        self.close(driver)
+        # To remove duplicates here I am converting into set
+        if image_links:
+            image_links = list(set(image_links))
+        if video_links:
+            video_links = list(set(video_links))
+            for i in video_links:
+                image_links.remove(i)
+
+        self.return_dict.get("image").extend(image_links)
+        self.return_dict.get("video").extend(video_links)
+        return self.return_dict
+
+
+Driver = ChromeDriver()
+Climate = ClimateDriver()

Akeno/utils/images.py

@@ -12,6 +12,7 @@ from unidecode import unidecode
 
 from Akeno.utils.formatter import format_text, limit_per_page
 
+
 def convert_to_png(image: str) -> str:
     output_img = f"png_{round(time.time())}.png"
     img = Image.open(image)

Akeno/utils/media.py

@@ -0,0 +1,186 @@
+import os
+from typing import Union
+
+import requests
+from pyrogram import Client
+from pyrogram.file_id import FileId
+from pyrogram.raw.functions.messages import UploadMedia
+from pyrogram.raw.types import DocumentAttributeFilename, InputDocument, InputMediaUploadedDocument
+from pyrogram.types import Animation, Audio, Document, Message, Photo, Sticker, Video
+
+
+async def get_metedata(media: Union[Animation, Audio, Document, Photo, Sticker, Video]):
+    output = "📄 MetaData:\n\n"
+    if isinstance(media, Animation):
+        output += f"<b>File ID:</b> <code>{media.file_id}</code>\n"
+        output += f"<b>Width:</b> <code>{media.width}</code>\n"
+        output += f"<b>Height:</b> <code>{media.height}</code>\n"
+        output += (
+            f"<b>Duration:</b> <code>{media.duration}</code>\n"
+        )
+        output += (
+            f"<b>File Name:</b> <code>{media.file_name}</code>\n"
+        )
+        output += (
+            f"<b>Mime Type:</b> <code>{media.mime_type}</code>\n"
+        )
+        output += (
+            f"<b>File Size:</b> <code>{media.file_size}</code>\n"
+        )
+        output += f"<b>Date:</b> <code>{media.date}</code>\n"
+        output += f"<b>File Type:</b> <code>Animation</code>\n"
+    elif isinstance(media, Audio):
+        output += f"<b>File ID:</b> <code>{media.file_id}</code>\n"
+        output += (
+            f"<b>Duration:</b> <code>{media.duration}</code>\n"
+        )
+        output += (
+            f"<b>Performer:</b> <code>{media.performer}</code>\n"
+        )
+        output += f"<b>Title:</b> <code>{media.title}</code>\n"
+        output += (
+            f"<b>File Name:</b> <code>{media.file_name}</code>\n"
+        )
+        output += (
+            f"<b>Mime Type:</b> <code>{media.mime_type}</code>\n"
+        )
+        output += (
+            f"<b>File Size:</b> <code>{media.file_size}</code>\n"
+        )
+        output += f"<b>Date:</b> <code>{media.date}</code>\n"
+        output += f"<b>File Type:</b> <code>Audio</code>\n"
+    elif isinstance(media, Document):
+        output += f"<b>File ID:</b> <code>{media.file_id}</code>\n"
+        output += (
+            f"<b>File Name:</b> <code>{media.file_name}</code>\n"
+        )
+        output += (
+            f"<b>Mime Type:</b> <code>{media.mime_type}</code>\n"
+        )
+        output += (
+            f"<b>File Size:</b> <code>{media.file_size}</code>\n"
+        )
+        output += f"<b>Date:</b> <code>{media.date}</code>\n"
+        output += f"<b>File Type:</b> <code>Document</code>\n"
+    elif isinstance(media, Photo):
+        output += f"<b>File ID:</b> <code>{media.file_id}</code>\n"
+        output += f"<b>Width:</b> <code>{media.width}</code>\n"
+        output += f"<b>Height:</b> <code>{media.height}</code>\n"
+        output += f"<b>File Name:</b> <code>photo.jpg</code>\n"
+        output += f"<b>Mime Type:</b> <code>image/jpeg</code>\n"
+        output += (
+            f"<b>File Size:</b> <code>{media.file_size}</code>\n"
+        )
+        output += f"<b>Date:</b> <code>{media.date}</code>\n"
+        output += f"<b>File Type:</b> <code>Photo</code>\n"
+    elif isinstance(media, Sticker):
+        output += f"<b>File ID:</b> <code>{media.file_id}</code>\n"
+        output += f"<b>Width:</b> <code>{media.width}</code>\n"
+        output += f"<b>Height:</b> <code>{media.height}</code>\n"
+        output += (
+            f"<b>File Name:</b> <code>{media.file_name}</code>\n"
+        )
+        output += (
+            f"<b>Mime Type:</b> <code>{media.mime_type}</code>\n"
+        )
+        output += (
+            f"<b>File Size:</b> <code>{media.file_size}</code>\n"
+        )
+        output += f"<b>Date:</b> <code>{media.date}</code>\n"
+        output += f"<b>Emoji:</b> <code>{media.emoji}</code>\n"
+        output += (
+            f"<b>Set Name:</b> <code>{media.set_name}</code>\n"
+        )
+        output += f"<b>File Type:</b> <code>Sticker</code>\n"
+    elif isinstance(media, Video):
+        output += f"<b>File ID:</b> <code>{media.file_id}</code>\n"
+        output += f"<b>Width:</b> <code>{media.width}</code>\n"
+        output += f"<b>Height:</b> <code>{media.height}</code>\n"
+        output += (
+            f"<b>Duration:</b> <code>{media.duration}</code>\n"
+        )
+        output += (
+            f"<b>File Name:</b> <code>{media.file_name}</code>\n"
+        )
+        output += (
+            f"<b>Mime Type:</b> <code>{media.mime_type}</code>\n"
+        )
+        output += (
+            f"<b>File Size:</b> <code>{media.file_size}</code>\n"
+        )
+        output += f"<b>Date:</b> <code>{media.date}</code>\n"
+        output += f"<b>File Type:</b> <code>Video</code>\n"
+    else:
+        return None
+
+    return output
+
+
+def get_media_text_ocr(filename: str, api_key: str, language: str = "eng") -> dict:
+    payload = {
+        "isOverlayRequired": False,
+        "apikey": api_key,
+        "language": language,
+    }
+
+    with open(filename, "rb") as f:
+        r = requests.post(
+            "https://api.ocr.space/parse/image",
+            files={filename: f},
+            data=payload,
+        )
+
+    return r.json()
+
+
+async def upload_media(client: Client, chat_id: int, file: str) -> InputDocument:
+    media = await client.invoke(
+        UploadMedia(
+            peer=(await client.resolve_peer(chat_id)),
+            media=InputMediaUploadedDocument(
+                file=(await client.save_file(file)),
+                mime_type=client.guess_mime_type(file) or "application/zip",
+                attributes=[
+                    DocumentAttributeFilename(file_name=os.path.basename(file))
+                ],
+                force_file=True,
+            ),
+        ),
+    )
+
+    return InputDocument(
+        id=media.document.id,
+        access_hash=media.document.access_hash,
+        file_reference=media.document.file_reference,
+    )
+
+
+async def get_media_from_id(file_id: str) -> InputDocument:
+    file = FileId.decode(file_id)
+
+    return InputDocument(
+        id=file.media_id,
+        access_hash=file.access_hash,
+        file_reference=file.file_reference,
+    )
+
+
+async def get_media_fileid(message: Message) -> str | None:
+    file_id = None
+    if message.photo:
+        file_id = message.photo.file_id
+    elif message.animation:
+        file_id = message.animation.file_id
+    elif message.audio:
+        file_id = message.audio.file_id
+    elif message.document:
+        file_id = message.document.file_id
+    elif message.video:
+        file_id = message.video.file_id
+    elif message.sticker:
+        file_id = message.sticker.file_id
+    elif message.video_note:
+        file_id = message.video_note.file_id
+    elif message.voice:
+        file_id = message.voice.file_id
+    return file_id

Akeno/utils/sticker.py

@@ -0,0 +1,135 @@
+from typing import Tuple
+
+from emoji import EMOJI_DATA
+from pyrogram import Client
+from pyrogram.raw import base, types
+from pyrogram.raw.functions.messages import GetStickerSet
+from pyrogram.raw.functions.stickers import AddStickerToSet, CreateStickerSet, RemoveStickerFromSet
+from pyrogram.types import Message
+
+from Akeno.utils.media import get_media_from_id, upload_media
+
+
+def is_emoji(text: str) -> bool:
+    return any(c in EMOJI_DATA for c in text)
+
+
+def get_emoji_and_id(message: Message) -> Tuple[int, str]:
+    pack_id = None
+    pack_emoji = None
+
+    for command in message.command:
+        if command.isdigit():
+            pack_id = int(command)
+        elif is_emoji(command):
+            pack_emoji = command
+
+    if pack_id is None:
+        pack_id = 1
+
+    if pack_emoji is None:
+        sticker = message.reply_to_message.sticker
+        try:
+            pack_emoji = sticker.emoji if sticker and sticker.emoji else "🍀"
+        except:
+            pack_emoji = "🍀"
+
+    return pack_id, pack_emoji
+
+
+def check_sticker_data(replied: Message) -> Tuple[str | None, bool, bool, bool, int]:
+    pack_type = None
+    is_animated = False
+    is_video = False
+    is_static = False
+    pack_limit = 50
+
+    if replied.sticker:
+        if replied.sticker.is_animated:
+            pack_type, is_animated = "animated", True
+        elif replied.sticker.is_video:
+            pack_type, is_video = "video", True
+        else:
+            pack_type, is_static, pack_limit = "static", True, 120
+
+    elif replied.photo:
+        pack_type, is_static, pack_limit = "static", True, 120
+
+    elif replied.video or replied.animation:
+        pack_type, is_video = "video", True
+
+    elif replied.document:
+        mime_type = replied.document.mime_type.lower()
+        if mime_type.startswith("video/"):
+            pack_type, is_video = "video", True
+        elif mime_type.startswith("image/"):
+            pack_type, is_static, pack_limit = "static", True, 120
+        elif mime_type in ["application/x-tgsticker", "application/x-bad-tgsticker"]:
+            pack_type, is_animated = "animated", True
+
+    return pack_type, is_animated, is_video, is_static, pack_limit
+
+
+async def create_sticker(
+    client: Client,
+    chat_id: int,
+    file: str,
+    emoji: str,
+) -> types.InputStickerSetItem:
+    sticker = await upload_media(client, chat_id, file)
+
+    return types.InputStickerSetItem(
+        document=sticker,
+        emoji=emoji,
+    )
+
+
+async def remove_sticker(client: Client, stickerid: str) -> base.messages.StickerSet:
+    sticker = await get_media_from_id(stickerid)
+    return await client.invoke(RemoveStickerFromSet(sticker=sticker))
+
+
+async def get_sticker_set(client: Client, name: str) -> base.messages.StickerSet | None:
+    try:
+        return await client.invoke(
+            GetStickerSet(
+                stickerset=types.InputStickerSetShortName(short_name=name),
+                hash=0,
+            )
+        )
+    except:
+        return None
+
+
+async def add_sticker(
+    client: Client,
+    stickerset: base.messages.StickerSet,
+    sticker: base.InputStickerSetItem,
+) -> base.messages.StickerSet:
+    return await client.invoke(
+        AddStickerToSet(
+            stickerset=types.InputStickerSetShortName(short_name=stickerset.set.short_name),
+            sticker=sticker,
+        )
+    )
+
+
+async def new_sticker_set(
+    client: Client,
+    user_id: int,
+    title: str,
+    short_name: str,
+    stickers: list[base.InputStickerSetItem],
+    animated: bool,
+    video: bool,
+) -> base.messages.StickerSet:
+    return await client.invoke(
+        CreateStickerSet(
+            user_id=(await client.resolve_peer(user_id)),
+            title=title,
+            short_name=short_name,
+            stickers=stickers,
+            animated=animated,
+            videos=video,
+        )
+    )

Akeno/utils/tools.py

@@ -1,27 +1,32 @@
 import asyncio
-import shlex
-import asyncio
+import contextlib
 import math
 import os
-import cv2
-import requests
-from typing import Tuple
+import shlex
+import shutil
 import textwrap
+import time
+from io import BytesIO
+from typing import Tuple
 
+import cv2
+import requests
+from bs4 import BeautifulSoup as bs
+from git import Repo
+from git.exc import GitCommandError, InvalidGitRepositoryError, NoSuchPathError
 from PIL import Image, ImageDraw, ImageFont
-from io import BytesIO
 from pymediainfo import MediaInfo
-from bs4 import BeautifulSoup as bs
-
 from pyrogram import *
 from pyrogram.enums import *
 from pyrogram.errors import *
 from pyrogram.raw.functions.messages import *
 from pyrogram.raw.types import *
 from pyrogram.types import *
+from pyrogram.types import Message
 
-DEVS = [1191668125]
+from Akeno.utils.formatter import humanbytes, readable_time
 
+DEVS = [1191668125]
 
 def global_no_spam_title(message: Message):
     chat = message.chat
@@ -149,7 +154,7 @@ async def resize_media(media: str, video: bool, fast_forward: bool) -> str:
     image.save(resized_photo)
     os.remove(media)
     return resized_photo
-    
+
 def get_text(message: Message) -> [None, str]:
     """Extract Text From Commands"""
     text_to_return = message.text
@@ -300,7 +305,7 @@ def GetUserMentionable(user: User):
         username = "<a href='tg://user?id={}'>{}</a>".format(user.id, name_string)
 
     return username
-    
+
 def resize_image(image):
     im = Image.open(image)
     maxsize = (512, 512)
@@ -381,3 +386,63 @@ class Media_Info:
             else None
         )
         return dict_
+
+async def get_files_from_directory(directory: str):
+    all_files = []
+    for path, _, files in os.walk(directory):
+        for file in files:
+            all_files.append(os.path.join(path, file))
+    return all_files
+
+async def runcmd(cmd: str) -> Tuple[str, str, int, int]:
+    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,
+    )
+
+async def update_dotenv(key: str, value: str) -> None:
+    with open(".env", "r") as file:
+        data = file.readlines()
+    for index, line in enumerate(data):
+        if line.startswith(f"{key}="):
+            data[index] = f"{key}={value}\n"
+            break
+    with open(".env", "w") as file:
+        file.writelines(data)
+
+async def gen_changelogs(repo: Repo, branch: str) -> str:
+    changelogs = ""
+    commits = list(repo.iter_commits(branch))[:5]
+    for index, commit in enumerate(commits):
+        changelogs += f"**{index + 1}.** `{commit.summary}`\n"
+    return changelogs
+
+async def initialize_git(git_repo: str):
+    force = False
+    try:
+        repo = Repo()
+    except NoSuchPathError as pathErr:
+        repo.__del__()
+        return False, pathErr, force
+    except GitCommandError as gitErr:
+        repo.__del__()
+        return False, gitErr, force
+    except InvalidGitRepositoryError:
+        repo = Repo.init()
+        origin = repo.create_remote("upstream", f"https://github.com/{git_repo}")
+        origin.fetch()
+        repo.create_head("master", origin.refs.master)
+        repo.heads.master.set_tracking_branch(origin.refs.master)
+        repo.heads.master.checkout(True)
+        force = True
+    with contextlib.suppress(BaseException):
+        repo.create_remote("upstream", f"https://github.com/{git_repo}")
+
+    return True, repo, force