diff --git a/.gitignore b/.gitignore index 8893307..c1440e8 100644 --- a/.gitignore +++ b/.gitignore @@ -4,4 +4,5 @@ **/dist/ **/.env **/.venv/ -site/ \ No newline at end of file +site/ +.cursor/ \ No newline at end of file diff --git a/README.md b/README.md index 724b0d2..4bdc165 100644 --- a/README.md +++ b/README.md @@ -202,7 +202,7 @@ List fields in search params are serialized to comma-separated strings when you | `gopro_api/api/models.py` | Pydantic request/response models | | `gopro_api/api/__init__.py` | Re-exports `GoProAPI`, `AsyncGoProAPI` | | `gopro_api/config.py` | pydantic-settings `Settings`, `GP_ACCESS_TOKEN` | -| `gopro_api/cli.py` | `gopro-api` CLI | +| `gopro_api/cli/` | `gopro-api` CLI | | `setup.py` | Package metadata, dependencies, console entry point | ## CI and releases diff --git a/docs/api/cli.md b/docs/api/cli.md index 98be663..3fad183 100644 --- a/docs/api/cli.md +++ b/docs/api/cli.md @@ -19,3 +19,11 @@ Built with [Typer](https://typer.tiangolo.com/); the Typer application is expose ::: gopro_api.cli.info_command ::: gopro_api.cli.pull_command + +## Printers + +::: gopro_api.cli.search.SearchPrinter + +::: gopro_api.cli.info.InfoPrinter + +::: gopro_api.cli.pull.PullPrinter diff --git a/gopro_api/api/__init__.py b/gopro_api/api/__init__.py index dd08c30..b293913 100644 --- a/gopro_api/api/__init__.py +++ b/gopro_api/api/__init__.py @@ -7,5 +7,4 @@ from .gopro import GoProAPI from .async_gopro import AsyncGoProAPI - __all__ = ["GoProAPI", "AsyncGoProAPI"] diff --git a/gopro_api/api/models.py b/gopro_api/api/models.py index 33071ca..5cd8ee3 100644 --- a/gopro_api/api/models.py +++ b/gopro_api/api/models.py @@ -7,7 +7,6 @@ from pydantic import BaseModel, ConfigDict, Field, field_serializer, model_serializer - DEFAULT_PROCESSING_STATES: List[str] = [ "rendering", "pretranscoding", diff --git a/gopro_api/cli.py b/gopro_api/cli.py deleted file mode 100644 index c4ad1a8..0000000 --- a/gopro_api/cli.py +++ /dev/null @@ -1,371 +0,0 @@ -"""Command-line interface for gopro-api.""" - -from __future__ import annotations - -import asyncio -import json -import os -import sys -from datetime import datetime -from importlib.metadata import PackageNotFoundError, version as package_version -from typing import Optional - -import typer - -from gopro_api.api.models import ( - DEFAULT_FIELDS, - CapturedRange, - GoProMediaSearchItem, - GoProMediaSearchParams, - GoProMediaSearchResponse, -) -from gopro_api.client import AsyncGoProClient -from gopro_api.config import GP_ACCESS_TOKEN -from gopro_api.exceptions import NoVariationsError -from gopro_api.utils import is_video_filename, pull_assets_for_response - -app = typer.Typer( - name="gopro-api", - help="CLI for the unofficial GoPro cloud API (api.gopro.com).", - no_args_is_help=True, -) - - -def _version() -> str: - try: - return package_version("gopro-api") - except PackageNotFoundError: - return "0.0.0" - - -def _version_callback(value: Optional[bool]) -> None: - if value: - typer.echo(f"gopro-api {_version()}") - raise typer.Exit() - - -@app.callback() -def _main_callback( - ctx: typer.Context, - timeout: float = typer.Option( - 60.0, - "--timeout", - help="HTTP timeout in seconds (default: 60)", - ), - show_version: Optional[bool] = typer.Option( - None, - "--version", - callback=_version_callback, - is_eager=True, - help="Show version and exit.", - ), -) -> None: - del show_version # handled by eager callback (--version exits before commands) - ctx.ensure_object(dict) - ctx.obj["timeout"] = timeout - - -def _parse_dt(raw: str) -> datetime: - """Parse a CLI date or datetime string. - - Args: - raw: ``YYYY-MM-DD`` or full ISO 8601 string (``Z`` normalized to offset). - - Returns: - Parsed naive or aware ``datetime`` as produced by ``fromisoformat``. - - Raises: - ValueError: If the string is not a valid ISO date/datetime. - """ - raw = raw.strip() - if len(raw) == 10 and raw[4] == "-" and raw[7] == "-": - return datetime.fromisoformat(raw) - return datetime.fromisoformat(raw.replace("Z", "+00:00")) - - -def _validate_positive_px(value: Optional[int], flag: str) -> Optional[int]: - if value is None: - return None - if value <= 0: - raise typer.BadParameter("must be a positive integer", param_hint=flag) - return value - - -def _require_token() -> None: - """Ensure ``GP_ACCESS_TOKEN`` is configured. - - Raises: - typer.Exit: With code ``2`` if the token is missing. - """ - if not GP_ACCESS_TOKEN: - typer.echo( - "error: GP_ACCESS_TOKEN is not set. " - "Add it to your environment or a .env file.", - err=True, - ) - raise typer.Exit(2) - - -def _print_search_plain_header() -> None: - cols = list(DEFAULT_FIELDS) - print("\t".join(cols)) - - -def _format_search_item_plain(item: GoProMediaSearchItem) -> str: - row = item.model_dump(mode="json") - cells = ["" if row.get(c) is None else str(row[c]) for c in DEFAULT_FIELDS] - return "\t".join(cells) - - -def _print_search_plain_page( - page_result: GoProMediaSearchResponse, - *, - print_header: bool = True, -) -> None: - pages = page_result.pages - print( - f"# _pages: current_page={pages.current_page} per_page={pages.per_page} " - f"total_items={pages.total_items} total_pages={pages.total_pages}", - ) - if page_result.embedded.errors: - for err in page_result.embedded.errors: - print( - f"# _embedded.errors: {json.dumps(err, ensure_ascii=False)}", - file=sys.stderr, - ) - if print_header: - _print_search_plain_header() - for item in page_result.embedded.media: - print(_format_search_item_plain(item)) - - -async def _run_search( - *, - timeout: float, - start: str, - end: str, - page: int, - per_page: int, - all_pages: bool, - json_out: bool, -) -> None: - _require_token() - start_dt = _parse_dt(start) - end_dt = _parse_dt(end) - - async with AsyncGoProClient(timeout=timeout) as client: - if all_pages: - all_pages_payload: list[dict] = [] - first_plain_page = True - async for page_result in client.iter_nonempty_search_pages( - start_dt, - end_dt, - per_page=per_page, - start_page=page, - ): - if json_out: - all_pages_payload.append( - page_result.model_dump(by_alias=True, mode="json"), - ) - else: - _print_search_plain_page( - page_result, - print_header=first_plain_page, - ) - first_plain_page = False - if json_out: - print(json.dumps(all_pages_payload, indent=2)) - return - - params = GoProMediaSearchParams( - captured_range=CapturedRange(start=start_dt, end=end_dt), - page=page, - per_page=per_page, - ) - page_result = await client.search(params) - if json_out: - print( - json.dumps( - page_result.model_dump(by_alias=True, mode="json"), - indent=2, - ), - ) - else: - _print_search_plain_page(page_result) - - -@app.command( - "search", - help=( - "List media in a capture date range (tab-separated fields; " - "use --json for raw API payloads)" - ), -) -def search_command( - ctx: typer.Context, - *, - start: str = typer.Option( - ..., - "--start", - help="Range start: YYYY-MM-DD or ISO datetime", - ), - end: str = typer.Option( - ..., - "--end", - help=( - "Range end: YYYY-MM-DD or ISO datetime " - "(API treats range as in query string)" - ), - ), - page: int = typer.Option(1, "--page", help="Page number (default: 1)"), - per_page: int = typer.Option(30, "--per-page", help="Page size (default: 30)"), - all_pages: bool = typer.Option( - False, - "--all-pages", - help="Keep requesting pages until a page returns no media", - ), - json_out: bool = typer.Option( - False, - "--json", - help="Print full API JSON (with --all-pages: list of page payloads)", - ), -) -> None: - """Run search against the cloud API and print results.""" - timeout = ctx.obj["timeout"] - asyncio.run( - _run_search( - timeout=timeout, - start=start, - end=end, - page=page, - per_page=per_page, - all_pages=all_pages, - json_out=json_out, - ), - ) - - -async def _run_info( - *, - timeout: float, - media_id: str, - json_out: bool, -) -> None: - _require_token() - async with AsyncGoProClient(timeout=timeout) as client: - meta = await client.download(media_id) - if json_out: - print( - json.dumps( - meta.model_dump(by_alias=True, mode="json"), - indent=2, - ), - ) - else: - print(meta.filename) - media_list = ( - meta.embedded.variations - if is_video_filename(meta.filename) - else meta.embedded.files - ) - for idx, media_item in enumerate(media_list): - print( - f" {idx:>3} {media_item.width}x{media_item.height} " - f"{media_item.url}", - ) - - -@app.command("info", help="Show download metadata (URLs, sizes) for one media id") -def info_command( - ctx: typer.Context, - media_id: str = typer.Argument(..., help="Media id from search"), - json_out: bool = typer.Option(False, "--json", help="Print full API JSON"), -) -> None: - """Fetch and display download metadata for ``media_id``.""" - asyncio.run( - _run_info(timeout=ctx.obj["timeout"], media_id=media_id, json_out=json_out), - ) - - -async def _run_pull( - *, - timeout: float, - media_id: str, - destination: str, - height: Optional[int], - width: Optional[int], -) -> None: - _require_token() - async with AsyncGoProClient(timeout=timeout) as client: - meta = await client.download(media_id) - try: - assets = pull_assets_for_response( - meta, - target_height=height, - target_width=width, - ) - except NoVariationsError as exc: - typer.echo(f"error: {exc}", err=True) - raise typer.Exit(2) from exc - - os.makedirs(destination, exist_ok=True) - await asyncio.gather( - *( - client.download_url_to_path( - asset.url, - os.path.join(destination, filename), - ) - for filename, asset in assets.items() - ) - ) - - -@app.command("pull", help="Download files from a media id") -def pull_command( - ctx: typer.Context, - media_id: str = typer.Argument(..., help="Media id from search"), - destination: str = typer.Argument(..., help="Path to save the file"), - height: Optional[int] = typer.Option( - None, - "--height", - metavar="PX", - help=( - "For video: pick the variation whose height is closest to PX " - "(default: tallest)" - ), - ), - width: Optional[int] = typer.Option( - None, - "--width", - metavar="PX", - help=( - "For video: pick the variation whose width is closest to PX " - "(default: tallest)" - ), - ), -) -> None: - """Download all resolved files for ``media_id`` into ``destination``.""" - height = _validate_positive_px(height, "--height") - width = _validate_positive_px(width, "--width") - asyncio.run( - _run_pull( - timeout=ctx.obj["timeout"], - media_id=media_id, - destination=destination, - height=height, - width=width, - ), - ) - - -def main(argv: Optional[list[str]] = None) -> None: - """CLI entrypoint: parse ``argv`` and run the selected command. - - Args: - argv: Argument list (defaults to ``sys.argv[1:]`` when ``None``). - """ - app(args=argv) - - -if __name__ == "__main__": - main() diff --git a/gopro_api/cli/__init__.py b/gopro_api/cli/__init__.py new file mode 100644 index 0000000..2abf5bc --- /dev/null +++ b/gopro_api/cli/__init__.py @@ -0,0 +1,11 @@ +"""gopro_api.cli — public surface for the gopro-api CLI. + +Importing this package registers all Typer commands on ``app``. +""" + +from .app import app, main +from .search import search_command +from .info import info_command +from .pull import pull_command + +__all__ = ["app", "main", "search_command", "info_command", "pull_command"] diff --git a/gopro_api/cli/__main__.py b/gopro_api/cli/__main__.py new file mode 100644 index 0000000..7c25c40 --- /dev/null +++ b/gopro_api/cli/__main__.py @@ -0,0 +1,5 @@ +"""Entry point for ``python -m gopro_api.cli``.""" + +from .app import main + +main() diff --git a/gopro_api/cli/_common.py b/gopro_api/cli/_common.py new file mode 100644 index 0000000..52d5ada --- /dev/null +++ b/gopro_api/cli/_common.py @@ -0,0 +1,119 @@ +"""Shared helpers: token gating, date parsing, pixel validation, Rich table builder.""" + +from __future__ import annotations + +from datetime import datetime +from typing import Optional + +import typer +from rich.table import Table + +from gopro_api.config import GP_ACCESS_TOKEN + +_FIELD_LABELS: dict[str, str] = { + "type": "media", + "file_extension": "type", + "file_size": "size", + "item_count": "items", +} + + +def _renamed_field(name: str) -> str: + """Return the display label for an API field name. + + Args: + name: Internal API field name. + + Returns: + Human-readable column label, or ``name`` unchanged when no mapping exists. + """ + return _FIELD_LABELS.get(name, name) + + +def _require_token() -> None: + """Ensure ``GP_ACCESS_TOKEN`` is configured. + + Raises: + typer.Exit: With code ``2`` if the token is missing. + """ + if not GP_ACCESS_TOKEN: + typer.secho( + "error: GP_ACCESS_TOKEN is not set. " + "Add it to your environment or a .env file.", + fg=typer.colors.RED, + bold=True, + err=True, + ) + raise typer.Exit(2) + + +def _parse_dt(raw: str) -> datetime: + """Parse a CLI date or datetime string. + + Args: + raw: ``YYYY-MM-DD`` or full ISO 8601 string (``Z`` normalized to offset). + + Returns: + Parsed naive or aware ``datetime`` as produced by ``fromisoformat``. + + Raises: + ValueError: If the string is not a valid ISO date/datetime. + """ + raw = raw.strip() + if len(raw) == 10 and raw[4] == "-" and raw[7] == "-": + return datetime.fromisoformat(raw) + return datetime.fromisoformat(raw.replace("Z", "+00:00")) + + +def _validate_positive_px(value: Optional[int], flag: str) -> Optional[int]: + """Validate that a pixel dimension option is a positive integer. + + Args: + value: The integer provided by the user, or ``None`` if omitted. + flag: CLI flag name shown in the error message (e.g. ``"--height"``). + + Returns: + The validated value, or ``None`` when not provided. + + Raises: + typer.BadParameter: If ``value`` is zero or negative. + """ + if value is None: + return None + if value <= 0: + raise typer.BadParameter("must be a positive integer", param_hint=flag) + return value + + +def _build_basic_table( + headers: list[str], *, fold_cols: tuple[str, ...] = ("url",) +) -> Table: + """Build a Rich Table with the given column headers. + + Args: + headers: Ordered list of column names to add. + fold_cols: Column names that should use ``overflow="fold"``; defaults to + ``("url",)`` so long URLs wrap inside their cell. + + Returns: + A ``rich.table.Table`` ready for row insertion. + """ + table = Table(show_header=True, header_style="bold") + for h in headers: + if h in fold_cols: + table.add_column(h, overflow="fold") + else: + table.add_column(h) + return table + + +def _yes_no(value: bool) -> str: + """Format a boolean as a human-readable string. + + Args: + value: Boolean to format. + + Returns: + ``"yes"`` when ``True``, ``"no"`` when ``False``. + """ + return "yes" if value else "no" diff --git a/gopro_api/cli/app.py b/gopro_api/cli/app.py new file mode 100644 index 0000000..1443ee8 --- /dev/null +++ b/gopro_api/cli/app.py @@ -0,0 +1,77 @@ +"""Typer application instance, root callback, and CLI entrypoint.""" + +from __future__ import annotations + +from importlib.metadata import PackageNotFoundError, version as package_version +from typing import Optional + +import typer + +app = typer.Typer( + name="gopro-api", + help="CLI for the unofficial GoPro cloud API (api.gopro.com).", + no_args_is_help=True, +) + + +def _version() -> str: + """Return the installed package version string. + + Returns: + Version string from package metadata, or ``"0.0.0"`` when not installed. + """ + try: + return package_version("gopro-api") + except PackageNotFoundError: + return "0.0.0" + + +def _version_callback(value: Optional[bool]) -> None: + """Print the package version and exit when ``--version`` is passed. + + Args: + value: ``True`` when the ``--version`` flag is present; ``None`` otherwise. + + Raises: + typer.Exit: Immediately after printing the version string. + """ + if value: + typer.echo(f"gopro-api {_version()}") + raise typer.Exit() + + +@app.callback() +def _main_callback( + ctx: typer.Context, + timeout: float = typer.Option( + 60.0, + "--timeout", + help="HTTP timeout in seconds (default: 60)", + ), + show_version: Optional[bool] = typer.Option( + None, + "--version", + callback=_version_callback, + is_eager=True, + help="Show version and exit.", + ), +) -> None: + """Store shared CLI options in the Typer context for all subcommands. + + Args: + ctx: Typer context used to pass ``timeout`` to subcommands. + timeout: HTTP timeout in seconds forwarded to ``AsyncGoProClient``. + show_version: Consumed by the eager ``--version`` callback before this runs. + """ + del show_version # handled by eager callback (--version exits before commands) + ctx.ensure_object(dict) + ctx.obj["timeout"] = timeout + + +def main(argv: Optional[list[str]] = None) -> None: + """CLI entrypoint: parse ``argv`` and run the selected command. + + Args: + argv: Argument list (defaults to process arguments when ``None``). + """ + app(args=argv) diff --git a/gopro_api/cli/info.py b/gopro_api/cli/info.py new file mode 100644 index 0000000..b8d1401 --- /dev/null +++ b/gopro_api/cli/info.py @@ -0,0 +1,199 @@ +"""Info command: InfoPrinter class, async runner, and Typer callback.""" + +from __future__ import annotations + +import asyncio +import json + +import typer +from rich.console import Console + +from gopro_api.api.models import ( + GoProMediaDownloadFile, + GoProMediaDownloadResponse, + GoProMediaDownloadSidecarFile, + GoProMediaDownloadVariation, +) +from gopro_api.client import AsyncGoProClient +from gopro_api.utils import is_video_filename + +from .app import app +from ._common import _build_basic_table, _require_token, _yes_no + + +class InfoPrinter: + """Handles Rich table and TSV rendering for the info command.""" + + VARIATION_HEADERS = ["idx", "label", "quality", "type", "dim", "available", "url"] + FILE_HEADERS = ["idx", "item", "camera", "dim", "available", "url"] + SIDECAR_HEADERS = ["idx", "label", "type", "fps", "available", "url"] + + def __init__(self, console: Console | None = None) -> None: + """Initialize with an optional Rich console. + + Args: + console: Console used for Rich output; a default soft-wrap console is + created when ``None``. + """ + self._console = console or Console(soft_wrap=True) + + def variation_cells(self, idx: int, v: GoProMediaDownloadVariation) -> list[str]: + """Build row cells for a video variation entry. + + Args: + idx: Zero-based row index shown in the ``idx`` column. + v: Variation object from the download metadata. + + Returns: + Ordered list of strings matching ``VARIATION_HEADERS``. + """ + return [ + str(idx), + v.label, + v.quality, + v.type, + f"{v.width}x{v.height}", + _yes_no(v.available), + v.url, + ] + + def file_cells(self, idx: int, f: GoProMediaDownloadFile) -> list[str]: + """Build row cells for a multi-lens file entry. + + Args: + idx: Zero-based row index shown in the ``idx`` column. + f: File object from the download metadata. + + Returns: + Ordered list of strings matching ``FILE_HEADERS``. + """ + return [ + str(idx), + str(f.item_number), + f.camera_position, + f"{f.width}x{f.height}", + _yes_no(f.available), + f.url, + ] + + def sidecar_cells(self, idx: int, s: GoProMediaDownloadSidecarFile) -> list[str]: + """Build row cells for a sidecar file entry. + + Args: + idx: Zero-based row index shown in the ``idx`` column. + s: Sidecar file object from the download metadata. + + Returns: + Ordered list of strings matching ``SIDECAR_HEADERS``. + """ + return [ + str(idx), + s.label, + s.type, + str(s.fps), + _yes_no(s.available), + s.url, + ] + + def print_rich(self, meta: GoProMediaDownloadResponse) -> None: + """Print a Rich table of variations or files, plus sidecars when present. + + Args: + meta: Download metadata response from the GoPro API. + """ + typer.secho(meta.filename, bold=True) + if is_video_filename(meta.filename): + table = _build_basic_table(self.VARIATION_HEADERS) + for idx, v in enumerate(meta.embedded.variations): + table.add_row(*self.variation_cells(idx, v)) + else: + table = _build_basic_table(self.FILE_HEADERS) + for idx, f in enumerate(meta.embedded.files): + table.add_row(*self.file_cells(idx, f)) + self._console.print(table) + if meta.embedded.sidecar_files: + typer.secho("sidecars", bold=True) + sidecars = _build_basic_table(self.SIDECAR_HEADERS) + for idx, s in enumerate(meta.embedded.sidecar_files): + sidecars.add_row(*self.sidecar_cells(idx, s)) + self._console.print(sidecars) + + def print_tsv(self, meta: GoProMediaDownloadResponse) -> None: + """Print tab-separated rows of variations or files, plus sidecars when present. + + Args: + meta: Download metadata response from the GoPro API. + """ + typer.echo(f"# filename: {meta.filename}") + if is_video_filename(meta.filename): + typer.echo("\t".join(self.VARIATION_HEADERS)) + for idx, v in enumerate(meta.embedded.variations): + typer.echo("\t".join(self.variation_cells(idx, v))) + else: + typer.echo("\t".join(self.FILE_HEADERS)) + for idx, f in enumerate(meta.embedded.files): + typer.echo("\t".join(self.file_cells(idx, f))) + if meta.embedded.sidecar_files: + typer.echo("# sidecars") + typer.echo("\t".join(self.SIDECAR_HEADERS)) + for idx, s in enumerate(meta.embedded.sidecar_files): + typer.echo("\t".join(self.sidecar_cells(idx, s))) + + +async def _run_info( + *, + timeout: float, + media_id: str, + json_out: bool, + tsv: bool, +) -> None: + """Fetch download metadata for ``media_id`` and render the output. + + Args: + timeout: HTTP timeout in seconds passed to ``AsyncGoProClient``. + media_id: GoPro cloud media identifier. + json_out: When ``True``, emit raw API JSON instead of a table. + tsv: When ``True``, emit tab-separated values instead of a Rich table. + """ + _require_token() + async with AsyncGoProClient(timeout=timeout) as client: + meta = await client.download(media_id) + if json_out: + typer.echo( + json.dumps( + meta.model_dump(by_alias=True, mode="json"), + indent=2, + ), + ) + elif tsv: + InfoPrinter().print_tsv(meta) + else: + InfoPrinter().print_rich(meta) + + +@app.command( + "info", + help=( + "Show download metadata for one media id " + "(Rich table by default; --tsv for tab-separated; --json for raw API)" + ), +) +def info_command( + ctx: typer.Context, + media_id: str = typer.Argument(..., help="Media id from search"), + tsv: bool = typer.Option( + False, + "--tsv", + help="Print tab-separated values for scripting", + ), + json_out: bool = typer.Option(False, "--json", help="Print full API JSON"), +) -> None: + """Fetch and display download metadata for ``media_id``.""" + asyncio.run( + _run_info( + timeout=ctx.obj["timeout"], + media_id=media_id, + json_out=json_out, + tsv=tsv, + ), + ) diff --git a/gopro_api/cli/pull.py b/gopro_api/cli/pull.py new file mode 100644 index 0000000..bc13ba3 --- /dev/null +++ b/gopro_api/cli/pull.py @@ -0,0 +1,182 @@ +"""Pull command: PullPrinter class, async runner, and Typer callback.""" + +from __future__ import annotations + +import asyncio +import os +from typing import Optional + +import typer +from rich.console import Console + +from gopro_api.client import AsyncGoProClient +from gopro_api.exceptions import NoVariationsError +from gopro_api.utils import DownloadAsset, pull_assets_for_response + +from .app import app +from ._common import _build_basic_table, _require_token, _validate_positive_px, _yes_no + + +class PullPrinter: + """Handles Rich table and TSV rendering for the pull command.""" + + HEADERS = ["filename", "dim", "available", "url"] + + def __init__(self, console: Console | None = None) -> None: + """Initialize with an optional Rich console. + + Args: + console: Console used for Rich output; a default soft-wrap console is + created when ``None``. + """ + self._console = console or Console(soft_wrap=True) + + def summary_cells(self, filename: str, asset: DownloadAsset) -> list[str]: + """Build row cells for a single download asset. + + Args: + filename: Local filename that the asset will be saved as. + asset: Resolved download asset containing URL and dimensions. + + Returns: + Ordered list of strings matching ``HEADERS``. + """ + return [ + filename, + f"{asset.width}x{asset.height}", + _yes_no(asset.available), + asset.url, + ] + + def print_rich(self, assets: dict[str, DownloadAsset], destination: str) -> None: + """Print a Rich summary table of all assets to be downloaded. + + Args: + assets: Mapping of local filename to resolved download asset. + destination: Target directory path shown in the header line. + """ + typer.secho( + f"Pulling {len(assets)} file(s) to {destination}", + bold=True, + ) + table = _build_basic_table(self.HEADERS) + for filename, asset in assets.items(): + table.add_row(*self.summary_cells(filename, asset)) + self._console.print(table) + + def print_tsv(self, assets: dict[str, DownloadAsset], destination: str) -> None: + """Print a tab-separated summary of all assets for scripting. + + Args: + assets: Mapping of local filename to resolved download asset. + destination: Target directory path shown in the comment header. + """ + typer.echo(f"# destination: {destination}") + typer.echo("\t".join(self.HEADERS)) + for filename, asset in assets.items(): + typer.echo("\t".join(self.summary_cells(filename, asset))) + + +async def _run_pull( + *, + timeout: float, + media_id: str, + destination: str, + height: Optional[int], + width: Optional[int], + tsv: bool, +) -> None: + """Resolve assets for ``media_id``, print a summary, and download the files. + + Args: + timeout: HTTP timeout in seconds passed to ``AsyncGoProClient``. + media_id: GoPro cloud media identifier. + destination: Local directory path where files will be saved. + height: Preferred variation height in pixels; ``None`` selects the tallest. + width: Preferred variation width in pixels; used when ``height`` is unset. + tsv: When ``True``, emit a tab-separated summary instead of a Rich table. + + Raises: + typer.Exit: With code ``2`` if no variations are available for the media. + """ + _require_token() + printer = PullPrinter() + async with AsyncGoProClient(timeout=timeout) as client: + meta = await client.download(media_id) + try: + assets = pull_assets_for_response( + meta, + target_height=height, + target_width=width, + ) + except NoVariationsError as exc: + typer.secho(f"error: {exc}", fg=typer.colors.RED, bold=True, err=True) + raise typer.Exit(2) from exc + + if tsv: + printer.print_tsv(assets, destination) + else: + printer.print_rich(assets, destination) + + os.makedirs(destination, exist_ok=True) + await asyncio.gather( + *( + client.download_url_to_path( + asset.url, + os.path.join(destination, filename), + ) + for filename, asset in assets.items() + ) + ) + + typer.secho(f"Done. ({len(assets)} file(s))", fg=typer.colors.GREEN) + + +@app.command( + "pull", + help=( + "Download files from a media id (prints a Rich summary by default; " + "--tsv for tab-separated)" + ), +) +def pull_command( # pylint: disable=too-many-positional-arguments + ctx: typer.Context, + media_id: str = typer.Argument(..., help="Media id from search"), + destination: str = typer.Argument(..., help="Path to save the file"), + height: Optional[int] = typer.Option( + None, + "--height", + metavar="PX", + help=( + "For video: pick the variation whose height is closest to PX " + "(default: tallest)" + ), + ), + width: Optional[int] = typer.Option( + None, + "--width", + metavar="PX", + help=( + "For video: pick the variation whose width is closest to PX " + "(default: tallest)" + ), + ), + tsv: bool = typer.Option( + False, + "--tsv", + help="Print tab-separated summary instead of the Rich table", + ), +) -> None: + """Download all resolved files for ``media_id`` into ``destination``.""" + height = _validate_positive_px(height, "--height") + width = _validate_positive_px(width, "--width") + asyncio.run( + _run_pull( + timeout=ctx.obj["timeout"], + media_id=media_id, + destination=destination, + height=height, + width=width, + tsv=tsv, + ), + ) diff --git a/gopro_api/cli/search.py b/gopro_api/cli/search.py new file mode 100644 index 0000000..5a6bc73 --- /dev/null +++ b/gopro_api/cli/search.py @@ -0,0 +1,372 @@ +"""Search command: SearchPrinter class, async runner, and Typer callback.""" + +from __future__ import annotations + +import asyncio +import json +from dataclasses import dataclass +from datetime import datetime +from typing import Optional + +import typer +from rich.console import Console +from rich.filesize import decimal as format_decimal_size +from rich.table import Table + +from gopro_api.api.models import ( + DEFAULT_FIELDS, + CapturedRange, + GoProMediaSearchItem, + GoProMediaSearchParams, + GoProMediaSearchResponse, +) +from gopro_api.client import AsyncGoProClient + +from .app import app +from ._common import _parse_dt, _renamed_field, _require_token + + +@dataclass(frozen=True, slots=True) +class _SearchParams: + """Bundled search parameters passed from the Typer callback to the async runner.""" + + start: str + end: str + page: int + per_page: int + all_pages: bool + json_out: bool + tsv: bool + + +class SearchPrinter: + """Handles all search output formatting: Rich table, TSV, and JSON key renaming.""" + + def __init__(self, console: Console | None = None) -> None: + """Initialize with an optional Rich console. + + Args: + console: Console used for Rich output; a default soft-wrap console is + created when ``None``. + """ + self._console = console or Console(soft_wrap=True) + + def page_meta_line(self, page: GoProMediaSearchResponse) -> str: + """Format the pagination metadata comment line for a search page. + + Args: + page: Search response containing pagination details. + + Returns: + A ``# _pages:`` comment string with current page, per-page, total items, + and total pages. + """ + pages = page.pages + return ( + f"# _pages: current_page={pages.current_page} per_page={pages.per_page} " + f"total_items={pages.total_items} total_pages={pages.total_pages}" + ) + + def emit_embedded_errors(self, page: GoProMediaSearchResponse) -> None: + """Print any embedded API errors to stderr as yellow comment lines. + + Args: + page: Search response whose ``_embedded.errors`` list is checked. + """ + if page.embedded.errors: + for err in page.embedded.errors: + typer.secho( + f"# _embedded.errors: {json.dumps(err, ensure_ascii=False)}", + fg=typer.colors.YELLOW, + err=True, + ) + + def cells_plain(self, item: GoProMediaSearchItem) -> list[str]: + """Build raw string cells for TSV output. + + Args: + item: A single media item from the search response. + + Returns: + Ordered list of strings for each field in ``DEFAULT_FIELDS``; + missing values are represented as empty strings. + """ + row = item.model_dump(mode="json") + return ["" if row.get(c) is None else str(row[c]) for c in DEFAULT_FIELDS] + + def cells_rich(self, item: GoProMediaSearchItem) -> list[str]: + """Build human-formatted cells for Rich table output. + + File sizes are formatted with decimal SI units; all other values are + stringified as-is. + + Args: + item: A single media item from the search response. + + Returns: + Ordered list of display strings for each field in ``DEFAULT_FIELDS``. + """ + row = item.model_dump(mode="json") + cells: list[str] = [] + for c in DEFAULT_FIELDS: + val = row.get(c) + if val is None: + cells.append("") + elif c == "file_size": + cells.append(format_decimal_size(int(val))) + else: + cells.append(str(val)) + return cells + + def make_table(self) -> Table: + """Build an empty Rich Table with the default search columns. + + Returns: + A ``rich.table.Table`` configured with per-column overflow settings, + ready to receive rows via ``add_row``. + """ + table = Table(show_header=True, header_style="bold") + for name in DEFAULT_FIELDS: + col_kw: dict = {} + if name == "filename": + col_kw["overflow"] = "ellipsis" + col_kw["max_width"] = 40 + elif name == "captured_at": + col_kw["overflow"] = "ellipsis" + col_kw["max_width"] = 28 + elif name == "id": + col_kw["overflow"] = "fold" + elif name == "type": + col_kw["overflow"] = "ellipsis" + col_kw["max_width"] = 14 + table.add_column(_renamed_field(name), **col_kw) + return table + + def print_table(self, table: Table) -> None: + """Print a Rich table to the console. + + Args: + table: Fully populated Rich table to render. + """ + self._console.print(table) + + def print_tsv_page( + self, page: GoProMediaSearchResponse, *, header: bool = True + ) -> None: + """Print a TSV-formatted search page to stdout. + + Args: + page: Search response page to render. + header: When ``True`` (default), emit the column-name header row first; + set to ``False`` for subsequent pages in ``--all-pages`` mode. + """ + typer.echo(self.page_meta_line(page)) + self.emit_embedded_errors(page) + if header: + typer.echo("\t".join(_renamed_field(c) for c in DEFAULT_FIELDS)) + for item in page.embedded.media: + typer.echo("\t".join(self.cells_plain(item))) + + def print_rich_page(self, page: GoProMediaSearchResponse) -> None: + """Print a single Rich-formatted search page with metadata and a table. + + Args: + page: Search response page to render. + """ + self.emit_embedded_errors(page) + typer.echo(self.page_meta_line(page)) + table = self.make_table() + for item in page.embedded.media: + table.add_row(*self.cells_rich(item)) + self._console.print(table) + + def append_rich_rows(self, table: Table, page: GoProMediaSearchResponse) -> None: + """Append a page's media rows to an existing Rich table. + + Used in ``--all-pages`` mode to accumulate rows across pages before a + single final render. + + Args: + table: Rich table to append rows to. + page: Search response page whose media items are appended. + """ + self.emit_embedded_errors(page) + for item in page.embedded.media: + table.add_row(*self.cells_rich(item)) + + def rename_payload(self, payload: dict) -> dict: + """Apply display-name aliases to a raw API JSON payload. + + Renames keys inside ``_embedded.media`` items according to + ``_FIELD_LABELS`` so that JSON output uses the same names as the table + headers. + + Args: + payload: Raw API payload dict (typically from ``model_dump``). + + Returns: + The same ``payload`` dict with ``_embedded.media`` keys renamed in-place. + """ + embedded = payload.get("_embedded") + if isinstance(embedded, dict): + media = embedded.get("media") + if isinstance(media, list): + embedded["media"] = [ + ( + {_renamed_field(k): v for k, v in it.items()} + if isinstance(it, dict) + else it + ) + for it in media + ] + return payload + + +async def _collect_all_pages( + *, + client: AsyncGoProClient, + printer: SearchPrinter, + params: _SearchParams, + start_dt: datetime, + end_dt: datetime, +) -> None: + """Stream all non-empty search pages and render them incrementally. + + Args: + client: Open ``AsyncGoProClient`` to use for API calls. + printer: ``SearchPrinter`` instance used for rendering. + params: Bundled search options controlling output format and pagination. + start_dt: Parsed capture-range start datetime. + end_dt: Parsed capture-range end datetime. + """ + all_pages_payload: list[dict] = [] + first_plain_page = True + rich_table: Optional[Table] = None + last_page: Optional[GoProMediaSearchResponse] = None + async for page_result in client.iter_nonempty_search_pages( + start_dt, + end_dt, + per_page=params.per_page, + start_page=params.page, + ): + last_page = page_result + if params.json_out: + all_pages_payload.append( + printer.rename_payload( + page_result.model_dump(by_alias=True, mode="json"), + ), + ) + elif params.tsv: + printer.print_tsv_page(page_result, header=first_plain_page) + first_plain_page = False + else: + if rich_table is None: + rich_table = printer.make_table() + printer.append_rich_rows(rich_table, page_result) + if params.json_out: + typer.echo(json.dumps(all_pages_payload, indent=2)) + elif not params.tsv and rich_table is not None and last_page is not None: + typer.echo(printer.page_meta_line(last_page)) + printer.print_table(rich_table) + + +async def _run_search(*, timeout: float, params: _SearchParams) -> None: + """Execute a media search against the GoPro cloud API and print results. + + Args: + timeout: HTTP timeout in seconds passed to ``AsyncGoProClient``. + params: Bundled search options (date range, pagination, output format). + """ + _require_token() + printer = SearchPrinter() + start_dt = _parse_dt(params.start) + end_dt = _parse_dt(params.end) + + async with AsyncGoProClient(timeout=timeout) as client: + if params.all_pages: + await _collect_all_pages( + client=client, + printer=printer, + params=params, + start_dt=start_dt, + end_dt=end_dt, + ) + return + + search_params = GoProMediaSearchParams( + captured_range=CapturedRange(start=start_dt, end=end_dt), + page=params.page, + per_page=params.per_page, + ) + page_result = await client.search(search_params) + if params.json_out: + typer.echo( + json.dumps( + printer.rename_payload( + page_result.model_dump(by_alias=True, mode="json"), + ), + indent=2, + ), + ) + elif params.tsv: + printer.print_tsv_page(page_result) + else: + printer.print_rich_page(page_result) + + +@app.command( + "search", + help=( + "List media in a capture date range (Rich table by default; " + "--tsv for tab-separated fields; --json for raw API payloads)" + ), +) +def search_command( # pylint: disable=too-many-arguments + ctx: typer.Context, + *, + start: str = typer.Option( + ..., + "--start", + help="Range start: YYYY-MM-DD or ISO datetime", + ), + end: str = typer.Option( + ..., + "--end", + help=( + "Range end: YYYY-MM-DD or ISO datetime " + "(API treats range as in query string)" + ), + ), + page: int = typer.Option(1, "--page", help="Page number (default: 1)"), + per_page: int = typer.Option(30, "--per-page", help="Page size (default: 30)"), + all_pages: bool = typer.Option( + False, + "--all-pages", + help="Keep requesting pages until a page returns no media", + ), + tsv: bool = typer.Option( + False, + "--tsv", + help="Print tab-separated values (header row + metadata line) for scripting", + ), + json_out: bool = typer.Option( + False, + "--json", + help="Print full API JSON (with --all-pages: list of page payloads)", + ), +) -> None: + """Run search against the cloud API and print results.""" + asyncio.run( + _run_search( + timeout=ctx.obj["timeout"], + params=_SearchParams( + start=start, + end=end, + page=page, + per_page=per_page, + all_pages=all_pages, + json_out=json_out, + tsv=tsv, + ), + ), + )