通用 API 快速入门

const response = await fetch('https://api.kie.ai/api/v1/chat/credit', {
  method: 'GET',
  headers: {
    'Authorization': 'Bearer YOUR_API_KEY'
  }
});

const result = await response.json();
console.log('Current credits:', result.data);

import requests

url = "https://api.kie.ai/api/v1/chat/credit"
headers = {
    "Authorization": "Bearer YOUR_API_KEY"
}

response = requests.get(url, headers=headers)
result = response.json()

print(f"Current credits: {result['data']}")

const response = await fetch('https://api.kie.ai/api/v1/common/download-url', {
  method: 'POST',
  headers: {
    'Authorization': 'Bearer YOUR_API_KEY',
    'Content-Type': 'application/json'
  },
  body: JSON.stringify({
    url: 'https://tempfile.1f6cxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxbd98'
  })
});

const result = await response.json();
console.log('Download URL:', result.data);

import requests

url = "https://api.kie.ai/api/v1/common/download-url"
headers = {
    "Authorization": "Bearer YOUR_API_KEY",
    "Content-Type": "application/json"
}

payload = {
    "url": "https://tempfile.1f6cxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxbd98"
}

response = requests.post(url, json=payload, headers=headers)
result = response.json()

print(f"Download URL: {result['data']}")

import requests
import time
import os
from datetime import datetime
from typing import Optional

class KieAIClient:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = 'https://api.kie.ai'
        self.headers = {
            'Authorization': f'Bearer {api_key}'
        }
    
    def get_credits(self) -> int:
        """获取当前账户积分"""
        response = requests.get(
            f'{self.base_url}/api/v1/chat/credit',
            headers=self.headers
        )
        
        if not response.ok:
            raise Exception(f'Failed to get credits: {response.text}')
        
        result = response.json()
        return result['data']
    
    def get_download_url(self, file_url: str) -> str:
        """获取生成文件的临时下载链接"""
        response = requests.post(
            f'{self.base_url}/api/v1/common/download-url',
            headers={**self.headers, 'Content-Type': 'application/json'},
            json={'url': file_url}
        )
        
        if not response.ok:
            raise Exception(f'Failed to get download URL: {response.text}')
        
        result = response.json()
        return result['data']
    
    def download_file(self, file_url: str, output_path: str) -> None:
        """从 kie.ai URL 下载文件"""
        # 获取下载链接
        download_url = self.get_download_url(file_url)
        
        # 下载文件
        response = requests.get(download_url)
        
        if not response.ok:
            raise Exception(f'Failed to download file: {response.text}')
        
        # 保存到文件
        os.makedirs(os.path.dirname(output_path), exist_ok=True)
        with open(output_path, 'wb') as f:
            f.write(response.content)
        
        print(f'File downloaded to: {output_path}')
    
    def check_credits_and_warn(self, threshold: int = 10) -> bool:
        """检查积分，如果低于阈值则发出警告"""
        credits = self.get_credits()
        
        if credits < threshold:
            print(f'⚠️  Low credits warning: {credits} credits remaining')
            return False
        
        print(f'✓ Credits available: {credits}')
        return True

# 使用示例
def main():
    client = KieAIClient('YOUR_API_KEY')
    
    # 操作前监控积分
    def run_with_credit_check():
        has_enough_credits = client.check_credits_and_warn(threshold=20)
        
        if not has_enough_credits:
            print('Insufficient credits. Please recharge your account.')
            return
        
        print('Credits verified. Proceeding with operations...')
    
    # 下载生成的文件
    def download_generated_files(file_urls: list):
        for i, file_url in enumerate(file_urls):
            try:
                client.download_file(
                    file_url,
                    f'./downloads/file-{i + 1}.mp4'
                )
                print(f'✓ Downloaded file {i + 1}/{len(file_urls)}')
            except Exception as e:
                print(f'✗ Failed to download file {i + 1}: {e}')
    
    # 定期积分监控
    def monitor_credits(interval_minutes: int = 60):
        while True:
            try:
                credits = client.get_credits()
                timestamp = datetime.now().isoformat()
                print(f'[{timestamp}] Current credits: {credits}')
                
                if credits < 50:
                    # 发送警报 (邮件, webhook 等)
                    print('ALERT: Credits below 50!')
            except Exception as e:
                print(f'Credit check failed: {e}')
            
            time.sleep(interval_minutes * 60)
    
    # 示例执行
    print('Checking credits...')
    run_with_credit_check()
    
    print('\nDownloading files...')
    file_urls = [
        'https://tempfile.1f6cxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxbd98',
        'https://tempfile.2f7dxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxcd99'
    ]
    download_generated_files(file_urls)

if __name__ == '__main__':
    main()