df667e0db8
- web/: Local chat UI (Express + WS → Codex bridge) - openwebui/: Preset, pipelines, knowledge manifest - Dockerfile + docker-compose.yml - Updated README with 3 frontend options - CLI-agnostic: works with Codex, Claude Code, Kiro, Gemini
109 lines
3.4 KiB
Python
109 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
|
import os
|
|
import sys
|
|
import json
|
|
import urllib.request
|
|
import urllib.error
|
|
|
|
# Gainsight PX API configuration
|
|
# Region determines the base URL (US or EU)
|
|
PX_REGION = os.environ.get("GAINSIGHT_PX_REGION", "US").upper()
|
|
PX_API_KEY = os.environ.get("GAINSIGHT_PX_API_KEY")
|
|
|
|
if PX_REGION == "EU":
|
|
BASE_URL = "https://eu-api.aptrinsic.com/v1"
|
|
else:
|
|
BASE_URL = "https://api.aptrinsic.com/v1"
|
|
|
|
def make_request(method, endpoint, data=None):
|
|
if not PX_API_KEY:
|
|
print(json.dumps({"error": "GAINSIGHT_PX_API_KEY environment variable is missing."}))
|
|
sys.exit(1)
|
|
|
|
url = f"{BASE_URL}{endpoint}"
|
|
headers = {
|
|
"X-APITOKEN": PX_API_KEY,
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json"
|
|
}
|
|
|
|
req_data = None
|
|
if data:
|
|
req_data = json.dumps(data).encode("utf-8")
|
|
|
|
req = urllib.request.Request(url, data=req_data, headers=headers, method=method)
|
|
|
|
try:
|
|
with urllib.request.urlopen(req) as response:
|
|
return json.loads(response.read().decode("utf-8"))
|
|
except urllib.error.HTTPError as e:
|
|
err_msg = e.read().decode("utf-8")
|
|
try:
|
|
parsed_err = json.loads(err_msg)
|
|
return {"error": f"HTTP {e.code}", "details": parsed_err}
|
|
except:
|
|
return {"error": f"HTTP {e.code}", "details": err_msg}
|
|
except Exception as e:
|
|
return {"error": str(e)}
|
|
|
|
def get_user(user_id):
|
|
"""Retrieve a specific user by their identifyId."""
|
|
return make_request("GET", f"/users/{user_id}")
|
|
|
|
def get_account(account_id):
|
|
"""Retrieve a specific account by its id."""
|
|
return make_request("GET", f"/accounts/{account_id}")
|
|
|
|
def search_users(email):
|
|
"""Search for users by email (requires query payload)."""
|
|
payload = {
|
|
"filter": {
|
|
"operator": "AND",
|
|
"conditions": [
|
|
{
|
|
"name": "email",
|
|
"operator": "EQ",
|
|
"value": email
|
|
}
|
|
]
|
|
}
|
|
}
|
|
return make_request("POST", "/users/query", data=payload)
|
|
|
|
def track_event(user_id, event_name, properties=None):
|
|
"""Track a custom event for a specific user."""
|
|
payload = {
|
|
"identifyId": user_id,
|
|
"eventName": event_name,
|
|
"properties": properties or {}
|
|
}
|
|
# Note: tracking usually happens via a different endpoint or batch API,
|
|
# but for simplicity assuming a standard REST event ingestion if available.
|
|
return make_request("POST", "/events/custom", data=payload)
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) < 2:
|
|
print(json.dumps({"error": "Missing action. Use: get_user, get_account, search_user, track_event"}))
|
|
sys.exit(1)
|
|
|
|
action = sys.argv[1]
|
|
|
|
if action == "get_user" and len(sys.argv) == 3:
|
|
print(json.dumps(get_user(sys.argv[2]), indent=2))
|
|
|
|
elif action == "get_account" and len(sys.argv) == 3:
|
|
print(json.dumps(get_account(sys.argv[2]), indent=2))
|
|
|
|
elif action == "search_user" and len(sys.argv) == 3:
|
|
print(json.dumps(search_users(sys.argv[2]), indent=2))
|
|
|
|
elif action == "track_event" and len(sys.argv) >= 4:
|
|
user_id = sys.argv[2]
|
|
event_name = sys.argv[3]
|
|
props = json.loads(sys.argv[4]) if len(sys.argv) > 4 else {}
|
|
print(json.dumps(track_event(user_id, event_name, props), indent=2))
|
|
|
|
else:
|
|
print(json.dumps({"error": f"Unknown action or missing arguments: {action}"}))
|
|
sys.exit(1)
|