Coverage for /opt/hostedtoolcache/Python/3.12.10/x64/lib/python3.12/site-packages/netflix_open_content_helper/cli.py: 89%
95 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-04-17 17:05 +0000
1import subprocess
2import webbrowser
3from pathlib import Path
4from typing import Annotated, Optional
6import typer
7from rich.progress import track
9from netflix_open_content_helper import CONFIG, __version__
10from netflix_open_content_helper.parser import parse_shotfile
13def download_from_s3(
14 s3_uri: str, s3_path: str, dest_path: str = ".", dry_run: bool = False
15) -> None:
16 """
17 Download a file from AWS S3.
19 Args:
20 s3_uri (str): The base S3 URI.
21 s3_path (str): The specific path to the file in S3.
22 dest_path (str): The destination path for the downloaded file.
23 dry_run (bool): If true, show what would be done, but do not do it.
24 """
25 commands = [
26 "aws",
27 "s3",
28 "cp",
29 "--quiet",
30 "--no-sign-request",
31 f"{s3_uri}/{s3_path}",
32 dest_path,
33 ]
34 if dry_run:
35 print(f"dry-run: {' '.join(commands)}")
36 else:
37 subprocess.run(commands, check=True)
40def version_callback(value: bool) -> None:
41 """Display the version of the package."""
42 if value:
43 typer.echo(f"Netflix Open Content Helper, version {__version__}")
44 raise typer.Exit()
47app = typer.Typer()
50@app.callback()
51def common(
52 version: bool = typer.Option(
53 False,
54 "--version",
55 is_eager=True,
56 help="Show the version of the package.",
57 callback=version_callback,
58 ),
59) -> None:
60 """A utility for interacting with Netflix Open Content media."""
61 pass
64@app.command()
65def browse() -> None:
66 """
67 Open a web browser to the Netflix Open Content URL.
68 """
69 NETFLIX_OPEN_CONTENT_URL = CONFIG["netflix_open_content_url"]
70 # Check if the URL is configured
71 if not NETFLIX_OPEN_CONTENT_URL:
72 raise ValueError(
73 "Netflix Open Content URL is not configured. Check the config file."
74 )
75 # Check if the URL is valid
76 if not NETFLIX_OPEN_CONTENT_URL.startswith(("http://", "https://")):
77 raise ValueError(
78 f"Invalid URL format for url {NETFLIX_OPEN_CONTENT_URL}. Should start with 'http://' or 'https://'."
79 )
80 # Open the URL in the default web browser
81 # This will open the URL in a new tab if the browser is already open
82 # or in a new window if the browser is not open
83 # Note: This will not work in a headless environment
84 # such as a server without a GUI
85 # or in a terminal without a web browser
86 webbrowser.open_new(NETFLIX_OPEN_CONTENT_URL)
89@app.command()
90def download(
91 name: Annotated[
92 str, typer.Argument(help="The name of the project to download from.")
93 ],
94 frame_start: Annotated[
95 int,
96 typer.Option("--frame-start", "-fs", help="The start frame for the download."),
97 ] = 1,
98 frame_end: Annotated[
99 int, typer.Option("--frame-end", "-fe", help="The end frame for the download.")
100 ] = 1,
101 force: Annotated[
102 bool,
103 typer.Option(
104 "--force",
105 "-f",
106 help="Force download/overwrite of files that already exist.",
107 ),
108 ] = False,
109 dry_run: Annotated[
110 bool,
111 typer.Option(
112 "--dry-run",
113 "-n",
114 help="Show what would be done, but do not do it.",
115 ),
116 ] = False,
117 rename: Annotated[
118 Optional[str],
119 typer.Option(help="A new name for the downloaded frames. Ex. name.%04d.ext."),
120 ] = "",
121 renumber: Annotated[
122 Optional[int],
123 typer.Option(
124 help="A new start frame for the downloaded frames (with rename). Ex. 1001."
125 ),
126 ] = None,
127) -> None:
128 """Download frames from Netflix Open Content project NAME to the current directory."""
130 typer.echo(f"Downloading: {name} frames {frame_start}-{frame_end}")
131 # Validate the frame range
132 if frame_start < 1 or frame_end < 1:
133 raise ValueError(
134 f"Frame numbers ({frame_start}, {frame_end}) must be positive integers."
135 )
136 if frame_start > frame_end:
137 raise ValueError(
138 f"Start frame ({frame_start}) must be less than or equal to end frame ({frame_end})."
139 )
141 # Check if the AWS CLI is installed
142 test_commands = ["aws", "--version"]
143 try:
144 subprocess.run(test_commands, check=True, capture_output=True)
145 except subprocess.CalledProcessError as exc:
146 raise OSError(
147 "AWS CLI is not installed. Please install it to use this feature."
148 ) from exc
150 # Obtain the asset configuration, conform to lower-case name
151 assets = [d for d in CONFIG["assets"] if d["name"] == name.lower()]
152 if not assets:
153 print(f"Asset {name} not found in config.")
154 list_assets()
155 raise ValueError(f"Asset '{name}' not found in config. Check asset name.")
157 asset = assets[0]
158 # Check if the S3 URI is configured for the asset
159 s3_uri = asset["s3_uri"]
161 if not s3_uri:
162 raise ValueError(
163 f"S3 URI is not configured for '{name}'. Check the config file."
164 )
165 # Check if the S3 URI is valid
166 if not s3_uri.startswith("s3://"):
167 raise ValueError(f"Invalid S3 URI format {s3_uri}. Must start with 's3://'.")
168 s3_basename = asset["s3_basename"]
169 if not s3_basename:
170 raise ValueError(
171 f"S3 basename is not configured for '{name}'. Check the config file."
172 )
173 # Check if the S3 basename is valid
174 if "%" not in s3_basename:
175 raise ValueError(
176 f"Invalid S3 basename format '{s3_basename}'. Must contain a frame substitution wildcard like %04d. Check the config file."
177 )
178 # check if the rename syntax is valid.
179 if rename and "%" not in rename:
180 raise ValueError(
181 f"Invalid rename format '{rename}'. Must contain a frame substitution wildcard like %04d."
182 )
183 # Generate the S3 path for each frame
184 if renumber:
185 if not rename:
186 raise ValueError("Option --renumber requires --rename.")
187 renumber_offset = renumber - frame_start
189 for value in track(range(frame_start, frame_end + 1), "Downloading Frames..."):
190 # Generate the S3 path
191 s3_path = s3_basename % value
192 frame_path = Path(s3_path)
193 if rename:
194 rename_value = value + renumber_offset if renumber else value
195 rename_path = rename % rename_value
196 frame_path = Path(rename_path)
197 # check if the frame exists on disk already
198 if Path(frame_path.name).is_file() and not force:
199 print(f"file {frame_path.name} exists, skipping. Use --force to overwrite.")
200 continue
202 # Download the content from S3, renaming if requested
203 dest_path = rename_path if rename else "."
204 download_from_s3(s3_uri, s3_path, dest_path=dest_path, dry_run=dry_run)
207@app.command()
208def get(
209 shotfile: Annotated[
210 str,
211 typer.Argument(
212 help="A file containing a list of Shots to download. CSV, JSON, and YAML formats are supported."
213 ),
214 ],
215 force: Annotated[
216 bool,
217 typer.Option(
218 "--force",
219 "-f",
220 help="Force download/overwrite of files that already exist.",
221 ),
222 ] = False,
223 dry_run: Annotated[
224 bool,
225 typer.Option(
226 "--dry-run",
227 "-n",
228 help="Show what would be done, but do not do it.",
229 ),
230 ] = False,
231) -> None:
232 """Get a list of Shots from a file. CSV, JSON, and YAML formats are supported."""
233 shots = parse_shotfile(shotfile=shotfile)
234 # TODO: using the typer progress bar due to the rich progress bar not nesting easily with multiples
235 with typer.progressbar(shots, label="Getting Shots...", length=len(shots)) as bar:
236 for shot in shots:
237 download(
238 name=shot.project,
239 frame_end=shot.frame_end,
240 frame_start=shot.frame_start,
241 rename=shot.name,
242 renumber=shot.renumber,
243 dry_run=dry_run,
244 force=force,
245 )
246 bar.update(1)
249@app.command("list")
250def list_assets(
251 only_frames: bool = typer.Option(True, help="Only list assets with frame content."),
252) -> None:
253 """
254 List available Netflix Open Content.
256 Some open content assets may not have frame content.
258 """
259 message = "Available content"
260 if only_frames:
261 message += " with frames:"
262 else:
263 message += ":"
264 typer.echo(message)
265 for asset in sorted(CONFIG["assets"], key=lambda x: x["name"]):
266 if only_frames and not asset.get("s3_uri"):
267 continue
268 typer.echo(f"- {asset['name']:<20}: {asset['description']}")
271if __name__ == "__main__":
272 app()