strava-utils/get_token.py at master · GigiLoria5/strava-utils · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
#!/usr/bin/env python3
"""One-shot Strava OAuth helper: writes credentials to .env.

Your app's "Authorization Callback Domain" must be exactly `localhost`
(https://www.strava.com/settings/api).

Usage:
  python3 get_token.py
"""

from __future__ import annotations

import getpass
import os
import sys
import urllib.parse
import webbrowser
from http.server import BaseHTTPRequestHandler, HTTPServer

from strava import ENV_FILE, exchange_tokens, load_env

PORT = 8080
SCOPE = "activity:read_all,activity:write"  # write is required by import_gpx.py uploads
DONE_HTML = b"<html><body style='font:16px sans-serif'>Done. Back to the terminal.</body></html>"


class Callback(BaseHTTPRequestHandler):
    params: dict[str, list[str]] | None = None

    def do_GET(self) -> None:
        query = urllib.parse.urlparse(self.path).query
        if "code" in query or "error" in query:
            Callback.params = urllib.parse.parse_qs(query)
        self.send_response(200)
        self.send_header("Content-Type", "text/html")
        self.end_headers()
        self.wfile.write(DONE_HTML)

    def log_message(self, *args: object) -> None:
        pass


def wait_for_callback() -> dict[str, list[str]]:
    server = HTTPServer(("127.0.0.1", PORT), Callback)
    while Callback.params is None:
        server.handle_request()
    server.server_close()
    return Callback.params


def authorize_url(client_id: str) -> str:
    return "https://www.strava.com/oauth/authorize?" + urllib.parse.urlencode(
        {
            "client_id": client_id,
            "redirect_uri": f"http://localhost:{PORT}",
            "response_type": "code",
            "approval_prompt": "force",
            "scope": SCOPE,
        }
    )


def write_env(client_id: str, client_secret: str, refresh_token: str) -> None:
    ENV_FILE.write_text(
        f"STRAVA_CLIENT_ID={client_id}\n"
        f"STRAVA_CLIENT_SECRET={client_secret}\n"
        f"STRAVA_REFRESH_TOKEN={refresh_token}\n"
    )
    ENV_FILE.chmod(0o600)


def main() -> None:
    if ENV_FILE.exists() and input(f"{ENV_FILE} exists. Overwrite? [y/N] ").strip().lower() != "y":
        sys.exit("Aborted.")

    load_env()
    client_id = os.environ.get("STRAVA_CLIENT_ID") or input("Client ID: ").strip()
    client_secret = os.environ.get("STRAVA_CLIENT_SECRET") or getpass.getpass("Client Secret: ").strip()
    if not (client_id and client_secret):
        sys.exit("Client ID and Client Secret are both required.")

    url = authorize_url(client_id)
    print(f"\nOpening browser. If nothing happens, paste this URL yourself:\n{url}\n")
    webbrowser.open(url)
    print(f"Waiting for the redirect on http://localhost:{PORT} ...")

    params = wait_for_callback()
    if "error" in params:
        sys.exit(f"Strava returned an error: {params['error'][0]} (did you click Authorize?)")
    granted = params.get("scope", [""])[0].split(",")
    if missing := [s for s in SCOPE.split(",") if s not in granted]:
        print(f"Warning: {', '.join(missing)} not granted (got '{','.join(granted)}'). "
              "Private activities and/or uploads will fail.")

    tokens = exchange_tokens(
        client_id, client_secret, grant_type="authorization_code", code=params["code"][0]
    )

    write_env(client_id, client_secret, tokens["refresh_token"])
    print(f"\nWrote {ENV_FILE}. Scripts read it automatically.")


if __name__ == "__main__":
    main()