タニタHealthPlanetAPIから体重と体脂肪率を取得し、FitbitAPIへ送った話

これを作った

どちらもAPIへのPOSTはドキュメント通りにやればよく、OAuth2.0のトークン払い出しが一番大変だった

■HealthPlanetAPI
[トークン払い出し]https://www.healthplanet.jp/apis_account.do
[ドキュメント]https://www.healthplanet.jp/apis/api.html

■FitbitAPI
[トークン払い出し]https://www.fitbit.com/dev
[ドキュメント]https://dev.fitbit.com/build/reference/web-api/
[体重のAPIドキュメント]https://dev.fitbit.com/build/reference/web-api/body/create-weight-log/
[体脂肪のAPIドキュメント]https://dev.fitbit.com/build/reference/web-api/body/create-bodyfat-log/

・アプリ作成のところやトークン払い出しの最初のところで、アプリケーションタイプを選択するが、自分以外のユーザーを扱いたい場合はどちらも「Server」を選択する。
・Clientを選択すると、APIのユーザーを入れるURI部分に、ユーザー名を入れる代わりに「-」（ハイフン）を入れることで処理を簡略化できる。また、クライアントシークレットも必要ない。
・アプリ作成の段階でアプリケーションタイプを「Personal」選択すると、Step2のAuthorizatonURLを他人が踏んでも、HTTPエラーが返される。
・最初はApplicationTypeをPersonalで開発して、個人用ならClientに切り替えて、他人のユーザーIDも扱いたいならServerを選択するとよさそう。

■プログラム

fitbitAPIのアクセストークンの寿命が8時間で、切れたらリフレッシュトークンで新しいリフレッシュトークンと新しいアクセストークンを払い出す必要がある。そして保存する。

body_data_sync_aws.py

import json
import boto3
import urllib.parse
import urllib.request
from datetime import datetime, timedelta

# AWS Systems Manager (SSM) クライアント
ssm_client = boto3.client("ssm")

# ============================================================
# JSONデータの取得・更新 (AWS SSM パラメータストア)
# ============================================================

def load_user_data():
    """AWS Systems Manager Parameter Store からユーザーデータを取得"""
    try:
        response = ssm_client.get_parameter(Name="********", WithDecryption=True)
        return json.loads(response["Parameter"]["Value"])
    except ssm_client.exceptions.ParameterNotFound:
        print("Error: Parameter not found in SSM")
        return {"users": []}

def save_user_data(data):
    """AWS Systems Manager Parameter Store にユーザーデータを保存"""
    ssm_client.put_parameter(
        Name="HealthPlanetFitbitUsers",
        Value=json.dumps(data, ensure_ascii=False, indent=4),
        Type="String",
        Overwrite=True
    )

def get_user_list():
    """ユーザー名のリストを取得"""
    data = load_user_data()
    return [user["user"] for user in data["users"]]

def get_user_credentials(user_name):
    """指定ユーザーのクレデンシャルを取得"""
    data = load_user_data()
    print(data)
    for user in data["users"]:
        if user["user"] == user_name:
            return (
                user.get("fitbit_user_id"),
                user.get("fitbit_refresh_token"),
                user.get("fitbit_access_token"),
                data.get("fitbit_client_id"),
                user.get("health_planet_access_token"),
            )
    return None, None, None, None, None

def update_fitbit_tokens(fb_user_id, fb_access_token, fb_refresh_token):
    """Fitbit のアクセストークンとリフレッシュトークンを更新"""
    data = load_user_data()
    for user in data["users"]:
        if user["fitbit_user_id"] == fb_user_id:
            user["fitbit_access_token"] = fb_access_token
            user["fitbit_refresh_token"] = fb_refresh_token
            break
    save_user_data(data)


# ============================================================
# HealthPlanetからデータ取得
# ============================================================
def fetch_healthplanet_data(hp_access_token):
    url = "https://www.healthplanet.jp/status/innerscan.json"
    date_str = (datetime.now() - timedelta(days=1)).strftime("%Y%m%d")
    
    params = {
        "access_token": hp_access_token,
        "date": "1",
        "from": f"{date_str}000000",
        "to": f"{date_str}235959",
        "tag": "6021,6022"
    }
    req = urllib.request.Request(url, data=urllib.parse.urlencode(params).encode("utf-8"), method="POST")

    try:
        with urllib.request.urlopen(req) as response:
            data = json.loads(response.read().decode("utf-8"))
            weight, fat = None, None
            for item in reversed(data.get("data", [])):
                if item["tag"] == "6021":
                    weight = item["keydata"]
                elif item["tag"] == "6022":
                    fat = item["keydata"]
            return weight, fat
    except urllib.error.URLError as e:
        print(f"Error fetching HealthPlanet data: {e}")
    return None, None

# ============================================================
# Fitbit API操作
# ============================================================
def log_weight_to_fitbit(access_token, user_id, weight):
    date = datetime.today().strftime("%Y-%m-%d")
    url = f"https://api.fitbit.com/1/user/{user_id}/body/log/weight.json?weight={weight}&date={date}"
    fitbit_request(url, access_token)

def log_body_fat_to_fitbit(access_token, user_id, body_fat):
    date = datetime.today().strftime("%Y-%m-%d")
    url = f"https://api.fitbit.com/1/user/{user_id}/body/log/fat.json?fat={body_fat}&date={date}"
    fitbit_request(url, access_token)

def fitbit_request(url, access_token):
    headers = {"Authorization": f"Bearer {access_token}", "Accept": "application/json"}
    req = urllib.request.Request(url, method="POST", headers=headers)
    try:
        with urllib.request.urlopen(req) as response:
            print("Fitbit Body Data Sync Success!(Weight or Fat)")
            return response.read().decode("utf-8")
    except urllib.error.HTTPError as e:
        print(f"HTTP Error: {e.code} {e.reason}")
    return None

def refresh_fitbit_token(fb_user_id, client_id, old_refresh_token):
    url = "https://api.fitbit.com/oauth2/token"
    headers = {
        "Authorization": "Basic ***",
        "Content-Type": "application/x-www-form-urlencoded"
    }
    data = {
        "grant_type": "refresh_token",
        "client_id": client_id,
        "refresh_token": old_refresh_token
    }
    req = urllib.request.Request(url, data=urllib.parse.urlencode(data).encode("utf-8"), headers=headers, method="POST")
    try:
        with urllib.request.urlopen(req) as response:
            res_data = json.loads(response.read().decode("utf-8"))
            update_fitbit_tokens(fb_user_id, res_data["access_token"], res_data["refresh_token"])
            return True
    except:
        print("Failed to refresh Fitbit token.")
    return False

# ============================================================
# メイン処理
# ============================================================
def lambda_handler(event, context):
    user_list = get_user_list()
    for user in user_list:
        print("user is " + user)
        fb_user_id, fb_refresh_token, fb_access_token, client_id, hp_access_token = get_user_credentials(user)
        print(fb_user_id, fb_refresh_token, fb_access_token, client_id, hp_access_token)
        weight, body_fat = fetch_healthplanet_data(hp_access_token)
        print(weight, body_fat)

        if weight is not None:
            if fitbit_request(f"https://api.fitbit.com/1/user/{fb_user_id}/profile.json", fb_access_token):
                log_weight_to_fitbit(fb_access_token, fb_user_id, weight)
                log_body_fat_to_fitbit(fb_access_token, fb_user_id, body_fat)
            elif refresh_fitbit_token(fb_user_id, client_id, fb_refresh_token):
                print("weight is None in Python")
                fb_user_id, fb_refresh_token, fb_access_token, _, _ = get_user_credentials(user)
                log_weight_to_fitbit(fb_access_token, fb_user_id, weight)
                log_body_fat_to_fitbit(fb_access_token, fb_user_id, body_fat)
            else:
                print(f"{user}: fitbit token is not available.")
        else:
            print(f"{user}: No HealthPlanet data available.")

ユーザーデータのJSON

user_data.json

{
    "users": [
        {
            "user": "****",
            "fitbit_user_id": "***",
            "health_planet_access_token": "********",
            "fitbit_access_token": "*******",
            "fitbit_refresh_token": "********"
        },
        {
            "user": "****",
            "fitbit_user_id": "***",
            "health_planet_access_token": "********",
            "fitbit_access_token": "*******",
            "fitbit_refresh_token": "********"
        },
        {
            "user": "****",
            "fitbit_user_id": "***",
            "health_planet_access_token": "********",
            "fitbit_access_token": "*******",
            "fitbit_refresh_token": "********"
        }
    ],
    "fitbit_client_id": "*****",
    "fitbit_client_secret": "************"
}