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)