200 lines
7.8 KiB
Python
200 lines
7.8 KiB
Python
import time
|
||
import logging
|
||
import requests
|
||
import json
|
||
from datetime import datetime
|
||
from KeLing.Kit.KlCommon import KlCommon
|
||
from KeLing.Kit.KlErrorCode import KlErrorCode
|
||
|
||
# 配置日志
|
||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
|
||
log = logging.getLogger(__name__)
|
||
|
||
class KlAccount(KlCommon):
|
||
BASE_URL = "https://api.klingai.com"
|
||
ACCOUNT_COSTS_PATH = "/account/costs"
|
||
|
||
@staticmethod
|
||
def query_resource_packs(start_time, end_time, resource_pack_name=None):
|
||
"""
|
||
查询账号下资源包列表及余量
|
||
|
||
:param start_time: 查询的开始时间，Unix时间戳、单位ms
|
||
:param end_time: 查询的结束时间，Unix时间戳、单位ms
|
||
:param resource_pack_name: 资源包名称，用于精准指定查询某个资源包，可为None
|
||
:return: 资源包信息列表
|
||
:raises Exception: 异常信息
|
||
"""
|
||
# 获取JWT令牌
|
||
jwt = KlAccount.get_jwt()
|
||
|
||
# 构建URL
|
||
url = f"{KlAccount.BASE_URL}{KlAccount.ACCOUNT_COSTS_PATH}?start_time={start_time}&end_time={end_time}"
|
||
|
||
# 如果提供了资源包名称，则添加到URL中
|
||
if resource_pack_name:
|
||
url += f"&resource_pack_name={resource_pack_name}"
|
||
|
||
# 发送GET请求
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"Authorization": f"Bearer {jwt}"
|
||
}
|
||
|
||
log.info(f"查询账户资源包请求URL：{url}")
|
||
response = requests.get(url, headers=headers)
|
||
|
||
# 检查响应状态码
|
||
if response.status_code != 200:
|
||
raise Exception(f"请求失败，状态码：{response.status_code}")
|
||
|
||
# 解析响应
|
||
response_body = response.text
|
||
try:
|
||
response_json = json.loads(response_body)
|
||
except json.JSONDecodeError as e:
|
||
raise Exception(f"响应解析失败：{str(e)}")
|
||
|
||
log.info(f"查询账户资源包响应：{response_body}")
|
||
|
||
# 检查响应状态
|
||
code = response_json.get("code")
|
||
if code != 0:
|
||
message = response_json.get("message", "未知错误")
|
||
solution = KlErrorCode.get_solution_by_code(code)
|
||
error_msg = f"查询账户资源包失败：[{code}] {message} - {solution}"
|
||
raise Exception(error_msg)
|
||
|
||
# 解析资源包信息
|
||
resource_pack_info_list = []
|
||
data = response_json.get("data", {})
|
||
|
||
# 检查data中的code
|
||
data_code = data.get("code")
|
||
if data_code != 0:
|
||
data_msg = data.get("msg", "未知错误")
|
||
raise Exception(f"查询账户资源包失败：{data_msg}")
|
||
|
||
# 获取资源包列表
|
||
resource_pack_array = data.get("resource_pack_subscribe_infos", [])
|
||
for resource_pack_json in resource_pack_array:
|
||
resource_pack_info = ResourcePackInfo()
|
||
|
||
resource_pack_info.resource_pack_name = resource_pack_json.get("resource_pack_name")
|
||
resource_pack_info.resource_pack_id = resource_pack_json.get("resource_pack_id")
|
||
resource_pack_info.resource_pack_type = resource_pack_json.get("resource_pack_type")
|
||
resource_pack_info.total_quantity = resource_pack_json.get("total_quantity", 0.0)
|
||
resource_pack_info.remaining_quantity = resource_pack_json.get("remaining_quantity", 0.0)
|
||
resource_pack_info.purchase_time = resource_pack_json.get("purchase_time", 0)
|
||
resource_pack_info.effective_time = resource_pack_json.get("effective_time", 0)
|
||
resource_pack_info.invalid_time = resource_pack_json.get("invalid_time", 0)
|
||
resource_pack_info.status = resource_pack_json.get("status")
|
||
|
||
resource_pack_info_list.append(resource_pack_info)
|
||
|
||
return resource_pack_info_list
|
||
|
||
@staticmethod
|
||
def print_resource_pack_info(resource_pack_info_list):
|
||
"""
|
||
打印资源包信息
|
||
|
||
:param resource_pack_info_list: 资源包信息列表
|
||
"""
|
||
if not resource_pack_info_list:
|
||
log.info("未查询到资源包信息")
|
||
return
|
||
|
||
log.info("资源包信息列表：")
|
||
for info in resource_pack_info_list:
|
||
log.info("----------------------------------------")
|
||
log.info(f"资源包名称: {info.resource_pack_name}")
|
||
log.info(f"资源包ID: {info.resource_pack_id}")
|
||
log.info(f"资源包类型: {KlAccount.get_resource_pack_type_desc(info.resource_pack_type)}")
|
||
log.info(f"总量: {info.total_quantity}")
|
||
log.info(f"余量: {info.remaining_quantity}")
|
||
log.info(f"购买时间: {KlAccount.format_timestamp(info.purchase_time)}")
|
||
log.info(f"生效时间: {KlAccount.format_timestamp(info.effective_time)}")
|
||
log.info(f"失效时间: {KlAccount.format_timestamp(info.invalid_time)}")
|
||
log.info(f"状态: {KlAccount.get_status_desc(info.status)}")
|
||
log.info("----------------------------------------")
|
||
|
||
@staticmethod
|
||
def get_resource_pack_type_desc(type_str):
|
||
"""
|
||
获取资源包类型描述
|
||
|
||
:param type_str: 资源包类型
|
||
:return: 资源包类型描述
|
||
"""
|
||
if type_str == "decreasing_total":
|
||
return "总量递减型"
|
||
elif type_str == "constant_period":
|
||
return "周期恒定型"
|
||
else:
|
||
return type_str
|
||
|
||
@staticmethod
|
||
def get_status_desc(status):
|
||
"""
|
||
获取资源包状态描述
|
||
|
||
:param status: 资源包状态
|
||
:return: 资源包状态描述
|
||
"""
|
||
status_map = {
|
||
"toBeOnline": "待生效",
|
||
"online": "生效中",
|
||
"expired": "已到期",
|
||
"runOut": "已用完"
|
||
}
|
||
return status_map.get(status, status)
|
||
|
||
@staticmethod
|
||
def format_timestamp(timestamp):
|
||
"""
|
||
格式化时间戳
|
||
|
||
:param timestamp: 时间戳（毫秒）
|
||
:return: 格式化后的时间字符串
|
||
"""
|
||
if timestamp == 0:
|
||
return "N/A"
|
||
# 转换毫秒时间戳为秒
|
||
seconds = timestamp / 1000
|
||
return datetime.fromtimestamp(seconds).strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
class ResourcePackInfo:
|
||
"""资源包信息类"""
|
||
def __init__(self):
|
||
self.resource_pack_name = None # 资源包名称
|
||
self.resource_pack_id = None # 资源包ID
|
||
self.resource_pack_type = None # 资源包类型
|
||
self.total_quantity = 0.0 # 总量
|
||
self.remaining_quantity = 0.0 # 余量
|
||
self.purchase_time = 0 # 购买时间
|
||
self.effective_time = 0 # 生效时间
|
||
self.invalid_time = 0 # 失效时间
|
||
self.status = None # 资源包状态
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
# 获取当前时间
|
||
current_time = int(time.time() * 1000) # 毫秒时间戳
|
||
# 设置查询时间范围为过去30天到当前
|
||
startTime = current_time - 30 * 24 * 60 * 60 * 1000
|
||
endTime = current_time
|
||
|
||
# 查询所有资源包
|
||
resource_pack_info_list = KlAccount.query_resource_packs(startTime, endTime)
|
||
|
||
# 打印资源包信息
|
||
KlAccount.print_resource_pack_info(resource_pack_info_list)
|
||
|
||
# 如果需要查询特定资源包，可以使用以下代码
|
||
# specific_pack_name = "视频生成-10000条"
|
||
# specific_resource_pack_info_list = KlAccount.query_resource_packs(startTime, endTime, specific_pack_name)
|
||
# KlAccount.print_resource_pack_info(specific_resource_pack_info_list)
|
||
|
||
except Exception as e:
|
||
log.error(f"查询账户资源包异常: {str(e)}", exc_info=True) |