export const remotePythonCreditsScript = String.raw` def credits_parse_time(value): if not isinstance(value, str) or not value: return None try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) return parsed.replace(tzinfo=timezone.utc) if parsed.tzinfo is None else parsed.astimezone(timezone.utc) except ValueError: return None def credits_window_start(value, end_at): matched = re.fullmatch(r"([1-9][0-9]*)([mhd])", str(value or "")) if matched is None: raise RuntimeError("credits active window must use a duration such as 24h") seconds = int(matched.group(1)) * {"m": 60, "h": 3600, "d": 86400}[matched.group(2)] return end_at - timedelta(seconds=seconds) def credits_all_users(token): users = [] page = 1 while True: data = ensure_success(curl_api("GET", f"/api/v1/admin/users?page={page}&page_size=100", bearer=token), "list users") batch = extract_items(data) users.extend(item for item in batch if isinstance(item, dict)) total = int(data.get("total", len(users))) if isinstance(data, dict) else len(users) if not batch or len(batch) < 100 or page * 100 >= total: return users page += 1 def credits_active_usage(token, since, active_start, active_end, active_start_utc, active_end_utc, timezone_name): if since: end_at = datetime.now(timezone.utc) start_at = credits_window_start(since, end_at) selection_kind = "active-users" else: start_at = credits_parse_time(active_start_utc) end_at = credits_parse_time(active_end_utc) selection_kind = "active-window" if start_at is None or end_at is None or end_at <= start_at: raise RuntimeError("credits active window end must be after start") user_usage = {} page = 1 while True: path = f"/api/v1/admin/usage?page={page}&page_size=100&start_date={start_at.date().isoformat()}&end_date={end_at.date().isoformat()}&timezone=UTC&sort_by=created_at&sort_order=desc" data = ensure_success(curl_api("GET", path, bearer=token), "list active user usage") batch = extract_items(data) reached_start = False for row in batch: if not isinstance(row, dict): continue created_at = credits_parse_time(row.get("created_at")) if created_at is not None and created_at < start_at: reached_start = True continue user_id = row.get("user_id") if row.get("user_id") is not None else row.get("userId") if user_id is not None and created_at is not None and start_at <= created_at < end_at: usage = user_usage.setdefault(str(user_id), { "requestCount": 0, "inputTokens": 0, "outputTokens": 0, "tokenRows": 0, "missingTokenRows": 0, }) usage["requestCount"] += 1 has_tokens = row.get("input_tokens") is not None and row.get("output_tokens") is not None if has_tokens: usage["tokenRows"] += 1 else: usage["missingTokenRows"] += 1 usage["inputTokens"] += int(row.get("input_tokens") or 0) usage["outputTokens"] += int(row.get("output_tokens") or 0) total = int(data.get("total", 0)) if isinstance(data, dict) else 0 if not batch or len(batch) < 100 or reached_start or page * 100 >= total: return user_usage, start_at, end_at, { "kind": selection_kind, "since": since, "timezone": timezone_name if selection_kind == "active-window" else "UTC", "startAt": active_start if selection_kind == "active-window" else start_at.isoformat(), "endAt": active_end if selection_kind == "active-window" else end_at.isoformat(), "startAtUtc": start_at.isoformat().replace("+00:00", "Z"), "endAtUtc": end_at.isoformat().replace("+00:00", "Z"), } page += 1 def credits_excluded_selector(user, payload): user_id = str(user.get("id")) if user.get("id") is not None else "" email = str(user.get("email") or "").strip().lower() for selector in payload.get("excludedUsers") or []: normalized = str(selector).strip().lower() if normalized == user_id or normalized == email: return str(selector) return None def credits_usage_fields(usage): if not isinstance(usage, dict): return { "requestCount": None, "inputTokens": None, "outputTokens": None, "totalTokens": None, "tokenRows": None, "missingTokenRows": None, "tokenCompleteness": "unavailable", } input_tokens = int(usage.get("inputTokens") or 0) output_tokens = int(usage.get("outputTokens") or 0) missing_rows = int(usage.get("missingTokenRows") or 0) return { "requestCount": int(usage.get("requestCount") or 0), "inputTokens": input_tokens, "outputTokens": output_tokens, "totalTokens": input_tokens + output_tokens, "tokenRows": int(usage.get("tokenRows") or 0), "missingTokenRows": missing_rows, "tokenCompleteness": "complete" if missing_rows == 0 else "partial", } def credits_selected_users(token, payload): users = credits_all_users(token) users_by_id = {str(item.get("id")): item for item in users if item.get("id") is not None} users_by_email = {str(item.get("email") or "").strip().lower(): item for item in users if item.get("email")} active_since = payload.get("activeSince") active_start = payload.get("activeStart") active_end = payload.get("activeEnd") if active_since or (active_start and active_end): usage_by_user, _, _, selection = credits_active_usage( token, active_since, active_start, active_end, payload.get("activeStartUtc"), payload.get("activeEndUtc"), payload.get("timezone"), ) selected = [] excluded = [] for user_id, usage in usage_by_user.items(): user = users_by_id.get(user_id) usage_fields = credits_usage_fields(usage) excluded_selector = credits_excluded_selector(user, payload) if isinstance(user, dict) else None if not isinstance(user, dict): excluded.append({"userId": user_id, "reason": "user-not-found", **usage_fields}) elif excluded_selector is not None: excluded.append({ "userId": user.get("id"), "email": user.get("email"), "reason": "yaml-excluded-user", "matchedSelector": excluded_selector, **usage_fields, }) elif user.get("role") == "admin": excluded.append({"userId": user.get("id"), "email": user.get("email"), "reason": "admin-excluded", **usage_fields}) elif user.get("status") != "active" or user.get("deleted_at") is not None: excluded.append({"userId": user.get("id"), "email": user.get("email"), "reason": "inactive-or-deleted", **usage_fields}) else: selected.append((user, usage_fields)) return selected, excluded, selection selected = [] excluded = [] seen = set() for selector in payload.get("users") or []: user = users_by_id.get(str(selector)) or users_by_email.get(str(selector).strip().lower()) if not isinstance(user, dict): raise RuntimeError("exact user selector not found: " + str(selector)) user_id = str(user.get("id")) if user_id in seen: raise RuntimeError("duplicate user selector: " + str(selector)) seen.add(user_id) excluded_selector = credits_excluded_selector(user, payload) if excluded_selector is not None: excluded.append({ "userId": user.get("id"), "email": user.get("email"), "reason": "yaml-excluded-user", "matchedSelector": excluded_selector, **credits_usage_fields(None), }) else: selected.append((user, credits_usage_fields(None))) return selected, excluded, {"kind": "exact-users", "selectors": payload.get("users") or []} def run_credits_grant(): payload = json.loads(base64.b64decode(PAYLOAD_B64).decode("utf-8")) if PAYLOAD_B64 else {} _, token, _ = login() selected, excluded, selection = credits_selected_users(token, payload) amount = profit_decimal(payload.get("amountUsd")) confirm = payload.get("confirm") is True batch_identity = json.dumps({ "users": sorted(str(value) for value in (payload.get("users") or [])), "amountUsd": profit_decimal_text(amount), "reason": payload.get("reason"), }, ensure_ascii=False, sort_keys=True, separators=(",", ":")) batch_key = hashlib.sha256(batch_identity.encode("utf-8")).hexdigest()[:24] items = [] write_succeeded = 0 write_failed = 0 reconciled = 0 for user, usage_fields in sorted(selected, key=lambda item: (str(item[0].get("email") or ""), int(item[0].get("id") or 0))): before = profit_decimal(user.get("balance")) after = before + amount item = { "userId": user.get("id"), "email": user.get("email"), "role": user.get("role"), "status": user.get("status"), **usage_fields, "balanceBefore": profit_decimal_text(before), "amountUsd": profit_decimal_text(amount), "balanceAfter": profit_decimal_text(after), "write": {"attempted": False, "status": "dry-run", "error": None}, "reconciled": None, } if confirm: try: ensure_success(curl_api("POST", f"/api/v1/admin/users/{user.get('id')}/balance", bearer=token, payload={ "balance": float(amount), "operation": "add", "notes": payload.get("reason"), }, idempotency_key=f"unidesk-credit-{batch_key}-{user.get('id')}"), "add user compensation balance") item["write"] = {"attempted": True, "status": "succeeded", "error": None} write_succeeded += 1 actual = ensure_success(curl_api("GET", f"/api/v1/admin/users/{user.get('id')}", bearer=token), "read user balance after grant") actual_balance = profit_decimal(actual.get("balance") if isinstance(actual, dict) else None) item["actualBalance"] = profit_decimal_text(actual_balance) item["reconciled"] = actual_balance == after reconciled += 1 if item["reconciled"] else 0 except Exception as exc: item["write"] = {"attempted": True, "status": "failed", "error": str(exc)[:500]} item["reconciled"] = False write_failed += 1 items.append(item) selected_count = len(items) return { "ok": not confirm or (write_failed == 0 and reconciled == selected_count), "mode": "confirmed" if confirm else "dry-run", "mutation": confirm and write_succeeded > 0, "selection": selection, "reason": payload.get("reason"), "items": items, "excluded": excluded, "confirmationUserIds": [item.get("userId") for item in items], "summary": { "selected": selected_count, "excluded": len(excluded), "amountUsd": profit_decimal_text(amount), "totalAmountUsd": profit_decimal_text(amount * selected_count), "writeAttempted": selected_count if confirm else 0, "writeSucceeded": write_succeeded, "writeFailed": write_failed, "reconciled": reconciled, }, "valuesPrinted": False, } `;