Sync Open WebUI Feedback → Portkey

pip install requests
# Save the script from the Python section as owui_to_portkey.py
python3 owui_to_portkey.py --since $(date -d '2 hours ago' +%s 2>/dev/null || python - <<'PY'
import time
print(int(time.time() - 7200))
PY
)

#!/usr/bin/env python3
"""Manual sync from Open WebUI to Portkey feedback API."""

import argparse
import json
import os
import sys
from typing import Any, Dict, Iterable, List, Optional

import requests

OW_BASE = os.getenv("OPENWEBUI_BASE_URL", "").rstrip("/")
OW_TOKEN = os.getenv("OPENWEBUI_TOKEN", "")
PK_KEY = os.getenv("PORTKEY_API_KEY", "")
PK_BASE = os.getenv("PORTKEY_BASE_URL", "https://api.portkey.ai/v1").rstrip("/")
PK_URL = f"{PK_BASE}/feedback"


def die(message: str) -> None:
    print(message, file=sys.stderr)
    sys.exit(1)


def fetch_feedbacks(admin: bool = True) -> List[Dict[str, Any]]:
    if not OW_BASE or not OW_TOKEN:
        die("Missing OPENWEBUI_BASE_URL or OPENWEBUI_TOKEN")

    path = "/api/v1/evaluations/feedbacks/all/export" if admin else "/api/v1/evaluations/feedbacks/user"
    url = f"{OW_BASE}{path}"
    headers = {"Authorization": f"Bearer {OW_TOKEN}", "Content-Type": "application/json"}

    response = requests.get(url, headers=headers, timeout=30)
    if response.status_code == 401:
        cookie_headers = {"Cookie": f"token={OW_TOKEN}", "Content-Type": "application/json"}
        response = requests.get(url, headers=cookie_headers, timeout=30)

    response.raise_for_status()
    payload = response.json()
    if not isinstance(payload, list):
        die("Unexpected Open WebUI response format")
    return payload


def within_since(item: Dict[str, Any], since_epoch: Optional[int]) -> bool:
    if not since_epoch:
        return True
    timestamp = int(item.get("updated_at") or item.get("created_at") or 0)
    return timestamp >= since_epoch


def map_item(feedback: Dict[str, Any]) -> Optional[Dict[str, Any]]:
    data = feedback.get("data") or {}
    meta = feedback.get("meta") or {}

    try:
        rating = int(data.get("rating"))
    except (TypeError, ValueError):
        return None

    if rating not in (1, -1):
        return None

    chat_id = meta.get("chat_id") or "unknown_chat"
    message_id = meta.get("message_id") or feedback.get("id") or "unknown_msg"
    trace_id = f"{chat_id}:{message_id}"

    metadata: Dict[str, Any] = {**data, **meta}

    snapshot_chat = (feedback.get("snapshot") or {}).get("chat")
    if isinstance(snapshot_chat, dict) and snapshot_chat.get("id"):
        metadata["snapshot_chat_id"] = snapshot_chat["id"]

    for key in ("message_index", "user_id", "version"):
        if key in feedback and feedback[key] is not None:
            metadata[key] = feedback[key]

    return {"trace_id": trace_id, "value": rating, "weight": 1, "metadata": metadata}


def post_to_portkey(item: Dict[str, Any]) -> None:
    if not PK_KEY:
        die("Missing PORTKEY_API_KEY")

    response = requests.post(
        PK_URL,
        headers={"x-portkey-api-key": PK_KEY, "Content-Type": "application/json"},
        json=item,
        timeout=30,
    )

    if not response.ok:
        print(f"[Portkey] {response.status_code} {response.text}", file=sys.stderr)
    else:
        print(f"[Portkey] OK {response.json()}")


def main() -> None:
    parser = argparse.ArgumentParser(description="Sync Open WebUI feedback to Portkey")
    parser.add_argument("--user-scope", action="store_true", help="Use per-user endpoint")
    parser.add_argument("--since", type=int, help="Only sync items updated_at >= EPOCH_SECONDS")
    parser.add_argument("--dry-run", action="store_true", help="Print mapped payloads only")
    args = parser.parse_args()

    feedbacks = fetch_feedbacks(admin=not args.user_scope)
    filtered = [item for item in feedbacks if within_since(item, args.since)]
    mapped = [mapped for item in filtered if (mapped := map_item(item))]

    print(f"Fetched {len(feedbacks)}; {len(filtered)} within --since; {len(mapped)} mapped 👍/👎.")

    if args.dry_run:
        print(json.dumps(mapped, indent=2))
        return

    for payload in mapped:
        post_to_portkey(payload)


if __name__ == "__main__":
    main()