56 lines
1.9 KiB
Python
56 lines
1.9 KiB
Python
import dataclasses
|
|
import datetime
|
|
import logging
|
|
from collections.abc import Iterator, Mapping
|
|
|
|
import requests
|
|
|
|
# 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,
|
|
}
|