fix: new_task usage

Author: NTFSvolume

src/mega/__main__.py

@@ -1,12 +1,15 @@
 import argparse
 import asyncio
+import logging
 from pathlib import Path
 from pprint import pprint
 
 from mega import env
 from mega.client import MegaNzClient
 from mega.utils import setup_logger
 
+logger = logging.getLogger(__name__)
+
 
 async def run() -> None:
     setup_logger(__name__)
@@ -29,8 +32,9 @@ async def run() -> None:
     async with MegaNzClient() as mega:
         await mega.login(env.EMAIL, env.PASSWORD)
         stats = await mega.get_account_stats()
-        pprint(stats)  # noqa: T203
-        public_handle, public_key = mega.parse_file_url(args.url)
+        pprint(stats.dump())  # noqa: T203
+        public_handle, public_key = mega.parse_folder_url(args.url)
+        logger.info(f"Downloading {args.url}")
         with mega.show_progress_bar():
             await mega.download_public_folder(public_handle, public_key, args.output_dir)
 

src/mega/core.py

@@ -140,7 +140,7 @@ async def _upload(self, file_path: str | PathLike[str], dest_node_id: NodeID) ->
         file_path = Path(file_path)
         file_size = file_path.stat().st_size
 
-        with progress.new_task(file_path.name, file_size):
+        with progress.new_task(file_path.name, file_size, "UP"):
             file_id, crypto = await upload.upload(self._api, file_path, file_size)
             return await upload.finish_upload(
                 self._api,
@@ -168,7 +168,7 @@ async def _download_file(
         name = output_name or Attributes.parse(decrypt_attr(b64_url_decode(file_info._at), crypto.key)).name
         output_path = Path(output_folder or Path()) / name
 
-        with progress.new_task(output_path.name, file_info.size):
+        with progress.new_task(output_path.name, file_info.size, "DOWN"):
             async with self._api.download(file_info.url) as response:
                 return await download.stream_download(
                     response.content,

src/mega/progress.py

@@ -28,14 +28,21 @@ def __call__(self, description: str, total: float, kind: Literal["UP", "DOWN"])
 current_hook: ContextVar[ProgressHook] = ContextVar("current_hook", default=lambda _: None)
 
 
+def _truncate_desc(desc: str, length: int = 30, placeholder: str = "...") -> str:
+    if len(desc) < length:
+        return desc
+
+    return f"{desc[: length - len(placeholder)]}{placeholder}"
+
+
 @contextlib.contextmanager
 def new_task(description: str, total: float, kind: Literal["UP", "DOWN"]) -> Generator[None]:
     factory = _PROGRESS_HOOK_FACTORY.get()
     if factory is None:
         yield
         return
 
-    with factory(description, total, kind) as progress_hook:
+    with factory(_truncate_desc(description), total, kind) as progress_hook:
         token = current_hook.set(progress_hook)
         try:
             yield
@@ -69,7 +76,7 @@ async def task(name: str) -> None:
         total = 1e6 * random.randint(200, 2000)
         max_step = int(total / 20)
         kind = random.choice(("UP", "DOWN"))
-        with new_task(name, total, kind=kind):
+        with new_task(name, total, kind):
             advance = current_hook.get()
             done = 0
             while done < total: