pikasTech-unidesk/scripts/src/platform-infra-sub2api-codex/remote-python-credits.ts

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()
    exact_selectors = payload.get("users") or list((payload.get("allocations") or {}).keys())
    for selector in exact_selectors:
        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)))
    selection_kind = "exact-allocations" if payload.get("allocations") else "exact-users"
    return selected, excluded, {"kind": selection_kind, "selectors": exact_selectors}

def credits_ranked_amounts(selected, payload):
    tiers = payload.get("rankTiers") or []
    ranked = sorted(selected, key=lambda item: (
        -int(item[1].get("totalTokens") or 0),
        -int(item[1].get("requestCount") or 0),
        int(item[0].get("id") or 0),
    ))
    amounts = {}
    tier_index = 0
    tier_used = 0
    for rank, (user, _) in enumerate(ranked, start=1):
        while tier_index < len(tiers) - 1:
            count = tiers[tier_index].get("count")
            if count is None or tier_used < int(count):
                break
            tier_index += 1
            tier_used = 0
        tier = tiers[tier_index]
        amounts[str(user.get("id"))] = profit_decimal(tier.get("amountUsd"))
        tier_used += 1
    return ranked, amounts

def credits_exact_amounts(selected, payload):
    allocations = payload.get("allocations") or {}
    amounts = {}
    for user, _ in selected:
        user_id = str(user.get("id"))
        email = str(user.get("email") or "").strip().lower()
        matched = None
        for selector, value in allocations.items():
            normalized = str(selector).strip().lower()
            if normalized == user_id or normalized == email:
                matched = value
                break
        if matched is None:
            raise RuntimeError("allocation missing for selected user: " + user_id)
        amounts[user_id] = profit_decimal(matched)
    return amounts

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)
    uniform_amount = profit_decimal(payload.get("amountUsd")) if payload.get("amountUsd") is not None else None
    rank_tiers = payload.get("rankTiers") or []
    allocations = payload.get("allocations") or {}
    if rank_tiers:
        selected, amount_by_user = credits_ranked_amounts(selected, payload)
    elif allocations:
        amount_by_user = credits_exact_amounts(selected, payload)
    else:
        amount_by_user = {str(user.get("id")): uniform_amount for user, _ in selected}
    confirm = payload.get("confirm") is True
    exact_plan = {
        str(user.get("id")): profit_decimal_text(amount_by_user[str(user.get("id"))])
        for user, _ in selected
    }
    if rank_tiers or allocations:
        batch_identity_data = {
            "allocations": dict(sorted(exact_plan.items())),
            "reason": payload.get("reason"),
        }
    else:
        batch_identity_data = {
            "users": sorted(str(value) for value in (payload.get("users") or [])),
            "amountUsd": profit_decimal_text(uniform_amount),
            "reason": payload.get("reason"),
        }
    batch_identity = json.dumps(batch_identity_data, 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
    ranked_mode = bool(rank_tiers)
    ordered = selected if ranked_mode else sorted(selected, key=lambda item: (str(item[0].get("email") or ""), int(item[0].get("id") or 0)))
    for rank, (user, usage_fields) in enumerate(ordered, start=1):
        amount = amount_by_user[str(user.get("id"))]
        before = profit_decimal(user.get("balance"))
        after = before + amount
        item = {
            "rank": rank if ranked_mode else None,
            "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)
    total_amount = sum((amount_by_user[str(item.get("userId"))] for item in items), profit_decimal(0))
    summary_amount = profit_decimal_text(uniform_amount) if uniform_amount is not None else "varies"
    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],
        "confirmationAllocations": exact_plan if rank_tiers or allocations else {},
        "summary": {
            "selected": selected_count, "excluded": len(excluded), "amountUsd": summary_amount,
            "totalAmountUsd": profit_decimal_text(total_amount), "writeAttempted": selected_count if confirm else 0,
            "writeSucceeded": write_succeeded, "writeFailed": write_failed, "reconciled": reconciled,
        },
        "valuesPrinted": False,
    }
`;