From 083c6848ebfdf494e3411d9053bf44a37d522517 Mon Sep 17 00:00:00 2001 From: Welbert Castro Date: Sat, 9 May 2026 16:35:35 -0300 Subject: [PATCH 1/5] feat(cli)!: default to Rich tables; add --tsv; format sizes & rename columns Rework all CLI presentations to match Typer/Rich conventions: - search/info/pull now print Rich tables by default, with consistent styling via typer.echo / typer.secho (red bold errors, yellow API warnings on stderr) and an opt-in --tsv flag for tab-separated machine output. - search: human-readable file sizes via rich.filesize.decimal; column labels renamed (type->media, file_extension->type, file_size->size, item_count->items) consistently across rich, --tsv, and --json; ids are always shown in full (overflow=fold). - info: bold filename header + variations/files table with optional sidecars table; URLs preserved via overflow=fold. - pull: pre-download summary table + green completion line. BREAKING CHANGE: gopro-api search plain output now renders a Rich table by default; previous tab-separated rows (with the "# _pages" metadata line and header) require the new --tsv flag. JSON output for search also renames per-item keys (type->media, file_extension->type, file_size->size, item_count->items) inside _embedded.media[*]. Co-authored-by: Cursor --- gopro_api/cli.py | 355 +++++++++++++++++++++++++++++++++++++++++------ 1 file changed, 311 insertions(+), 44 deletions(-) diff --git a/gopro_api/cli.py b/gopro_api/cli.py index c4ad1a8..e1f648f 100644 --- a/gopro_api/cli.py +++ b/gopro_api/cli.py @@ -5,16 +5,22 @@ 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 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, + GoProMediaDownloadFile, + GoProMediaDownloadResponse, + GoProMediaDownloadSidecarFile, + GoProMediaDownloadVariation, GoProMediaSearchItem, GoProMediaSearchParams, GoProMediaSearchResponse, @@ -22,7 +28,7 @@ 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 +from gopro_api.utils import DownloadAsset, is_video_filename, pull_assets_for_response app = typer.Typer( name="gopro-api", @@ -98,45 +104,134 @@ def _require_token() -> None: typer.Exit: With code ``2`` if the token is missing. """ if not GP_ACCESS_TOKEN: - typer.echo( + 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 _print_search_plain_header() -> None: - cols = list(DEFAULT_FIELDS) - print("\t".join(cols)) +def _search_item_cells(item: GoProMediaSearchItem) -> list[str]: + row = item.model_dump(mode="json") + return ["" if row.get(c) is None else str(row[c]) for c in DEFAULT_FIELDS] -def _format_search_item_plain(item: GoProMediaSearchItem) -> str: +def _search_item_cells_rich(item: GoProMediaSearchItem) -> list[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) + 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 _print_search_plain_page( - page_result: GoProMediaSearchResponse, - *, - print_header: bool = True, -) -> None: +def _format_search_item_plain(item: GoProMediaSearchItem) -> str: + return "\t".join(_search_item_cells(item)) + + +def _pages_meta_line(page_result: GoProMediaSearchResponse) -> str: pages = page_result.pages - print( + return ( f"# _pages: current_page={pages.current_page} per_page={pages.per_page} " - f"total_items={pages.total_items} total_pages={pages.total_pages}", + f"total_items={pages.total_items} total_pages={pages.total_pages}" ) + + +def _emit_search_embedded_errors(page_result: GoProMediaSearchResponse) -> None: if page_result.embedded.errors: for err in page_result.embedded.errors: - print( + typer.secho( f"# _embedded.errors: {json.dumps(err, ensure_ascii=False)}", - file=sys.stderr, + fg=typer.colors.YELLOW, + err=True, ) + + +_FIELD_LABELS: dict[str, str] = { + "type": "media", + "file_extension": "type", + "file_size": "size", + "item_count": "items", +} + + +def _renamed_field(name: str) -> str: + return _FIELD_LABELS.get(name, name) + + +def _rename_media_item_keys(item: dict) -> dict: + return {_renamed_field(k): v for k, v in item.items()} + + +def _rename_search_payload(payload: dict) -> dict: + embedded = payload.get("_embedded") + if isinstance(embedded, dict): + media = embedded.get("media") + if isinstance(media, list): + embedded["media"] = [ + _rename_media_item_keys(it) if isinstance(it, dict) else it + for it in media + ] + return payload + + +def _make_search_table() -> Table: + 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_search_plain_header() -> None: + typer.echo("\t".join(_renamed_field(c) for c in DEFAULT_FIELDS)) + + +def _print_search_plain_page( + page_result: GoProMediaSearchResponse, + *, + print_header: bool = True, +) -> None: + typer.echo(_pages_meta_line(page_result)) + _emit_search_embedded_errors(page_result) if print_header: _print_search_plain_header() for item in page_result.embedded.media: - print(_format_search_item_plain(item)) + typer.echo(_format_search_item_plain(item)) + + +def _print_search_rich_page(page_result: GoProMediaSearchResponse) -> None: + _emit_search_embedded_errors(page_result) + typer.echo(_pages_meta_line(page_result)) + table = _make_search_table() + for item in page_result.embedded.media: + table.add_row(*_search_item_cells_rich(item)) + Console(soft_wrap=True).print(table) + + +def _append_search_rich_rows(table: Table, page_result: GoProMediaSearchResponse) -> None: + _emit_search_embedded_errors(page_result) + for item in page_result.embedded.media: + table.add_row(*_search_item_cells_rich(item)) async def _run_search( @@ -148,6 +243,7 @@ async def _run_search( per_page: int, all_pages: bool, json_out: bool, + tsv: bool, ) -> None: _require_token() start_dt = _parse_dt(start) @@ -157,24 +253,36 @@ async def _run_search( if all_pages: 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=per_page, start_page=page, ): + last_page = page_result if json_out: all_pages_payload.append( - page_result.model_dump(by_alias=True, mode="json"), + _rename_search_payload( + page_result.model_dump(by_alias=True, mode="json"), + ), ) - else: + elif tsv: _print_search_plain_page( page_result, print_header=first_plain_page, ) first_plain_page = False + else: + if rich_table is None: + rich_table = _make_search_table() + _append_search_rich_rows(rich_table, page_result) if json_out: - print(json.dumps(all_pages_payload, indent=2)) + typer.echo(json.dumps(all_pages_payload, indent=2)) + elif not tsv and rich_table is not None and last_page is not None: + typer.echo(_pages_meta_line(last_page)) + Console(soft_wrap=True).print(rich_table) return params = GoProMediaSearchParams( @@ -184,21 +292,25 @@ async def _run_search( ) page_result = await client.search(params) if json_out: - print( + typer.echo( json.dumps( - page_result.model_dump(by_alias=True, mode="json"), + _rename_search_payload( + page_result.model_dump(by_alias=True, mode="json"), + ), indent=2, ), ) - else: + elif tsv: _print_search_plain_page(page_result) + else: + _print_search_rich_page(page_result) @app.command( "search", help=( - "List media in a capture date range (tab-separated fields; " - "use --json for raw API payloads)" + "List media in a capture date range (Rich table by default; " + "--tsv for tab-separated fields; --json for raw API payloads)" ), ) def search_command( @@ -224,6 +336,11 @@ def search_command( "--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", @@ -241,50 +358,180 @@ def search_command( per_page=per_page, all_pages=all_pages, json_out=json_out, + tsv=tsv, ), ) +_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 _build_basic_table(headers: list[str], *, fold_cols: tuple[str, ...] = ("url",)) -> Table: + 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: + return "yes" if value else "no" + + +def _info_variation_cells(idx: int, v: GoProMediaDownloadVariation) -> list[str]: + return [ + str(idx), + v.label, + v.quality, + v.type, + f"{v.width}x{v.height}", + _yes_no(v.available), + v.url, + ] + + +def _info_file_cells(idx: int, f: GoProMediaDownloadFile) -> list[str]: + return [ + str(idx), + str(f.item_number), + f.camera_position, + f"{f.width}x{f.height}", + _yes_no(f.available), + f.url, + ] + + +def _info_sidecar_cells(idx: int, s: GoProMediaDownloadSidecarFile) -> list[str]: + return [ + str(idx), + s.label, + s.type, + str(s.fps), + _yes_no(s.available), + s.url, + ] + + +def _print_info_rich(meta: GoProMediaDownloadResponse) -> None: + typer.secho(meta.filename, bold=True) + console = Console(soft_wrap=True) + if is_video_filename(meta.filename): + table = _build_basic_table(_VARIATION_HEADERS) + for idx, v in enumerate(meta.embedded.variations): + table.add_row(*_info_variation_cells(idx, v)) + else: + table = _build_basic_table(_FILE_HEADERS) + for idx, f in enumerate(meta.embedded.files): + table.add_row(*_info_file_cells(idx, f)) + console.print(table) + if meta.embedded.sidecar_files: + typer.secho("sidecars", bold=True) + sidecars = _build_basic_table(_SIDECAR_HEADERS) + for idx, s in enumerate(meta.embedded.sidecar_files): + sidecars.add_row(*_info_sidecar_cells(idx, s)) + console.print(sidecars) + + +def _print_info_tsv(meta: GoProMediaDownloadResponse) -> None: + typer.echo(f"# filename: {meta.filename}") + if is_video_filename(meta.filename): + typer.echo("\t".join(_VARIATION_HEADERS)) + for idx, v in enumerate(meta.embedded.variations): + typer.echo("\t".join(_info_variation_cells(idx, v))) + else: + typer.echo("\t".join(_FILE_HEADERS)) + for idx, f in enumerate(meta.embedded.files): + typer.echo("\t".join(_info_file_cells(idx, f))) + if meta.embedded.sidecar_files: + typer.echo("# sidecars") + typer.echo("\t".join(_SIDECAR_HEADERS)) + for idx, s in enumerate(meta.embedded.sidecar_files): + typer.echo("\t".join(_info_sidecar_cells(idx, s))) + + async def _run_info( *, timeout: float, media_id: str, json_out: bool, + tsv: bool, ) -> None: _require_token() async with AsyncGoProClient(timeout=timeout) as client: meta = await client.download(media_id) if json_out: - print( + typer.echo( json.dumps( meta.model_dump(by_alias=True, mode="json"), indent=2, ), ) + elif tsv: + _print_info_tsv(meta) 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}", - ) + _print_info_rich(meta) -@app.command("info", help="Show download metadata (URLs, sizes) for one media id") +@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), + _run_info( + timeout=ctx.obj["timeout"], + media_id=media_id, + json_out=json_out, + tsv=tsv, + ), + ) + + +_PULL_HEADERS = ["filename", "dim", "available", "url"] + + +def _pull_summary_cells(filename: str, asset: DownloadAsset) -> list[str]: + return [ + filename, + f"{asset.width}x{asset.height}", + _yes_no(asset.available), + asset.url, + ] + + +def _print_pull_rich(assets: dict[str, DownloadAsset], destination: str) -> None: + typer.secho( + f"Pulling {len(assets)} file(s) to {destination}", + bold=True, ) + table = _build_basic_table(_PULL_HEADERS) + for filename, asset in assets.items(): + table.add_row(*_pull_summary_cells(filename, asset)) + Console(soft_wrap=True).print(table) + + +def _print_pull_tsv(assets: dict[str, DownloadAsset], destination: str) -> None: + typer.echo(f"# destination: {destination}") + typer.echo("\t".join(_PULL_HEADERS)) + for filename, asset in assets.items(): + typer.echo("\t".join(_pull_summary_cells(filename, asset))) async def _run_pull( @@ -294,6 +541,7 @@ async def _run_pull( destination: str, height: Optional[int], width: Optional[int], + tsv: bool, ) -> None: _require_token() async with AsyncGoProClient(timeout=timeout) as client: @@ -305,9 +553,14 @@ async def _run_pull( target_width=width, ) except NoVariationsError as exc: - typer.echo(f"error: {exc}", err=True) + typer.secho(f"error: {exc}", fg=typer.colors.RED, bold=True, err=True) raise typer.Exit(2) from exc + if tsv: + _print_pull_tsv(assets, destination) + else: + _print_pull_rich(assets, destination) + os.makedirs(destination, exist_ok=True) await asyncio.gather( *( @@ -319,8 +572,16 @@ async def _run_pull( ) ) + typer.secho(f"Done. ({len(assets)} file(s))", fg=typer.colors.GREEN) -@app.command("pull", help="Download files from a media id") + +@app.command( + "pull", + help=( + "Download files from a media id (prints a Rich summary by default; " + "--tsv for tab-separated)" + ), +) def pull_command( ctx: typer.Context, media_id: str = typer.Argument(..., help="Media id from search"), @@ -343,6 +604,11 @@ def pull_command( "(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") @@ -354,6 +620,7 @@ def pull_command( destination=destination, height=height, width=width, + tsv=tsv, ), ) @@ -362,7 +629,7 @@ 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``). + argv: Argument list (defaults to process arguments when ``None``). """ app(args=argv) From 356cbc70b3482e183a4246de2ac5b544b4419d20 Mon Sep 17 00:00:00 2001 From: Welbert Castro Date: Sat, 9 May 2026 16:35:44 -0300 Subject: [PATCH 2/5] chore: ignore .cursor/ for local IDE skills and rules Co-authored-by: Cursor --- .gitignore | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) 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 From 2fd10af8ba720c30c4eb58c4f1906813d9d28a59 Mon Sep 17 00:00:00 2001 From: Welbert Castro Date: Sat, 9 May 2026 17:19:51 -0300 Subject: [PATCH 3/5] refactor(cli): split cli.py into cli/ package Move all CLI logic from the single gopro_api/cli.py module into a dedicated gopro_api/cli/ package with one file per command (search, info, pull) plus shared helpers in _common.py. Co-authored-by: Cursor --- README.md | 2 +- gopro_api/cli.py | 638 -------------------------------------- gopro_api/cli/__init__.py | 11 + gopro_api/cli/__main__.py | 3 + gopro_api/cli/_common.py | 81 +++++ gopro_api/cli/app.py | 57 ++++ gopro_api/cli/info.py | 148 +++++++++ gopro_api/cli/pull.py | 142 +++++++++ gopro_api/cli/search.py | 248 +++++++++++++++ 9 files changed, 691 insertions(+), 639 deletions(-) delete mode 100644 gopro_api/cli.py create mode 100644 gopro_api/cli/__init__.py create mode 100644 gopro_api/cli/__main__.py create mode 100644 gopro_api/cli/_common.py create mode 100644 gopro_api/cli/app.py create mode 100644 gopro_api/cli/info.py create mode 100644 gopro_api/cli/pull.py create mode 100644 gopro_api/cli/search.py 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/gopro_api/cli.py b/gopro_api/cli.py deleted file mode 100644 index e1f648f..0000000 --- a/gopro_api/cli.py +++ /dev/null @@ -1,638 +0,0 @@ -"""Command-line interface for gopro-api.""" - -from __future__ import annotations - -import asyncio -import json -import os -from datetime import datetime -from importlib.metadata import PackageNotFoundError, version as package_version -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, - GoProMediaDownloadFile, - GoProMediaDownloadResponse, - GoProMediaDownloadSidecarFile, - GoProMediaDownloadVariation, - 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 DownloadAsset, 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.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 _search_item_cells(item: GoProMediaSearchItem) -> list[str]: - row = item.model_dump(mode="json") - return ["" if row.get(c) is None else str(row[c]) for c in DEFAULT_FIELDS] - - -def _search_item_cells_rich(item: GoProMediaSearchItem) -> list[str]: - 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 _format_search_item_plain(item: GoProMediaSearchItem) -> str: - return "\t".join(_search_item_cells(item)) - - -def _pages_meta_line(page_result: GoProMediaSearchResponse) -> str: - pages = page_result.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_search_embedded_errors(page_result: GoProMediaSearchResponse) -> None: - if page_result.embedded.errors: - for err in page_result.embedded.errors: - typer.secho( - f"# _embedded.errors: {json.dumps(err, ensure_ascii=False)}", - fg=typer.colors.YELLOW, - err=True, - ) - - -_FIELD_LABELS: dict[str, str] = { - "type": "media", - "file_extension": "type", - "file_size": "size", - "item_count": "items", -} - - -def _renamed_field(name: str) -> str: - return _FIELD_LABELS.get(name, name) - - -def _rename_media_item_keys(item: dict) -> dict: - return {_renamed_field(k): v for k, v in item.items()} - - -def _rename_search_payload(payload: dict) -> dict: - embedded = payload.get("_embedded") - if isinstance(embedded, dict): - media = embedded.get("media") - if isinstance(media, list): - embedded["media"] = [ - _rename_media_item_keys(it) if isinstance(it, dict) else it - for it in media - ] - return payload - - -def _make_search_table() -> Table: - 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_search_plain_header() -> None: - typer.echo("\t".join(_renamed_field(c) for c in DEFAULT_FIELDS)) - - -def _print_search_plain_page( - page_result: GoProMediaSearchResponse, - *, - print_header: bool = True, -) -> None: - typer.echo(_pages_meta_line(page_result)) - _emit_search_embedded_errors(page_result) - if print_header: - _print_search_plain_header() - for item in page_result.embedded.media: - typer.echo(_format_search_item_plain(item)) - - -def _print_search_rich_page(page_result: GoProMediaSearchResponse) -> None: - _emit_search_embedded_errors(page_result) - typer.echo(_pages_meta_line(page_result)) - table = _make_search_table() - for item in page_result.embedded.media: - table.add_row(*_search_item_cells_rich(item)) - Console(soft_wrap=True).print(table) - - -def _append_search_rich_rows(table: Table, page_result: GoProMediaSearchResponse) -> None: - _emit_search_embedded_errors(page_result) - for item in page_result.embedded.media: - table.add_row(*_search_item_cells_rich(item)) - - -async def _run_search( - *, - timeout: float, - start: str, - end: str, - page: int, - per_page: int, - all_pages: bool, - json_out: bool, - tsv: 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 - 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=per_page, - start_page=page, - ): - last_page = page_result - if json_out: - all_pages_payload.append( - _rename_search_payload( - page_result.model_dump(by_alias=True, mode="json"), - ), - ) - elif tsv: - _print_search_plain_page( - page_result, - print_header=first_plain_page, - ) - first_plain_page = False - else: - if rich_table is None: - rich_table = _make_search_table() - _append_search_rich_rows(rich_table, page_result) - if json_out: - typer.echo(json.dumps(all_pages_payload, indent=2)) - elif not tsv and rich_table is not None and last_page is not None: - typer.echo(_pages_meta_line(last_page)) - Console(soft_wrap=True).print(rich_table) - 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: - typer.echo( - json.dumps( - _rename_search_payload( - page_result.model_dump(by_alias=True, mode="json"), - ), - indent=2, - ), - ) - elif tsv: - _print_search_plain_page(page_result) - else: - _print_search_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( - 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.""" - 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, - tsv=tsv, - ), - ) - - -_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 _build_basic_table(headers: list[str], *, fold_cols: tuple[str, ...] = ("url",)) -> Table: - 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: - return "yes" if value else "no" - - -def _info_variation_cells(idx: int, v: GoProMediaDownloadVariation) -> list[str]: - return [ - str(idx), - v.label, - v.quality, - v.type, - f"{v.width}x{v.height}", - _yes_no(v.available), - v.url, - ] - - -def _info_file_cells(idx: int, f: GoProMediaDownloadFile) -> list[str]: - return [ - str(idx), - str(f.item_number), - f.camera_position, - f"{f.width}x{f.height}", - _yes_no(f.available), - f.url, - ] - - -def _info_sidecar_cells(idx: int, s: GoProMediaDownloadSidecarFile) -> list[str]: - return [ - str(idx), - s.label, - s.type, - str(s.fps), - _yes_no(s.available), - s.url, - ] - - -def _print_info_rich(meta: GoProMediaDownloadResponse) -> None: - typer.secho(meta.filename, bold=True) - console = Console(soft_wrap=True) - if is_video_filename(meta.filename): - table = _build_basic_table(_VARIATION_HEADERS) - for idx, v in enumerate(meta.embedded.variations): - table.add_row(*_info_variation_cells(idx, v)) - else: - table = _build_basic_table(_FILE_HEADERS) - for idx, f in enumerate(meta.embedded.files): - table.add_row(*_info_file_cells(idx, f)) - console.print(table) - if meta.embedded.sidecar_files: - typer.secho("sidecars", bold=True) - sidecars = _build_basic_table(_SIDECAR_HEADERS) - for idx, s in enumerate(meta.embedded.sidecar_files): - sidecars.add_row(*_info_sidecar_cells(idx, s)) - console.print(sidecars) - - -def _print_info_tsv(meta: GoProMediaDownloadResponse) -> None: - typer.echo(f"# filename: {meta.filename}") - if is_video_filename(meta.filename): - typer.echo("\t".join(_VARIATION_HEADERS)) - for idx, v in enumerate(meta.embedded.variations): - typer.echo("\t".join(_info_variation_cells(idx, v))) - else: - typer.echo("\t".join(_FILE_HEADERS)) - for idx, f in enumerate(meta.embedded.files): - typer.echo("\t".join(_info_file_cells(idx, f))) - if meta.embedded.sidecar_files: - typer.echo("# sidecars") - typer.echo("\t".join(_SIDECAR_HEADERS)) - for idx, s in enumerate(meta.embedded.sidecar_files): - typer.echo("\t".join(_info_sidecar_cells(idx, s))) - - -async def _run_info( - *, - timeout: float, - media_id: str, - json_out: bool, - tsv: bool, -) -> None: - _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: - _print_info_tsv(meta) - else: - _print_info_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, - ), - ) - - -_PULL_HEADERS = ["filename", "dim", "available", "url"] - - -def _pull_summary_cells(filename: str, asset: DownloadAsset) -> list[str]: - return [ - filename, - f"{asset.width}x{asset.height}", - _yes_no(asset.available), - asset.url, - ] - - -def _print_pull_rich(assets: dict[str, DownloadAsset], destination: str) -> None: - typer.secho( - f"Pulling {len(assets)} file(s) to {destination}", - bold=True, - ) - table = _build_basic_table(_PULL_HEADERS) - for filename, asset in assets.items(): - table.add_row(*_pull_summary_cells(filename, asset)) - Console(soft_wrap=True).print(table) - - -def _print_pull_tsv(assets: dict[str, DownloadAsset], destination: str) -> None: - typer.echo(f"# destination: {destination}") - typer.echo("\t".join(_PULL_HEADERS)) - for filename, asset in assets.items(): - typer.echo("\t".join(_pull_summary_cells(filename, asset))) - - -async def _run_pull( - *, - timeout: float, - media_id: str, - destination: str, - height: Optional[int], - width: Optional[int], - tsv: bool, -) -> 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.secho(f"error: {exc}", fg=typer.colors.RED, bold=True, err=True) - raise typer.Exit(2) from exc - - if tsv: - _print_pull_tsv(assets, destination) - else: - _print_pull_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( - 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, - ), - ) - - -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) - - -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..b982e5d --- /dev/null +++ b/gopro_api/cli/__main__.py @@ -0,0 +1,3 @@ +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..dfbf82d --- /dev/null +++ b/gopro_api/cli/_common.py @@ -0,0 +1,81 @@ +"""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 _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]: + 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: + 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: + 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..371dbac --- /dev/null +++ b/gopro_api/cli/app.py @@ -0,0 +1,57 @@ +"""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: + 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 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..0cdd962 --- /dev/null +++ b/gopro_api/cli/info.py @@ -0,0 +1,148 @@ +"""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: + self._console = console or Console(soft_wrap=True) + + def variation_cells(self, idx: int, v: GoProMediaDownloadVariation) -> list[str]: + 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]: + 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]: + return [ + str(idx), + s.label, + s.type, + str(s.fps), + _yes_no(s.available), + s.url, + ] + + def print_rich(self, meta: GoProMediaDownloadResponse) -> None: + 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: + 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: + _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..a671dc4 --- /dev/null +++ b/gopro_api/cli/pull.py @@ -0,0 +1,142 @@ +"""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: + self._console = console or Console(soft_wrap=True) + + def summary_cells(self, filename: str, asset: DownloadAsset) -> list[str]: + 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: + 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: + 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: + _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( + 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..2a9f2b8 --- /dev/null +++ b/gopro_api/cli/search.py @@ -0,0 +1,248 @@ +"""Search command: SearchPrinter class, async runner, and Typer callback.""" + +from __future__ import annotations + +import asyncio +import json +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 + + +class SearchPrinter: + """Handles all search output formatting: Rich table, TSV, and JSON key renaming.""" + + def __init__(self, console: Console | None = None) -> None: + self._console = console or Console(soft_wrap=True) + + def page_meta_line(self, page: GoProMediaSearchResponse) -> str: + 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: + 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]: + 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]: + 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: + 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: + self._console.print(table) + + def print_tsv_page( + self, page: GoProMediaSearchResponse, *, header: bool = True + ) -> None: + 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: + 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: + 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: + 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 _run_search( + *, + timeout: float, + start: str, + end: str, + page: int, + per_page: int, + all_pages: bool, + json_out: bool, + tsv: bool, +) -> None: + _require_token() + printer = SearchPrinter() + 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 + 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=per_page, + start_page=page, + ): + last_page = page_result + if json_out: + all_pages_payload.append( + printer.rename_payload( + page_result.model_dump(by_alias=True, mode="json"), + ), + ) + elif 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 json_out: + typer.echo(json.dumps(all_pages_payload, indent=2)) + elif not 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) + 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: + typer.echo( + json.dumps( + printer.rename_payload( + page_result.model_dump(by_alias=True, mode="json"), + ), + indent=2, + ), + ) + elif 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( + 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.""" + 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, + tsv=tsv, + ), + ) From ad164a112bc5a6957efa5c1632cd59803c430f36 Mon Sep 17 00:00:00 2001 From: Welbert Castro Date: Sat, 9 May 2026 17:38:32 -0300 Subject: [PATCH 4/5] docs(cli): add google-style docstrings and expose printer classes in mkdocs Add missing Google-style docstrings to every public and private function and method in the cli package (_common, app, info, pull, __main__). Expand docs/api/cli.md with a Printers section referencing SearchPrinter, InfoPrinter and PullPrinter so mkdocstrings renders them in the API site. Co-authored-by: Cursor --- docs/api/cli.md | 8 ++++++ gopro_api/api/__init__.py | 1 - gopro_api/api/models.py | 1 - gopro_api/cli/__main__.py | 2 ++ gopro_api/cli/_common.py | 38 +++++++++++++++++++++++++++++ gopro_api/cli/app.py | 20 +++++++++++++++ gopro_api/cli/info.py | 51 +++++++++++++++++++++++++++++++++++++++ gopro_api/cli/pull.py | 42 +++++++++++++++++++++++++++++++- 8 files changed, 160 insertions(+), 3 deletions(-) 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/__main__.py b/gopro_api/cli/__main__.py index b982e5d..7c25c40 100644 --- a/gopro_api/cli/__main__.py +++ b/gopro_api/cli/__main__.py @@ -1,3 +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 index dfbf82d..52d5ada 100644 --- a/gopro_api/cli/_common.py +++ b/gopro_api/cli/_common.py @@ -19,6 +19,14 @@ 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) @@ -58,6 +66,18 @@ def _parse_dt(raw: str) -> datetime: 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: @@ -68,6 +88,16 @@ def _validate_positive_px(value: Optional[int], flag: str) -> Optional[int]: 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: @@ -78,4 +108,12 @@ def _build_basic_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 index 371dbac..1443ee8 100644 --- a/gopro_api/cli/app.py +++ b/gopro_api/cli/app.py @@ -15,6 +15,11 @@ 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: @@ -22,6 +27,14 @@ def _version() -> str: 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() @@ -43,6 +56,13 @@ def _main_callback( 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 diff --git a/gopro_api/cli/info.py b/gopro_api/cli/info.py index 0cdd962..b8d1401 100644 --- a/gopro_api/cli/info.py +++ b/gopro_api/cli/info.py @@ -29,9 +29,24 @@ class InfoPrinter: 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, @@ -43,6 +58,15 @@ def variation_cells(self, idx: int, v: GoProMediaDownloadVariation) -> list[str] ] 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), @@ -53,6 +77,15 @@ def file_cells(self, idx: int, f: GoProMediaDownloadFile) -> list[str]: ] 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, @@ -63,6 +96,11 @@ def sidecar_cells(self, idx: int, s: GoProMediaDownloadSidecarFile) -> list[str] ] 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) @@ -81,6 +119,11 @@ def print_rich(self, meta: GoProMediaDownloadResponse) -> None: 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)) @@ -104,6 +147,14 @@ async def _run_info( 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) diff --git a/gopro_api/cli/pull.py b/gopro_api/cli/pull.py index a671dc4..bc13ba3 100644 --- a/gopro_api/cli/pull.py +++ b/gopro_api/cli/pull.py @@ -23,9 +23,24 @@ class PullPrinter: 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}", @@ -34,6 +49,12 @@ def summary_cells(self, filename: str, asset: DownloadAsset) -> list[str]: ] 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, @@ -44,6 +65,12 @@ def print_rich(self, assets: dict[str, DownloadAsset], destination: str) -> None 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(): @@ -59,6 +86,19 @@ async def _run_pull( 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: @@ -99,7 +139,7 @@ async def _run_pull( "--tsv for tab-separated)" ), ) -def pull_command( +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"), From 5ded69f7c48fc995dea490ef2d45b95fbd0cba15 Mon Sep 17 00:00:00 2001 From: Welbert Castro Date: Sat, 9 May 2026 17:38:45 -0300 Subject: [PATCH 5/5] refactor(cli): extract _SearchParams dataclass and _collect_all_pages helper Introduce _SearchParams (frozen dataclass) to bundle the 7 search options, reducing _run_search from 8 keyword arguments to 2. Extract the all-pages streaming loop into _collect_all_pages to bring locals below the pylint R0914 threshold. Suppress R0913 on the Typer callback search_command via an inline pylint disable (framework callbacks must declare every CLI option as a parameter and cannot be refactored further). Co-authored-by: Cursor --- gopro_api/cli/search.py | 244 ++++++++++++++++++++++++++++++---------- 1 file changed, 184 insertions(+), 60 deletions(-) diff --git a/gopro_api/cli/search.py b/gopro_api/cli/search.py index 2a9f2b8..5a6bc73 100644 --- a/gopro_api/cli/search.py +++ b/gopro_api/cli/search.py @@ -4,6 +4,8 @@ import asyncio import json +from dataclasses import dataclass +from datetime import datetime from typing import Optional import typer @@ -24,13 +26,41 @@ 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} " @@ -38,6 +68,11 @@ def page_meta_line(self, page: GoProMediaSearchResponse) -> str: ) 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( @@ -47,10 +82,30 @@ def emit_embedded_errors(self, page: GoProMediaSearchResponse) -> None: ) 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: @@ -64,6 +119,12 @@ def cells_rich(self, item: GoProMediaSearchItem) -> list[str]: 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 = {} @@ -82,11 +143,23 @@ def make_table(self) -> Table: 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: @@ -95,6 +168,11 @@ def print_tsv_page( 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() @@ -103,80 +181,125 @@ def print_rich_page(self, page: GoProMediaSearchResponse) -> None: 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 + ( + {_renamed_field(k): v for k, v in it.items()} + if isinstance(it, dict) + else it + ) for it in media ] return payload -async def _run_search( +async def _collect_all_pages( *, - timeout: float, - start: str, - end: str, - page: int, - per_page: int, - all_pages: bool, - json_out: bool, - tsv: bool, + 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(start) - end_dt = _parse_dt(end) + start_dt = _parse_dt(params.start) + end_dt = _parse_dt(params.end) async with AsyncGoProClient(timeout=timeout) as client: - if all_pages: - 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=per_page, - start_page=page, - ): - last_page = page_result - if json_out: - all_pages_payload.append( - printer.rename_payload( - page_result.model_dump(by_alias=True, mode="json"), - ), - ) - elif 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 json_out: - typer.echo(json.dumps(all_pages_payload, indent=2)) - elif not 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) + if params.all_pages: + await _collect_all_pages( + client=client, + printer=printer, + params=params, + start_dt=start_dt, + end_dt=end_dt, + ) return - params = GoProMediaSearchParams( + search_params = GoProMediaSearchParams( captured_range=CapturedRange(start=start_dt, end=end_dt), - page=page, - per_page=per_page, + page=params.page, + per_page=params.per_page, ) - page_result = await client.search(params) - if json_out: + page_result = await client.search(search_params) + if params.json_out: typer.echo( json.dumps( printer.rename_payload( @@ -185,7 +308,7 @@ async def _run_search( indent=2, ), ) - elif tsv: + elif params.tsv: printer.print_tsv_page(page_result) else: printer.print_rich_page(page_result) @@ -198,7 +321,7 @@ async def _run_search( "--tsv for tab-separated fields; --json for raw API payloads)" ), ) -def search_command( +def search_command( # pylint: disable=too-many-arguments ctx: typer.Context, *, start: str = typer.Option( @@ -233,16 +356,17 @@ def search_command( ), ) -> 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, - tsv=tsv, + 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, + ), ), )