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): end_at = datetime.now(timezone.utc) start_at = credits_window_start(since, end_at) 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 None or 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 page += 1 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") if active_since: requests, start_at, end_at = credits_active_usage(token, active_since) selected = [] excluded = [] for user_id, request_count in requests.items(): user = users_by_id.get(user_id) if not isinstance(user, dict): excluded.append({"userId": user_id, "reason": "user-not-found"}) 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, {"kind": "active-users", "since": active_since, "startAt": start_at.isoformat(), "endAt": end_at.isoformat()} selected = [] 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) selected.append((user, None)) return selected, [], {"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, } `;