From c40986a7586232d709616c796aa0dec0bbebcf8f Mon Sep 17 00:00:00 2001 From: Jorge Silva Date: Tue, 7 Jul 2026 14:22:05 +0100 Subject: [PATCH] fix(igpsport): use mobile auth API for international login New passport accounts fail on the legacy cookie endpoint. Unify international and China on bearer-token auth and queryMyActivity. Co-authored-by: Cursor --- src/intervalssync/igpsport/core.py | 92 +++++++--------------------- src/intervalssync/igpsport/region.py | 11 ++-- tests/test_igpsport_core.py | 61 ++++++++++-------- 3 files changed, 63 insertions(+), 101 deletions(-) diff --git a/src/intervalssync/igpsport/core.py b/src/intervalssync/igpsport/core.py index 748e1f3..3f57cc3 100644 --- a/src/intervalssync/igpsport/core.py +++ b/src/intervalssync/igpsport/core.py @@ -5,9 +5,8 @@ front-end can render it however it likes. Flow (see CLAUDE.md for the full API notes): - 1. Log in to iGPSPORT; the `loginToken` cookie is URL-decoded and reused as a - Bearer token for the gateway API. - 2. List activities (PascalCase keys: RideId / Title / StartTime). + 1. Log in to iGPSPORT via the mobile auth API; reuse access_token as Bearer. + 2. List activities via queryMyActivity (camelCase rideId / title / startTime). 3. Resolve the FIT download URL: try queryActivityDetail, fall back to getDownloadUrl. 4. Download the .fit file and upload it to intervals.icu (basic auth with the @@ -20,7 +19,6 @@ from datetime import date, datetime, timedelta from pathlib import Path from typing import Callable -from urllib.parse import unquote import requests @@ -38,7 +36,6 @@ from .region import INTERNATIONAL, IgpRegionConfig, resolve_region LOGIN_URL = INTERNATIONAL.login_url -ACTIVITY_LIST_URL = INTERNATIONAL.activity_list_url or "" GATEWAY = INTERNATIONAL.gateway_base # iGPSPORT reports activity start times as "YYYY-MM-DD HH:MM:SS". IGP_TIME_FORMAT = "%Y-%m-%d %H:%M:%S" @@ -127,51 +124,33 @@ def login( ) -> dict[str, str]: """Authenticate and return auth headers carrying the Bearer token.""" cfg = resolve_region(region.name if isinstance(region, IgpRegionConfig) else region) - - if cfg.name == "china": - return _login_china(session, user, password, cfg) - return _login_international(session, user, password, cfg) + try: + return _login_bearer_token(session, cfg.login_url, user, password) + except AuthError as exc: + if cfg.name == "international": + raise AuthError(f"{exc}{_INTL_LOGIN_RETRY_HINT}") from exc + raise _CHINA_REGION_HINT = ( " If your account is registered on app.igpsport.cn, set region to China in Settings." ) - -def _login_international( - session: requests.Session, - user: str, - password: str, - region: IgpRegionConfig, -) -> dict[str, str]: - """International: loginToken cookie URL-decoded as Bearer.""" - resp = session.post(region.login_url, json={"username": user, "password": password}) - if not resp.ok: - raise AuthError(f"iGPSPORT login failed: HTTP {resp.status_code}") - - token = session.cookies.get("loginToken") - if not token: - hint = "" - try: - body = resp.json() - if isinstance(body, dict) and body.get("Code") == 403: - hint = _CHINA_REGION_HINT - except ValueError: - pass - raise AuthError(f"iGPSPORT login did not return a loginToken cookie.{hint}") - - return {"Authorization": f"Bearer {unquote(token)}"} +_INTL_LOGIN_RETRY_HINT = ( + " Check your email and password (international accounts use your email)." + + _CHINA_REGION_HINT +) -def _login_china( +def _login_bearer_token( session: requests.Session, + login_url: str, user: str, password: str, - region: IgpRegionConfig, ) -> dict[str, str]: - """China: access_token from JSON response body.""" + """Mobile API login; returns Bearer headers from access_token JSON field.""" resp = session.post( - region.login_url, + login_url, json={"appId": "igpsport-web", "username": user, "password": password}, ) try: @@ -200,44 +179,15 @@ def list_activities( ) -> list[Activity]: """Return the most recent activities, newest first, capped at max_activities.""" cfg = resolve_region(region.name if isinstance(region, IgpRegionConfig) else region) + return _list_activities_query(session, max_activities, cfg) - if cfg.name == "china": - return _list_activities_china(session, max_activities, cfg) - return _list_activities_international(session, max_activities, cfg) - - -def _list_activities_international( - session: requests.Session, - max_activities: int, - region: IgpRegionConfig, -) -> list[Activity]: - """International ActivityList — PascalCase RideId / Title / StartTime.""" - resp = session.get( - region.activity_list_url, - params={"pageNo": 1, "pageSize": max_activities}, - ) - resp.raise_for_status() - - items = resp.json().get("item", [])[:max_activities] - activities: list[Activity] = [] - for item in items: - ride_id = item["RideId"] - activities.append( - Activity( - ride_id=ride_id, - title=item.get("Title") or f"iGPSPORT {ride_id}", - start_time=item.get("StartTime") or item.get("StartDate") or "unknown date", - ) - ) - return activities - -def _list_activities_china( +def _list_activities_query( session: requests.Session, max_activities: int, region: IgpRegionConfig, ) -> list[Activity]: - """China queryMyActivity — camelCase rideId in data.rows.""" + """queryMyActivity — camelCase rideId in data.rows (China and new intl accounts).""" resp = session.get( region.activity_query_url, params={ @@ -255,7 +205,9 @@ def _list_activities_china( for item in rows[:max_activities]: if not isinstance(item, dict): continue - ride_id = item.get("rideId") or item.get("RideId") + ride_id = item.get("rideId") + if ride_id is None: + ride_id = item.get("RideId") if ride_id is None: continue activities.append( diff --git a/src/intervalssync/igpsport/region.py b/src/intervalssync/igpsport/region.py index 8d50e6e..680fc1b 100644 --- a/src/intervalssync/igpsport/region.py +++ b/src/intervalssync/igpsport/region.py @@ -12,8 +12,7 @@ class IgpRegionConfig: name: IgpRegionName login_url: str - activity_list_url: str | None - activity_query_url: str | None + activity_query_url: str gateway_base: str mobile_api_base: str accept_language: str @@ -37,9 +36,10 @@ def update_interval_url(self) -> str: INTERNATIONAL = IgpRegionConfig( name="international", - login_url="https://i.igpsport.com/Auth/Login", - activity_list_url="https://i.igpsport.com/Activity/ActivityList", - activity_query_url=None, + login_url="https://prod.en.igpsport.com/service/auth/account/login", + activity_query_url=( + "https://prod.en.igpsport.com/service/web-gateway/web-analyze/activity/queryMyActivity" + ), gateway_base="https://prod.en.igpsport.com/service/web-gateway/web-analyze/activity", mobile_api_base="https://prod.en.igpsport.com/service/mobile/api", accept_language="en", @@ -48,7 +48,6 @@ def update_interval_url(self) -> str: CHINA = IgpRegionConfig( name="china", login_url="https://prod.zh.igpsport.com/service/auth/account/login", - activity_list_url=None, activity_query_url=( "https://prod.zh.igpsport.com/service/web-gateway/web-analyze/activity/queryMyActivity" ), diff --git a/tests/test_igpsport_core.py b/tests/test_igpsport_core.py index 05cbe17..aed1b1c 100644 --- a/tests/test_igpsport_core.py +++ b/tests/test_igpsport_core.py @@ -120,38 +120,37 @@ def test_set_activity_type(monkeypatch, status, expected): assert core.set_activity_type("i1", "MountainBikeRide", "key") is expected -def test_login_builds_bearer_from_cookie(monkeypatch): - class FakeCookies: - def get(self, name): - return "abc%20def" if name == "loginToken" else None - +def test_login_builds_bearer_from_access_token(): class FakeSession: - cookies = FakeCookies() + headers: dict[str, str] = {} def post(self, *a, **k): - return FakeResponse(status=200) + return FakeResponse( + status=200, + json_data={"data": {"access_token": "intl.jwt.token"}}, + ) - headers = core.login(FakeSession(), "user", "pass") - assert headers == {"Authorization": "Bearer abc def"} # %20 -> space + def update(self, headers): + self.headers.update(headers) + session = FakeSession() + headers = core.login(session, "user+tag@example.com", "igp&1%!") + assert headers == {"Authorization": "Bearer intl.jwt.token"} -def test_login_raises_without_cookie(monkeypatch): - class FakeSession: - cookies = type("C", (), {"get": lambda self, n: None})() +def test_login_raises_when_access_token_missing(): + class FakeSession: def post(self, *a, **k): - return FakeResponse(status=200, json_data={"Code": 403}) + return FakeResponse(status=200, json_data={"message": "Password error"}) - with pytest.raises(core.AuthError, match="loginToken cookie"): - core.login(FakeSession(), "user", "pass") + with pytest.raises(core.AuthError, match="Password error"): + core.login(FakeSession(), "user", "pass", region="china") -def test_login_hints_china_region_when_intl_auth_fails(): +def test_login_international_adds_retry_hint_on_failure(): class FakeSession: - cookies = type("C", (), {"get": lambda self, n: None})() - def post(self, *a, **k): - return FakeResponse(status=200, json_data={"Code": 403}) + return FakeResponse(status=200, json_data={"message": "Password error"}) with pytest.raises(core.AuthError, match="app.igpsport.cn"): core.login(FakeSession(), "user", "pass") @@ -234,12 +233,24 @@ class FakeSession: def get(self, url, params=None): captured["url"] = url captured["params"] = params - items = [{"RideId": i, "Title": f"Ride {i}", "StartTime": "2026-05-28 19:20:42"} - for i in range(50)] - return FakeResponse(json_data={"item": items, "total": 59}) + rows = [ + { + "rideId": i, + "title": f"Ride {i}", + "startTime": "2026-05-28 19:20:42", + } + for i in range(50) + ] + return FakeResponse(json_data={"data": {"rows": rows}}) acts = core.list_activities(FakeSession(), 50) - assert captured["params"] == {"pageNo": 1, "pageSize": 50} + assert "queryMyActivity" in captured["url"] + assert captured["params"] == { + "pageNo": "1", + "pageSize": "50", + "sort": "1", + "reqType": "0", + } assert len(acts) == 50 assert acts[0].ride_id == 0 @@ -247,8 +258,8 @@ def get(self, url, params=None): def test_list_activities_caps_when_server_returns_extra(): class FakeSession: def get(self, url, params=None): - items = [{"RideId": i, "Title": "", "StartTime": ""} for i in range(20)] - return FakeResponse(json_data={"item": items, "total": 20}) + rows = [{"rideId": i, "title": "", "startTime": ""} for i in range(20)] + return FakeResponse(json_data={"data": {"rows": rows}}) acts = core.list_activities(FakeSession(), 5) assert len(acts) == 5