Data Collection Automation Guide

A skill for automating research data collection, survey deployment, and data pipeline management. Covers survey platform APIs, automated data retrieval, quality checks, ETL pipelines, and scheduling for longitudinal studies.

Survey Platform APIs

Qualtrics API

import os
import json
import urllib.request
import time


def export_qualtrics_responses(survey_id: str,
                                file_format: str = "csv") -> str:
    """
    Export survey responses from Qualtrics via API.

    Args:
        survey_id: The Qualtrics survey ID (SV_...)
        file_format: Export format (csv, json, spss)
    """
    api_token = os.environ["QUALTRICS_API_TOKEN"]
    data_center = os.environ["QUALTRICS_DATACENTER"]
    base_url = f"https://{data_center}.qualtrics.com/API/v3"

    headers = {
        "X-API-TOKEN": api_token,
        "Content-Type": "application/json"
    }

    # Step 1: Start export
    export_data = json.dumps({
        "format": file_format,
        "compress": False
    }).encode("utf-8")

    req = urllib.request.Request(
        f"{base_url}/surveys/{survey_id}/export-responses",
        data=export_data,
        headers=headers
    )
    response = json.loads(urllib.request.urlopen(req).read())
    progress_id = response["result"]["progressId"]

    # Step 2: Poll for completion
    status = "inProgress"
    while status == "inProgress":
        time.sleep(2)
        req = urllib.request.Request(
            f"{base_url}/surveys/{survey_id}/export-responses/{progress_id}",
            headers=headers
        )
        check = json.loads(urllib.request.urlopen(req).read())
        status = check["result"]["status"]

    file_id = check["result"]["fileId"]

    # Step 3: Download file
    req = urllib.request.Request(
        f"{base_url}/surveys/{survey_id}/export-responses/{file_id}/file",
        headers=headers
    )
    file_data = urllib.request.urlopen(req).read()

    output_path = f"responses_{survey_id}.{file_format}"
    with open(output_path, "wb") as f:
        f.write(file_data)

    return output_path

REDCap API

def export_redcap_records(api_url: str, fields: list[str] = None) -> list:
    """
    Export records from a REDCap project.

    Args:
        api_url: REDCap API endpoint URL
        fields: List of field names to export (None = all fields)
    """
    api_token = os.environ["REDCAP_API_TOKEN"]

    data = {
        "token": api_token,
        "content": "record",
        "format": "json",
        "type": "flat"
    }

    if fields:
        data["fields"] = ",".join(fields)

    encoded = urllib.parse.urlencode(data).encode("utf-8")
    req = urllib.request.Request(api_url, data=encoded)
    response = urllib.request.urlopen(req)

    return json.loads(response.read())