#!/usr/bin/env python3
"""
Siphox Health API Workflow Script
This script automates the complete Siphox SAI workflow:
1. Creates a user in Siphox admin account
2. Uploads PDF blood test data
3. Adds test data for user account manually
4. Adds questionnaire data
5. Sends initial message and follow-up
6. Updates conversation with notes
7. Lists all conversations and messages
Requirements:
- SIPHOX_API_KEY in .env file
- requests library (pip install requests)
- python-dotenv library (pip install python-dotenv)
- boto3 library (pip install boto3)
"""
import os
import json
import time
import logging
import threading
from datetime import datetime
from typing import Dict, Any, Optional, List, Literal
from pathlib import Path
import random
import requests
import boto3
from botocore.exceptions import ClientError
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
# Configure logging
def setup_logging():
"""Setup logging to output both to console and file"""
# Create log filename with timestamp
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
log_file = f"./outputs/sai_workflow_{timestamp}.txt"
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(message)s',
handlers=[
logging.FileHandler(log_file, mode='w'),
logging.StreamHandler() # Also output to console
]
)
return str(log_file)
# Logger instance
logger = logging.getLogger(__name__)
class S3Manager:
"""Simple S3 manager for uploading files and generating presigned URLs"""
def __init__(self, bucket_name: str, profile_name: str = "sp"):
self.bucket_name = bucket_name
self.session = boto3.Session(profile_name=profile_name)
self.s3_client = self.session.client('s3')
def upload_file(self, local_file_path: str, s3_key: str) -> bool:
"""Upload a file to S3 bucket"""
try:
self.s3_client.upload_file(local_file_path, self.bucket_name, s3_key)
logger.info(f"ā
File uploaded to S3: s3://{self.bucket_name}/{s3_key}")
return True
except ClientError as e:
logger.error(f"ā Failed to upload file to S3: {e}")
return False
def generate_presigned_url(self, s3_key: str, expiration: int = 3600) -> Optional[str]:
"""Generate a presigned URL for the S3 object"""
try:
url = self.s3_client.generate_presigned_url(
'get_object',
Params={'Bucket': self.bucket_name, 'Key': s3_key},
ExpiresIn=expiration
)
return url
except ClientError as e:
logger.error(f"ā Failed to generate presigned URL: {e}")
return None
def upload_and_get_url(self, local_file_path: str, s3_key: str) -> Optional[str]:
"""Upload file and return presigned URL"""
if self.upload_file(local_file_path, s3_key):
return self.generate_presigned_url(s3_key)
return None
class SiphoxAPI:
"""Simple and maintainable Siphox API client"""
def __init__(self):
self.api_key = os.getenv('SIPHOX_API_KEY')
if not self.api_key:
raise ValueError("SIPHOX_API_KEY not found in environment variables")
self.base_url = "https://connect.siphoxhealth.com"
self.headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
def _make_request(self, method: str, endpoint: str, data: Dict = None, files: Dict = None) -> Dict[str, Any]:
"""Make HTTP request with error handling"""
url = f"{self.base_url}{endpoint}"
headers = self.headers.copy()
if files:
# Remove Content-Type for file uploads
headers.pop("Content-Type", None)
try:
response = requests.request(
method=method,
url=url,
headers=headers,
json=data if not files else None,
files=files,
timeout=30
)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
logger.error(f"API request failed: {e}")
if hasattr(e, 'response') and e.response is not None:
logger.error(f"Response: {e.response.text}")
raise
def create_user(self, email: str, first_name: str, last_name: str, dob: str = None, gender: str = None) -> Dict[str, Any]:
"""Create a new user"""
data = {
"email": email,
"name": f"{first_name} {last_name}",
"dateOfBirth": dob,
"gender": gender
}
return self._make_request("POST", "/api/v1/customer", data)
def get_user(self, user_id: str) -> Dict[str, Any]:
"""Get a user"""
return self._make_request("GET", f"/api/v1/customers/{user_id}")
def add_user_data(
self,
customer_id: str,
type: Literal['file', 'quiz', 'test'],
quiz_name: str = None,
date: str = None,
s3_file_url: str = None,
questionnaire_data: List[Dict[str, Any]] = None,
test_data: List[Dict[str, Any]] = None,
async_: bool = False
) -> Dict[str, Any]:
"""Add file data using S3 presigned URL"""
if not date:
date = datetime.utcnow().isoformat() + "Z"
data = {
"type": type,
"userId": customer_id,
"date": date,
"quizName": quiz_name,
"s3FileUrl": s3_file_url,
"biomarkerData": test_data,
"quizData": questionnaire_data,
"async": async_
}
logger.info(f"Adding file data to Siphox: {data}")
return self._make_request("POST", "/api/v1/customer/add-data", data)
def get_job_status(self, job_id: str) -> Dict[str, Any]:
"""Get job status"""
return self._make_request("GET", f"/api/v1/customer/add-data/jobs/{job_id}")
def send_message(self, user_id: str, message: str, chat_id: str = None) -> Dict[str, Any]:
"""Send message to user (creates conversation if none exists)"""
data = {"userMessage": message}
if chat_id:
data["chatId"] = chat_id
return self._make_request("POST", f"/api/v1/sai/chats/{user_id}", data)
def update_conversation_notes(self, chat_id: str, notes: str, tags: List[str]) -> Dict[str, Any]:
"""Update conversation with notes"""
data = {"notes": notes, "tags": tags}
return self._make_request("PATCH", f"/api/v1/sai/chats/{chat_id}", data)
def list_conversations(
self,
user_id: Optional[str] = None,
startDate: Optional[str] = None,
endDate: Optional[str] = None,
minLength: Optional[str] = None,
tags: List[str] = None,
page: Optional[str] = None,
limit: Optional[str] = None
) -> Dict[str, Any]:
"""List all conversations for account or specific user"""
endpoint = "/api/v1/sai/chats"
if user_id:
endpoint += f"?userId={user_id}"
if startDate:
endpoint += f"?startDate={startDate}"
if endDate:
endpoint += f"?endDate={endDate}"
if minLength:
endpoint += f"?minLength={minLength}"
if tags:
endpoint += f"?tags={','.join(tags)}"
if page:
endpoint += f"?page={page}"
if limit:
endpoint += f"?limit={limit}"
return self._make_request("GET", endpoint)
def get_conversation_messages(self, conversation_id: str) -> Dict[str, Any]:
"""Get all messages in a conversation"""
return self._make_request("GET", f"/api/v1/conversations/{conversation_id}/messages")
def reformat_questionnaire_data(questionnaire_data: Dict[str, Any]) -> Dict[str, Any]:
"""Reformat questionnaire data"""
formatted_data = []
for question in questionnaire_data:
formatted_data.append({
"question": question,
"answer": str(questionnaire_data.get(question))
})
return formatted_data
def create_sample_questionnaire_data() -> Dict[str, Any]:
"""Create sample questionnaire data"""
quiz = {
"age": 30,
"gender": "male",
"weight": 70,
"height": 175,
"activityLevel": "moderate",
"dietType": "balanced",
"sleepHours": 7,
"stressLevel": "moderate",
"supplements": ["vitamin_d", "omega_3"],
"medicalConditions": [],
"goals": ["general_health", "energy_improvement"]
}
return reformat_questionnaire_data(quiz)
def create_sample_test_data() -> Dict[str, Any]:
"""Create sample test data"""
test_data = [{
"name": "Total Cholesterol",
"result": 185,
"unit_of_measure": "mg/dL",
"simpleLabel": "Cholesterol",
"range_minimum": 0,
"range_maximum": 200
},
{
"name": "HDL Cholesterol",
"result": 55,
"unit_of_measure": "mg/dL",
"range_minimum": 40,
"range_maximum": 60
}
]
return test_data
def add_data_for_user(api: SiphoxAPI):
s3_manager = S3Manager("sai-api-playground")
# Configuration
user_email = f"test-{random.randint(1000, 9999)}@example.com"
user_first_name = "Test"
user_last_name = "User"
user_dob = "1990-01-01"
user_gender = "M"
# pdf_path = "/Users/jordanmoradian/workdev/playground/data/blood_test_pdf/inside-tracker-report-1.pdf"
pdf_path = "/Users/jordanmoradian/workdev/playground/data/blood_test_pdf/siphox.pdf"
try:
# Step 1: Create user
logger.info("\nš Creating user...")
user_response = api.create_user(user_email, user_first_name, user_last_name, user_dob, user_gender)
user_id = user_response.get('_id') or user_response.get('internalUserId')
logger.info(f"ā
User created with ID: {user_id} \n {user_response}")
# Step 2: Upload PDF to S3 and add file data
logger.info("\nš Uploading blood test PDF to S3...")
pdf_filename = Path(pdf_path).name
s3_key = f"blood-tests/{user_id}/{pdf_filename}"
# Upload to S3 and get presigned URL
s3_file_url = s3_manager.upload_and_get_url(pdf_path, s3_key)
if not s3_file_url:
raise Exception("Failed to upload PDF to S3")
logger.info(f"š¤ Adding file data to Siphox...")
response = api.add_user_data(user_id, 'file', s3_file_url=s3_file_url, async_=True)
job_id = response.get('jobId')
# Step 3: Get async job completion status
logger.info("\nš Polling for async job completion status...")
max_retries = 48
retries = 0
while retries < max_retries:
job_response = api.get_job_status(job_id)
status = job_response.get('status')
if status == 'completed':
logger.info(f"ā
Job completed: {job_response}")
break
elif status == 'failed':
logger.error(f"ā File upload failed: {job_response.get('error')}")
break
else:
logger.info(f"{retries + 1}: Job has not resolved yet. Polling again...")
time.sleep(5)
retries += 1
# Step 3: Add questionnaire data
logger.info("\nš Adding questionnaire data...")
questionnaire_data = create_sample_questionnaire_data()
questionnaire_response = api.add_user_data(user_id, 'quiz', quiz_name="health-assessment", questionnaire_data=questionnaire_data)
logger.info(f"ā
Questionnaire data added: {questionnaire_response}")
# Step 4: Add test data
logger.info("\nš Adding test data...")
test_data = create_sample_test_data()
test_response = api.add_user_data(user_id, 'test', test_data=test_data)
logger.info(f"ā
Test data added: {test_response}")
return user_id
except Exception as e:
logger.error(f"\nā User set up failed: {e}")
raise
def run_user_conversation(api: SiphoxAPI, user_id: str):
try:
# Step 4: Send initial message
logger.info("\nš¬ Sending initial message...")
initial_message = "Hello! I just uploaded my blood test results. Can you help me understand my health status and provide recommendations?"
message_response = api.send_message(user_id, initial_message)
chat_id = message_response.get('chatId')
logger.info(f"ā
Initial message sent. Conversation ID: {chat_id} \n {message_response}")
# Step 5: Send follow-up message
logger.info("\nš¬ Sending follow-up message...")
followup_message = "Could you also provide specific supplement recommendations based on my results?"
followup_response = api.send_message(user_id, followup_message, chat_id=chat_id)
logger.info(f"ā
Follow-up message sent: {followup_response}")
# Step 6: Update conversation with notes
logger.info("\nš Updating conversation notes...")
notes = "Initial consultation - user interested in supplement recommendations and health optimization"
notes_response = api.update_conversation_notes(chat_id, notes, tags=["supplement_recommendations"])
logger.info(f"ā
Notes updated: {notes_response}")
# Step 7: List all conversations
logger.info("\nš Listing all conversations...")
conversations = api.list_conversations(user_id=user_id)
# Step 8: Display conversation details
logger.info(f"ā
Conversation details: {conversations}")
except Exception as e:
logger.error(f"\nā User conversation management failed: {e}")
raise
def main():
"""Main workflow execution"""
# Setup logging
log_file = setup_logging()
logger.info(f"Starting Siphox Workflow...")
logger.info(f"Output will be saved to: {log_file}")
# Initialize API client and S3 manager
api = SiphoxAPI()
user_id = add_data_for_user(api)
logger.info(f"ā
User successfully added: {user_id}")
run_user_conversation(api, user_id)
logger.info(f"ā
User conversation successfully run: {user_id}")
logger.info("\nš Workflow completed successfully!")
if __name__ == "__main__":
main()