feat(projekt-matching): langfuse project, keys, score configs; admin API in pod script

projekt-matching/deploy/setup_langfuse.py

@@ -0,0 +1,164 @@
+#!/usr/bin/env python3
+"""Create Langfuse org/project/keys + score configs for projekt-matching.
+
+Uses the Langfuse v3.195 API. IMPORTANT deviation from the original plan:
+the Organization Management API (Bearer ADMIN_API_KEY, /api/admin/...) is an
+Enterprise-only feature in this self-hosted OSS build -- confirmed live via
+`curl -H "Authorization: Bearer $ADMIN_API_KEY" .../api/admin/organizations`
+-> 403 {"error":"This feature is not available on your current plan."}. The
+route exists (not a 404 / wrong-URL guess), it is just plan-gated.
+
+Fallback used instead (still API-driven, no browser): log in as the
+bootstrap admin user (LANGFUSE_INIT_USER_EMAIL/PASSWORD from the pod
+script) via the NextAuth credentials flow to get a session cookie, then call
+the same internal tRPC endpoints the Langfuse web UI itself uses
+(organizations.create, projects.create, projectApiKeys.create) to create the
+org, project and a project API key pair. Score configs ARE available on the
+public, documented REST API (POST /api/public/score-configs) once we have a
+project key pair, so those use requests+Basic auth like the public API
+elsewhere in this repo.
+
+Idempotent: reuses an existing "projekt-matching-org"/"projekt-matching" org
++ project (looked up via the session endpoint) instead of erroring out, and
+score-config creation is safe to retry (Langfuse allows same-name configs;
+re-running this script when configs already exist will just add duplicates,
+so check the UI/API before rerunning after a successful first run).
+
+Prints the project keys; Step 3 (create_pod_langflow.sh) wires them into the
+Langflow container env."""
+from __future__ import annotations
+
+import sys
+
+import requests
+
+BASE = "http://127.0.0.1:8091"
+ADMIN_BEARER = {"Authorization": "Bearer admin-pm-3f61c2a89d4e"}
+LOGIN_EMAIL = "admin@example.com"
+LOGIN_PASSWORD = "langfuse-admin-pw"
+ORG_NAME = "projekt-matching-org"
+PROJECT_NAME = "projekt-matching"
+SCORE_CONFIGS = (
+    ("extraction-correct",
+     "Must/Nice/Misc korrekt aus der Ausschreibung abgeleitet?"),
+    ("matching-correct",
+     "✅/❌/❔-Bewertungen gegen den Lebenslauf korrekt?"),
+)
+SCORE_CATEGORIES = [
+    {"label": "correct", "value": 1},
+    {"label": "partially-correct", "value": 0.5},
+    {"label": "wrong", "value": 0},
+]
+
+
+def try_admin_api() -> dict | None:
+    """Attempt the documented Organization Management API. Returns the
+    project key pair dict on success, None if the feature is unavailable
+    (403) so the caller can fall back."""
+    r = requests.post(f"{BASE}/api/admin/organizations",
+                       json={"name": ORG_NAME}, headers=ADMIN_BEARER,
+                       timeout=30)
+    if r.status_code == 403:
+        print("Admin Organization Management API is plan-gated (403) on "
+              "this self-hosted OSS build -- falling back to the session "
+              "+ tRPC method.", file=sys.stderr)
+        return None
+    if r.status_code == 404:
+        print("Admin API route missing (404) -- falling back to the "
+              "session + tRPC method.", file=sys.stderr)
+        return None
+    r.raise_for_status()
+    org = r.json()
+    org_key = requests.post(
+        f"{BASE}/api/admin/organizations/{org['id']}/apiKeys", json={},
+        headers=ADMIN_BEARER, timeout=30)
+    org_key.raise_for_status()
+    ok = org_key.json()
+    org_auth = (ok["publicKey"], ok["secretKey"])
+    proj = requests.post(f"{BASE}/api/public/projects",
+                          json={"name": PROJECT_NAME, "retention": 0},
+                          auth=org_auth, timeout=30)
+    proj.raise_for_status()
+    project = proj.json()
+    keys = requests.post(
+        f"{BASE}/api/public/projects/{project['id']}/apiKeys", json={},
+        auth=org_auth, timeout=30)
+    keys.raise_for_status()
+    return keys.json()
+
+
+def session_login() -> requests.Session:
+    s = requests.Session()
+    csrf = s.get(f"{BASE}/api/auth/csrf", timeout=15).json()["csrfToken"]
+    r = s.post(f"{BASE}/api/auth/callback/credentials", data={
+        "email": LOGIN_EMAIL, "password": LOGIN_PASSWORD, "csrfToken": csrf,
+        "callbackUrl": f"{BASE}/", "json": "true",
+    }, timeout=15)
+    r.raise_for_status()
+    sess = s.get(f"{BASE}/api/auth/session", timeout=15).json()
+    if not sess.get("user"):
+        sys.exit(f"Login failed: {sess}")
+    return s, sess
+
+
+def trpc(s: requests.Session, path: str, payload: dict) -> dict:
+    r = s.post(f"{BASE}/api/trpc/{path}", json={"json": payload},
+               timeout=30)
+    if r.status_code >= 400:
+        sys.exit(f"tRPC {path} failed ({r.status_code}): {r.text[:500]}")
+    return r.json()["result"]["data"]["json"]
+
+
+def via_session_trpc() -> dict:
+    s, sess = session_login()
+    org_id = project_id = None
+    for org in sess["user"]["organizations"]:
+        if org["name"] == ORG_NAME or org["id"] == ORG_NAME:
+            org_id = org["id"]
+            for proj in org["projects"]:
+                if proj["name"] == PROJECT_NAME:
+                    project_id = proj["id"]
+    if org_id is None:
+        org = trpc(s, "organizations.create", {"name": ORG_NAME})
+        org_id = org["id"]
+        print(f"Created organization {ORG_NAME} ({org_id})", file=sys.stderr)
+    else:
+        print(f"Reusing existing organization {ORG_NAME} ({org_id})",
+              file=sys.stderr)
+    if project_id is None:
+        proj = trpc(s, "projects.create",
+                    {"orgId": org_id, "name": PROJECT_NAME})
+        project_id = proj["id"]
+        print(f"Created project {PROJECT_NAME} ({project_id})",
+              file=sys.stderr)
+    else:
+        print(f"Reusing existing project {PROJECT_NAME} ({project_id})",
+              file=sys.stderr)
+    keys = trpc(s, "projectApiKeys.create", {"projectId": project_id})
+    return keys
+
+
+def create_score_configs(pk: str, sk: str) -> None:
+    for name, desc in SCORE_CONFIGS:
+        r = requests.post(f"{BASE}/api/public/score-configs", auth=(pk, sk),
+                           json={"name": name, "dataType": "CATEGORICAL",
+                                 "description": desc,
+                                 "categories": SCORE_CATEGORIES},
+                           timeout=30)
+        r.raise_for_status()
+        print(f"Score config '{name}' created: {r.json()['id']}",
+              file=sys.stderr)
+
+
+def main() -> None:
+    keys = try_admin_api()
+    if keys is None:
+        keys = via_session_trpc()
+    pk, sk = keys["publicKey"], keys["secretKey"]
+    create_score_configs(pk, sk)
+    print(f"PM_LANGFUSE_PUBLIC_KEY={pk}")
+    print(f"PM_LANGFUSE_SECRET_KEY={sk}")
+
+
+if __name__ == "__main__":
+    main()