refactor: replace typer with cyclopts

Author: NTFSvolume

pyproject.toml

@@ -38,17 +38,13 @@ version = "2.0.3"
 
 [project.optional-dependencies]
 cli = [
-    "async-mega-py[default]",
-    "typer-slim>=0.21.1"
-]
-default = [
-    "python-dotenv>=1.2.1",
-    "rich>=14.0.0"
+    "cyclopts>=4.10.1",
+    "python-dotenv>=1.2.1"
 ]
 
 [project.scripts]
-async-mega-py = "mega.cli:main"
-mega-py = "mega.cli:main"
+async-mega-py = "mega.__main__:app"
+mega-py = "mega.__main__:app"
 
 [project.urls]
 Homepage = "https://github.com/NTFSvolume/mega.py"
@@ -131,6 +127,5 @@ dev = [
     "pytest-asyncio>=0.25.0",
     "pytest>=8.3.5",
     "rich>=14.0.0",
-    "ruff>=0.15.0",
-    "typer-slim>=0.21.1"
+    "ruff>=0.15.0"
 ]

src/mega/__main__.py

@@ -1,4 +1,147 @@
-from mega.cli import main
+from __future__ import annotations
 
-if __name__ == "__main__":
-    main()
+import contextlib
+import json
+import logging
+from pathlib import Path
+from typing import TYPE_CHECKING, Annotated
+
+import yarl
+from cyclopts import App, Parameter
+
+from mega import __version__, env
+from mega.api import LOG_HTTP_TRAFFIC
+from mega.client import MegaNzClient
+from mega.transfer_it import TransferItClient
+from mega.utils import Site, setup_logger
+
+if TYPE_CHECKING:
+    from collections.abc import AsyncGenerator
+
+
+logger = logging.getLogger("mega")
+
+Verbose = Annotated[
+    int,
+    Parameter(
+        name=["-v", "--verbose"], count=True, help="Increase verbosity (-v shows debug logs,  -vv shows HTTP traffic)"
+    ),
+]
+
+
+def parse_verbose(count: int) -> None:
+    if count > 1:
+        LOG_HTTP_TRAFFIC.set(True)
+
+    level = logging.DEBUG if count else logging.INFO
+    setup_logger(level)
+
+
+app = App(
+    help=(
+        "CLI app for the [bold black]Mega.nz[/bold black] and [bold black]Transfer.it[/bold black].\n"
+        f"Set [bold green]{env.EMAIL.name}[/bold green] and [bold green]{env.PASSWORD.name}[/bold green]\n"
+        "enviroment variables to use them as credentials for Mega"
+    ),
+    version=__version__,
+)
+CWD = Path.cwd()
+
+
+@contextlib.asynccontextmanager
+async def connect() -> AsyncGenerator[MegaNzClient]:
+    async with MegaNzClient() as mega:
+        await mega.login(env.EMAIL, env.PASSWORD)
+        with mega.progress_bar:
+            yield mega
+
+
+async def transfer_it(url: str, output_dir: Path) -> None:
+    async with TransferItClient() as client:
+        with client.progress_bar:
+            transfer_id = client.parse_url(url)
+            logger.info(f"Downloading '{url}'")
+            results = await client.download_transfer(transfer_id, output_dir)
+            logger.info(
+                f"Download of '{url}' finished. Successful = {len(results.success)}, failed = {len(results.fails)}"
+            )
+
+
+@app.command()
+async def download(url: str, output_dir: Path = CWD, *, verbose: Verbose = 0) -> None:
+    """Download a public file or folder by its URL (transfer.it / mega.nz)"""
+    parse_verbose(verbose)
+    site = Site(yarl.URL(url).origin())
+    if site is Site.TRANSFER_IT:
+        return await transfer_it(url, output_dir)
+
+    async with connect() as mega:
+        parsed_url = mega.parse_url(url)
+        if parsed_url.is_folder:
+            await download_folder(mega, url, output_dir)
+        else:
+            await download_file(mega, url, output_dir)
+
+
+@app.command()
+async def dump(output_dir: Path = CWD, *, verbose: Verbose = 0) -> None:
+    """Dump a copy of your filesystem to disk"""
+    parse_verbose(verbose)
+    async with connect() as mega:
+        fs = await mega.get_filesystem()
+        out = output_dir / "filesystem.json"
+        out.parent.mkdir(exist_ok=True)
+        logger.info(f"Creating filesystem dump at '{out!s}'")
+        out.write_text(json.dumps(fs.dump(), indent=2, ensure_ascii=False))
+
+
+@app.command()
+async def stats(*, verbose: Verbose = 0) -> None:
+    """Show account stats"""
+    parse_verbose(verbose)
+    async with connect() as mega:
+        stats = await mega.get_account_stats()
+        logger.info(f"Account stats for {env.EMAIL or 'TEMP ACCOUNT'}:")
+        logger.info(stats.storage.dump())
+        logger.info(stats.balance.dump())
+        fs = await mega.get_filesystem()
+        metrics = {root.attributes.name: stats.metrics[root.id] for root in (fs.root, fs.inbox, fs.trash_bin)}
+        logger.info(metrics)
+
+
+@app.command()
+async def upload(file_path: Path, *, verbose: Verbose = 0) -> None:
+    """Upload a file to your account"""
+    parse_verbose(verbose)
+    async with connect() as mega:
+        if not env.EMAIL:
+            logger.warning("Files uploaded by a temp account can not be exported")
+
+        folder = await mega.create_folder("uploaded by mega.py")
+        logger.info(f'Uploading "{file_path!s}"')
+        file = await mega.upload(file_path, folder.id)
+        path = (await mega.get_filesystem()).absolute_path(file.id)
+        logger.info(f'File uploaded to your cloud. Path = "{path}"')
+        if not env.EMAIL:
+            return
+
+        link = await mega.export(file)
+        logger.info(f'Public link for "{file_path!s}": {link}')
+
+
+async def download_file(mega: MegaNzClient, url: str, output: Path) -> None:
+    public_handle, public_key = mega.parse_file_url(url)
+    logger.info(f"Downloading {url}")
+    path = await mega.download_public_file(public_handle, public_key, output)
+    logger.info(f'Download of {url} finished. File save at "{path!s}"')
+
+
+async def download_folder(mega: MegaNzClient, url: str, output: Path) -> None:
+    public_handle, public_key, root_node = mega.parse_folder_url(url)
+    logger.info(f"Downloading {url}")
+    results = await mega.download_public_folder(public_handle, public_key, output, root_node)
+    logger.info(f"Download of '{url}' finished. Successful = {len(results.success)}, failed = {len(results.fails)}")
+
+
+if __name__ == "__main_":
+    app()