diff --git a/update_tracks.py b/update_tracks.py index 750a237..94d6877 100644 --- a/update_tracks.py +++ b/update_tracks.py @@ -16,11 +16,11 @@ ACCESS_TOKEN = os.getenv("GROUNDED_TOKEN") def get_author_name(data: dict) -> str: """ Extracts the author's display name from a nested track data dictionary. - + Parameters: data (dict): A mapping representing track/result data. Expected shapes include {"author": {"user": {"name": ..., "nickname": ...}}} or {"author": {"name": ..., "nickname": ...}}. - + Returns: str: The author's `name` if present, otherwise the author's `nickname`, otherwise "Kinetic Oasis". """ @@ -30,47 +30,51 @@ def get_author_name(data: dict) -> str: async def update_tracks() -> None: """ - Fetch tracks from the Grounded Labs cloud, detect new or changed public tracks compared to the local TRACKS mapping, augment changed entries with author and reduced SVG content, and persist the merged, sorted track list to custom_components/oasis_mini/pyoasiscontrol/tracks.json. - + Fetch tracks from the Grounded Labs cloud, detect new or changed public tracks + compared to the local TRACKS mapping, augment changed entries with author and + reduced SVG content, and persist the merged, sorted track list to + `custom_components/oasis_mini/pyoasiscontrol/tracks.json`. + Side effects: - May print error or status messages to stdout. - Writes the updated tracks JSON file. - - Ensures the OasisCloudClient session is closed and returns early on errors or unexpected data. + - Ensures the OasisCloudClient session is closed and returns early on errors or + unexpected data. """ client = OasisCloudClient(access_token=ACCESS_TOKEN) - try: - data = await client.async_get_tracks() - except Exception as ex: - print(type(ex).__name__, ex) - await client.session.close() - return + try: + data = await client.async_get_tracks() + except Exception as ex: + print(type(ex).__name__, ex) + return - if not isinstance(data, list): - print("Unexpected result:", data) - return + if not isinstance(data, list): + print("Unexpected result:", data) + return - updated_tracks: dict[int, dict[str, Any]] = {} - for result in filter(lambda d: d["public"], data): - if ( - (track_id := result["id"]) not in TRACKS - or any( - result[field] != TRACKS[track_id].get(field) - for field in ("name", "image", "png_image") - ) - or TRACKS[track_id].get("author") != get_author_name(result) - ): - print(f"Updating track {track_id}: {result['name']}") - track_info = await client.async_get_track_info(int(track_id)) - if not track_info: - print("No track info") - break - result["author"] = get_author_name(result) - result["reduced_svg_content_new"] = track_info.get( - "reduced_svg_content_new" - ) - updated_tracks[track_id] = result - await client.session.close() + updated_tracks: dict[int, dict[str, Any]] = {} + for result in filter(lambda d: d["public"], data): + if ( + (track_id := result["id"]) not in TRACKS + or any( + result[field] != TRACKS[track_id].get(field) + for field in ("name", "image", "png_image") + ) + or TRACKS[track_id].get("author") != get_author_name(result) + ): + print(f"Updating track {track_id}: {result['name']}") + track_info = await client.async_get_track_info(int(track_id)) + if not track_info: + print("No track info") + break + result["author"] = get_author_name(result) + result["reduced_svg_content_new"] = track_info.get( + "reduced_svg_content_new" + ) + updated_tracks[track_id] = result + finally: + await client.async_close() if not updated_tracks: print("No updated tracks") @@ -87,4 +91,4 @@ async def update_tracks() -> None: if __name__ == "__main__": - asyncio.run(update_tracks()) \ No newline at end of file + asyncio.run(update_tracks())