LongArticleTaskServer/tasks/monitor_tasks/kimi_balance.py

import traceback
from typing import Dict
from applications.api import feishu_robot
from applications.utils import AsyncHttPClient

# const
BALANCE_LIMIT_THRESHOLD = 100.0


async def check_kimi_balance() -> Dict:
    url = "https://api.moonshot.cn/v1/users/me/balance"
    headers = {
        "Authorization": "Bearer sk-5DqYCa88kche6nwIWjLE1p4oMm8nXrR9kQMKbBolNAWERu7q",
        "Content-Type": "application/json; charset=utf-8",
    }
    async with AsyncHttPClient() as client:
        response = await client.get(url, headers=headers)

    try:
        balance = response["data"]["available_balance"]
        if balance < BALANCE_LIMIT_THRESHOLD:
            await feishu_robot.bot(
                title="kimi余额小于 {} 块".format(BALANCE_LIMIT_THRESHOLD),
                detail={"balance": balance},
            )
    except Exception as e:
        error_stack = traceback.format_exc()
        await feishu_robot.bot(
            title="kimi余额接口处理失败，数据结构异常",
            detail={"error": str(e), "error_msg": error_stack},
        )

    return response