51 lines
1.9 KiB
Python
51 lines
1.9 KiB
Python
import datetime
|
|
import logging
|
|
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"
|
|
|
|
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"
|
|
headers = {
|
|
"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)
|
|
return
|
|
data = response.json()
|
|
# Check that 'data' key exists in the JSON response.
|
|
assignments = data.get("data", [])
|
|
for assignment in assignments:
|
|
assignment_data = assignment.get("data", {})
|
|
# Only yield if unlocked_at is available.
|
|
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"))
|
|
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
|
|
}
|