red-future/data-engine

Fork 0

Files

lmk 812693caae 新增快手平台和对应的接口

2026-06-01 14:08:17 +08:00

20 KiB

Raw Blame History

数据引擎 - 项目记忆 (MEMORY.md)

项目概述

Golang + GoFrame v2 + PostgreSQL 的配置化数据同步引擎。通过 api_datasource_platform + api_interface 两表驱动，支持多平台多接口的数据同步，无需重新编译代码。

核心表结构

见 sql/init_core_tables.sql：

api_datasource_platform — 数据源平台配置
api_interface — API 接口配置（request_config / response_config / table_definition 三大 JSONB）
sync_tracker — 同步跟踪（last_sync_time, sync_status）
sync_task_log — 同步任务日志（用于补偿重试）

认证类型（auth_type）模板

1. OAuth2（腾讯广告）

auth_type: OAUTH2
token: access_token 值
client_id / client_secret: OAuth2 凭证
auth_config: {
  "token_in_query": true,         // token 放 URL 查询参数
  "query_key": "access_token",    // 参数名
  "refresh_token": "xxx",         // 刷新 token
  "extra_query_params": {
    "timestamp": "{timestamp}",
    "nonce": "{nonce}"
  }
}

安全校验：applyAuthHeader 中 OAUTH2/TOKEN 模式会设 Authorization: Bearer {token}，但 token_in_query=true 时跳过 Header，转加 URL 参数。

2. API_KEY + 签名（快手电商）

auth_type: API_KEY
token/access_token: 调用凭证
auth_config: {
  "sign_algorithm": "md5_upper",       // 签名算法: md5 / md5_upper
  "app_key": "xxx",
  "app_secret": "xxx",
  "token_in_query": true,
  "query_key": "access_token",
  "extra_query_params": {
    "timestamp": "{timestamp_ms}",
    "signMethod": "MD5"
  }
}

签名过程：所有查询参数按 key 排序 → k1=v1&k2=v2&...&key=app_secret → MD5 摘要 → 追加 sign 参数。 {timestamp_ms} 替换为毫秒时间戳，{timestamp} 替换为秒时间戳，{nonce} 替换为随机数。

3. TOKEN（简单 Bearer Token）

auth_type: TOKEN
token: "xxx"

applyAuthHeader 设 Authorization: Bearer {token}。

4. SIGN（仅签名，无额外 token）

auth_type: SIGN
auth_config: { "app_key": "...", "app_secret": "..." }

applyAuthHeader 不设特殊 Header，仅通过 applySignature 追加签名参数。PlatformManager 从 auth_config 读取 app_key/app_secret。

request_config 字段完整参考

字段	类型	含义	适用平台
`parameters_location`	string	`"query"`=URL查询参数，不设则POST走JSON body	通用
`page_param`	string	分页参数名，默认`"page"`	通用
`page_size_param`	string	每页条数参数名，默认`"page_size"`	通用
`page_size`	int	每页条数，覆盖 config.yml 全局默认值，需按各平台最大限制设置（如快手最大50，腾讯100）	通用
`cursor_pagination`	bool	是否游标分页（true=游标，false/无=普通分页）	通用
`time_field`	string	增量时间字段名	通用
`time_field_mode`	string	`"range"`=beginTime/endTime模式（快手），`"filtering"`=filtering数组模式（腾讯，默认）	通用
`prefetch`	object	预取配置（见下方）	需遍历实体的接口
`body_wrapper_field`	string	业务参数包装字段名，如`"param"`	快手
`exclude_from_wrapper`	string[]	body_wrapper 时不包装的字段	快手
`top_level_params`	string[]	保留在顶层的字段（备用）	通用
`fields`	string[]	API 的 fields 参数	腾讯

prefetch 预取配置

"prefetch": {
  "url": "/advertiser/get",        // 预取接口的相对路径
  "method": "GET",                 // HTTP 方法
  "response_path": "data.list",    // 从响应中取列表的 JSON 路径
  "target_param": "account_id",    // 预取值注入到数据接口的参数名
  "value_field": "account_id"      // 从预取列表里取哪个字段的值
}

流程：先分页拉取预取接口 → 提取所有 ID → 逐个 ID 并发拉取数据接口 → 将 ID 注入每行数据。

预取阶段分页：自动适配预取来源接口的分页方式：

游标分页（cursor_pagination: true）：首次 cursor=""，后续从响应 cursor_field 取值，直到 "" 或 "nomore" 停止
普通分页：读取 page_info.total_page 遍历全部页码

预取阶段参数构建：使用 buildReqBody(prefetchIface) 构建，自动处理 body_wrapper_field 包装（param JSON）、时间过滤（增量时传 beginTime/endTime）、分页参数名。响应解析使用 prefetchIface.ResponseConfig，正确识别各种 list_path（data.orderList、data.items、data.cpsOrderList 等）。

预取阶段时间过滤：增量同步时自动复用预取来源接口的 time_field + time_field_mode（如果来源支持增量，prefetch 会自动带上时间范围只拉增量数据）。

response_config 字段完整参考

字段	类型	含义
`success_field`	string	成功标识字段名，默认`"code"`
`success_value`	number	成功标识值，默认`0`
`message_field`	string	错误信息字段名，默认`"message"`
`list_path`	string	数据列表在 JSON 中的路径，如`"data.list"`、`"data.orderList"`
`cursor_field`	string	游标字段路径，如`"data.cursor"`
`cursor_end_marker`	string	游标结束标记，如`"nomore"`（代码中硬判断此字符串）
`single_record`	bool	单记录模式：`list_path` 指向的路径是单个对象而非数组，会自动包装为单元素数组

响应解析逻辑：parseRespExt 函数

检查 success_field 的值是否等于 success_value | 注意：success_value 仅支持数字类型（内部通过 toFloat64 转换），不支持字符串如 "SUCCESS" 或布尔值 true
按 list_path 遍历到数据列表数组 | 支持路径如 data.orderList，最后一段是数组 OK（代码有兼容处理）
每条数据调用 flattenRow 展平嵌套 field（orderBaseInfo.oid → oid）
提取 cursor_field 用于下一页，为空或等于 cursor_end_marker 时停止循环
从 page_info.total_page 获取总页数（普通分页用）

游标分页首次请求：代码自动传 cursor=""（首页不传游标），由服务端返回第一页数据 + 下一页游标值。

单记录模式："single_record": true 时，list_path 指向单个对象（如 "data"），自动包装成单元素数组用于统一处理。

table_definition 字段完整参考

字段	类型	含义
`table_name`	string	目标表名，如`"kuaishou_order_list"`
`columns`	array	列定义数组，每个元素 `{"name":"字段名","type":"PG类型","comment":"说明"}`
`conflict_keys`	string[]	upsert 冲突键，如`["oid"]`或`["image_id","account_id"]`

注意：

列名必须和 API 响应中的字段名完全一致（camelCase/snake_case 取决于 API）
嵌套字段会被 flattenRow 自动展平到顶层
raw_data 列自动包含，存储完整原始 JSON
自动建表时会额外添加 id/tenant_id/creator/created_at/updater/updated_at/deleted_at 审计字段
必须写全：columns 中需列出 API 响应中所有平铺的标量字段（字符串/数字/布尔），数组和嵌套对象字段保留在 raw_data 中即可

同步策略原则

API 支持时间过滤（有 beginTime/endTime 或 filtering 参数）→ 配 time_field + time_field_mode，实现增量同步
API 不支持时间过滤 → 不配 time_field，走全量同步
详情/子接口（通过 prefetch 遍历）：prefetch 阶段自动复用预取来源接口的 time_field_mode（如果来源接口支持增量，prefetch 也会带上时间过滤只拉增量数据）

全量同步流程

SyncByConfig(platformCode, interfaceCode, isFullSync=false)
  │
  ├─ getLastSyncTime() → 无记录返回 0
  │   lastSyncTime=0 → 全量
  │
  ├─ markSyncRunning() 标记 running
  │
  ├─ buildReqBody(page=1, cursor="", lastSyncTime=0)
  │   ├─ 游标分页: cursor="" 首次空游标
  │   ├─ range 模式: beginTime=90天前, endTime=now
  │   └─ filtering 模式: 不添加时间过滤
  │
  ├─ API → parseRespExt → flattenRow → savePage(upsert)
  │
  └─ 游标循环 → 直到 cursor="" 或 "nomore"
       updateSyncTime() 记录最大时间戳

增量同步流程

SyncByConfig(platformCode, interfaceCode, isFullSync=false)
  │
  ├─ getLastSyncTime() → 返回上次 maxTime
  │   lastSyncTime>0 → 增量
  │
  ├─ buildReqBody(cursor="", lastSyncTime=xxx)
  │   ├─ range 模式: beginTime=lastSyncTime, endTime=now
  │   └─ filtering 模式: filtering=[{field=time_field, operator=GREATER_EQUALS, values=[lastSyncTime]}]
  │
  └─ 后续同全量（只拉取时间范围内的数据）

并发保护

内存锁（同一进程内）

SyncByConfig 入口使用 sync.Map（syncRunningMap）记录正在执行的接口。若同一接口的同步请求再次到达（如调度器重叠），直接跳过并打印警告：

WARN 接口 [kuaishou/order_list] 正在同步中，跳过重复请求

无论接口类型（prefetch/非prefetch、腾讯/快手、现有/新增）均自动生效。

异常中断检测（DB 状态）

如果 sync_tracker.sync_status = "running"，说明上次同步异常中断（进程崩溃），自动回退全量。

补偿机制

compensation.go 定时扫描 sync_task_log 中 status="failed" 的任务，自动调用 SyncByConfig 重试（增量），支持指数退避（5, 15, 30, 60, 120 分钟）。

自动同步调度

sync_scheduler.go 的 runAutoSync 遍历所有 ACTIVE 平台下的所有配置了 table_definition 的活跃接口，自动调用同步。

调度器时序：使用 for { runAutoSync(); time.Sleep(interval) } 模式，而非 time.NewTicker。每次同步完全结束后才开始计时 interval，避免前一次未跑完就启动下一次导致重叠。无论全量跑 30 分钟还是 70 分钟，下一次都在「本次完成时间 + interval」后执行。

新增平台 + 接口的操作步骤

操作步骤

在 sql/ 下创建 seed_data_{平台编码}.sql
先执行 sql/init_core_tables.sql（仅首次需要）
执行新建的 seed SQL 文件（INSERT 平台 → INSERT 接口）
配置认证凭据（token / app_key / app_secret 等）
重启服务，管理后台验证：http://localhost:3002/admin

种子 SQL 标准模板

模板 A：OAuth2 + 普通分页（腾讯广告风格）

-- =============================================
-- 1. 创建 {平台编码} 平台
-- =============================================
INSERT INTO api_datasource_platform (
    tenant_id, creator, created_at, updater, updated_at,
    platform_code, platform_name, description, status,
    api_base_url, auth_type,
    token, client_id, client_secret,
    auth_config,
    rate_limit_per_minute, rate_limit_per_hour,
    concurrency_limit, request_timeout_ms, max_retries, retry_delay_ms
) VALUES (
    1, 'admin', NOW(), 'admin', NOW(),
    '{平台编码}', '{平台名称}', '{描述}', 'ACTIVE',
    '{API_BASE_URL}', 'OAUTH2',
    '{access_token}', '{client_id}', '{client_secret}',
    '{
        "token_in_query": true,
        "query_key": "access_token",
        "refresh_token": "{refresh_token}",
        "extra_query_params": {
            "timestamp": "{timestamp}",
            "nonce": "{nonce}"
        }
    }'::jsonb,
    60, 3600, 10, 30000, 3, 1000
);

-- 2. 创建 {接口名} 接口（普通分页 + 增量 filtering）
INSERT INTO api_interface (
    tenant_id, creator, created_at, updater, updated_at,
    platform_id, name, code, url, method, status, auth_type,
    request_config, response_config, table_definition
) VALUES (
    1, 'admin', NOW(), 'admin', NOW(),
    (SELECT id FROM api_datasource_platform WHERE platform_code = '{平台编码}'),
    '{接口名称}', '{接口编码}',
    '{相对路径}', 'GET', 'active', 'inherit',
    '{
        "parameters_location": "query",
        "page": 1,
        "page_size": 100,
        "page_param": "page",
        "page_size_param": "page_size",
        "pagination_mode": "PAGINATION_MODE_NORMAL",
        "time_field": "last_modified_time",
        "fields": ["field1", "field2"]
    }'::jsonb,
    '{
        "success_field": "code",
        "success_value": 0,
        "message_field": "message",
        "list_path": "data.list"
    }'::jsonb,
    '{
        "table_name": "{表名}",
        "columns": [
            {"name": "id", "type": "BIGINT", "comment": "主键"},
            {"name": "name", "type": "VARCHAR(200)", "comment": "名称"},
            {"name": "created_time", "type": "BIGINT", "comment": "创建时间"}
        ],
        "conflict_keys": ["id"]
    }'::jsonb
);

模板 B：OAuth2 + prefetch 预取（遍历账户拉取子数据，腾讯广告风格）

-- 预取接口（先拉取实体列表）
INSERT INTO api_interface (
    tenant_id, creator, created_at, updater, updated_at,
    platform_id, name, code, url, method, status, auth_type,
    request_config, response_config, table_definition
) VALUES (
    1, 'admin', NOW(), 'admin', NOW(),
    (SELECT id FROM api_datasource_platform WHERE platform_code = '{平台编码}'),
    '{实体列表名称}', '{实体编码}',
    '{实体列表相对路径}', 'GET', 'active', 'inherit',
    '{
        "parameters_location": "query",
        "page": 1,
        "page_size": 100,
        "page_param": "page",
        "page_size_param": "page_size",
        "fields": ["id", "name"]
    }'::jsonb,
    '{
        "success_field": "code",
        "success_value": 0,
        "list_path": "data.list"
    }'::jsonb,
    '{
        "table_name": "{实体表名}",
        "columns": [
            {"name": "id", "type": "BIGINT", "comment": "实体ID"},
            {"name": "name", "type": "VARCHAR(200)", "comment": "实体名称"}
        ],
        "conflict_keys": ["id"]
    }'::jsonb
);

-- 数据接口（遍历每个实体拉取数据）
INSERT INTO api_interface (
    tenant_id, creator, created_at, updater, updated_at,
    platform_id, name, code, url, method, status, auth_type,
    request_config, response_config, table_definition
) VALUES (
    1, 'admin', NOW(), 'admin', NOW(),
    (SELECT id FROM api_datasource_platform WHERE platform_code = '{平台编码}'),
    '{数据接口名称}', '{数据接口编码}',
    '{数据接口相对路径}', 'GET', 'active', 'inherit',
    '{
        "parameters_location": "query",
        "page": 1,
        "page_size": 100,
        "page_param": "page",
        "page_size_param": "page_size",
        "time_field": "last_modified_time",
        "prefetch": {
            "url": "{实体列表相对路径}",
            "method": "GET",
            "response_path": "data.list",
            "target_param": "{参数名}",
            "value_field": "{取值字段}"
        }
    }'::jsonb,
    '{
        "success_field": "code",
        "success_value": 0,
        "list_path": "data.list"
    }'::jsonb,
    '{
        "table_name": "{数据表名}",
        "columns": [
            {"name": "id", "type": "BIGINT", "comment": "ID"},
            {"name": "{参数名}", "type": "BIGINT", "comment": "实体ID（预取注入）"},
            {"name": "created_time", "type": "BIGINT", "comment": "创建时间"}
        ],
        "conflict_keys": ["id", "{参数名}"]
    }'::jsonb
);

模板 C：API_KEY + MD5 签名 + 游标分页 + param 包装（快手风格）

INSERT INTO api_datasource_platform (
    tenant_id, creator, created_at, updater, updated_at,
    platform_code, platform_name, description, status,
    api_base_url, auth_type,
    auth_config,
    rate_limit_per_minute, rate_limit_per_hour,
    concurrency_limit, request_timeout_ms, max_retries, retry_delay_ms
) VALUES (
    1, 'admin', NOW(), 'admin', NOW(),
    '{平台编码}', '{平台名称}', '{描述}', 'ACTIVE',
    '{API_BASE_URL}', 'API_KEY',
    '{
        "sign_algorithm": "md5_upper",
        "app_key": "{YOUR_APP_KEY}",
        "app_secret": "{YOUR_APP_SECRET}",
        "token_in_query": true,
        "query_key": "access_token",
        "extra_query_params": {
            "timestamp": "{timestamp_ms}",
            "signMethod": "MD5"
        }
    }'::jsonb,
    100, 3600, 5, 30000, 3, 1000
);

INSERT INTO api_interface (
    tenant_id, creator, created_at, updater, updated_at,
    platform_id, name, code, url, method, status, auth_type,
    request_config, response_config, table_definition
) VALUES (
    1, 'admin', NOW(), 'admin', NOW(),
    (SELECT id FROM api_datasource_platform WHERE platform_code = '{平台编码}'),
    '{接口名称}', '{接口编码}',
    '{相对路径}', 'GET', 'active', 'inherit',
    '{
        "parameters_location": "query",
        "page_param": "cursor",
        "page_size_param": "pageSize",
        "cursor_pagination": true,
        "method": "{API方法名}",
        "version": 1,
        "signMethod": "MD5",
        "pageSize": 50,
        "time_field": "updateTime",
        "time_field_mode": "range",
        "body_wrapper_field": "param",
        "exclude_from_wrapper": ["method", "version", "signMethod"],
        "{其他业务参数}": {值}
    }'::jsonb,
    '{
        "success_field": "result",
        "success_value": 1,
        "list_path": "data.{数组字段名}",
        "cursor_field": "data.cursor",
        "cursor_end_marker": "nomore"
    }'::jsonb,
    '{
        "table_name": "{表名}",
        "columns": [
            {"name": "oid", "type": "BIGINT", "comment": "订单ID"},
            {"name": "createTime", "type": "BIGINT", "comment": "创建时间"},
            {"name": "updateTime", "type": "BIGINT", "comment": "更新时间"},
            {"name": "{字段名}", "type": "{类型}", "comment": "{说明}"}
        ],
        "conflict_keys": ["oid"]
    }'::jsonb
);

模板 D：POST + JSON Body（腾讯音频风格，无 prefetch，无增量）

INSERT INTO api_interface (
    tenant_id, creator, created_at, updater, updated_at,
    platform_id, name, code, url, method, status, auth_type,
    request_config, response_config, table_definition
) VALUES (
    1, 'admin', NOW(), 'admin', NOW(),
    (SELECT id FROM api_datasource_platform WHERE platform_code = '{平台编码}'),
    '{接口名称}', '{接口编码}',
    '{相对路径}', 'POST', 'active', 'inherit',
    '{
        "page": 1,
        "page_size": 100,
        "page_param": "page",
        "page_size_param": "page_size",
        "fields": ["field1", "field2"]
    }'::jsonb,
    '{
        "success_field": "code",
        "success_value": 0,
        "message_field": "message",
        "list_path": "data.list"
    }'::jsonb,
    '{
        "table_name": "{表名}",
        "columns": [
            {"name": "id", "type": "BIGINT", "comment": "ID"},
            {"name": "name", "type": "VARCHAR(200)", "comment": "名称"}
        ],
        "conflict_keys": ["id"]
    }'::jsonb
);

注意：POST 方法且无 parameters_location: "query" 时，参数以 JSON body 形式发送。

新增接口步骤总结

确定平台：已有平台直接跳到第3步，否则先新增平台（参考对应认证类型的模板）
确定平台 auth_type：OAUTH2 / TOKEN / API_KEY / SIGN → 复制对应模板的 auth_config
分析 API：
- 请求方式（GET/POST）
- 分页方式（普通分页 → page + total_page / 游标分页 → cursor + cursor_end）
- 成功标识（code=0 / result=1 / errcode=0 等）
- 数据列表路径（data.list / data.orderList / data.records 等）
- 数据字段名（注意大小写，嵌套结构会被展开）
- 是否需要遍历实体（prefetch）
- 增量时间字段（字段名 + 值类型）
选择模板并填入对应值
执行 seed SQL
在管理后台验证：http://{host}:3002/admin

20 KiB Raw Blame History Unescape Escape