From 240e4ffdd4be58c0e967872a65bb192053a6f583 Mon Sep 17 00:00:00 2001 From: TPKutch Date: Mon, 16 Mar 2026 11:25:30 -0500 Subject: [PATCH] updated oura webhook checker and json uploader to accommodate webhooks --- source/checkers/api/oura.py | 196 +++++++++++++++++++++++++++--------- source/uploaders/json.py | 28 ++++-- 2 files changed, 166 insertions(+), 58 deletions(-) diff --git a/source/checkers/api/oura.py b/source/checkers/api/oura.py index 011d0a3e..f74e0fc3 100644 --- a/source/checkers/api/oura.py +++ b/source/checkers/api/oura.py @@ -1,5 +1,6 @@ import os import json +import time from asyncio.staggered import staggered_race import pandas as pd @@ -171,18 +172,17 @@ class OuraOAuthWebhookChecker(OuraOAuthBaseChecker): checker_name = "OuraOAuthWebhookChecker" # Optional settings for where to find data on the remote listener server - # - webhook_post_folder_name = 'webhook_posts' + webhook_post_folder_name = "webhook_posts" webhook_post_folder_path = None - participant_map_file_name = 'participant_map.json' + participant_map_file_name = "participant_map.json" participant_map_file_path = None - auth_token_file_name = 'oura_tokens.json' + auth_token_file_name = "oura_tokens.json" auth_token_file_path = None # Name of file in staging to keep track of the most recent event for each patient - event_track_datatype = 'webhook_times' + event_track_datatype = "webhook_times" stub_config = """ [parser.init.source] @@ -199,9 +199,47 @@ class OuraOAuthWebhookChecker(OuraOAuthBaseChecker): """ source_location = toml.loads(stub_config) + def _json_safe(self, obj): + """Recursively convert Path-like objects into JSON-serializable types.""" + if isinstance(obj, Path): + return str(obj) + if isinstance(obj, dict): + return {k: self._json_safe(v) for k, v in obj.items()} + if isinstance(obj, (list, tuple)): + return [self._json_safe(v) for v in obj] + return obj + + def _state_file_path(self) -> str: + """ + Our framework sometimes treats state_path as a file path, sometimes as a directory. + If it's a directory, store upload state as /upload_state.json + """ + # If state_path ends with .json, assume it's a file path + if isinstance(self.state_path, str) and self.state_path.endswith(".json"): + return self.state_path + return os.path.join(self.state_path, "upload_state.json") + + def _load_upload_state(self) -> None: + """ + Ensure self.upload_state exists and self._upload_state_path points to the on-disk JSON. + """ + p = self._state_file_path() + os.makedirs(os.path.dirname(p), exist_ok=True) + + if os.path.exists(p): + try: + with open(p, "r") as f: + self.upload_state = json.load(f) + except json.JSONDecodeError: + self.upload_state = {"success": [], "failure": [], "skipped": []} + else: + self.upload_state = {"success": [], "failure": [], "skipped": []} + + self._upload_state_path = p + @property def source_data_path(self): - return Path(self.source_location['listener_data']) + return Path(self.source_location["listener_data"]) def build_remote_path(self, fullpath, ending): if fullpath: # Prefer full explicit path if given @@ -234,24 +272,46 @@ def _map_user_to_participant(self, user_id): try: mapped_id = user_map[user_id] except KeyError: - self.debug(f'User ID {user_id} not found in the participant map') + self.debug(f"User ID {user_id} not found in the participant map") raise KeyError("Could not find user!") # Make sure user IDs are not sent via error logging else: self.debug(f"User ID: {user_id}, mapped participant: {mapped_id}") return mapped_id def iter_webhooks(self): + """ + Generator yielding (remote_file_path, filename). + + NOTE: Kept original structure, but now includes a fast look_back_duration cutoff + using lexicographic comparison of the filename timestamp. + """ + # Fast cutoff string for filenames: "YYYY-MM-DDTHH-MM-SS" + start_cutoff = (pd.Timestamp.now(tz="UTC") - pd.Timedelta(self.look_back_duration)).strftime( + "%Y-%m-%dT%H-%M-%S" + ) + user_dirs = self.sftp.listdir(self.webhook_post_path.as_posix()) self.info(f"Users found in webhook dir: {user_dirs}") + for user in user_dirs: user_path = Path(self.webhook_post_path, user) modality_dirs = self.sftp.listdir(user_path.as_posix()) + for modality in modality_dirs: modality_path = Path(user_path, modality) - webhooks_jsons = [f for f - in self.sftp.listdir(modality_path.as_posix()) - if f.endswith(".json")] + + webhooks_jsons = [f for f in self.sftp.listdir(modality_path.as_posix()) if f.endswith(".json")] + + # NEW: sort newest-first so we can break early when we hit older-than-cutoff + webhooks_jsons = sorted(webhooks_jsons, reverse=True) + for filename in webhooks_jsons: + # filename looks like: "2026-03-11T10-29-08.json" + base = filename[:-5] # strip ".json" + if base < start_cutoff: + # since sorted desc, everything else will be older too + break + yield Path(modality_path, filename), filename def process_webhooks(self): @@ -268,16 +328,20 @@ def process_webhooks(self): payload = self._read_json_file(file_path.as_posix()) results = self._handle_payload(payload, tokens, filename) except Exception as e: - import sys, traceback + import traceback + self.warning(f"[ERROR] Failed on {file_path}: {str(e)}") - failures.append({ - "file": file_path, - "error": str(e), - "traceback": traceback.format_exc(), - "timestamp": datetime.now().timestamp() - }) + failures.append( + { + "file": file_path, + "error": str(e), + "traceback": traceback.format_exc(), + "timestamp": datetime.now().timestamp(), + } + ) else: to_process.extend(results) + return to_process, failures def _handle_payload(self, payload, tokens, timestamp_clean): @@ -301,37 +365,56 @@ def _handle_payload(self, payload, tokens, timestamp_clean): response.raise_for_status() data = response.json() + # --- required by JSONDocInjectorUploader --- + payload_out = data if isinstance(data, dict) else {"data": data} + + payload_out["patient_id"] = participant_id + payload_out["doc_type"] = data_type + payload_out["document_id"] = object_id # stable unique identifier + payload_out["event_time"] = event_time # optional but useful + + # choose a date for grouping (best effort) + payload_out["date"] = ( + payload_out.get("day") + or payload_out.get("summary_date") + or (event_time[:10] if isinstance(event_time, str) and len(event_time) >= 10 else None) + or timestamp_clean[:10] # from filename like YYYY-MM-DD... + ) + # Save the downloaded data in the local staging directory for further processing stage_dir = self.local_staging_path / participant_id / data_type stage_dir.mkdir(parents=True, exist_ok=True) staged_file = stage_dir / timestamp_clean with open(staged_file, "w") as f: - f.write(json.dumps(data, indent=2)) + f.write(json.dumps(payload_out, indent=2)) staged_files = [staged_file] - # Update the time of the most recent event for this patient in the event time tracker - patient_event_file = self.local_staging_path / participant_id / self.event_track_datatype / timestamp_clean - if os.path.exists(patient_event_file): - with open(patient_event_file, "r") as f: - existing_event_times = json.load(f) - else: - existing_event_times = {} - - if data_type in existing_event_times: - existing_event_times[data_type] = existing_event_times[data_type] + event_time - else: - existing_event_times[data_type] = event_time + event_payload = { + "patient_id": participant_id, + "doc_type": self.event_track_datatype, # "webhook_times" + "date": payload_out["date"], # same day grouping + "document_id": timestamp_clean, # unique per webhook file (good enough) + "data_type": data_type, + "object_id": object_id, + "event_time": event_time, + "event_type": payload.get("event_type"), + "user_id": user_id, + } - # If an event time was updated make sure the event tracker file is appended to the list - with open(patient_event_file, "w") as f: - json.dump(existing_event_times, f, indent=2) - staged_files.append(patient_event_file) + event_dir = self.local_staging_path / participant_id / self.event_track_datatype + event_dir.mkdir(parents=True, exist_ok=True) + event_file = event_dir / f"{timestamp_clean}.json" + with open(event_file, "w") as f: + json.dump(event_payload, f, indent=2) + staged_files = [staged_file.as_posix(), event_file.as_posix()] return staged_files # find new webhook posts that haven't been processed and query the api for the data def check(self): + # NEW: make sure upload_state exists for save()/clean() + self._load_upload_state() try: self.make_connection() @@ -343,34 +426,48 @@ def check(self): # Make sure to upload files are unique (since there could be multiple even tracker updates) to_process = list(set(to_process)) + to_process = [str(p) for p in to_process] - self.info(f'Found data from new webhook events!') - self.info(f'Experienced {len(failures)} failures during data retrieval') + self.info("Found data from new webhook events!") + self.info(f"Experienced {len(failures)} failures during data retrieval") return {"to do": to_process, "failure": failures} def save(self, completed): - to_save = completed.get("success", []) + # Ensure upload_state exists even if save() is called without check() + if not hasattr(self, "upload_state"): + self._load_upload_state() + + to_save = completed.get("success", []) or [] if to_save: - self.info(f"[Save] to_save contents: {to_save[0]}") + self.info(f"[Save] to_save sample keys: {list(to_save[0].keys())}") self.info(f"[Save] Marking {len(to_save)} files as uploaded") for item in to_save: - self.upload_state["success"].append({ - "key": item["key"], - "uploaded": item["uploaded"], - "timestamp": item["timestamp"] - }) + # Robustly infer a "key" for bookkeeping if the pipeline didn't supply one + key = item.get("key") or item.get("destination") or item.get("filename") + + uploaded = item.get("uploaded", True) + ts = item.get("timestamp", time.time()) + + # Store the raw item too (helps debugging / future schema changes) + record = { + "key": key, + "uploaded": uploaded, + "timestamp": ts, + "raw": item, + } + self.upload_state.setdefault("success", []).append(record) - # write once after all are added - with open(self.state_path, "w") as f: - json.dump(self.upload_state, f, indent=2) + with open(self._upload_state_path, "w") as f: + json.dump(self._json_safe(self.upload_state), f, indent=2) def clean(self): if not hasattr(self, "sftp"): self.make_connection() - with open(self.state_path, "r") as f: - self.upload_state = json.load(f) + + # NEW: load upload_state from disk + self._load_upload_state() # clean outdated log entries cleaned_log = super().clean_outdated(self.upload_state) @@ -379,7 +476,7 @@ def clean(self): final_log = super().clean_old_success(cleaned_log) # save cleaned upload log - with open(self.state_path, "w") as f: + with open(self._upload_state_path, "w") as f: json.dump(final_log, f, indent=2) @@ -515,7 +612,6 @@ def patients(self): simple_tokens = {patient: data['access_token'] for patient, data in patient_tokens.items()} return simple_tokens - class OuraOAuthStreamChecker(OuraOAuthBaseChecker, OuraAPIStreamChecker): """ Checker designed to periodically fetch data from Oura API, using OAuth tokens which are continuously refreshed diff --git a/source/uploaders/json.py b/source/uploaders/json.py index 26d1eb51..f91de9f9 100644 --- a/source/uploaders/json.py +++ b/source/uploaders/json.py @@ -2,6 +2,7 @@ import os import re import json +import time from pathlib import Path from datetime import datetime @@ -66,7 +67,7 @@ def make_target_path(self, payload, **kwargs): context = copy.deepcopy(payload) context.update(kwargs) - path = self.new_format.format(**context) + path = self.target_location['new_format'].format(**context) return Path(path) def inject_payload(self, target_file, payload): @@ -82,12 +83,12 @@ def inject_payload(self, target_file, payload): else: contents = [] - payload_id = payload[self.doc_id_key] + payload_id = payload[self.target_location['doc_id_key']] # Determine if this payload needs to be appended or replace an existing payload placement_loc = None for idx, doc in enumerate(contents): - this_id = doc[self.doc_id_key] + this_id = doc[self.target_location['doc_id_key']] if payload_id == this_id: placement_loc = idx break @@ -99,11 +100,20 @@ def inject_payload(self, target_file, payload): contents[placement_loc] = payload # Write the updated file contents to the file - with open(target_file, 'w') as json_file: - json_file.write(contents) + with open(target_file, "w") as json_file: + json.dump(contents, json_file, indent=2) def upload(self, ready): + if isinstance(ready, dict): + ready = ( + ready.get("to upload") + or ready.get("to do") + or ready.get("todo") + or ready.get("to_do") + or [] + ) + successes = [] errors = [] @@ -126,7 +136,7 @@ def upload(self, ready): error_dict = { "type": "upload failure", "filename": filename, - "destination": destination, + "destination": str(destination), "error": str(e), "trace": traceback.format_exception(*sys.exc_info()), "timestamp": datetime.now().timestamp() @@ -139,6 +149,8 @@ def upload(self, ready): successes.append({ "type": "upload success", "filename": filename, - "destination": destination, + "destination": str(destination), "timestamp": datetime.now().timestamp() - }) \ No newline at end of file + }) + + return {"success": successes, "failure": errors, "skipped": []}