diff --git a/personal_data/fetchers/withings.py b/personal_data/fetchers/withings.py new file mode 100644 index 0000000..0f8e954 --- /dev/null +++ b/personal_data/fetchers/withings.py @@ -0,0 +1,102 @@ +"""Withings API fetcher. + +Supports downloading activity summary from the [Withings +API](https://developer.withings.com/api-reference/) using the [non-official +Withings API Python Client](https://pypi.org/project/withings-api/). +""" + +import withings_api +from withings_api.common import get_measure_value, MeasureType, CredentialsType +import datetime + +import dataclasses +import logging +import re +from collections.abc import Iterator +import subprocess + +import bs4 +import requests_util + +import personal_data.html_util +from personal_data import secrets +from personal_data.data import DeduplicateMode, Scraper + +from .. import parse_util +import pickle +from pathlib import Path + +logger = logging.getLogger(__name__) + +CREDENTIALS_FILE = Path('secrets/withings_oath_creds') + +def save_credentials(credentials: CredentialsType) -> None: + """Save credentials to a file.""" + logger.info("Saving credentials in: %s", CREDENTIALS_FILE) + with open(CREDENTIALS_FILE, "wb") as file_handle: + pickle.dump(credentials, file_handle) + + +def load_credentials() -> CredentialsType: + """Load credentials from a file.""" + logger.info("Using credentials saved in: %s", CREDENTIALS_FILE) + try: + with open(CREDENTIALS_FILE, "rb") as file_handle: + return pickle.load(file_handle) + except FileNotFoundError: + return None + +@dataclasses.dataclass(frozen=True) +class WithingsActivityScraper(Scraper): + dataset_name = 'withings_activity' + deduplicate_mode = DeduplicateMode.BY_ALL_COLUMNS + + @staticmethod + def requires_cfscrape() -> bool: + return False + + def oauth_flow(self) -> CredentialsType: + if creds := load_credentials(): + return creds + + auth = withings_api.WithingsAuth( + client_id=secrets.WITHINGS_CLIENTID, + consumer_secret=secrets.WITHINGS_SECRET, + callback_uri=secrets.WITHINGS_CALLBACK_URI, + scope=( + withings_api.AuthScope.USER_ACTIVITY, + withings_api.AuthScope.USER_METRICS, + withings_api.AuthScope.USER_INFO, + withings_api.AuthScope.USER_SLEEP_EVENTS, + ), + ) + + authorize_url = auth.get_authorize_url() + + subprocess.run(['firefox', '--new-tab', authorize_url]) + credentials_code = input('Please insert your code here: ').strip() + + creds = auth.get_credentials(credentials_code) + save_credentials(creds) + return creds + + def scrape(self): + credentials = self.oauth_flow() + + # Now you are ready to make calls for data. + api = withings_api.WithingsApi(credentials) + + start = datetime.date.today() - datetime.timedelta(days = 200) + end = datetime.date.today() + + activity_result = api.measure_get_activity( + startdateymd=start, + enddateymd=end, + ) + for activity in activity_result.activities: + sample = dict(activity) + sample['date'] = activity.date.date() + del sample['timezone'], sample['is_tracker'] + yield sample + del activity, sample + diff --git a/personal_data/secrets.py b/personal_data/secrets.py index 14ef980..fa383b4 100644 --- a/personal_data/secrets.py +++ b/personal_data/secrets.py @@ -40,3 +40,8 @@ MAILGUN_RECIPIENT = load_secret('MAILGUN_RECIPIENT') JELLYFIN_URL = load_secret('JELLYFIN_URL') JELLYFIN_USERNAME = load_secret('JELLYFIN_USERNAME') JELLYFIN_PASSWORD = load_secret('JELLYFIN_PASSWORD') + +# Withings +WITHINGS_CLIENTID = load_secret('WITHINGS_CLIENTID') +WITHINGS_SECRET = load_secret('WITHINGS_SECRET') +WITHINGS_CALLBACK_URI = load_secret('WITHINGS_CALLBACK_URI') diff --git a/personal_data/util.py b/personal_data/util.py index b0a800f..748f825 100644 --- a/personal_data/util.py +++ b/personal_data/util.py @@ -1,6 +1,7 @@ import csv import datetime import decimal +import _csv import io import logging import typing @@ -180,7 +181,7 @@ def extend_csv_file( try: dicts = load_csv_file(csv_file) - except FileNotFoundError as e: + except (FileNotFoundError, _csv.Error) as e: logger.info('Creating file: %s', csv_file) dicts = []