103 lines
3.0 KiB
Python
103 lines
3.0 KiB
Python
|
"""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
|
||
|
|
|