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_requests = {} 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: user_requests[str(user_id)] = user_requests.get(str(user_id), 0) + 1 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_requests, 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_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): requests, _, _, 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, request_count in requests.items(): user = users_by_id.get(user_id) 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"}) elif excluded_selector is not None: excluded.append({ "userId": user.get("id"), "email": user.get("email"), "requestCount": request_count, "reason": "yaml-excluded-user", "matchedSelector": excluded_selector, }) elif user.get("role") == "admin": excluded.append({"userId": user.get("id"), "email": user.get("email"), "reason": "admin-excluded"}) 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"}) else: selected.append((user, request_count)) 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"), "requestCount": None, "reason": "yaml-excluded-user", "matchedSelector": excluded_selector, }) else: selected.append((user, 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, request_count 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"), "requestCount": request_count, "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, } `;