📚 课程目录第4章队列数据采集系统对接

4. 队列数据采集系统对接

临床研究队列的数据往往并不直接产生在分析环境里，而是先进入 EDC（Electronic Data Capture，电子数据采集）系统。REDCap 与 OpenClinica 是两类常见选择：前者以项目化、表单化和成熟的 API 著称，后者更强调符合 CDISC 与法规要求的临床数据管理流程。队列工程要做的，不是手工导出 CSV 再上传，而是通过 API 建立可重复、可增量、可审计的拉取通道。本节围绕“认证—发现—拉取—增量—落库”这条主线，说明如何用 requests 与 pycap 完成自动化同步。

一、先理解 EDC API 的两种交互模型

REDCap 的 API 是典型的“动作型”接口：所有请求都发往同一个 /api/ 端点，通过 content 参数声明要做什么，例如 content=metadata 取数据字典、content=record 取记录、content=file 取上传文件。认证靠请求体中的 token，返回格式由 format=json 控制。OpenClinica 则更接近 REST 风格，资源路径区分研究对象、事件、表单与数据项，认证通常使用用户名密码换取会话，再以会话标识访问 /OpenClinica/rest/ 下的资源。理解这一差异，决定了代码是“一个端点加参数”还是“多资源路径加会话”。

pycap 是对 REDCap API 的 Python 封装，核心对象是 Project。它把 token、URL 和常用动作封装成方法，例如 project.export_records()、project.metadata、project.export_file()。对于 OpenClinica，社区封装不如 REDCap 统一，实践中更常见的是用 requests.Session 自行维护会话并解析返回的 XML 或 JSON。因此本节把 pycap 作为 REDCap 的主路径，把 requests 作为通用底座与 OpenClinica 的实现方式。

二、用 pycap 拉取 REDCap 数据

REDCap 项目管理员在“API”页面生成 token 后，即可在 Python 中建立项目对象。下面示例中的 URL、token 与字段名均为教学占位，实际使用时替换为真实值：

import pycap
import pandas as pd

project = pycap.Project(
    url="https://redcap.example.org/api/",
    token="ABCDEF0123456789ABCDEF0123456789"
)

# 1. 取数据字典，确认字段名与表单结构
meta = project.metadata
fields = [f["field_name"] for f in meta]
print("字段数:", len(fields))

# 2. 拉取全部记录，返回 list[dict]
records = project.export_records(
    format="json",
    raw_or_label="raw",
    export_checkbox_label=True
)
df = pd.DataFrame(records)
print(df.shape)

这里有两个关键点。第一，raw_or_label="raw" 返回编码值而非标签，适合后续统计；若需要可读标签，改为 label。第二，复选框字段在 REDCap 中会展开为多个 字段名___选项 列，导出后需要按数据字典还原为分类变量。若项目启用了纵向事件，export_records 返回的每条记录会带 redcap_event_name，此时主键应视为 record_id + redcap_event_name 的组合。

三、用 requests 对接 OpenClinica

OpenClinica 的 REST 接口通常先建立会话，再按资源路径取数。下面示例演示登录、取研究对象与表单数据的基本形态，路径与参数需按实际部署版本调整：

import requests

BASE = "https://oc.example.org/OpenClinica"
s = requests.Session()
s.auth = ("researcher", "password")

# 1. 校验会话
r = s.get(f"{BASE}/rest/clinicaldata/json/view/S_DEMO/SE_DEMO",
          params={"studyOID": "S_DEMO"})
r.raise_for_status()

# 2. 取某研究对象的表单数据
subject = "SS_001"
r2 = s.get(f"{BASE}/rest/clinicaldata/json/view/S_DEMO/SE_DEMO/{subject}")
data = r2.json()
print(data["ClinicalData"]["SubjectData"]["StudyEventData"].keys())

与 REDCap 不同，OpenClinica 返回结构嵌套较深，通常需要按 StudyEventData → FormData → ItemGroupData → ItemData 逐层展开，再拍平成一行一记录的表。建议先写一个 flatten_oc() 函数，把嵌套 JSON 转为扁平字典，再统一进入落库流程。

四、增量同步：不要每次全量拉取

全量拉取在队列早期可行，但记录数上千后既慢又浪费。增量同步的核心是维护一个“上次同步水位”。REDCap 的 export_records 支持按 dateRangeBegin 与 dateRangeEnd 过滤，可基于记录的创建或修改时间拉取；OpenClinica 则可通过审计事件或按研究对象状态筛选。更稳妥的做法是本地维护 sync_state 表，记录每个项目的 last_sync_time 与已同步的 record_id 集合。

from datetime import datetime, timedelta

last_sync = datetime(2025, 3, 1, 0, 0, 0)
now = datetime.now()

new_records = project.export_records(
    format="json",
    dateRangeBegin=last_sync.strftime("%Y-%m-%d %H:%M:%S"),
    dateRangeEnd=now.strftime("%Y-%m-%d %H:%M:%S")
)

# 按 record_id 去重后写入
seen = set()
for rec in new_records:
    rid = rec["record_id"]
    if rid in seen:
        continue
    seen.add(rid)
    upsert_to_db(rec)

update_sync_state(project_name="demo", last_sync_time=now)

增量同步还要处理三类边界情况：一是记录被修改但时间戳未变，需辅以字段级哈希比对；二是删除的记录不会出现在增量结果中，需定期做一次主键对账；三是网络中断导致部分写入，需用事务或幂等写入保证不重复。建议每次同步都记录 batch_id、拉取条数、写入条数与失败条数，形成可追溯的同步日志。

五、字段映射与数据质量检查

EDC 字段名往往带有项目缩写，直接进入分析会降低可读性。应在 staging 层建立映射表，把 redcap_event_name、record_id、各表单字段映射为分析变量名，并统一日期格式与缺失值编码。同步完成后至少做三项检查：主键是否唯一、必填字段缺失率是否异常、数值字段是否落在合理范围。若发现某次同步缺失率突然升高，应优先排查 API 返回是否被截断或权限是否变更，而不是直接进入分析。

建议把 token 与密码放入环境变量或密钥管理服务，不要硬编码在脚本中；同步脚本应支持 dry-run 模式，先打印将写入的条数与主键，再执行真实写入。

把上述环节串起来，队列工程就拥有了一个可重复执行的 EDC 同步通道：认证负责安全接入，发现负责理解结构，拉取负责获取数据，增量负责控制成本，落库与检查负责保证质量。下一阶段即可在此基础上构建清洗规则与衍生变量。