diff --git a/personal_data/fetchers/wanikani_lessons.py b/personal_data/fetchers/wanikani_lessons.py index a104486..7d38df0 100644 --- a/personal_data/fetchers/wanikani_lessons.py +++ b/personal_data/fetchers/wanikani_lessons.py @@ -1,50 +1,55 @@ +import dataclasses import datetime import logging +from collections.abc import Iterator, Mapping + import requests -import dataclasses -from typing import Iterator, Mapping + # Import the base Scraper class; adjust the import if your code structure differs. from personal_data.data import Scraper logger = logging.getLogger(__name__) + @dataclasses.dataclass(frozen=True) class WaniKaniLessonsFetcher(Scraper): api_token: str @staticmethod def dataset_name() -> str: - return "wanikani_lessons" + return 'wanikani_lessons' def scrape(self) -> Iterator[Mapping[str, object]]: """ Fetch assignments from the WaniKani API and yield a dict for each assignment with a non-null unlocked_at timestamp. """ - url = "https://api.wanikani.com/v2/assignments" + url = 'https://api.wanikani.com/v2/assignments' headers = { - "Authorization": f"Bearer {self.api_token}", - "Wanikani-Revision": "20170710" + 'Authorization': f'Bearer {self.api_token}', + 'Wanikani-Revision': '20170710', } response = requests.get(url, headers=headers) if response.status_code != 200: - logger.error("Error retrieving assignments: %s", response.text) + logger.error('Error retrieving assignments: %s', response.text) return data = response.json() # Check that 'data' key exists in the JSON response. - assignments = data.get("data", []) + assignments = data.get('data', []) for assignment in assignments: - assignment_data = assignment.get("data", {}) + assignment_data = assignment.get('data', {}) # Only yield if unlocked_at is available. - unlocked_at = assignment_data.get("unlocked_at") + unlocked_at = assignment_data.get('unlocked_at') if unlocked_at: # Convert unlocked_at ISO8601 string (assume 'Z' for UTC) to a datetime object. try: - dt = datetime.datetime.fromisoformat(unlocked_at.replace("Z", "+00:00")) + dt = datetime.datetime.fromisoformat( + unlocked_at.replace('Z', '+00:00'), + ) except Exception as e: logger.error("Error parsing unlocked_at '%s': %s", unlocked_at, e) continue yield { - "subject_id": assignment_data.get("subject_id"), - "unlocked_at": dt + 'subject_id': assignment_data.get('subject_id'), + 'unlocked_at': dt, }