接口说明¶

def get_workers():
    FLOWER_API_HOST = 'http://xxxx:5555'
    url = f"{FLOWER_API_HOST}/workers?json=1"
    headers = {"content-type": "application/json",
               "Authorization": "Basic aWRxxxxxxxjM="}
    response = requests.request("get", url, json={}, headers=headers)
    if response.status_code not in [200, 201]:
        # print(f"调用{url}失败，返回信息如下:")
        if response.status_code:
            print(f"状态码：{response.status_code}")
        if response.text:
            print(response.text)
        return 0
    workers_info = response.json()
    worker_status_active = {}
    for w in workers_info['data']:
        if "download" in w['hostname']:
            worker_name=w['hostname'].split("@")[1]
            worker_status_active[worker_name]=[w['status'],w['active']]
    return worker_status_active

import requests
API_HOST = "http://localhost:8000"
TOKEN = "970e8f84xxxxxxx56050ba"
def call_api(method:str, api_path:str, params:any):
    #print(params)
    url = f"{API_HOST}{api_path}"
    headers = {"content-type": "application/json",
               "Authorization": f"Token {TOKEN}"}
    response = requests.request(method, url, json=params, headers=headers)
    if response.status_code not in [200, 201]:
        print(f"调用{url}失败，返回信息如下:")
        if response.status_code:
            print(f"状态码：{response.status_code}")
        if response.text:
            print(response.text.encode('utf-8'))
        return 0
    return response.json()


def download_runs(runs, worker, store_path,download_method, repeated=False):
    run_list = runs
    # Prepare API parameters
    params = {
        "worker": f"{worker}",
        "run_list": run_list,
        "download_max": len(run_list),
        "store_path": store_path,
        "download_method":download_method,
        "repeated_download" : repeated
    }
    # Call the API
    res = call_api("post", "/Task/base_download", params)

def run_download_status(runs):
    run_list = runs
    # Prepare API parameters
    params = {
        "runs": f"{runs}",
    }
    # Call the API
    res = call_api("post", "Base/downloadstatus", params)
    return res["undownload_runs"]

def run_filesize(runs):
    run_list = runs
    # Prepare API parameters
    params = {
        "runs": f"{runs}",
    }
    # Call the API
    res = call_api("post", "Base/filesize", params)
    return res["data"]["total_filesize"]