red-future/data-engine
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
32bd6c8895bba163ec6725223f2095534609206c
data-engine/docs/CONFIG_GUIDE.md

19 KiB
Raw Blame History

数据引擎 — 平台与接口配置指南

适用版本：HDWL data-engine
阅读对象：实施工程师、运维人员
前置知识：了解 JSON 基本格式即可

目录

  1. 概念说明
  2. 平台管理 (api_datasource_platform)
  3. 接口管理 (api_interface)
  4. 实战：新增一个平台
  5. 常见问题

1. 概念说明

平台 (Platform)  →  对接的第三方系统（如：快手电商、腾讯广告、钉钉）
  └── 接口 (Interface) → 平台提供的具体 API（如：订单列表、商品详情）
        └── 表 (Table) → 数据最终存储的数据库表

举例：

  • 平台 = 快手电商
    • 接口1 = 订单列表 → 数据存入 kuaishou_order_list 表
    • 接口2 = 商品列表 → 数据存入 kuaishou_item_list 表
    • 接口3 = 售后单列表 → 数据存入 kuaishou_refund_list 表

系统会按照配置的调度周期，自动拉取每个接口的数据并存入对应的数据库表。

2. 平台管理

平台配置存储在 api_datasource_platform 表中，定义了一个第三方平台的基本信息和认证方式。

2.1 基础字段

字段 类型 必填 说明 示例
platform_code VARCHAR ✅ 平台唯一编码，不能重复 kuaishou、tencent
platform_name VARCHAR ✅ 平台显示名称 快手电商、腾讯广告
description VARCHAR ❌ 平台描述 快手电商开放平台数据同步
status VARCHAR ✅ 状态，ACTIVE=启用，INACTIVE=停用 ACTIVE
api_base_url VARCHAR ✅ API 域名，接口地址会拼接在此 URL 后面 https://openapi.kwaixiaodian.com

2.2 认证类型 (auth_type)

类型 说明 适用场景
TOKEN 简单 Token 认证，放在 Header 中 内部系统
API_KEY API Key 认证，Token 放在请求参数中 快手电商、钉钉
OAUTH2 OAuth 2.0 认证 腾讯广告
SIGN 签名认证，使用 appKey + appSecret 签名 部分开放平台
APP_SIGNATURE 应用签名认证，基于请求体 MD5 钉钉智能薪酬

2.3 认证配置 (auth_config)

auth_config 是一个 JSON 对象，根据不同的 auth_type 配置不同内容。

API_KEY 类型配置（快手电商示例）

{
  "sign_algorithm": "md5",
  "app_key": "ks651333099611149957",
  "app_secret": "JPUXG2CS3I7tqRWbKaLrYQ",
  "sign_secret": "7bc51baab818cf86e121a48d99ff3fe4",
  "token_in_query": true,
  "query_key": "access_token",
  "extra_query_params": {
    "timestamp": "{timestamp_ms}"
  }
}
字段 说明
sign_algorithm 签名算法，可选 md5（小写输出）或 md5_upper（大写输出）或 HMAC_SHA256
app_key 平台分配的 AppKey（也叫 AppId、client_id），快手签名必填
app_secret 平台分配的 AppSecret，用于 OAuth 获取 Token
sign_secret 签名专用密钥（快手特有），平台分配，在"应用详情"中查看
token_in_query true=Token 放在 URL 参数中；false=Token 放在 Header 中
query_key Token 的参数名，默认 access_token
extra_query_params 额外需要拼接的参数，支持 {timestamp}（秒）、{timestamp_ms}（毫秒）、{nonce} 占位符

快手签名算法（仅供参考，代码已实现）：

  1. 所有参数按字母排序，拼接成 key1=value1&key2=value2&...
  2. 末尾追加 &signSecret=你的signSecret
  3. 对整个字符串取 MD5

SIGN 类型配置

{
  "sign_algorithm": "md5",
  "app_key": "your_app_key",
  "app_secret": "your_app_secret"
}

2.4 限流与重试字段

字段 类型 默认值 说明
rate_limit_per_minute INT 100 每分钟最大请求次数，防止触发平台限流
rate_limit_per_hour INT 3600 每小时最大请求次数
concurrency_limit INT 5 并发请求数
request_timeout_ms INT 30000 单次请求超时时间（毫秒）
max_retries INT 3 请求失败后的最大重试次数
retry_delay_ms INT 1000 重试间隔（毫秒），每次翻倍

2.5 Token 与 API Key 字段

字段 说明 快手示例
token 认证 Token / Refresh Token 快手 OAuth 返回的 refresh_token
api_key API Key / Access Token 快手 OAuth 返回的 access_token
client_id OAuth2 Client ID 腾讯广告的 client_id
client_secret OAuth2 Client Secret 腾讯广告的 client_secret

3. 接口管理

接口配置存储在 api_interface 表中，定义了一个平台下每个 API 的请求方式、参数、响应解析规则和存储表结构。

3.1 基础字段

字段 类型 必填 说明
platform_id INT ✅ 所属平台的 ID（关联 api_datasource_platform.id）
name VARCHAR ✅ 接口名称，如"订单列表"
code VARCHAR ✅ 接口编码，唯一标识，如 order_list
url VARCHAR ✅ API 路径，会拼接在平台 api_base_url 后面
method VARCHAR ✅ HTTP 方法，GET 或 POST
status VARCHAR ✅ 状态，active=启用，inactive=停用
auth_type VARCHAR ✅ 认证方式，inherit=继承平台配置

3.2 请求配置 (request_config)

request_config 是一个 JSON 对象，定义如何构造 API 请求参数。

3.2.1 完整字段说明

{
  "page_param": "cursor",
  "page_size_param": "pageSize",
  "cursor_pagination": true,
  "pagination_mode": "offset",
  "page_size": 50,
  "initial_cursor": "",
  "method": "open.order.cursor.list",
  "version": 1,
  "signMethod": "MD5",
  "time_field": "updateTime",
  "time_field_mode": "range",
  "full_sync_start_time": 1700000000000,
  "queryType": 2,
  "parameters_location": "query",
  "body_wrapper_field": "param",
  "exclude_from_wrapper": ["method", "version", "signMethod"],
  "prefetch": { ... },
  "recursive": { ... },
  "row_inject": ["statisticsMonth"],
  "max_recursive_depth": 20
}

3.2.2 分页配置

字段 说明 示例
page_param 页码/游标的参数名 cursor、page、offset
page_size_param 每页条数的参数名 pageSize、page_size
page_size 每页请求条数，覆盖全局配置 50
cursor_pagination true=游标分页；不配置=普通页码分页 true
pagination_mode offset=偏移量分页（offset=(page-1)*size）；不配置=页码分页 offset
initial_cursor 游标分页时的初始游标值，不配则为空字符串 ""

三种分页模式对比：

模式 配置方式 适用平台 翻页方式
游标分页 cursor_pagination: true 快手、钉钉 响应返回 cursor，下次请求带上
普通分页 默认 腾讯广告 响应返回 totalPages，按页码翻
hasMore 分页 has_more_field 在 response_config 钉钉 响应有 hasMore 字段判断是否有下一页

3.2.3 时间过滤配置

用于增量同步——只拉取上次同步之后新增/更新的数据。

字段 说明
time_field 时间字段名，如 updateTime、createTime
time_field_mode 时间模式：range=快手模式(beginTime/endTime)，filtering=腾讯模式(filtering 数组)
full_sync_start_time 首次全量同步的起始时间戳（Unix 毫秒），不配则用全局 default_lookback_days
queryType 快手专用：1=按创建时间查(90天内)，2=按更新时间查(240天内)

全局配置 config.yml 中 sync.default_lookback_days 控制默认回溯天数（默认 89 天）。

3.2.4 业务参数

除了上述系统级字段外，request_config 中其他字段都会作为业务参数发送给 API。

快手示例（orderViewStatus、cpsType、sort 等是业务参数）：

{
  "orderViewStatus": 1,
  "cpsType": 1,
  "sort": 1,
  "pageSize": 50
}

3.2.5 body_wrapper_field（快手专用）

快手 API 要求所有业务参数打包成一个 JSON 字符串，放在 param 字段中发送。

{
  "body_wrapper_field": "param",
  "exclude_from_wrapper": ["method", "version", "signMethod"]
}

效果：

  • 业务参数 orderViewStatus=1, pageSize=50, sort=1 → 打包为 param={"orderViewStatus":1,"pageSize":50,"sort":1}
  • method、version、signMethod 保持在顶层，不被打包

3.2.6 预取配置 (prefetch)

用于需要先获取实体列表，再逐个查详情的场景。

{
  "prefetch": {
    "url": "/open/order/cursor/list",
    "method": "GET",
    "response_path": "data.orderList",
    "target_param": "oid",
    "value_field": "oid"
  }
}
字段 说明
url 预取数据的 API 路径
method HTTP 方法
response_path 从响应中提取实体列表的 JSON 路径，如 data.orderList
target_param 将提取的值传给目标接口时使用的参数名
value_field 从实体中取哪个字段的值，不配则传整个实体对象

工作流程示例（订单详情）：

  1. 先调用 /open/order/cursor/list 获取所有订单 ID 列表
  2. 对每个订单 ID，调用 /open/order/detail?oid=xxx 获取详情
  3. 所有详情数据合并存入 kuaishou_order_detail 表

3.2.7 递归遍历配置 (recursive)

用于树形结构数据（如钉钉部门树）。

{
  "recursive": {
    "key_field": "dept_id",
    "target_param": "parent_id"
  }
}
字段 说明
key_field 当前节点的 ID 字段名
target_param 传给下级查询的参数名
max_recursive_depth 最大递归深度，默认 20

3.2.8 参数位置控制

字段 说明
parameters_location query=参数放 URL 查询字符串；不配=GET 放 URL，POST 放 Body

3.2.9 字段注入 (row_inject)

如果响应中缺少请求参数中的某些字段，可以用 row_inject 注入到每行数据中。

{
  "row_inject": ["statisticsMonth"]
}

3.3 响应配置 (response_config)

response_config 是一个 JSON 对象，定义如何解析 API 响应，判断成功/失败，提取数据。

{
  "success_field": "result",
  "success_value": 1,
  "message_field": "error_msg",
  "list_path": "data.orderList",
  "cursor_field": "data.cursor",
  "cursor_end_marker": "nomore",
  "single_record": false,
  "has_more_field": "data.hasMore"
}
字段 说明 默认值
success_field 判断成功的字段名 code
success_value 成功的值（数字） 0
message_field 错误消息的字段名 message
list_path 数据列表的 JSON 路径，如 data.orderList data
cursor_field 游标值的 JSON 路径，如 data.cursor 无
cursor_end_marker 游标结束标记，收到此值表示到底 nomore
single_record true=响应是单条记录(详情接口)，会自动包装成数组 false
has_more_field hasMore 分页模式的判断字段路径 无

成功判断逻辑

读取响应的 success_field 字段 → 转成数字 → 和 success_value 比较
相等 = 成功
不相等 = 失败，取 message_field 作为错误消息

快手示例：result 字段值 = 1 表示成功 腾讯示例：code 字段值 = 0 表示成功

数据提取逻辑

1. 按 list_path 逐层进入 JSON
2. 最后一段如果是数组 → 直接作为列表
3. 最后一段是对象 → 找对象里的 list/orderList 字段
4. 如果 single_record=true → 将对象包装为单元素数组
5. 每行数据自动展平（子对象字段合并到顶层）
6. 每行附加 raw_data 字段（原始 JSON）

3.4 表结构定义 (table_definition)

table_definition 是 JSON 对象，定义目标表的表名、列结构和冲突处理。

{
  "table_name": "kuaishou_order_list",
  "columns": [
    { "name": "oid", "type": "BIGINT", "comment": "订单ID" },
    { "name": "status", "type": "INT", "comment": "订单状态码" },
    { "name": "createTime", "type": "BIGINT", "comment": "创建时间" },
    { "name": "updateTime", "type": "BIGINT", "comment": "更新时间" },
    { "name": "totalFee", "type": "BIGINT", "comment": "总金额(分)" },
    { "name": "buyerNick", "type": "VARCHAR(100)", "comment": "买家昵称" },
    { "name": "itemTitle", "type": "VARCHAR(300)", "comment": "商品标题" }
  ],
  "conflict_keys": ["oid"]
}
字段 说明
table_name 数据库表名，系统会自动创建
columns 列定义数组
columns[].name 列名，必须与 API 响应中的字段名一致
columns[].type PostgreSQL 数据类型：BIGINT、INT、VARCHAR(n)、TEXT、BOOLEAN、JSONB
columns[].comment 列的注释说明
conflict_keys 唯一键，插入时冲突则更新（UPSERT），如 ["oid"]

自动建表：系统首次同步时会自动创建表。
自动过滤：只写入 columns 中定义的字段，多余的忽略。
raw_data：每行会自动附带一个 raw_data 字段，保存完整的原始 JSON 响应。

4. 实战：新增一个平台

以"快手电商"为例，完整步骤：

第 1 步：创建平台配置

在 api_datasource_platform 表插入一条记录：

INSERT INTO api_datasource_platform (
  tenant_id, platform_code, platform_name, description, status,
  api_base_url, auth_type, token, api_key, auth_config,
  rate_limit_per_minute, request_timeout_ms, max_retries, retry_delay_ms
) VALUES (
  1,                    -- 租户 ID
  'kuaishou',           -- 平台编码（唯一）
  '快手电商',           -- 名<><E5908D>
  '快手电商开放平台数据同步',
  'ACTIVE',             -- 状态
  'https://openapi.kwaixiaodian.com',  -- API 域名
  'API_KEY',            -- 认证类型
  '你的refresh_token',  -- Token（OAuth 返回）
  '你的access_token',   -- API Key（OAuth 返回）
  '{...}',              -- auth_config（见 2.3 节）
  100, 30000, 3, 1000   -- 限流/超时/重试
);

第 2 步：创建接口配置

对每个需要同步的 API，在 api_interface 表插入一条记录：

INSERT INTO api_interface (
  tenant_id, platform_id, name, code, url, method, status, auth_type,
  request_config, response_config, table_definition
) VALUES (
  1,
  (SELECT id FROM api_datasource_platform WHERE platform_code='kuaishou'),
  '订单列表',           -- 名称
  'order_list',         -- 编码（唯一）
  '/open/order/cursor/list',  -- API 路径
  'GET',                -- HTTP 方法
  'active',             -- 状态
  'inherit',            -- 认证（继承平台）
  '{...}',              -- request_config（见 3.2 节）
  '{...}',              -- response_config（见 3.3 节）
  '{...}'               -- table_definition（见 3.4 节）
);

第 3 步：验证

系统启动后会自动：

  1. 根据 table_definition 创建数据库表
  2. 按调度周期拉取数据
  3. 写入表中

5. 常见问题

Q1：怎么判断数据有没有同步成功？

查看 sync_tracker 表，每个接口有一条记录：

  • sync_status = success 表示成功
  • last_sync_time = 上次同步的时间戳
  • 查看 sync_task_log 表可以看到每次同步的详细日志

Q2：怎么修改同步频率？

编辑 config.yml：

sync:
  sync_interval_minutes: 60   # 改这个，单位分钟

Q3：接口报"签名校验失败"怎么排查？

  1. 确认 app_key、app_secret、sign_secret 和快手后台一致
  2. 确认 access_token 没有过期
  3. 确认 sign_algorithm 配置正确（快手用 md5）
  4. 查看日志中的 "签名原文" 和 "签名值"，对比快手官方签名工具的结果

Q4：想抽取历史数据怎么办？

在接口的 request_config 中设置 full_sync_start_time（Unix 毫秒时间戳），或修改 config.yml 的 sync.default_lookback_days。

Q5：数据重复怎么办？

配置 table_definition 中的 conflict_keys，系统会用 UPSERT 模式（冲突时自动更新，不产生重复数据）。

Q6：如何停用某个接口？

UPDATE api_interface SET status = 'inactive' WHERE code = '接口编码';

📅 最后更新：2026-06-16
📝 如有疑问，请联系开发团队

6. 同步机制说明

6.1 同步生命周期

第 1 次同步（全量）
  lastSyncTime = 0 → 从 default_lookback_days 天前开始拉取
  时间分片：每 7 天一个分片（如 90 天 ≈ 13 个分片循环）
  完成后：记录 lastSyncTime = 数据中最大的更新时间
                          ↓
第 N 次同步（增量）
  读取 lastSyncTime → beginTime = lastSyncTime
  只拉取上次同步之后新增/更新的数据
  完成后：更新 lastSyncTime
                          ↓
（每隔 sync_interval_minutes 分钟循环一次）

6.2 配置项

编辑 config.yml：

sync:
  page_size: 100                  # 每次分页请求条数
  concurrency: 5                  # 同步并发数
  retry_count: 3                  # 最大重试次数
  sync_interval_minutes: 60       # 自动同步间隔（分钟），增量频率
  compensation_interval_seconds: 300  # 失败重试扫描间隔（秒）
  auto_sync_enabled: true         # 是否启用自动同步
  sync_timeout_minutes: 120       # 单次同步超时（分钟）
  default_lookback_days: 89       # 首次全量回溯天数（可按接口覆盖）
  default_tenant_id: 1            # 租户 ID
配置项 说明 建议值
sync_interval_minutes 增量同步频率 60（1小时）
default_lookback_days 首次全量拉多少天 快手限制 90 天内，建议 89
compensation_interval_seconds 失败后多久重试 300（5分钟）
auto_sync_enabled 是否自动跑 生产环境 true

6.3 覆盖单个接口的回溯天数

在接口 request_config 中加入 full_sync_start_time（Unix 毫秒）

{ "full_sync_start_time": 1769200000000 }

6.4 查看同步状态

-- 所有接口的同步状态
SELECT platform_code, interface_code, sync_status, 
       to_timestamp(last_sync_time) AS last_sync_at
FROM sync_tracker ORDER BY platform_code, interface_code;

-- 最近失败记录
SELECT * FROM sync_task_log WHERE status = 'failed' 
ORDER BY start_time DESC LIMIT 20;
  • sync_status = 'success'：上次成功
  • sync_status = 'running'：正在跑
  • last_sync_time = 0：还没完成过同步

6.5 手动重新全量

把 last_sync_time 置 0，下次自动变全量：

UPDATE sync_tracker SET last_sync_time = 0 
WHERE platform_code = 'kuaishou' AND interface_code = 'order_list';

📅 最后更新：2026-06-16

Reference in New Issue View Git Blame Copy Permalink