From 5b381b07236ed02dc7fc6b8780e464fbc94cc44a Mon Sep 17 00:00:00 2001 From: ben_29 Date: Fri, 20 Oct 2023 00:01:47 +0800 Subject: [PATCH] fix: #484 keep gps data decode --- requirements.txt | 1 + run_page/keep_sync.py | 57 ++++++++++++++++++++++--------------------- 2 files changed, 30 insertions(+), 28 deletions(-) diff --git a/requirements.txt b/requirements.txt index b04ec294f98..da5d2237234 100644 --- a/requirements.txt +++ b/requirements.txt @@ -25,4 +25,5 @@ tzlocal fit-tool haversine==2.8.0 garth +pycryptodome diff --git a/run_page/keep_sync.py b/run_page/keep_sync.py index c632d0ae42c..ad34200ce54 100755 --- a/run_page/keep_sync.py +++ b/run_page/keep_sync.py @@ -7,6 +7,7 @@ import zlib from collections import namedtuple from datetime import datetime, timedelta +from Crypto.Cipher import AES import eviltransform import gpxpy @@ -21,6 +22,10 @@ RUN_DATA_API = "https://api.gotokeep.com/pd/v3/stats/detail?dateUnit=all&type=running&lastDate={last_date}" RUN_LOG_API = "https://api.gotokeep.com/pd/v3/runninglog/{run_id}" +# AES Decrypt key +key = b"56fe59;82g:d873c" +iv = b"2346892432920300" + # If your points need trans from gcj02 to wgs84 coordinate which use by Mappbox TRANS_GCJ02_TO_WGS84 = True @@ -64,8 +69,12 @@ def get_single_run_data(session, headers, run_id): return r.json() -def decode_runmap_data(text): - run_points_data = zlib.decompress(base64.b64decode(text), 16 + zlib.MAX_WBITS) +def decode_runmap_data(text, is_geo=False): + _bytes = base64.b64decode(text) + if is_geo: + cipher = AES.new(key, AES.MODE_CBC, iv) + _bytes = cipher.decrypt(_bytes) + run_points_data = zlib.decompress(_bytes, 16 + zlib.MAX_WBITS) run_points_data = json.loads(run_points_data) return run_points_data @@ -80,33 +89,25 @@ def parse_raw_data_to_nametuple( keep_id = run_data["id"].split("_")[1] start_time = run_data["startTime"] - if run_data.get("vendor", {}).get("source", "") == "Keep" and run_data.get( - "rawDataURL" - ): - raw_data_url = run_data.get("rawDataURL") - r = session.get(raw_data_url) - if r.ok: - # string strart with `H4sIAAAAAAAA` --> decode and unzip - run_points_data = decode_runmap_data(r.text) - run_points_data_gpx = run_points_data - if TRANS_GCJ02_TO_WGS84: - run_points_data = [ - list(eviltransform.gcj2wgs(p["latitude"], p["longitude"])) - for p in run_points_data - ] - for i, p in enumerate(run_points_data_gpx): - p["latitude"] = run_points_data[i][0] - p["longitude"] = run_points_data[i][1] - else: - run_points_data = [ - [p["latitude"], p["longitude"]] for p in run_points_data - ] - if with_download_gpx: - if str(keep_id) not in old_gpx_ids: - gpx_data = parse_points_to_gpx(run_points_data_gpx, start_time) - download_keep_gpx(gpx_data, str(keep_id)) + if run_data["geoPoints"]: + run_points_data = decode_runmap_data(run_data["geoPoints"], True) + run_points_data_gpx = run_points_data + if TRANS_GCJ02_TO_WGS84: + run_points_data = [ + list(eviltransform.gcj2wgs(p["latitude"], p["longitude"])) + for p in run_points_data + ] + for i, p in enumerate(run_points_data_gpx): + p["latitude"] = run_points_data[i][0] + p["longitude"] = run_points_data[i][1] else: - print(f"ID {keep_id} retrieved gpx data failed") + run_points_data = [[p["latitude"], p["longitude"]] for p in run_points_data] + if with_download_gpx: + if str(keep_id) not in old_gpx_ids: + gpx_data = parse_points_to_gpx(run_points_data_gpx, start_time) + download_keep_gpx(gpx_data, str(keep_id)) + else: + print(f"ID {keep_id} no gps data") heart_rate = None if run_data["heartRate"]: heart_rate = run_data["heartRate"].get("averageHeartRate", None)