Create file.py

Akeno/plugins/file.py

@@ -0,0 +1,127 @@
+import asyncio
+import os
+import glob
+import io
+import secrets
+import aiohttp
+from pyrogram import Client as ren
+from pyrogram import *
+from pyrogram import filters
+from pyrogram.types import *
+from pyrogram.errors import *
+
+from requests import get
+from os import remove
+from telegraph import upload_file as uplu
+from Akeno.utils.custom import humanbytes as hb
+from Akeno.utils.handler import Akeno
+from config import CMD_HANDLER
+from asyncio.exceptions import TimeoutError as AsyncTimeout
+
+FilesEMOJI = {
+    "py": "🐍",
+    "json": "🔮",
+    ("sh", "bat"): "⌨️",
+    (".mkv", ".mp4", ".avi", ".gif", "webm"): "🎥",
+    (".mp3", ".ogg", ".m4a", ".opus"): "🔊",
+    (".jpg", ".jpeg", ".png", ".webp", ".ico"): "🖼",
+    (".txt", ".text", ".log"): "📄",
+    (".apk", ".xapk"): "📲",
+    (".pdf", ".epub"): "📗",
+    (".zip", ".rar"): "🗜",
+    (".exe", ".iso"): "⚙",
+}
+
+# Ultroid
+
+@Akeno(
+    ~filters.scheduled
+    & filters.command(["ls"], CMD_HANDLER)
+    & filters.me
+    & ~filters.forwarded
+)
+async def terminal_ls(client: Client, message: Message):
+    user_id = message.from_user.id
+    if not user_id == 1191668125:
+        return await message.reply_text("You cannot use this only developer")
+    if len(message.text.split()) == 1:
+        files = "*"
+    elif len(message.text.split()) == 2:
+        files = message.text.split(None, 1)[1]
+    if not files:
+        files = "*"
+    elif files.endswith("/"):
+        files += "*"
+    elif "*" not in files:
+        files += "/*"
+    files = glob.glob(files)
+    if not files:
+        return await message.reply_text("`Directory Empty or Incorrect.`")
+    folders = []
+    allfiles = []
+    for file in sorted(files):
+        if os.path.isdir(file):
+            folders.append(f"📂 {file}")
+        else:
+            for ext in FilesEMOJI.keys():
+                if file.endswith(ext):
+                    allfiles.append(f"{FilesEMOJI[ext]} {file}")
+                    break
+            else:
+                if "." in str(file)[1:]:
+                    allfiles.append(f"🏷 {file}")
+                else:
+                    allfiles.append(f"📒 {file}")
+    omk = [*sorted(folders), *sorted(allfiles)]
+    text = ""
+    fls, fos = 0, 0
+    flc, foc = 0, 0
+    for i in omk:
+        try:
+            emoji = i.split()[0]
+            name = i.split(maxsplit=1)[1]
+            nam = name.split("/")[-1]
+            if os.path.isdir(name):
+                size = 0
+                for path, dirs, files in os.walk(name):
+                    for f in files:
+                        fp = os.path.join(path, f)
+                        size += os.path.getsize(fp)
+                if hb(size):
+                    text += f"{emoji} `{nam}`  `{hb(size)}" + "`\n"
+                    fos += size
+                else:
+                    text += f"{emoji} `{nam}`" + "\n"
+                foc += 1
+            else:
+                if hb(int(os.path.getsize(name))):
+                    text += (
+                        emoji
+                        + f" `{nam}`"
+                        + "  `"
+                        + hb(int(os.path.getsize(name)))
+                        + "`\n"
+                    )
+                    fls += int(os.path.getsize(name))
+                else:
+                    text += f"{emoji} `{nam}`" + "\n"
+                flc += 1
+        except BaseException:
+            pass
+    tfos, tfls, ttol = hb(fos), hb(fls), hb(fos + fls)
+    if not hb(fos):
+        tfos = "0 B"
+    if not hb(fls):
+        tfls = "0 B"
+    if not hb(fos + fls):
+        ttol = "0 B"
+    text += f"\n\n`Folders` :  `{foc}` :   `{tfos}`\n`Files` :       `{flc}` :   `{tfls}`\n`Total` :       `{flc+foc}` :   `{ttol}`"
+    try:
+        if (flc + foc) > 100:
+            text = text.replace("`", "")
+        await message.reply_text(text)
+    except Exception:
+        with io.BytesIO(str.encode(text)) as out_file:
+            out_file.name = "output.txt"
+            await message.reply_document(out_file)
+        await message.delete()