pyalexatodo.cli

  1import asyncio
  2import json
  3import sys
  4from functools import wraps
  5from pathlib import Path
  6from typing import Any, cast
  7
  8try:
  9    import keyring
 10    import orjson
 11    import typer
 12    from rich.console import Console
 13except ImportError:
 14    print("Required packages for CLI usage are not installed. Please install pyalexatodo[cli] to install them. "
 15          "For example with pip: pip install \"pyalexatodo[cli]\"")
 16    sys.exit(1)
 17
 18from aioamazondevices import CannotAuthenticate, CannotConnect
 19from aioamazondevices.api import AmazonEchoApi
 20from aioamazondevices.exceptions import AmazonError, CannotRegisterDevice
 21from aiohttp import ClientSession
 22
 23from pyalexatodo.api import AlexaToDoAPI
 24from pyalexatodo.exceptions import ItemNotFoundException
 25from pyalexatodo.models.cli_settings import CliSettings
 26
 27app = typer.Typer()
 28console = Console()
 29
 30# Global variables to hold the API client and default list ID after initialization
 31alexa_list_api: AlexaToDoAPI
 32client_session: ClientSession
 33default_list_id: str = ""
 34
 35KEYRING_SERVICE = "alexalists-cli"
 36PASSWORD_KEY = "amazon-password"
 37
 38### Helper functions for file I/O and API initialization ###
 39
 40def read_from_file(data_file: str) -> dict[str, Any]:
 41    """Load stored login data from file."""
 42    if not data_file or not Path(data_file).exists():
 43        print(
 44            "Cannot find previous login data file: ",
 45            data_file,
 46        )
 47        return {}
 48
 49    with open(Path(data_file), "r") as f:
 50        return cast("dict[str, Any]", json.loads(f.read()))
 51
 52
 53def save_to_file(
 54    raw_data: str | dict[str, Any],
 55    content_type: str = "application/json",
 56) -> None:
 57    """Save login_data data to disk."""
 58    if not raw_data:
 59        return
 60
 61    try:
 62        fullpath = Path(get_outputpath("login_data.json"))
 63
 64        # Create main output directory and timestamp subdirectory
 65        output_dir = fullpath.parent
 66        output_dir.mkdir(parents=True, exist_ok=True)
 67
 68        # Convert dict to JSON string if needed
 69        if isinstance(raw_data, dict):
 70            json_data = raw_data
 71        else:
 72            # Assume it's a JSON string
 73            json_data = orjson.loads(raw_data)
 74
 75        data = orjson.dumps(
 76            json_data,
 77            option=orjson.OPT_INDENT_2,
 78        ).decode("utf-8")
 79
 80        print(f"Saving data to {fullpath}")
 81
 82        with open(fullpath, "w", encoding="utf-8") as file:
 83            file.write(data)
 84            file.write("\n")
 85    except Exception as e:
 86        print(f"Error saving login data: {e}")
 87
 88def get_outputpath(filename: str) -> str:
 89    """Get the absolute path for storing application files.
 90
 91    Args:t
 92        filename (str): The name of the file.
 93
 94    Returns:
 95        str: The absolute path to the file in the user's home directory.
 96    """
 97    return Path(Path.home(), ".pyalexatodo", filename).as_posix()
 98
 99
100async def init_api():
101    """Initialize the Alexa API using stored credentials and settings.
102
103    This function:
104    1. Loads the CLI settings from the config file
105    2. Retrieves credentials from the system keyring
106    3. Initializes and tests the Alexa login
107    4. Creates the API instance
108
109    Returns:
110        AlexaListAPI: An initialized API instance.
111
112    Raises:
113        FileNotFoundError: If the CLI settings file is not found.
114        SystemExit: If login fails or settings are invalid.
115    """
116    global alexa_list_api, default_list_id, client_session
117
118    try:
119        # Load CLI settings
120        with open(get_outputpath("cli_settings.json"), "r") as f:
121            settings = CliSettings.model_validate_json(f.read())
122
123            login_data_stored = read_from_file(get_outputpath("login_data.json"))
124
125            client_session = ClientSession()
126
127            password = keyring.get_password(
128                KEYRING_SERVICE, f"{settings.email}-{PASSWORD_KEY}"
129            )
130
131            if not password:
132                console.print(
133                    f"[bold red]No password found in keyring for {settings.email}. Please run setup option first.[/bold red]"
134                )
135                sys.exit(1)
136
137            amazon_echo_api = AmazonEchoApi(
138                client_session=client_session,
139                login_email=settings.email,
140                login_password=password,
141                login_data=login_data_stored,
142            )
143
144            try:
145                await amazon_echo_api.login.login_mode_stored_data()
146            except CannotAuthenticate:
147                console.print(
148                    f"[bold red]Cannot authenticate with {settings.email} credentials[/bold red]"
149                )
150                raise
151            except CannotConnect:
152                console.print(
153                    f"[bold red]Cannot connect to {amazon_echo_api.domain} Amazon host[/bold red]"
154                )
155                raise
156            except CannotRegisterDevice:
157                console.print(
158                    f"[bold red]Cannot register device for {settings.email}[/bold red]"
159                )
160                raise
161
162        alexa_list_api = AlexaToDoAPI(amazon_echo_api)
163        default_list_id = settings.default_list_id
164
165    except AmazonError:
166        console.print("[bold red]Login failed.[/bold red]")
167        sys.exit(1)
168    except FileNotFoundError:
169        console.print(
170            "CLI settings not found. Please run setup option first.", style="red"
171        )
172        sys.exit(1)
173
174### Decorators for CLI commands ###
175
176def with_alexa_api(func):
177    """Decorator that initializes the Alexa API connection before function execution.
178
179    Args:
180        func: The async function to wrap.
181
182    Returns:
183        wrapper: The wrapped function that handles API initialization and cleanup.
184
185    Example:
186        @with_alexa_api
187        async def my_function():
188            # Function will have access to initialized alexa_list_api
189            pass
190    """
191
192    @wraps(func)
193    async def wrapper(*args, **kwargs):
194        try:
195            with console.status("Logging into Alexa API..."):
196                await init_api()
197            return await func(*args, **kwargs)
198        finally:
199            if client_session:
200                await client_session.close()
201
202    return wrapper
203
204
205def cli_command(func):
206    """Decorator that wraps an async function to run in the asyncio event loop.
207
208    Args:
209        func: The async function to wrap.
210
211    Returns:
212        wrapper: The wrapped function that handles asyncio.run.
213
214    Example:
215        @cli_command
216        async def my_function():
217            # Function will run in asyncio event loop
218            pass
219    """
220
221    @wraps(func)
222    def wrapper(*args, **kwargs):
223        return asyncio.run(func(*args, **kwargs))
224
225    return wrapper
226
227### CLI Commands ###
228
229@app.command()
230def setup():
231    """Command to set up the Alexa Lists CLI with user credentials and preferences."""
232    asyncio.run(setup_async())
233
234
235async def setup_async():
236    """Async implementation of the setup command.
237
238    Guides the user through the setup process:
239    1. Collects Amazon credentials and OTP secret
240    2. Stores sensitive data in system keyring
241    3. Authenticates with Amazon
242    4. Lets user select default list
243    5. Saves non-sensitive settings to file
244
245    Raises:
246        Exception: If any step of the setup process fails
247    """
248    try:
249        # Welcome message
250        console.print("[bold blue]Welcome to the Alexa Lists CLI Setup![/bold blue]")
251        console.print("This will guide you through setting up your Alexa Lists CLI.")
252        console.print(
253            "You will need your Amazon account credentials and an OTP token for two-factor authentication.\n"
254        )
255        console.print(
256            "The password will be stored securely in your system's keyring.\n"
257        )
258
259        while True:
260            email = console.input("Enter your Amazon email: ").strip()
261            if email and "@" in email and "." in email:
262                break
263            console.print(
264                "[red]Invalid email format. Please enter a valid email address.[/red]"
265            )
266
267        while True:
268            password = console.input("Enter your Amazon password: ", password=True)
269            if password:  # Basic password length check
270                break
271            console.print("[red]Password cannot be empty.[/red]")
272
273        # Store sensitive data in keyring
274        keyring.set_password(KEYRING_SERVICE, f"{email}-{PASSWORD_KEY}", password)
275
276        client_session = ClientSession()
277
278        amazon_echo_api = AmazonEchoApi(
279            client_session=client_session, login_email=email, login_password=password
280        )
281
282        while True:
283            otp_token = console.input("Enter current OTP token: ")
284            if len(otp_token) == 6:  # Basic OTP token length check
285                break
286            console.print("[red]OTP token must be 6 digits.[/red]")
287
288        with console.status("[bold blue]Logging into Alexa API..."):
289            try:
290                login_data = await amazon_echo_api.login.login_mode_interactive(
291                    otp_token
292                )
293            except CannotAuthenticate:
294                console.print(
295                    f"[bold red]Cannot authenticate with {email} credentials[/bold red]"
296                )
297                raise
298            except CannotConnect:
299                console.print(
300                    f"[bold red]Cannot connect to {amazon_echo_api.domain} Amazon host[/bold red]"
301                )
302                raise
303            except CannotRegisterDevice:
304                console.print(
305                    f"[bold red]Cannot register device for {email}[/bold red]"
306                )
307                raise
308
309        with console.status("[bold blue]Saving login data to disk..."):
310            save_to_file(login_data)
311
312        console.print("[green]Logged in successfully![/green]")
313
314        # Get available lists
315        alexa_list_api = AlexaToDoAPI(amazon_echo_api)
316        with console.status("[bold blue]Fetching available lists..."):
317            lists = await alexa_list_api.get_lists()
318
319        # Display lists and get user selection
320        console.print("\n[bold]Available Lists:[/bold]")
321        for i, list_info in enumerate(lists):
322            console.print(f"  [{i}] {list_info.name}")
323
324        while True:
325            default_list_id = console.input(
326                "\nWhich is your default list? Enter the number: "
327            )
328            if default_list_id.isdigit() and 0 <= int(default_list_id) < len(lists):
329                default_list_id = lists[int(default_list_id)].id
330                break
331            console.print("[red]Invalid input. Please enter a valid list number.[/red]")
332
333        # Save non-sensitive settings
334        cli_settings = CliSettings(
335            email=email,
336            default_list_id=default_list_id,
337        )
338
339        settings_path = get_outputpath("cli_settings.json")
340        cli_settings_json = cli_settings.model_dump_json(indent=4)
341        with open(settings_path, "w") as f:
342            f.write(cli_settings_json)
343
344        console.print("[green]Settings and credentials saved successfully![/green]")
345
346    except AmazonError:
347        console.print("[bold red]Login failed.[/bold red]")
348        sys.exit(1)
349    except Exception:
350        console.print("[bold red]Error during setup:[/bold red]")
351        raise  # Typer will catch this and print the stack trace for debugging
352    finally:
353        if "alexa_login" in locals():
354            await client_session.close()
355
356
357@app.command()
358@cli_command
359@with_alexa_api
360async def list(list_id: str = ""):
361    """Fetch and display all items from a specified Alexa list.
362
363    Args:
364        list_id: The ID of the list to fetch items from.
365            If not provided, uses the default list.
366    """
367    if not list_id:
368        list_id = default_list_id
369
370    with console.status("Fetching list items from Alexa API..."):
371        list_items = await alexa_list_api.get_list_items(list_id)
372
373    for list_item in list_items:
374        line = typer.style(
375            f"[{'x' if list_item.is_checked else ' '}] {list_item.name}",
376            fg=typer.colors.GREEN if list_item.is_checked else typer.colors.RED,
377        )
378        typer.echo(line)
379
380
381@app.command()
382@cli_command
383@with_alexa_api
384async def check(item_name: str, list_id: str = ""):
385    """Toggle the checked status of an item in a specified Alexa list.
386
387    Args:
388        item_name: The name of the item to toggle.
389        list_id: The ID of the list containing the item.
390            If not provided, uses the default list.
391
392    Raises:
393        ItemNotFoundException: If the item is not found in the list.
394    """
395    if not list_id:
396        list_id = default_list_id
397
398    try:
399        with console.status("Fetching item from Alexa API and find item by name..."):
400            item = await alexa_list_api.get_item_by_name(list_id, item_name)
401
402        if item is None:
403            console.print(f'Item "{item_name}" not found.', style="red")
404            return
405
406        with console.status("Toggling item status..."):
407            await alexa_list_api.set_item_checked_status(
408                list_id, item.id, not item.is_checked, item.version
409            )
410
411        console.print(f'Item "{item_name}" toggled sucessfully.', style="green")
412    except ItemNotFoundException:
413        console.print(f'Item "{item_name}" not found.', style="red")
414
415
416@app.command()
417@cli_command
418@with_alexa_api
419async def add(item_name: str, list_id: str = ""):
420    """Add a new item to a specified Alexa list.
421
422    Args:
423        item_name: The name of the item to add.
424        list_id: The ID of the list to add the item to.
425            If not provided, uses the default list.
426    """
427    if not list_id:
428        list_id = default_list_id
429
430    with console.status("Adding item to list..."):
431        await alexa_list_api.add_item(list_id, item_name)
432
433    console.print(f'Item "{item_name}" added successfully.', style="green")
434
435
436@app.command()
437@cli_command
438@with_alexa_api
439async def remove(item_name: str, list_id: str = ""):
440    """Remove an item from a specified Alexa list.
441
442    Args:
443        item_name: The name of the item to remove.
444        list_id: The ID of the list to remove the item from.
445            If not provided, uses the default list.
446
447    Raises:
448        ItemNotFoundException: If the item is not found in the list.
449    """
450    if not list_id:
451        list_id = default_list_id
452
453    try:
454        with console.status("Fetching item from Alexa API and find item by name..."):
455            item = await alexa_list_api.get_item_by_name(list_id, item_name)
456
457        with console.status("Removing item from list..."):
458            await alexa_list_api.delete_item(list_id, item.id, item.version)
459
460        console.print(f'Item "{item_name}" removed successfully.', style="green")
461    except ItemNotFoundException:
462        console.print(f'Item "{item_name}" not found.', style="red")
463
464@app.command()
465@cli_command
466@with_alexa_api
467async def lists():
468    """Fetch and display all available Alexa lists."""
469    with console.status("Fetching available lists from Alexa API..."):
470        lists = await alexa_list_api.get_lists()
471
472    console.print("\n[bold]Available Lists:[/bold]")
473    for list_info in lists:
474        console.print(f"  - {list_info.name} (ID: {list_info.id})")
475
476if __name__ == "__main__":
477    app()