Python 和结构化 API 调用

Question

在运行用于将文件导出到文件夹以实现自动化目的的 python 脚本后，我收到的响应遇到了一些问题。该脚本应该从我们的数据中心找到一个文件，然后将该文件下载到一个文件夹中。每当我有一个工作脚本时，我计划使用 Windows 调度程序来协助自动化部分。

我调整了示例脚本的参数（第 8 行到第 20 行），也可以在此处找到（https://github.com/nrccua/file_export_sample/blob/master/file_export_sample.py）。但是当我运行脚本，我收到“没有文件可下载！”在终端中。我在这里缺少什么吗？我知道这是不准确的，因为我可以在我们的系统中看到所述文件。我已经使用了“Delivered”和“NotDeliverd”参数，但仍然在终端中收到相同的消息。

我测试了脚本的各个部分，并且能够成功登录和验证。所以我不确定我哪里错了。

脚本还有什么需要添加的吗？我正在使用 VS code 进行测试，workspace/acutal .py 文件中似乎没有问题或语法错误。

API 文档：https://helpcenter.encoura.org/hc/en-us/article_attachments/20370572326029

API Swagger：https://developers.encoura.org/

我期待收到可供下载的可用文件列表。相反，我收到一条消息，指出没有文件可供下载。

import requests
import json
import os
import re
from urllib.parse import urlparse, unquote
from pathlib import Path

# SET THE API URL
URL = "https://api.datalab.nrccua.org/v1"
# SET YOUR API KEY
API_KEY = "xxxx"
# SET YOUR ORG UID
ORGANIZATION_UID = "d1e01227-404d-439a-b7e8-12cfd11a6863"
# override this if you want
DOWNLOAD_DIR = os.path.dirname(__file__)
DOWNLOAD_DIR = Path(DOWNLOAD_DIR)
# SET USERNAME
USERNAME = "[email protected]"
# SET PASSWORD
PASSWORD = "xxxx"


def get_valid_filename(s):
    s = str(s).strip().replace(" ", "_")
    return re.sub(r"(?u)[^-\w.]", "", s)


# SET YOUR USERNAME AND PASSWORD
payload = {"userName": USERNAME, "password": PASSWORD, "acceptedTerms": True}

session = requests.Session()

# set the api key for the rest of the session
session.headers.update({"x-api-key": API_KEY})

# login
response_json = session.post(f"{URL}/login", data=json.dumps(payload)).json()

if "sessionToken" not in response_json:
    print(f"Couldn't find sessionToken in response json:\n {response_json}")

# set the authorization header for the rest of the session
session.headers.update({"Authorization": f"JWT {response_json['sessionToken']}"})

# payload to return list of files
get_exports_payload = {"status": "Delivered", "productKey": "score-reporter"}
response_json = session.get(
    f"{URL}/datacenter/exports", params=get_exports_payload, headers={"Organization": ORGANIZATION_UID},
).json()

# loop through results
files_to_download = []
for export in response_json:
    if "uid" in export:
        export_uid = export["uid"]
        # api route for download
        file_export_url = f"{URL}/datacenter/exports/{export_uid}/download"
        export_response_json = session.get(file_export_url, headers={"Organization": ORGANIZATION_UID}).json()
        if "downloadUrl" in export_response_json:
            files_to_download.append(export_response_json["downloadUrl"])

if len(files_to_download) == 0:
    print(f"No files to download!")
else:
    for file in files_to_download:
        parsed_url = urlparse(file)
        # get the file name from the url, unescape it, and then replace whitespace with underscore
        escaped_filename = get_valid_filename(unquote(os.path.basename(parsed_url.path)))
        download_path = DOWNLOAD_DIR / escaped_filename
        print(f"Downloading file from url {file}")
        # don't use the session here
        download_file_response = requests.get(file, allow_redirects=True, stream=True)
        if download_file_response.ok:
            print(f"Writing file to {download_path}.")
            with open(download_path, "wb") as f:
                # we are going to chunk the download because we don't know how large the files are
                for chunk in download_file_response.iter_content(chunk_size=1024):
                    if chunk:
                        f.write(chunk)
        else:
            print(f"There was an error retrieving {file} with status code {download_file_response.status_code}.")
            print(f"{download_file_response.content}")

Answer 1

目前尚不清楚问题到底是什么，但我可能会首先通过测试 Swagger 网站本身的 API 调用来确保您看到预期的项目。

Swagger 页面可让您使用您的帐户实际测试 API 调用，只需选择 Authorize 按钮并将您的信用信息放入其中：

您应该从那里测试

/datacenter/exports

和

/datacenter/exports/{uid}/download

API 调用，以确保返回的数据符合您的预期：

如果您发现问题，可以从那里开始。

如果看到所需的数据，则可以返回调试 Python 代码。我将通过简单地在循环中添加一些

print()

语句来开始调试，如下所示：

if "uid" in export:
    export_uid = export["uid"]
    print("#### Found uid:", export_uid)

...并且：

if "downloadUrl" in export_response_json:
    print("#### Found download for uid:", export_uid)
    files_to_download.append(export_response_json["downloadUrl"])

...并且：

# loop through results
files_to_download = []
print("#### response_json is length:", len(response_json))