You can now close this window.
" ) logger.info("Authorization code received successfully") else: error = params.get("error", ["Unknown error"])[0] OAuth2CallbackHandler.error_message = error self.send_response(400) self.send_header("Content-type", "text/html") self.end_headers() self.wfile.write( f"Error: {error}
".encode() ) logger.error(f"Authorization error: {error}") def log_message(self, format, *args): """Suppress default logging.""" pass class iHealthOAuth2Collector: """iHealth OAuth2 collector for weight and blood pressure data.""" def __init__(self): """Initialize the collector.""" self.client_id = IHEALTH_CLIENT_ID self.client_secret = IHEALTH_CLIENT_SECRET self.tokens = {} self.user_id = None self._load_tokens() # Get paths relative to script location script_dir = Path(__file__).parent.parent # kitzu directory self.kitzu_root = script_dir self.profile_path = self.kitzu_root / "data" / "unified" / "profile.json" def _load_tokens(self) -> None: """Load tokens from config file.""" if TOKENS_FILE.exists(): try: with open(TOKENS_FILE, "r") as f: self.tokens = json.load(f) self.user_id = self.tokens.get("user_id") logger.info(f"Tokens loaded from {TOKENS_FILE}") except Exception as e: logger.warning(f"Failed to load tokens: {e}") self.tokens = {} else: logger.info(f"No tokens file found at {TOKENS_FILE}") def _save_tokens(self) -> None: """Save tokens to config file.""" TOKENS_FILE.parent.mkdir(parents=True, exist_ok=True) with open(TOKENS_FILE, "w") as f: json.dump(self.tokens, f, indent=2) # Set restrictive permissions on tokens file TOKENS_FILE.chmod(0o600) logger.info(f"Tokens saved to {TOKENS_FILE}") def authorize(self) -> bool: """ Perform OAuth2 authorization flow. Opens browser to iHealth authorization page. After user approves, iHealth redirects to the HTTPS callback on moodhoney.com which displays the authorization code for manual entry. Returns: True if authorization successful, False otherwise """ logger.info("Starting OAuth2 authorization flow...") # Build authorization URL auth_params = { "client_id": self.client_id, "response_type": "code", "redirect_uri": REDIRECT_URI, "scope": " ".join(SCOPES), "state": "kitzu_auth", } auth_url = f"{IHEALTH_AUTH_URL}?{'&'.join(f'{k}={v}' for k, v in auth_params.items())}" logger.info(f"Opening browser to: {auth_url}") # Open browser for user to authorize webbrowser.open(auth_url) print("\n" + "=" * 60) print(" KITZU — iHealth Authorization") print("=" * 60) print("\n 1. A browser window has opened to iHealth.") print(" 2. Log in and approve access for Kitzu Wellness.") print(" 3. You'll be redirected to moodhoney.com with a code.") print(" 4. Copy the code and paste it below.\n") auth_code = input(" Paste authorization code here: ").strip() if not auth_code: logger.error("No authorization code entered") return False logger.info(f"Authorization code received: {auth_code[:20]}...") # Exchange authorization code for tokens return self._exchange_code_for_tokens(auth_code) def _exchange_code_for_tokens(self, auth_code: str) -> bool: """ Exchange authorization code for access and refresh tokens. Args: auth_code: Authorization code from OAuth callback Returns: True if successful, False otherwise """ logger.info("Exchanging authorization code for tokens...") token_params = { "client_id": self.client_id, "client_secret": self.client_secret, "grant_type": "authorization_code", "code": auth_code, "redirect_uri": REDIRECT_URI, } try: response = requests.post(IHEALTH_TOKEN_URL, params=token_params, timeout=10) response.raise_for_status() data = response.json() # Store tokens self.tokens = { "access_token": data.get("access_token"), "refresh_token": data.get("refresh_token"), "token_type": data.get("token_type", "Bearer"), "expires_in": data.get("expires_in", 3600), "user_id": data.get("UserID"), "timestamp": datetime.now().isoformat(), } self.user_id = self.tokens.get("user_id") self._save_tokens() logger.info(f"Tokens obtained successfully. User ID: {self.user_id}") return True except requests.RequestException as e: logger.error(f"Failed to exchange code for tokens: {e}") return False def _is_token_expired(self) -> bool: """Check if access token is expired.""" if "timestamp" not in self.tokens or "expires_in" not in self.tokens: return True issued_at = datetime.fromisoformat(self.tokens["timestamp"]) expires_at = issued_at + timedelta(seconds=self.tokens["expires_in"]) is_expired = datetime.now() >= expires_at if is_expired: logger.info("Access token has expired") return is_expired def refresh_access_token(self) -> bool: """ Refresh the access token using the refresh token. Returns: True if successful, False otherwise """ if "refresh_token" not in self.tokens: logger.error("No refresh token available") return False logger.info("Refreshing access token...") token_params = { "client_id": self.client_id, "client_secret": self.client_secret, "grant_type": "refresh_token", "refresh_token": self.tokens["refresh_token"], } try: response = requests.post(IHEALTH_TOKEN_URL, params=token_params, timeout=10) response.raise_for_status() data = response.json() # Update tokens self.tokens.update({ "access_token": data.get("access_token"), "refresh_token": data.get("refresh_token", self.tokens.get("refresh_token")), "expires_in": data.get("expires_in", 3600), "timestamp": datetime.now().isoformat(), }) self._save_tokens() logger.info("Access token refreshed successfully") return True except requests.RequestException as e: logger.error(f"Failed to refresh access token: {e}") return False def _ensure_valid_token(self) -> bool: """Ensure we have a valid access token.""" if not self.tokens.get("access_token"): logger.error("No access token available. Run --authorize first.") return False if self._is_token_expired(): return self.refresh_access_token() return True def collect_weight_data(self) -> List[Dict[str, Any]]: """ Collect weight data from iHealth API. Returns: List of weight readings """ if not self._ensure_valid_token(): return [] if not self.user_id: logger.error("No user ID available") return [] logger.info("Collecting weight data...") # Build request parameters start_time = int((datetime.now() - timedelta(days=30)).timestamp()) params = { "client_id": self.client_id, "client_secret": self.client_secret, "access_token": self.tokens.get("access_token"), "sc": start_time, "sv": "OpenApiWeight", "locale": LOCALE, "page_index": 1, } url = f"{IHEALTH_API_BASE}/{self.user_id}/weight.json" try: response = requests.get(url, params=params, timeout=10, verify=False) response.raise_for_status() data = response.json() # Parse weight data weight_list = data.get("WeightValue", []) logger.info(f"Retrieved {len(weight_list)} weight readings") return weight_list except requests.RequestException as e: logger.error(f"Failed to collect weight data: {e}") return [] def collect_bp_data(self) -> List[Dict[str, Any]]: """ Collect blood pressure data from iHealth API. Returns: List of blood pressure readings """ if not self._ensure_valid_token(): return [] if not self.user_id: logger.error("No user ID available") return [] logger.info("Collecting blood pressure data...") # Build request parameters start_time = int((datetime.now() - timedelta(days=30)).timestamp()) params = { "client_id": self.client_id, "client_secret": self.client_secret, "access_token": self.tokens.get("access_token"), "sc": start_time, "sv": "OpenApiBP", "locale": LOCALE, "page_index": 1, } url = f"{IHEALTH_API_BASE}/{self.user_id}/bp.json" try: response = requests.get(url, params=params, timeout=10, verify=False) response.raise_for_status() data = response.json() # Parse BP data bp_list = data.get("BPValue", []) logger.info(f"Retrieved {len(bp_list)} blood pressure readings") return bp_list except requests.RequestException as e: logger.error(f"Failed to collect blood pressure data: {e}") return [] def _load_profile(self) -> Dict[str, Any]: """Load the Kitzu profile.""" if self.profile_path.exists(): try: with open(self.profile_path, "r") as f: return json.load(f) except Exception as e: logger.warning(f"Failed to load profile: {e}") # Return default profile structure return { "vitals": { "weight": None, "blood_pressure": None, }, "weight_history": [], "bp_history": [], "weight_trend_7d": None, "bp_stats_30d": None, "last_sync": None, } def _save_profile(self, profile: Dict[str, Any]) -> None: """Save the Kitzu profile.""" self.profile_path.parent.mkdir(parents=True, exist_ok=True) with open(self.profile_path, "w") as f: json.dump(profile, f, indent=2) logger.info(f"Profile saved to {self.profile_path}") def _parse_weight_reading(self, reading: Dict[str, Any]) -> Optional[Dict[str, Any]]: """ Parse a weight reading from iHealth API response. Args: reading: Raw weight reading from API Returns: Parsed weight reading or None if parsing fails """ try: # iHealth returns weight in kg or lbs weight_value = reading.get("Weight") unit = reading.get("WeightUnit", "kg") timestamp = reading.get("MeasureTime", int(time.time())) if isinstance(timestamp, str): timestamp = int(timestamp) measure_date = datetime.fromtimestamp(timestamp).isoformat() return { "date": measure_date, "weight": weight_value, "unit": unit, "timestamp": timestamp, "date_key": datetime.fromtimestamp(timestamp).strftime("%Y-%m-%d %H:%M"), } except Exception as e: logger.warning(f"Failed to parse weight reading: {e}") return None def _parse_bp_reading(self, reading: Dict[str, Any]) -> Optional[Dict[str, Any]]: """ Parse a blood pressure reading from iHealth API response. Args: reading: Raw BP reading from API Returns: Parsed BP reading or None if parsing fails """ try: systolic = reading.get("SystolicValue") diastolic = reading.get("DiastolicValue") pulse = reading.get("PulseValue") timestamp = reading.get("MeasureTime", int(time.time())) if isinstance(timestamp, str): timestamp = int(timestamp) measure_date = datetime.fromtimestamp(timestamp).isoformat() return { "date": measure_date, "systolic": systolic, "diastolic": diastolic, "pulse": pulse, "timestamp": timestamp, "date_key": datetime.fromtimestamp(timestamp).strftime("%Y-%m-%d %H:%M"), } except Exception as e: logger.warning(f"Failed to parse BP reading: {e}") return None def _deduplicate_readings( self, existing: List[Dict[str, Any]], new: List[Dict[str, Any]], key_field: str = "date_key" ) -> List[Dict[str, Any]]: """ Deduplicate readings by date_key. Args: existing: Existing readings new: New readings to add key_field: Field to use as deduplication key Returns: Deduplicated list with new readings added """ existing_keys = {r.get(key_field): r for r in existing} for reading in new: key = reading.get(key_field) if key: existing_keys[key] = reading return list(existing_keys.values()) def _calculate_weight_trend(self, readings: List[Dict[str, Any]]) -> Optional[float]: """ Calculate weight trend over 7 days. Args: readings: Weight readings Returns: Trend value (positive = weight gain, negative = weight loss) or None """ if len(readings) < 2: return None try: # Sort by timestamp sorted_readings = sorted(readings, key=lambda x: x.get("timestamp", 0)) # Get readings from last 7 days cutoff = datetime.now() - timedelta(days=7) recent = [r for r in sorted_readings if datetime.fromisoformat(r.get("date", "")).replace(tzinfo=None) >= cutoff] if len(recent) < 2: return None # Calculate trend first_weight = recent[0].get("weight") last_weight = recent[-1].get("weight") if first_weight and last_weight: return last_weight - first_weight except Exception as e: logger.warning(f"Failed to calculate weight trend: {e}") return None def _calculate_bp_stats(self, readings: List[Dict[str, Any]]) -> Optional[Dict[str, Any]]: """ Calculate BP statistics over 30 days. Args: readings: BP readings Returns: BP statistics or None """ if not readings: return None try: # Sort by timestamp sorted_readings = sorted(readings, key=lambda x: x.get("timestamp", 0)) # Get readings from last 30 days cutoff = datetime.now() - timedelta(days=30) recent = [r for r in sorted_readings if datetime.fromisoformat(r.get("date", "")).replace(tzinfo=None) >= cutoff] if not recent: return None systolics = [r.get("systolic") for r in recent if r.get("systolic")] diastolics = [r.get("diastolic") for r in recent if r.get("diastolic")] if not systolics or not diastolics: return None return { "avg_systolic": sum(systolics) / len(systolics), "avg_diastolic": sum(diastolics) / len(diastolics), "min_systolic": min(systolics), "max_systolic": max(systolics), "min_diastolic": min(diastolics), "max_diastolic": max(diastolics), "reading_count": len(recent), } except Exception as e: logger.warning(f"Failed to calculate BP stats: {e}") return None def collect(self) -> bool: """ Collect weight and blood pressure data and update profile. Returns: True if successful, False otherwise """ logger.info("Starting data collection...") # Load profile profile = self._load_profile() # Collect weight data weight_readings = self.collect_weight_data() weight_parsed = [self._parse_weight_reading(r) for r in weight_readings] weight_parsed = [r for r in weight_parsed if r is not None] if weight_parsed: # Update profile with latest weight latest_weight = max(weight_parsed, key=lambda x: x.get("timestamp", 0)) profile["vitals"]["weight"] = { "value": latest_weight.get("weight"), "unit": latest_weight.get("unit"), "date": latest_weight.get("date"), } # Deduplicate and update history existing_weight = profile.get("weight_history", []) profile["weight_history"] = self._deduplicate_readings(existing_weight, weight_parsed) # Calculate trend profile["weight_trend_7d"] = self._calculate_weight_trend(profile["weight_history"]) logger.info(f"Updated weight: {latest_weight.get('weight')} {latest_weight.get('unit')}") # Collect BP data bp_readings = self.collect_bp_data() bp_parsed = [self._parse_bp_reading(r) for r in bp_readings] bp_parsed = [r for r in bp_parsed if r is not None] if bp_parsed: # Update profile with latest BP latest_bp = max(bp_parsed, key=lambda x: x.get("timestamp", 0)) profile["vitals"]["blood_pressure"] = { "systolic": latest_bp.get("systolic"), "diastolic": latest_bp.get("diastolic"), "pulse": latest_bp.get("pulse"), "date": latest_bp.get("date"), } # Deduplicate and update history existing_bp = profile.get("bp_history", []) profile["bp_history"] = self._deduplicate_readings(existing_bp, bp_parsed) # Calculate stats profile["bp_stats_30d"] = self._calculate_bp_stats(profile["bp_history"]) logger.info(f"Updated BP: {latest_bp.get('systolic')}/{latest_bp.get('diastolic')} mmHg") # Update last sync time profile["last_sync"] = datetime.now().isoformat() # Save profile self._save_profile(profile) logger.info("Data collection completed") return True def show_status(self) -> None: """Display token and data status.""" print("\n" + "=" * 60) print("iHealth OAuth2 Collector Status") print("=" * 60) # Token status print("\nToken Status:") if self.tokens.get("access_token"): print(f" Access Token: Present") print(f" User ID: {self.user_id}") if self.tokens.get("timestamp"): issued = datetime.fromisoformat(self.tokens["timestamp"]) print(f" Issued At: {issued.strftime('%Y-%m-%d %H:%M:%S')}") if self._is_token_expired(): print(f" Status: EXPIRED") else: print(f" Status: VALID") else: print(" No access token found. Run --authorize first.") # Profile data status print("\nProfile Data Status:") if self.profile_path.exists(): profile = self._load_profile() print(f" Profile Path: {self.profile_path}") if profile.get("vitals", {}).get("weight"): weight_data = profile["vitals"]["weight"] print(f" Latest Weight: {weight_data.get('value')} {weight_data.get('unit')}") print(f" Date: {weight_data.get('date')}") else: print(" No weight data") if profile.get("vitals", {}).get("blood_pressure"): bp_data = profile["vitals"]["blood_pressure"] print(f" Latest BP: {bp_data.get('systolic')}/{bp_data.get('diastolic')} mmHg") print(f" Date: {bp_data.get('date')}") else: print(" No blood pressure data") if profile.get("last_sync"): last_sync = datetime.fromisoformat(profile["last_sync"]) print(f" Last Sync: {last_sync.strftime('%Y-%m-%d %H:%M:%S')}") weight_count = len(profile.get("weight_history", [])) bp_count = len(profile.get("bp_history", [])) print(f" Weight Readings: {weight_count}") print(f" BP Readings: {bp_count}") if profile.get("weight_trend_7d") is not None: trend = profile["weight_trend_7d"] direction = "↑ gain" if trend > 0 else "↓ loss" if trend < 0 else "→ stable" print(f" 7-Day Trend: {abs(trend):.1f} {direction}") if profile.get("bp_stats_30d"): stats = profile["bp_stats_30d"] print(f" 30-Day Avg BP: {stats.get('avg_systolic'):.0f}/{stats.get('avg_diastolic'):.0f} mmHg") else: print(f" No profile found at {self.profile_path}") print("=" * 60 + "\n") def main(): """Main entry point.""" parser = argparse.ArgumentParser( description="iHealth OAuth2 Collector for weight and blood pressure data" ) parser.add_argument( "--authorize", action="store_true", help="Start OAuth2 authorization flow" ) parser.add_argument( "--collect", action="store_true", help="Collect weight and blood pressure data" ) parser.add_argument( "--status", action="store_true", help="Show token and data status" ) parser.add_argument( "--verbose", action="store_true", help="Enable verbose logging" ) args = parser.parse_args() if args.verbose: logging.getLogger().setLevel(logging.DEBUG) collector = iHealthOAuth2Collector() if args.authorize: success = collector.authorize() sys.exit(0 if success else 1) elif args.status: collector.show_status() sys.exit(0) elif args.collect or (not args.authorize and not args.status): success = collector.collect() sys.exit(0 if success else 1) else: parser.print_help() sys.exit(0) if __name__ == "__main__": main()