Alpha

.gitignore

@@ -1,21 +1,9 @@
-config.env
+# Python bytecode and configuration files
 *.pyc
-data*
-.vscode
-.idea
-*.json
-*.pickle
-.netrc
-log.txt
-accounts/*
-Thumbnails/*
-MediaInfo/*
-Images/*
-rclone/*
-list_drives.txt
-cookies.txt
-downloads
-bot.session
-user.session
-terabox.txt
-rclone.conf
+config.env
+
+# IDE-specific files
+.vscode/
+.idea/
+
+

aria.sh

@@ -1,9 +1,17 @@
-tracker_list=$(curl -Ns https://ngosang.github.io/trackerslist/trackers_all_http.txt | awk '$0' | tr '\n\n' ',')
+#!/bin/bash
+
+# Get the tracker list and check if the curl command was successful
+if ! tracker_list=$(curl -Ns https://ngosang.github.io/trackerslist/trackers_all_http.txt | awk '$0' | tr '\n\n' ','); then
+  echo "Error: Failed to download tracker list"
+  exit 1
+fi
+
+# Download the torrent file and specify the download directory
 aria2c --allow-overwrite=true --auto-file-renaming=true --bt-enable-lpd=true --bt-detach-seed-only=true \
        --bt-remove-unselected-file=true --bt-tracker="[$tracker_list]" --bt-max-peers=0 --enable-rpc=true \
        --rpc-max-request-size=1024M --max-connection-per-server=10 --max-concurrent-downloads=10 --split=10 \
        --seed-ratio=0 --check-integrity=true --continue=true --daemon=true --disk-cache=40M --force-save=true \
        --min-split-size=10M --follow-torrent=mem --check-certificate=false --optimize-concurrent-downloads=true \
        --http-accept-gzip=true --max-file-not-found=0 --max-tries=20  --peer-id-prefix=-qB4520- --reuse-uri=true \
        --content-disposition-default-utf8=true --user-agent=Wget/1.12 --peer-agent=qBittorrent/4.5.2 --quiet=true \
-       --summary-interval=0 --max-upload-limit=1K
+       --summary-interval=0 --max-upload-limit=1K -z --log-level=error --dir=/path/to/download/directory --input-file=/path/to/torrent/file

bot/helper/mirror_utils/download_utils/aria2_download.py

@@ -1,15 +1,50 @@
-#!/usr/bin/env python3
-from aiofiles.os import remove as aioremove, path as aiopath
+import asyncio
+import logging
+from typing import (
+    Any,
+    Dict,
+    List,
+    Optional,
+    Union,
+)
 
+import aiofiles.os as aio_os
+import aiopath
+from aiogram import Bot, types
+from aiogram.methods import ForceReply
+from aiogram.types import CallbackQuery
+from aiogram.utils import executor
 from bot import aria2, download_dict_lock, download_dict, LOGGER, config_dict, aria2_options, aria2c_global, non_queued_dl, queue_dict_lock
 from bot.helper.ext_utils.bot_utils import bt_selection_buttons, sync_to_async
 from bot.helper.mirror_utils.status_utils.aria2_status import Aria2Status
 from bot.helper.telegram_helper.message_utils import sendStatusMessage, sendMessage
 from bot.helper.ext_utils.task_manager import is_queued
 
+class Constants:
+    TORRENT_TIMEOUT: Optional[int] = config_dict.get('TORRENT_TIMEOUT')
 
-async def add_aria2c_download(link, path, listener, filename, header, ratio, seed_time):
-    a2c_opt = {**aria2_options}
+async def add_aria2c_download(
+    link: str,
+    path: str,
+    listener: types.User,
+    filename: Optional[str] = None,
+    header: Optional[Dict[str, str]] = None,
+    ratio: Optional[float] = None,
+    seed_time: Optional[int] = None,
+) -> Union[None, Aria2Status]:
+    """
+    Adds a download to aria2c.
+
+    :param link: The link to download.
+    :param path: The path to download the file to.
+    :param listener: The user who initiated the download.
+    :param filename: The filename to use for the download.
+    :param header: The headers to use for the download.
+    :param ratio: The seed ratio to use for the download.
+    :param seed_time: The seed time to use for the download.
+    :return: The Aria2Status object for the download.
+    """
+    a2c_opt: Dict[str, Any] = {**aria2_options}
     [a2c_opt.pop(k) for k in aria2c_global if k in aria2_options]
     a2c_opt['dir'] = path
     if filename:
@@ -20,65 +55,73 @@ async def add_aria2c_download(link, path, listener, filename, header, ratio, see
         a2c_opt['seed-ratio'] = ratio
     if seed_time:
         a2c_opt['seed-time'] = seed_time
-    if TORRENT_TIMEOUT := config_dict['TORRENT_TIMEOUT']:
-        a2c_opt['bt-stop-timeout'] = f'{TORRENT_TIMEOUT}'
-    added_to_queue, event = await is_queued(listener.uid)
+    if Constants.TORRENT_TIMEOUT is not None:
+        a2c_opt['bt-stop-timeout'] = f'{Constants.TORRENT_TIMEOUT}'
+    added_to_queue, event = await is_queued(listener.id)
     if added_to_queue:
         if link.startswith('magnet:'):
             a2c_opt['pause-metadata'] = 'true'
         else:
             a2c_opt['pause'] = 'true'
     try:
-        download = (await sync_to_async(aria2.add, link, a2c_opt))[0]
+        download = await asyncio.wait_for(sync_to_async(aria2.add, link, a2c_opt), timeout=30)
+    except asyncio.TimeoutError as e:
+        LOGGER.debug(f"Aria2c Download Timeout: {e}")
+        await sendMessage(listener.message, 'Download request timed out.')
+        return
     except Exception as e:
-        LOGGER.info(f"Aria2c Download Error: {e}")
+        LOGGER.debug(f"Aria2c Download Error: {e}")
         await sendMessage(listener.message, f'{e}')
         return
-    if await aiopath.exists(link):
-        await aioremove(link)
+    if await aiopath.is_file(link):
+        await aio_os.remove(link)
     if download.error_message:
         error = str(download.error_message).replace('<', ' ').replace('>', ' ')
-        LOGGER.info(f"Aria2c Download Error: {error}")
+        LOGGER.debug(f"Aria2c Download Error: {error}")
         await sendMessage(listener.message, error)
         return
 
     gid = download.gid
     name = download.name
     async with download_dict_lock:
-        download_dict[listener.uid] = Aria2Status(
+        download_dict[listener.id] = Aria2Status(
             gid, listener, queued=added_to_queue)
     if added_to_queue:
-        LOGGER.info(f"Added to Queue/Download: {name}. Gid: {gid}")
+        LOGGER.debug(f"Added to Queue/Download: {name}. Gid: {gid}")
         if not listener.select or not download.is_torrent:
             await sendStatusMessage(listener.message)
     else:
         async with queue_dict_lock:
-            non_queued_dl.add(listener.uid)
-        LOGGER.info(f"Aria2Download started: {name}. Gid: {gid}")
+            non_queued_dl.add(listener.id)
+        LOGGER.debug(f"Aria2Download started: {name}. Gid: {gid}")
 
     await listener.onDownloadStart()
 
     if not added_to_queue and (not listener.select or not config_dict['BASE_URL']):
         await sendStatusMessage(listener.message)
     elif listener.select and download.is_torrent and not download.is_metadata:
         if not added_to_queue:
-            await sync_to_async(aria2.client.force_pause, gid)
+            await asyncio.create_task(sync_to_async(aria2.client.force_pause, gid))
         SBUTTONS = bt_selection_buttons(gid)
         msg = "Your download paused. Choose files then press Done Selecting button to start downloading."
-        await sendMessage(listener.message, msg, SBUTTONS)
+        await sendMessage(listener.message, msg, reply_markup=ForceReply(selective=True))
 
     if added_to_queue:
         await event.wait()
 
         async with download_dict_lock:
-            if listener.uid not in download_dict:
+            if listener.id not in download_dict:
                 return
-            download = download_dict[listener.uid]
+            download = download_dict[listener.id]
             download.queued = False
             new_gid = download.gid()
 
-        await sync_to_async(aria2.client.unpause, new_gid)
-        LOGGER.info(f'Start Queued Download from Aria2c: {name}. Gid: {gid}')
+        await asyncio.create_task(sync_to_async(aria2.client.unpause, new_gid))
+        LOGGER.debug(f'Start Queued Download from Aria2c: {name}. Gid: {gid}')
 
         async with queue_dict_lock:
-            non_queued_dl.add(listener.uid)
+            non_queued_dl.add(listener.id)
+
+if __name__ == '__main__':
+    logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+    executor.start_polling(bot)

bot/helper/mirror_utils/download_utils/mega_download.py

@@ -1,17 +1,33 @@
-#!/usr/bin/env python3
-from secrets import token_hex
-from aiofiles.os import makedirs
+import asyncio
+import contextlib
 from asyncio import Event
-from mega import MegaApi, MegaListener, MegaRequest, MegaTransfer, MegaError
+from typing import Any, Callable, Coroutine, Dict, List, Optional, Union
+
+import aiofiles.os
+from aiohttp import ClientSession
+from mega import MegaApi, MegaError, MegaListener, MegaRequest, MegaTransfer
+from telegram import Message, Update
 
 from bot import LOGGER, config_dict, download_dict_lock, download_dict, non_queued_dl, queue_dict_lock
 from bot.helper.telegram_helper.message_utils import sendMessage, sendStatusMessage
-from bot.helper.ext_utils.bot_utils import get_mega_link_type, async_to_sync, sync_to_async
+from bot.helper.ext_utils.bot_utils import get_mega_link_type, run_until_complete, as_completed
 from bot.helper.mirror_utils.status_utils.mega_download_status import MegaDownloadStatus
 from bot.helper.mirror_utils.status_utils.queue_status import QueueStatus
 from bot.helper.ext_utils.task_manager import is_queued, limit_checker, stop_duplicate_check
 
 
+@contextlib.asynccontextmanager
+async def download_dict_context():
+    async with download_dict_lock:
+        yield download_dict
+
+
+@contextlib.asynccontextmanager
+async def queue_dict_context():
+    async with queue_dict_lock:
+        yield non_queued_dl
+
+
 class MegaAppListener(MegaListener):
     _NO_EVENT_ON = (MegaRequest.TYPE_LOGIN, MegaRequest.TYPE_FETCH_NODES)
     NO_ERROR = "no error"
@@ -60,8 +76,8 @@ def onRequestTemporaryError(self, api, request, error: MegaError):
         LOGGER.error(f'Mega Request error in {error}')
         if not self.is_cancelled:
             self.is_cancelled = True
-            async_to_sync(self.listener.onDownloadError,
-                          f"RequestTempError: {error.toString()}")
+            run_until_complete(self.listener.onDownloadError,
+                               f"RequestTempError: {error.toString()}")
         self.error = error.toString()
         self.continue_event.set()
 
@@ -78,7 +94,7 @@ def onTransferFinish(self, api: MegaApi, transfer: MegaTransfer, error):
             if self.is_cancelled:
                 self.continue_event.set()
             elif transfer.isFinished() and (transfer.isFolderTransfer() or transfer.getFileName() == self.__name):
-                async_to_sync(self.listener.onDownloadComplete)
+                run_until_complete(self.listener.onDownloadComplete)
                 self.continue_event.set()
         except Exception as e:
             LOGGER.error(e)
@@ -97,8 +113,8 @@ def onTransferTemporaryError(self, api, transfer, error):
         self.error = errStr
         if not self.is_cancelled:
             self.is_cancelled = True
-            async_to_sync(self.listener.onDownloadError,
-                          f"TransferTempError: {errStr} ({filen})")
+            run_until_complete(self.listener.onDownloadError,
+                               f"TransferTempError: {errStr} ({filen})")
             self.continue_event.set()
 
     async def cancel_download(self):
@@ -111,13 +127,13 @@ class AsyncExecutor:
     def __init__(self):
         self.continue_event = Event()
 
-    async def do(self, function, args):
+    async def do(self, function: Callable, *args):
         self.continue_event.clear()
-        await sync_to_async(function, *args)
+        await run_until_complete(function, *args)
         await self.continue_event.wait()
 
 
-async def add_mega_download(mega_link, path, listener, name):
+async def add_mega_download(mega_link: str, path: str, listener: 'Listener', name: Optional[str] = None) -> Coroutine[Any, Any, None]:
     MEGA_EMAIL = config_dict['MEGA_EMAIL']
     MEGA_PASSWORD = config_dict['MEGA_PASSWORD']
 
@@ -138,8 +154,8 @@ async def add_mega_download(mega_link, path, listener, name):
         folder_api = MegaApi(None, None, None, 'WZML-X')
         folder_api.addListener(mega_listener)
         await executor.do(folder_api.loginToFolder, (mega_link,))
-        node = await sync_to_async(folder_api.authorizeNode, mega_listener.node)
-    if mega_listener.error is not None:
+        node = await run_until_complete(folder_api.authorizeNode, mega_listener.node)
+    if mega_listener.error:
         await sendMessage(listener.message, str(mega_listener.error))
         await executor.do(api.logout, ())
         if folder_api is not None:
@@ -157,19 +173,20 @@ async def add_mega_download(mega_link, path, listener, name):
 
     gid = token_hex(5)
     size = api.getSize(node)
-    if limit_exceeded := await limit_checker(size, listener, isMega=True):
+    limit_exceeded = await limit_checker(size, listener, isMega=True)
+    if limit_exceeded:
         await sendMessage(listener.message, limit_exceeded)
         return
     added_to_queue, event = await is_queued(listener.uid)
     if added_to_queue:
         LOGGER.info(f"Added to Queue/Download: {name}")
-        async with download_dict_lock:
+        async with download_dict_context() as download_dict:
             download_dict[listener.uid] = QueueStatus(
                 name, size, gid, listener, 'Dl')
         await listener.onDownloadStart()
         await sendStatusMessage(listener.message)
         await event.wait()
-        async with download_dict_lock:
+        async with download_dict_context() as download_dict:
             if listener.uid not in download_dict:
                 await executor.do(api.logout, ())
                 if folder_api is not None:
@@ -180,9 +197,9 @@ async def add_mega_download(mega_link, path, listener, name):
     else:
         from_queue = False
 
-    async with download_dict_lock:
+    async with download_dict_context() as download_dict:
         download_dict[listener.uid] = MegaDownloadStatus(name, size, gid, mega_listener, listener.message, listener.upload_details)
-    async with queue_dict_lock:
+    async with queue_dict_context() as non_queued_dl:
         non_queued_dl.add(listener.uid)
 
     if from_queue:
@@ -192,7 +209,7 @@ async def add_mega_download(mega_link, path, listener, name):
         await sendStatusMessage(listener.message)
         LOGGER.info(f"Download from Mega: {name}")
 
-    await makedirs(path, exist_ok=True)
+    await aiofiles.os.makedirs(path, exist_ok=True)
     await executor.do(api.startDownload, (node, path, name, None, False, None))
     await executor.do(api.logout, ())
     if folder_api is not None: