1#!/usr/bin/env python3
2from __future__ import annotations
3 
4import argparse
5import base64
6import csv
7import datetime as dt
8import http.server
9import json
10import math
11import os
12import secrets as pysecrets
13import statistics
14import sys
15import threading
16import time
17import urllib.error
18import urllib.parse
19import urllib.request
20import webbrowser
21from pathlib import Path
22from typing import Any
23 
24 
25SCRIPT_DIR = Path(__file__).resolve().parent
26SKILL_DIR = SCRIPT_DIR.parent
27STATE_DIR = Path(os.environ.get("OURA_ANALYZER_HOME", "~/.local/share/oura-analyzer")).expanduser()
28DEFAULT_TOKEN_FILE = STATE_DIR / "secrets" / "oura_tokens.json"
29RAW_DIR = STATE_DIR / "data" / "raw"
30IMPORTED_DIR = STATE_DIR / "data" / "imported"
31REPORTS_DIR = STATE_DIR / "reports"
32API_BASE = "https://api.ouraring.com"
33AUTHORIZE_URL = "https://cloud.ouraring.com/oauth/authorize"
34TOKEN_URL = f"{API_BASE}/oauth/token"
35 
36 
37def load_dotenv() -> None:
38    env_path = STATE_DIR / ".env"
39    if not env_path.exists():
40        return
41    for line in env_path.read_text(encoding="utf-8").splitlines():
42        line = line.strip()
43        if not line or line.startswith("#") or "=" not in line:
44            continue
45        key, value = line.split("=", 1)
46        os.environ.setdefault(key.strip(), value.strip())
47 
48 
49def ensure_dirs() -> None:
50    for path in (RAW_DIR, IMPORTED_DIR, REPORTS_DIR, DEFAULT_TOKEN_FILE.parent):
51        path.mkdir(parents=True, exist_ok=True)
52 
53 
54def iso_now() -> str:
55    return dt.datetime.now(dt.timezone.utc).replace(microsecond=0).isoformat()
56 
57 
58def date_today() -> dt.date:
59    return dt.datetime.now(dt.timezone.utc).date()
60 
61 
62def parse_date(value: str) -> dt.date:
63    return dt.date.fromisoformat(value)
64 
65 
66def read_json(path: Path) -> Any:
67    return json.loads(path.read_text(encoding="utf-8"))
68 
69 
70def write_json(path: Path, data: Any) -> None:
71    path.parent.mkdir(parents=True, exist_ok=True)
72    path.write_text(json.dumps(data, indent=2, ensure_ascii=True) + "\n", encoding="utf-8")
73 
74 
75def resolve_token_file(path_arg: str | None) -> Path:
76    token_file = path_arg or os.environ.get("OURA_TOKEN_FILE")
77    if not token_file:
78        return DEFAULT_TOKEN_FILE
79    path = Path(token_file)
80    return path if path.is_absolute() else STATE_DIR / path
81 
82 
83def paths_command(_args: argparse.Namespace) -> None:
84    paths = {
85        "state_dir": STATE_DIR,
86        "env_file": STATE_DIR / ".env",
87        "token_file": DEFAULT_TOKEN_FILE,
88        "latest_api_pull": RAW_DIR / "latest_api_pull.json",
89        "latest_export": IMPORTED_DIR / "latest_export.json",
90        "latest_report": REPORTS_DIR / "latest_analysis.md",
91    }
92    for key, value in paths.items():
93        print(f"{key}={value}")
94 
95 
96def required_env(name: str) -> str:
97    value = os.environ.get(name)
98    if not value:
99        raise SystemExit(f"Missing {name}. Put it in .env or pass the equivalent CLI option.")
100    return value
101 
102 
103def http_json(
104    method: str,
105    url: str,
106    *,
107    headers: dict[str, str] | None = None,
108    form: dict[str, str] | None = None,
109    timeout: int = 30,
110) -> Any:
111    payload = None
112    request_headers = {"Accept": "application/json"}
113    if headers:
114        request_headers.update(headers)
115    if form is not None:
116        payload = urllib.parse.urlencode(form).encode("utf-8")
117        request_headers["Content-Type"] = "application/x-www-form-urlencoded"
118    request = urllib.request.Request(url, data=payload, headers=request_headers, method=method)
119    try:
120        with urllib.request.urlopen(request, timeout=timeout) as response:
121            return json.loads(response.read().decode("utf-8"))
122    except urllib.error.HTTPError as exc:
123        body = exc.read().decode("utf-8", errors="replace")
124        try:
125            parsed = json.loads(body)
126        except json.JSONDecodeError:
127            parsed = {"status": exc.code, "detail": body}
128        raise RuntimeError(f"HTTP {exc.code} for {url}: {json.dumps(parsed)}") from exc
129 
130 
131def build_basic_auth_header(client_id: str, client_secret: str) -> str:
132    raw = f"{client_id}:{client_secret}".encode("utf-8")
133    return "Basic " + base64.b64encode(raw).decode("ascii")
134 
135 
136class OAuthCallbackHandler(http.server.BaseHTTPRequestHandler):
137    server_version = "OuraAnalyzer/1.0"
138 
139    def do_GET(self) -> None:  # noqa: N802
140        parsed = urllib.parse.urlparse(self.path)
141        query = urllib.parse.parse_qs(parsed.query)
142        self.server.result = query  # type: ignore[attr-defined]
143        body = (
144            "<html><body><h1>Oura authorization received</h1>"
145            "<p>You can close this tab and return to the terminal.</p></body></html>"
146        ).encode("utf-8")
147        self.send_response(200)
148        self.send_header("Content-Type", "text/html; charset=utf-8")
149        self.send_header("Content-Length", str(len(body)))
150        self.end_headers()
151        self.wfile.write(body)
152 
153    def log_message(self, fmt: str, *args: Any) -> None:
154        return
155 
156 
157def wait_for_oauth_callback(redirect_uri: str, timeout_seconds: int = 180) -> dict[str, list[str]]:
158    parsed = urllib.parse.urlparse(redirect_uri)
159    if parsed.scheme != "http" or parsed.hostname not in {"127.0.0.1", "localhost"} or not parsed.port:
160        raise SystemExit("Redirect URI must be a local http://localhost:<port>/... or http://127.0.0.1:<port>/... URI for the built-in callback server.")
161 
162    server = http.server.ThreadingHTTPServer((parsed.hostname, parsed.port), OAuthCallbackHandler)
163    server.result = None  # type: ignore[attr-defined]
164    thread = threading.Thread(target=server.handle_request, daemon=True)
165    thread.start()
166 
167    deadline = time.time() + timeout_seconds
168    while time.time() < deadline:
169        result = getattr(server, "result", None)
170        if result is not None:
171            server.server_close()
172            return result
173        time.sleep(0.2)
174 
175    server.server_close()
176    raise SystemExit("Timed out waiting for the Oura OAuth callback.")
177 
178 
179def exchange_code_for_tokens(client_id: str, client_secret: str, code: str, redirect_uri: str) -> dict[str, Any]:
180    return http_json(
181        "POST",
182        TOKEN_URL,
183        headers={"Authorization": build_basic_auth_header(client_id, client_secret)},
184        form={
185            "grant_type": "authorization_code",
186            "code": code,
187            "redirect_uri": redirect_uri,
188        },
189    )
190 
191 
192def refresh_access_token(
193    client_id: str,
194    client_secret: str,
195    refresh_token: str,
196) -> dict[str, Any]:
197    return http_json(
198        "POST",
199        TOKEN_URL,
200        headers={"Authorization": build_basic_auth_header(client_id, client_secret)},
201        form={
202            "grant_type": "refresh_token",
203            "refresh_token": refresh_token,
204        },
205    )
206 
207 
208def token_expired(token_data: dict[str, Any], skew_seconds: int = 60) -> bool:
209    obtained_at = token_data.get("obtained_at")
210    expires_in = token_data.get("expires_in")
211    if not obtained_at or not expires_in:
212        return False
213    obtained = dt.datetime.fromisoformat(obtained_at)
214    expiry = obtained + dt.timedelta(seconds=int(expires_in))
215    return dt.datetime.now(dt.timezone.utc) >= (expiry - dt.timedelta(seconds=skew_seconds))
216 
217 
218def authorize_command(args: argparse.Namespace) -> None:
219    client_id = args.client_id or required_env("OURA_CLIENT_ID")
220    client_secret = args.client_secret or required_env("OURA_CLIENT_SECRET")
221    redirect_uri = args.redirect_uri or os.environ.get("OURA_REDIRECT_URI") or "http://localhost:8765/callback"
222    token_file = resolve_token_file(args.token_file)
223    state = pysecrets.token_urlsafe(24)
224    scopes = args.scopes or "email personal daily heartrate workout tag session spo2Daily"
225 
226    query = urllib.parse.urlencode(
227        {
228            "response_type": "code",
229            "client_id": client_id,
230            "redirect_uri": redirect_uri,
231            "scope": scopes,
232            "state": state,
233        }
234    )
235    auth_url = f"{AUTHORIZE_URL}?{query}"
236    print("Open this URL to authorize Oura:")
237    print(auth_url)
238    if not args.no_browser:
239        webbrowser.open(auth_url)
240 
241    callback = wait_for_oauth_callback(redirect_uri, timeout_seconds=args.timeout)
242    if callback.get("state", [None])[0] != state:
243        raise SystemExit("OAuth state mismatch.")
244    if "error" in callback:
245        raise SystemExit(f"Oura returned an authorization error: {callback['error'][0]}")
246 
247    code = callback.get("code", [None])[0]
248    if not code:
249        raise SystemExit("No authorization code returned by Oura.")
250 
251    tokens = exchange_code_for_tokens(client_id, client_secret, code, redirect_uri)
252    tokens["obtained_at"] = iso_now()
253    tokens["redirect_uri"] = redirect_uri
254    tokens["scopes_requested"] = scopes
255    write_json(token_file, tokens)
256    print(f"Saved tokens to {token_file}")
257 
258 
259def load_token_data(token_file: Path) -> dict[str, Any]:
260    if not token_file.exists():
261        raise SystemExit(f"Token file not found: {token_file}. Run authorize first.")
262    return read_json(token_file)
263 
264 
265def ensure_live_access_token(args: argparse.Namespace, token_file: Path) -> tuple[str, dict[str, Any]]:
266    token_data = load_token_data(token_file)
267    if token_expired(token_data):
268        client_id = args.client_id or required_env("OURA_CLIENT_ID")
269        client_secret = args.client_secret or required_env("OURA_CLIENT_SECRET")
270        refresh_token_value = token_data.get("refresh_token")
271        if not refresh_token_value:
272            raise SystemExit("Access token expired and no refresh token is available.")
273        refreshed = refresh_access_token(client_id, client_secret, refresh_token_value)
274        refreshed["obtained_at"] = iso_now()
275        refreshed["redirect_uri"] = token_data.get("redirect_uri")
276        refreshed["scopes_requested"] = token_data.get("scopes_requested")
277        write_json(token_file, refreshed)
278        token_data = refreshed
279    access_token = token_data.get("access_token")
280    if not access_token:
281        raise SystemExit("No access_token in token file.")
282    return access_token, token_data
283 
284 
285def api_get(token: str, path: str, params: dict[str, str] | None = None) -> Any:
286    url = f"{API_BASE}{path}"
287    if params:
288        url += "?" + urllib.parse.urlencode(params)
289    return http_json("GET", url, headers={"Authorization": f"Bearer {token}"})
290 
291 
292def fetch_paginated_collection(token: str, path: str, params: dict[str, str]) -> list[dict[str, Any]]:
293    results: list[dict[str, Any]] = []
294    next_token = None
295    while True:
296        page_params = dict(params)
297        if next_token:
298            page_params["next_token"] = next_token
299        payload = api_get(token, path, page_params)
300        batch = payload.get("data", [])
301        if not isinstance(batch, list):
302            raise RuntimeError(f"Unexpected payload for {path}: {payload}")
303        results.extend(batch)
304        next_token = payload.get("next_token")
305        if not next_token:
306            return results
307 
308 
309def pull_command(args: argparse.Namespace) -> None:
310    token_file = resolve_token_file(args.token_file)
311    access_token, _token_data = ensure_live_access_token(args, token_file)
312 
313    end_date = parse_date(args.end_date) if args.end_date else date_today()
314    start_date = parse_date(args.start_date) if args.start_date else (end_date - dt.timedelta(days=args.days - 1))
315    if start_date > end_date:
316        raise SystemExit("start_date must be on or before end_date.")
317 
318    base_params = {"start_date": start_date.isoformat(), "end_date": end_date.isoformat()}
319    pulled_at = iso_now()
320    dataset = {
321        "source": "oura_api_v2",
322        "pulled_at": pulled_at,
323        "date_range": base_params,
324        "endpoints": {},
325    }
326 
327    dataset["endpoints"]["personal_info"] = api_get(access_token, "/v2/usercollection/personal_info")
328    for name in ("daily_activity", "daily_sleep", "daily_readiness", "sleep"):
329        path = f"/v2/usercollection/{name}"
330        dataset["endpoints"][name] = fetch_paginated_collection(access_token, path, base_params)
331 
332    timestamp = dt.datetime.now().strftime("%Y%m%d-%H%M%S")
333    output_path = RAW_DIR / f"api-pull-{timestamp}.json"
334    latest_path = RAW_DIR / "latest_api_pull.json"
335    write_json(output_path, dataset)
336    write_json(latest_path, dataset)
337    print(f"Saved API pull to {output_path}")
338    print(f"Updated {latest_path}")
339 
340 
341def normalize_name(value: str) -> str:
342    lowered = value.lower()
343    chars = []
344    for ch in lowered:
345        chars.append(ch if ch.isalnum() else "_")
346    return "_".join(filter(None, "".join(chars).split("_")))
347 
348 
349def infer_dataset_key(path: Path) -> str:
350    return normalize_name(path.stem)
351 
352 
353def import_export_command(args: argparse.Namespace) -> None:
354    source_path = Path(args.path).expanduser().resolve()
355    if not source_path.exists():
356        raise SystemExit(f"Path does not exist: {source_path}")
357 
358    csv_paths = [source_path] if source_path.is_file() else sorted(source_path.rglob("*.csv"))
359    if not csv_paths:
360        raise SystemExit("No CSV files found in the provided export path.")
361 
362    imported: dict[str, Any] = {
363        "source": "oura_export_csv",
364        "imported_at": iso_now(),
365        "origin_path": str(source_path),
366        "files": {},
367    }
368 
369    for csv_path in csv_paths:
370        key = infer_dataset_key(csv_path)
371        with csv_path.open("r", encoding="utf-8-sig", newline="") as handle:
372            reader = csv.DictReader(handle)
373            rows = list(reader)
374        imported["files"][key] = {
375            "file_name": csv_path.name,
376            "path": str(csv_path),
377            "rows": rows,
378        }
379 
380    timestamp = dt.datetime.now().strftime("%Y%m%d-%H%M%S")
381    output_path = IMPORTED_DIR / f"export-import-{timestamp}.json"
382    latest_path = IMPORTED_DIR / "latest_export.json"
383    write_json(output_path, imported)
384    write_json(latest_path, imported)
385    print(f"Imported {len(csv_paths)} CSV files into {output_path}")
386    print(f"Updated {latest_path}")
387 
388 
389def safe_float(value: Any) -> float | None:
390    if value is None or value == "":
391        return None
392    if isinstance(value, (int, float)):
393        return float(value)
394    try:
395        return float(str(value).strip())
396    except ValueError:
397        return None
398 
399 
400def seconds_to_hours(seconds: float | None) -> float | None:
401    if seconds is None:
402        return None
403    return seconds / 3600.0
404 
405 
406def pick_first(mapping: dict[str, Any], keys: tuple[str, ...]) -> Any:
407    for key in keys:
408        if key in mapping and mapping[key] not in (None, ""):
409            return mapping[key]
410    return None
411 
412 
413def mean_or_none(values: list[float]) -> float | None:
414    return statistics.fmean(values) if values else None
415 
416 
417def median_or_none(values: list[float]) -> float | None:
418    return statistics.median(values) if values else None
419 
420 
421def correlation(xs: list[float], ys: list[float]) -> float | None:
422    if len(xs) < 3 or len(xs) != len(ys):
423        return None
424    mean_x = statistics.fmean(xs)
425    mean_y = statistics.fmean(ys)
426    num = sum((x - mean_x) * (y - mean_y) for x, y in zip(xs, ys))
427    den_x = math.sqrt(sum((x - mean_x) ** 2 for x in xs))
428    den_y = math.sqrt(sum((y - mean_y) ** 2 for y in ys))
429    if den_x == 0 or den_y == 0:
430        return None
431    return num / (den_x * den_y)
432 
433 
434def ymd(value: Any) -> str | None:
435    if value is None:
436        return None
437    text = str(value)
438    return text[:10] if len(text) >= 10 else None
439 
440 
441def prepare_api_days(dataset: dict[str, Any]) -> dict[str, dict[str, Any]]:
442    endpoints = dataset.get("endpoints", {})
443    by_day: dict[str, dict[str, Any]] = {}
444 
445    def touch(day: str) -> dict[str, Any]:
446        return by_day.setdefault(day, {"day": day})
447 
448    for row in endpoints.get("daily_readiness", []):
449        day = row.get("day")
450        if not day:
451            continue
452        target = touch(day)
453        target["readiness_score"] = row.get("score")
454        for key, value in row.get("contributors", {}).items():
455            target[f"readiness_{key}"] = value
456 
457    for row in endpoints.get("daily_sleep", []):
458