All new features done. To do bug hunting....

Powers/plugins/stickers.py

@@ -64,7 +64,7 @@ async def sticker_id_gib(c: Gojo, m: Message):
 async def kang(c:Gojo, m: Message):
     if not m.reply_to_message:
         return await m.reply_text("Reply to a sticker or image to kang it.")
-    elif not (m.reply_to_message.sticker or m.reply_to_message.photo or (m.reply_to_message.document and m.reply_to_message.document.mime_type.split("/")[0]=="image")):
+    elif not (m.reply_to_message.animation or m.reply_to_message.sticker or m.reply_to_message.photo or (m.reply_to_message.document and m.reply_to_message.document.mime_type.split("/")[0]in["image","video"])):
         return await m.reply_text("Reply to a sticker or image to kang it.")
     if not m.from_user:
         return await m.reply_text("You are anon admin, kang stickers in my pm.")
@@ -92,7 +92,7 @@ async def kang(c:Gojo, m: Message):
 
     # Get the corresponding fileid, resize the file if necessary
     try:
-        if is_requ or m.reply_to_message.photo or (m.reply_to_message.document and m.reply_to_message.document.mime_type.split("/")[0]=="image"):
+        if is_requ or m.reply_to_message.video or m.reply_to_message.photo or (m.reply_to_message.document and m.reply_to_message.document.mime_type.split("/")[0]=="image"):
             sizee = (await get_file_size(m.reply_to_message)).split()
             if (sizee[1] == "mb" and sizee > 10) or sizee[1] == "gb":
                 await m.reply_text("File size is too big")
@@ -113,8 +113,11 @@ async def kang(c:Gojo, m: Message):
         LOGGER.error(format_exc())
         return
     try:
-        if is_requ or not m.reply_to_message.sticker:
+        if is_requ or not m.reply_to_message.sticker or m.reply_to_message.animation or m.reply_to_message.video or (m.reply_to_message.document and m.reply_to_message.document.mime_type.split("/")[0] == "video"):
             # telegram doesn't allow animated and video sticker to be kanged as we do for normal stickers
+            if m.reply_to_message.animation or m.reply_to_message.video or (m.reply_to_message.document and m.reply_to_message.document.mime_type.split("/")[0] == "video"):
+                path = m.reply_to_message.download()
+                path = Vsticker(path)
             sticker = await create_sticker(
                 await upload_document(
                     c, path, m.chat.id
@@ -283,27 +286,40 @@ async def get_sticker_from_file(c: Gojo, m: Message):
         await m.reply_text("Reply to a sticker or file")
         return
     repl = m.reply_to_message
-    if not (repl.sticker or repl.photo or (repl.document and repl.document.mime_type.split("/")[0]=="image")):
+    to_vid = False
+    if not (repl.sticker.is_animated or repl.video or repl.sticker or repl.photo or (repl.document and repl.document.mime_type.split("/")[0] in ["image","video"])):
         await m.reply_text("I only support conversion of plain stickers and images for now")
         return
-    if repl.sticker and (repl.sticker.is_video or repl.sticker.is_animated):
-        await m.reply_text("I only support conversion of plain stickers for now")
-        return
+    if repl.video or (repl.document and repl.document.mime_type.split("/")[0]=="video"):
+        to_vid = True
     x = await m.reply_text("Converting...")
+    upp = await repl.download()
     if repl.sticker:
-        upp = await repl.download()
-        up = toimage(upp,is_direc=True)
-        await x.delete()
-        await m.reply_photo(up,caption=Caption)
-        os.remove(up)
-        return
+        if repl.sticker.is_animated:
+            up = tgs_to_gif(upp,True)
+            await x.delete()
+            await m.reply_animation(up,caption=Caption)
+        elif repl.sticker.is_video:
+            up = webm_to_gif(upp)
+            await x.delete()
+            await m.reply_animation(up,caption=Caption)
+        else:
+            up = toimage(upp,is_direc=True)
+            await x.delete()
+            await m.reply_photo(up,caption=Caption)
+            os.remove(up)
+            return
     elif repl.photo:
-        upp = await repl.download()
         up = tosticker(upp,is_direc=True)
         await x.delete()
-        await m.reply_sticker(up,caption=Caption)
+        await m.reply_sticker(up)
         os.remove(up)
         return
+    
+    elif to_vid:
+        up = Vsticker(upp)
+        await x.delete()
+        await m.reply_sticker(up)
 
         
 __PLUGIN__ = "sticker"

Powers/utils/sticker_help.py

@@ -12,6 +12,19 @@ from pyrogram.file_id import FileId
 from Powers.bot_class import Gojo
 
 
+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 get_sticker_set_by_name(
     client: Gojo, name: str
 ) -> raw.base.messages.StickerSet:
@@ -97,7 +110,34 @@ async def resize_file_to_sticker_size(file_path: str,length:int=512,width:int=51
     im.save(file_pathh)
     os.remove(file_path)
     return file_pathh
+
+async def tgs_to_gif(file, tgs=False, video=False):
+    if tgs:
+        cmd = f"lottie_convert.py '{file}' 'gojo_satoru.gif'"
+    elif video:
+        cmd = f"ffmpeg -i '{file}' -c copy 'gojo_satoru.gif'"
+    await runcmd(cmd)
+    os.remove(file)
+    return 'hellbot.gif'
+
+async def webm_to_gif(file):
+    cmd = f"ffmpeg -i '{file}' 'goJo.gif'"
+    await runcmd(cmd)
+    os.remove(file)
+    return "goJo.gif"
         
+async def Vsticker(file: str):
+    _width_ = file.file.width
+    _height_ = file.file.height
+    if _height_ > _width_:
+        _height_, _width_ = (512, -1)
+    else:
+        _height_, _width_ = (-1, 512)
+    await runcmd(
+        f"ffmpeg -to 00:00:02.900 -i '{file}' -vf scale={_width_}:{_height_} -c:v libvpx-vp9 -crf 30 -b:v 560k -maxrate 560k -bufsize 256k -an 'VideoSticker.webm'"
+    )
+    os.remove(file)
+    return "VideoSticker.webm"
 
 
 async def upload_document(
@@ -136,7 +176,7 @@ async def get_document_from_file_id(
     )
 
 
-async def draw_meame(image_path: str, text: str, sticker: bool, fiill: str) -> list:
+async def draw_meme(image_path: str, text: str, sticker: bool, fiill: str) -> list:
     _split = text.split(";", 1)
     if len(_split) == 2:
         lower_text = _split[1]

Powers/utils/web_helpers.py

@@ -1,13 +1,15 @@
 import json
 import os
+from traceback import format_exc
 from urllib import parse
 
 import yt_dlp
 from pyrogram.types import InlineKeyboardButton as IKB
 from pyrogram.types import InlineKeyboardMarkup as IKM
 from pyrogram.types import Message
+from youtubesearchpython.__future__ import Video, VideosSearch
 
-from Powers.bot_class import Gojo
+from Powers.bot_class import LOGGER, MESSAGE_DUMP, Gojo
 from Powers.utils.http_helper import *
 
 
@@ -34,98 +36,25 @@ async def get_file_size(file: Message):
             return f"{round(size)} gb"
 
 
-class GOJO_YTS:
-    """
-    A class to fetch link of from youtube using the name provided
-    Creadit: Hellbot
-    """
-    def __init__(self, search_terms: str, max_results=None):
-        self.search_terms = search_terms
-        self.max_results = max_results
-        self.videos = self._search()
-
-    def _search(self):
-        encoded_search = parse.quote_plus(self.search_terms)
-        BASE_URL = "https://youtube.com"
-        url = f"{BASE_URL}/results?search_query={encoded_search}"
-        response = requests.get(url).text
-        while "ytInitialData" not in response:
-            response = requests.get(url).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):
-        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", {})
-                res["id"] = video_data.get("videoId", None)
-                res["thumbnails"] = [
-                    thumb.get("url", None)
-                    for thumb in video_data.get("thumbnail", {}).get("thumbnails", [{}])
-                ]
-                res["title"] = (
-                    video_data.get("title", {}).get("runs", [[{}]])[0].get("text", None)
-                )
-                res["long_desc"] = (
-                    video_data.get("descriptionSnippet", {})
-                    .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", 0)
-                res["publish_time"] = video_data.get("publishedTimeText", {}).get(
-                    "simpleText", 0
-                )
-                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):
-        result = self.videos
-        if clear_cache:
-            self.videos = ""
-        return result
-
-    def to_json(self, clear_cache=True):
-        result = json.dumps({"videos": self.videos})
-        if clear_cache:
-            self.videos = ""
-        return result
-
 
 # Gets yt result of given query.
-async def song_search(query, max_results=1):
-    try:
-        results = json.loads(GOJO_YTS(query, max_results=max_results).to_json())
-    except KeyError:
-        return 
+async def song_search(query, is_direct, max_results=1):
     yt_dict = {}
+    try:
+        if is_direct:
+            vid = Video.getInfo(query)
+            query = vid["title"]
+        else:
+            query = query
+        videos = VideosSearch(query,max_results)
+        results = await videos.next()
+    except Exception as e:
+        LOGGER.error(e)
+        LOGGER.error(format_exc())
+        return e
            
     nums = 1
-    for i in results["videos"]:
+    for i in results["result"]:
         durr = i['duration'].split(":")
         if len(durr) == 3:
             hour_to_sec = int(durr[0])*60*60
@@ -136,12 +65,14 @@ async def song_search(query, max_results=1):
             total = minutes_to_sec + int(durr[1])
         if not (total > 600):
             dict_form = {
-                "link": f"https://www.youtube.com{i['url_suffix']}",
-                "title": i['title'],
-                "views": i['views'],
-                "channel": i['channel'],
-                "duration": i['duration'],
-                "thumbnail": i['thumbnails'][0]
+                "link": i["link"],
+                "title": i["title"],
+                "views": i["viewCount"]["short"],
+                "channel": i["channel"]["link"],
+                "duration": i["accessibility"]['duration'],
+                "thumbnail": i["richThumbnail"],
+                "published": i["publishedTime"],
+                "uploader": i ["channel"]["name"]
                 }
             yt_dict.update({nums: dict_form})
             nums += 1
@@ -200,51 +131,65 @@ async def youtube_downloader(c:Gojo,m:Message,query:str,is_direct:bool,type_:str
         video = True
         song = False
     ydl = yt_dlp.YoutubeDL(opts)
-    if is_direct:
-        query = query
-    else:
-        dicti = await song_search(query, 1)
-        if not dicti:
-            await m.reply_text("File with duration less than or equals to 5 minutes is allowed only")
-        try:
-            query = dicti[1]['link']
-        except KeyError:
-            z = "KeyError"
-            return z
-    FILE = ydl.extract_info(query,download=video)
-    if int(FILE['duration']) > 600:
+    dicti = await song_search(query, is_direct,1)
+    if not dicti and type(dicti) != str:
         await m.reply_text("File with duration less than or equals to 5 minutes is allowed only")
-        return 
-    f_name = FILE['title']
-    uploader = FILE['uploader']
-    up_url = FILE['uploader_url']
-    views = FILE['view_count']
+    elif type(dicti) == str:
+        await m.reply_text(dicti)
+        return
+    try:
+        query = dicti[1]['link']
+    except KeyError:
+        z = "KeyError"
+        return z
+    
+    f_name = dicti["title"]
+    views = dicti["views"]
+    up_url = dicti["channel"]
+    uploader = dicti["uploader"]
+    thumb = dicti["thumbnail"]
+    published_on = dicti["publishedON"]
+
+    FILE = ydl.extract_info(query,download=video)
     url = query
+    thumb_ = await c.send_photo(MESSAGE_DUMP, thumb)
+    thumb = await thumb_.download()
+    await thumb_.delete()
     if song:
         f_down = ydl.prepare_filename(FILE)
         f_path = f"{f_down}.mp3"
-        thumb = f"{f_down}.webp"
         ydl.download([query])
+        ext = ".mp3"
     elif video:
         f_path = open(f"{FILE['id']}.mp4","rb")
+        ext = ".mp4"
     cap = f"""
-✘ Name: `{f_name}`
-✘ Views: `{views}` 
+⤷ Name: `{f_name}`
+⤷ Views: `{views}`
+⤷ Published date: `{published_on}`
 """
     kb = IKM(
         [
             [
-                IKB(f"✘ {uploader.capitalize()} ✘",url=f"{up_url}"),
+                IKB(f"✘ {uploader.capitalize()} ✘",url=f"{up_url}")
+            ],
+            [
                 IKB(f"✘ Youtube url ✘", url=f"{url}")
             ]
         ]
     )
+
+    file_path = f_name.strip() + ext
+    os.rename(f_path,file_path)
+
+
     if video:
-        await m.reply_video(f_path,caption=cap,reply_markup=kb,duration=int(FILE['duration']))
-        os.remove(f"./{FILE['id']}.mp4")
+        await m.reply_video(file_path,caption=cap,reply_markup=kb,duration=int(FILE['duration']))
+        os.remove(file_path)
+        os.remove(thumb)
         return
     elif song:
-        await m.reply_audio(f_path,caption=cap,reply_markup=kb,duration=int(FILE['duration']),thumb=thumb,title=f_name)
+        await m.reply_audio(file_path,caption=cap,reply_markup=kb,duration=int(FILE['duration']),thumb=thumb,title=f_name)
         os.remove(f_path)
         os.remove(thumb)
         return

Powers/vars.py

@@ -15,7 +15,7 @@ class Config:
     API_ID = int(config("API_ID", default="123"))
     API_HASH = config("API_HASH", default=None)
     OWNER_ID = int(config("OWNER_ID", default=1344569458))
-    MESSAGE_DUMP = int(config("MESSAGE_DUMP", default=-100))
+    MESSAGE_DUMP = int(config("MESSAGE_DUMP"))
     DEV_USERS = [
         int(i)
         for i in config(
@@ -64,7 +64,7 @@ class Development:
     API_ID = 12345  # Your APP_ID from Telegram
     API_HASH = "YOUR API HASH"  # Your APP_HASH from Telegram
     OWNER_ID = 1344569458  # Your telegram user id defult to mine
-    MESSAGE_DUMP = -100  # Your Private Group ID for logs
+    MESSAGE_DUMP = -100845454887  # Your Private Group ID for logs
     DEV_USERS = []
     SUDO_USERS = []
     WHITELIST_USERS = []

README.md

@@ -172,6 +172,8 @@ If all works well, the bot should send a message to the MESSAGE_DUMP Group!--->
 `API_HASH` You can get your api hash [here](my.telegram.org)
 
 `DB_URI` Your [MongoDB](https://www.mongodb.com/) connection string.
+
+`MESSAGE_DUMP`: Event logs channel where the bot will send updates. Note that it should start with `-100`.
 </details>
 
 
@@ -201,8 +203,6 @@ If all works well, the bot should send a message to the MESSAGE_DUMP Group!--->
 
 `SUPPORT_GROUP`: Your Telegram support group chat username that users can contact in case of a problem.
 
-`MESSAGE_DUMP`: Event logs channel where the bot will send updates. Note that it should start with `-100`.
-
 `PREFIX_HANDLER`: Something like '/' to execute commands.
 
 `SUPPORT_CHANNEL`: Your Telegram support channel username where users can see bot updates.