Get a weekly NPS read-out by survey

import io, json, time, zipfile
from datetime import datetime, timedelta, timezone
from clickruntime import on_trigger, credentials, http, notify

DATACENTER = "iad1"
NPS_QUESTION = "QID1"

def qx(method, path, **kw):
    resp = http.request(method, f"https://{DATACENTER}.qualtrics.com/API/v3/{path}",
                        headers={"X-API-TOKEN": credentials.secret("QUALTRICS_API_TOKEN")}, **kw)
    resp.raise_for_status()
    return resp

def export(survey_id, since):
    job = qx("POST", f"surveys/{survey_id}/export-responses",
             json={"format": "json", "compress": True, "startDate": since}).json()["result"]
    while (prog := qx("GET", f"surveys/{survey_id}/export-responses/{job['progressId']}").json()["result"])["status"] != "complete":
        time.sleep(2)
    zf = zipfile.ZipFile(io.BytesIO(qx("GET", f"surveys/{survey_id}/export-responses/{prog['fileId']}/file").content))
    return json.loads(zf.read(zf.namelist()[0]))["responses"]

@on_trigger
def run(event):
    since = (datetime.now(timezone.utc) - timedelta(days=7)).strftime("%Y-%m-%dT%H:%M:%SZ")
    surveys = [s for s in qx("GET", "surveys").json()["result"]["elements"] if s["isActive"]]
    lines = []
    for s in surveys:
        scores = [int(r["values"][NPS_QUESTION]) for r in export(s["id"], since) if NPS_QUESTION in r["values"]]
        nps = round(100 * (sum(x >= 9 for x in scores) - sum(x <= 6 for x in scores)) / len(scores)) if scores else None
        lines.append(f"{s['name']}: {len(scores)} responses, NPS {nps if nps is not None else '-'}")
    notify.email_me("Qualtrics: last week", "\n".join(lines))
    return {"surveys": len(lines)}