Jmaa/personal-data
1
0
Fork 0
Code Actions Packages

Further cleanup
All checks were successful
Run Python tests (through Pytest) / Test (push) Successful in 36s
Details
Verify Python project can be installed, loaded and have version checked / Test (push) Successful in 31s
Details

Browse Source
This commit is contained in:
Jon Michael Aanes 2025-03-24 20:33:10 +01:00
parent 0d787532d4
commit 0cc55e44ff
1 changed files with 11 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
personal_data/fetchers/wanikani_lessons.py
View File

@ -1,5 +1,6 @@
import dataclasses import dataclasses
import datetime import datetime
import requests
import logging import logging
from collections.abc import Iterator, Mapping from collections.abc import Iterator, Mapping
from email.utils import parsedate_to_datetime from email.utils import parsedate_to_datetime
@ -65,15 +66,16 @@ class WaniKaniLessonsFetcher(Scraper):
yield data_item yield data_item
url = json_resp.get('pages', {}).get('next_url') url = json_resp.get('pages', {}).get('next_url')
def date_from_response(response) -> datetime.datetime:
if date_header := response.headers.get('Date'):
return parsedate_to_datetime(date_header)
return datetime.datetime.now(datetime.timezone.utc)
@dataclasses.dataclass(frozen=True) @dataclasses.dataclass(frozen=True)
class WaniKaniSummaryFetcher(Scraper): class WaniKaniSummaryFetcher(Scraper):
dataset_name: str = 'wanikani_summary' dataset_name: str = 'wanikani_summary'
deduplicate_mode = DeduplicateMode.BY_ALL_COLUMNS deduplicate_mode = DeduplicateMode.BY_ALL_COLUMNS
deduplicate_ignore_columns = ['time.current']
@staticmethod
def deduplicate_mode() -> DeduplicateMode:
return DeduplicateMode.BY_ALL_COLUMNS
def scrape(self) -> Iterator[dict]: def scrape(self) -> Iterator[dict]:
_setup_cache(self.session) _setup_cache(self.session)
@ -81,7 +83,7 @@ class WaniKaniSummaryFetcher(Scraper):
'Authorization': f'Bearer {secrets.wanikani_api_key()}', 'Authorization': f'Bearer {secrets.wanikani_api_key()}',
'Wanikani-Revision': '20170710', 'Wanikani-Revision': '20170710',
} }
response = self.session.get(URL_SUMMARY, headers=headers) response = requests.get(URL_SUMMARY, headers=headers)
response.raise_for_status() response.raise_for_status()
data = response.json() data = response.json()
@ -89,16 +91,8 @@ class WaniKaniSummaryFetcher(Scraper):
total_lessons = sum(len(lesson.get('subject_ids', [])) for lesson in lessons) total_lessons = sum(len(lesson.get('subject_ids', [])) for lesson in lessons)
reviews = data.get('data', {}).get('reviews', []) reviews = data.get('data', {}).get('reviews', [])
date_hdr = response.headers.get('Date')
try:
now = (
parsedate_to_datetime(date_hdr)
if date_hdr
else datetime.datetime.now(datetime.timezone.utc)
)
except Exception:
now = datetime.datetime.now(datetime.timezone.utc)
total_reviews = 0 total_reviews = 0
now = date_from_response(response)
for review in reviews: for review in reviews:
available_at_str = review.get('available_at') available_at_str = review.get('available_at')
if available_at_str: if available_at_str:
@ -110,7 +104,7 @@ class WaniKaniSummaryFetcher(Scraper):
del review del review
yield { yield {
'time': now, 'time.current': now,
'lessons_available': total_lessons, 'lessons.available': total_lessons,
'reviews_available': total_reviews, 'reviews.available': total_reviews,
} }