red-future/data-engine
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

数据引擎重构

Browse Source
This commit is contained in:
lmk
2026-07-16 14:34:06 +08:00
parent 802139c349
commit b45c4a2316
19 changed files with 1332 additions and 307 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/report/cron.go
+43 -40
View File
@@ -9,52 +9,55 @@ import (
"github.com/sirupsen/logrus"
)
// ExecDailyExtract 执行一次全量每日抽取（遍历所有启用业务+报表）
// 提取出独立函数，供定时任务和 HTTP 端点共同使用
func ExecDailyExtract(ctx context.Context) {
svc := GetService()
logrus.Info("[报表引擎] 开始全量每日抽取任务")
businesses, err := svc.GetAllBusinesses(ctx)
if err != nil {
logrus.Errorf("[报表引擎] 获取业务列表失败: %v", err)
return
}
yesterday := time.Now().AddDate(0, 0, -1).Format("2006-01-02")
for _, biz := range businesses {
reports, err := svc.GetAllReports(ctx, biz.BusinessCode)
if err != nil {
logrus.Errorf("[报表引擎] 获取报表列表失败 [%s]: %v", biz.BusinessCode, err)
continue
}
for _, rpt := range reports {
logrus.Infof("[报表引擎] 开始抽取 %s/%s 日期 %s", biz.BusinessCode, rpt.ReportCode, yesterday)
resp, err := svc.ExtractDailyData(ctx, biz.BusinessCode, rpt.ReportCode, yesterday, "system")
if err != nil {
logrus.Errorf("[报表引擎] 抽取失败 %s/%s %s: %v", biz.BusinessCode, rpt.ReportCode, yesterday, err)
continue
}
if resp.Success {
logrus.Infof("[报表引擎] 抽取成功 %s/%s %s 总数:%d 耗时:%dms",
biz.BusinessCode, rpt.ReportCode, yesterday, resp.TotalCount, resp.ExecTimeMs)
} else {
logrus.Errorf("[报表引擎] 抽取异常 %s/%s %s: %s", biz.BusinessCode, rpt.ReportCode, yesterday, resp.ErrorMsg)
}
}
}
logrus.Info("[报表引擎] 全量每日抽取任务完成")
}
// StartDailyExtractJob 启动每日自动抽取定时任务
// 每天凌晨 2:00 遍历所有已启用的业务+报表，抽取前一天数据
// 在 main.go 中调用：report.StartDailyExtractJob()
func StartDailyExtractJob() {
svc := GetService()
// 每天凌晨 2:00 执行
_, err := gcron.Add(gctx.New(), "0 0 2 * * *", func(ctx context.Context) {
logrus.Info("[报表引擎] 开始每日自动抽取任务")
// 获取所有启用业务
businesses, err := svc.GetAllBusinesses(ctx)
if err != nil {
logrus.Errorf("[报表引擎] 获取业务列表失败: %v", err)
return
}
yesterday := time.Now().AddDate(0, 0, -1).Format("2006-01-02")
for _, biz := range businesses {
// 获取业务下所有启用报表
reports, err := svc.GetAllReports(ctx, biz.BusinessCode)
if err != nil {
logrus.Errorf("[报表引擎] 获取报表列表失败 [%s]: %v", biz.BusinessCode, err)
continue
}
for _, rpt := range reports {
logrus.Infof("[报表引擎] 开始抽取 %s/%s 日期 %s", biz.BusinessCode, rpt.ReportCode, yesterday)
resp, err := svc.ExtractDailyData(ctx, biz.BusinessCode, rpt.ReportCode, yesterday, "system")
if err != nil {
logrus.Errorf("[报表引擎] 抽取失败 %s/%s %s: %v", biz.BusinessCode, rpt.ReportCode, yesterday, err)
continue
}
if resp.Success {
logrus.Infof("[报表引擎] 抽取成功 %s/%s %s 总数:%d 耗时:%dms",
biz.BusinessCode, rpt.ReportCode, yesterday, resp.TotalCount, resp.ExecTimeMs)
} else {
logrus.Errorf("[报表引擎] 抽取异常 %s/%s %s: %s", biz.BusinessCode, rpt.ReportCode, yesterday, resp.ErrorMsg)
}
}
}
logrus.Info("[报表引擎] 每日自动抽取任务完成")
ExecDailyExtract(ctx)
}, "daily-report-extract")
if err != nil {
config.yml
+13 -3
View File
@@ -9,6 +9,13 @@ rate:
limit: 200
burst: 300
# 报表引擎配置
report:
# 是否由本进程内部自动调度每日抽取（gcron "0 0 2 * * *"）
# true — 进程启动后自动在每天 02:00 执行全量抽取，无需外部依赖
# false — 不自启调度，由 PPGo_Job 通过 POST /report/extractAll 触发（推荐）
extract_enabled: false
# 数据同步配置
sync:
page_size: 100 # 每次分页请求条数
@@ -16,9 +23,12 @@ sync:
retry_count: 3 # 最大重试次数
sync_interval_minutes: 60 # 自动同步间隔（分钟）
compensation_interval_seconds: 300 # 补偿调度器扫描间隔（秒）
auto_sync_enabled: false # 是否启用自动同步　
# 是否由本进程内部自动循环执行全平台同步（每次完成后等待 sync_interval_minutes 再继续）
# true — 进程启动后自动在后台循环同步所有 ACTIVE 平台的接口数据
# false — 不自启调度，由 PPGo_Job 通过 POST /sync/ctrl/triggerAll 触发（推荐）
auto_sync_enabled: false
sync_timeout_minutes: 120 # 单次同步超时（分钟），全量超大表可适当调大
default_lookback_days: 89 # 全量同步默认回溯天数（快手限制90天内，留1天余量）（单接口可通过 request_config.full_sync_start_time 覆盖）
default_lookback_days: 10 # 全量同步默认回溯天数（快手限制90天内，留1天余量）（单接口可通过 request_config.full_sync_start_time 覆盖）
default_tenant_id: 1 # 自动同步使用的租户 ID（多租户部署时配置）
# Database.
@@ -55,7 +65,7 @@ redis:
writeTimeout: "30s" #TCP的Write操作超时时间，使用时间字符串例如30s/1m/1d
maxActive: 100
consul:
address: 192.168.3.135:8500
address: 192.168.3.37:8500
#k8s:
# apiServer: "https://192.168.3.37:6443"
# token: "eyJhbGciOiJSUzI1NiIsImtpZCI6IlR6X0QtelVDYkZPYnpjTDlfamZzOEFjbERJb2dUMTRqMjNfYUNncGcwRW8ifQ.eyJhdWQiOlsiaHR0cHM6Ly9rdWJlcm5ldGVzLmRlZmF1bHQuc3ZjLmNsdXN0ZXIubG9jYWwiLCJrM3MiXSwiZXhwIjoxODA3ODY3MTA1LCJpYXQiOjE3NzYzMzExMDUsImlzcyI6Imh0dHBzOi8va3ViZXJuZXRlcy5kZWZhdWx0LnN2Yy5jbHVzdGVyLmxvY2FsIiwianRpIjoiNjAwMGY0ODctZTQ4Ni00NTUxLWIyNjgtMzE1MWE1MDE5YjU4Iiwia3ViZXJuZXRlcy5pbyI6eyJuYW1lc3BhY2UiOiJrdWJlcm5ldGVzLWRhc2hib2FyZCIsInNlcnZpY2VhY2NvdW50Ijp7Im5hbWUiOiJkYXNoYm9hcmQtYWRtaW4iLCJ1aWQiOiJlZjAxY2IxNS0xNTc0LTRlYTYtODg4Ny03YjZhODY3OWFkYmIifX0sIm5iZiI6MTc3NjMzMTEwNSwic3ViIjoic3lzdGVtOnNlcnZpY2VhY2NvdW50Omt1YmVybmV0ZXMtZGFzaGJvYXJkOmRhc2hib2FyZC1hZG1pbiJ9.Z5MEy-dhWq6lruuR6QSI--cYRZzDoeEes78a4lLMtXa8lSLt1FZy16kNVn2pJli74RSk0kKS2F5GAatyuyGjF_yB7Tm1Tb7iCjzeUM7xRpdvXEmk4MUImlzwGeW31NXpwFrfR0oPS-9TLlbahEmsgwL0DUOC0CekSiToVbIduwEmmrB0FMPFayr5_wqGDmxdqtaAH3K-LJNCqPiAMqevsxTebrhyJSrThU-Pi7iJYm_sUd8VHKrRxrwDiGA77j4lfita_hZSdyrZe8qsrbGqomHxFWk9ZzcJJ9q7OLS13OXo-Xbcu-_TZcfDOreZhune9ctM2lmuOtmYnad47gYRcA"
controller/report/report_controller.go
+20
View File
@@ -13,6 +13,26 @@ import (
"github.com/sirupsen/logrus"
)
// ============================================================
// 批量操作（供 PPGo_Job 调度）
// ============================================================
type extractAllReq struct {
g.Meta `path:"/extractAll" method:"post" tags:"报表引擎" summary:"全量抽取" dc:"遍历所有启用业务+报表，抽取前一天数据"`
}
type extractAllRes struct {
Success bool `json:"success"`
Message string `json:"message"`
}
func (c *report) ExtractAll(ctx context.Context, req *extractAllReq) (*extractAllRes, error) {
ctx = ctxWithUser(ctx)
logrus.Info("[HTTP] 触发全量抽取")
reportSvc.ExecDailyExtract(ctx)
return &extractAllRes{Success: true, Message: "全量抽取已触发"}, nil
}
type report struct{}
var ReportController = new(report)
controller/sync/platform_sync_controller.go
+111
View File
@@ -3,7 +3,9 @@ package sync
import (
"context"
svc "dataengine/service/sync"
"fmt"
"gitea.redpowerfuture.com/red-future/common/beans"
"github.com/gogf/gf/v2/frame/g"
"github.com/sirupsen/logrus"
)
@@ -12,6 +14,37 @@ var PlatformSyncController = new(syncCtrl)
type syncCtrl struct{}
// RefreshTokenReq 刷新 Token 请求
type RefreshTokenReq struct {
g.Meta `path:"/refreshToken" method:"post" tags:"平台同步" summary:"刷新Token" dc:"刷新指定平台的认证Token（如腾讯OAuth2）"`
PlatformCode string `json:"platformCode" v:"required" dc:"平台编码"`
}
// RefreshTokenRes 刷新 Token 响应
type RefreshTokenRes struct {
Success bool `json:"success"`
Message string `json:"message"`
}
// RefreshToken 手动触发 Token 刷新
// 由 PPGo_Job 定时调度，定期刷新平台 Token（如腾讯 OAuth2 access_token）
func (c *syncCtrl) RefreshToken(ctx context.Context, req *RefreshTokenReq) (*RefreshTokenRes, error) {
ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1})
logrus.Infof("[HTTP] 触发 Token 刷新: platform=%s", req.PlatformCode)
pm := &svc.PlatformManager{}
platform, _, err := pm.GetPlatformWithInterfaces(ctx, req.PlatformCode)
if err != nil {
return nil, fmt.Errorf("读取平台配置失败: %w", err)
}
if err := svc.RefreshTencentToken(ctx, platform); err != nil {
return &RefreshTokenRes{Success: false, Message: err.Error()}, nil
}
return &RefreshTokenRes{Success: true, Message: "Token 刷新成功"}, nil
}
// TriggerSyncReq 触发同步请求
type TriggerSyncReq struct {
g.Meta `path:"/trigger" method:"post" tags:"平台同步" summary:"触发同步" dc:"根据平台编码和接口编码触发数据同步"`
@@ -31,6 +64,7 @@ type TriggerSyncRes struct {
// TriggerSync 触发同步
func (c *syncCtrl) TriggerSync(ctx context.Context, req *TriggerSyncReq) (*TriggerSyncRes, error) {
ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1})
logrus.Infof("触发同步: platform=%s, interface=%s, fullSync=%v", req.PlatformCode, req.InterfaceCode, req.FullSync)
result, err := svc.SyncByConfig(ctx, req.PlatformCode, req.InterfaceCode, req.FullSync)
if err != nil {
@@ -65,8 +99,85 @@ type QueryPlatformConfigRes struct {
} `json:"interfaces"`
}
// ============================================================
// 批量操作（供 PPGo_Job 调度）
// ============================================================
// TriggerAllSyncReq 触发全平台同步请求
type TriggerAllSyncReq struct {
g.Meta `path:"/triggerAll" method:"post" tags:"平台同步" summary:"触发全平台同步" dc:"遍历所有ACTIVE平台/接口执行同步"`
FullSync bool `json:"fullSync" dc:"true=强制全量同步，false=自动判断增量/全量"`
}
// TriggerAllSyncRes 触发全平台同步响应
type TriggerAllSyncRes struct {
Success bool `json:"success"`
Message string `json:"message"`
Results []svc.SyncRunItemResult `json:"results,omitempty"`
}
// TriggerAllSync 触发全平台同步（等价于内部自动同步的一次完整执行）
func (c *syncCtrl) TriggerAllSync(ctx context.Context, req *TriggerAllSyncReq) (*TriggerAllSyncRes, error) {
ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1})
logrus.Infof("[HTTP] 触发全平台同步, fullSync=%v", req.FullSync)
results := svc.TriggerAllSync(ctx, req.FullSync)
hasError := false
for _, r := range results {
if !r.Success {
hasError = true
break
}
}
res := &TriggerAllSyncRes{Results: results}
if hasError {
res.Success = false
res.Message = "全平台同步完成，部分失败"
} else {
res.Success = true
res.Message = "全平台同步完成"
}
return res, nil
}
// CompensateReq 触发补偿扫描请求
type CompensateReq struct {
g.Meta `path:"/compensate" method:"post" tags:"平台同步" summary:"触发补偿扫描" dc:"扫描失败任务并重试"`
}
// CompensateRes 触发补偿扫描响应
type CompensateRes struct {
Success bool `json:"success"`
Message string `json:"message"`
Result *svc.CompensateResult `json:"result,omitempty"`
}
// Compensate 触发补偿扫描（等价于内部补偿调度的一次完整执行）
func (c *syncCtrl) Compensate(ctx context.Context, req *CompensateReq) (*CompensateRes, error) {
ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1})
logrus.Info("[HTTP] 触发补偿扫描")
result := svc.TriggerCompensation(ctx)
res := &CompensateRes{Result: result}
if result.TotalFailed == 0 {
res.Success = true
res.Message = "补偿扫描完成，没有待补偿任务"
} else if result.Failed > 0 {
res.Success = false
res.Message = fmt.Sprintf("补偿扫描完成，共 %d 个失败任务，成功补偿 %d 个，失败 %d 个，已达最大重试 %d 个",
result.TotalFailed, result.Succeeded, result.Failed, result.MaxRetryReached)
} else {
res.Success = true
res.Message = fmt.Sprintf("补偿扫描完成，共 %d 个失败任务，成功补偿 %d 个",
result.TotalFailed, result.Succeeded)
}
return res, nil
}
// QueryConfig 查询配置
func (c *syncCtrl) QueryConfig(ctx context.Context, req *QueryPlatformConfigReq) (*QueryPlatformConfigRes, error) {
ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1})
pm := &svc.PlatformManager{}
platform, interfaces, err := pm.GetPlatformWithInterfaces(ctx, req.PlatformCode)
if err != nil {
dao/copydata/sync_task_log_dao.go
+2
View File
@@ -5,6 +5,7 @@ import (
consts "dataengine/consts/public"
dto "dataengine/model/dto/copydata"
entity "dataengine/model/entity/copydata"
"dataengine/utils"
"time"
"gitea.redpowerfuture.com/red-future/common/db/gfdb"
@@ -39,6 +40,7 @@ func (d *SyncTaskLogDao) Create(ctx context.Context, req *dto.CreateSyncTaskLogR
"request_params": req.RequestParams,
"retry_count": 0,
"duration_ms": 0,
"tenant_id": utils.GetCurrentTenantId(ctx),
}
r, err := gfdb.DB(ctx).Model(ctx, consts.SyncTaskLogTable).Data(data).Insert()
docs/USAGE.md
+411 -117
View File
@@ -10,8 +10,9 @@
2. **拉取数据**（分页请求、多步骤请求、并发处理）
3. **写入数据库**（批量 upsert）
4. **增量同步**（通过 `filtering` 按最后修改时间过滤）
5. **自动调度**（按配置的时间间隔循环执行）
5. **自动调度** → **推荐由 PPGo_Job 外部调度**（支持内部定时和外部 HTTP 触发两种模式）
6. **补偿重试**（失败的同步任务自动重试，退避递增）
7. **Token 自动刷新**（OAuth2 token 过期自动续期，不占用重试次数）
**不需要为每个平台写一行业务代码。**
@@ -32,7 +33,9 @@
### 2.2 初始化数据
执行 `sql/seed_data.sql` 创建腾讯广告平台+接口配置。如需清空重来：
执行 `sql/seed_data.sql` 创建腾讯广告平台 + 4 个接口配置。
如需清空重来：
```sql
ALTER SEQUENCE api_datasource_platform_id_seq RESTART WITH 1;
@@ -42,6 +45,13 @@ DELETE FROM api_datasource_platform;
\i sql/seed_data.sql
```
其他平台的 seed 数据：
- `sql/seed_data_dingtalk.sql` — 钉钉平台（部门、用户、考勤）
- `sql/seed_data_dingtalk_hrm.sql` — 钉钉 HRM
- `sql/seed_data_dingtalk_salary.sql` — 钉钉智能薪酬
- `sql/seed_data_kuaishou.sql` — 快手平台
- `sql/seed_data_report_kuaishou.sql` — 快手报表引擎演示数据
---
## 三、配置说明
@@ -53,26 +63,45 @@ DELETE FROM api_datasource_platform;
| `platform_code` | 平台编码（唯一） | `tencent` |
| `platform_name` | 平台名称 | `腾讯广告` |
| `api_base_url` | API 基础地址 | `https://api.e.qq.com/v3.0` |
| `auth_type` | 认证类型 | `OAUTH2` / `TOKEN` / `API_KEY` / `BASIC` |
| `token` | access_token | `xxxxx` |
| `auth_type` | 认证类型 | `OAUTH2` / `TOKEN` / `API_KEY` / `SIGN` / `APP_SIGNATURE` |
| `token` | access_token 明文 | `xxxxx` |
| `client_id` / `client_secret` | OAuth2 凭证 | `xxxxx` |
| `auth_config` | 自定义认证配置(JSONB) | 详见 3.1.1 |
| `auth_config` | 自定义认证配置 (JSONB) | 详见 3.1.1 |
| `rate_limit_per_minute` | 每分钟请求限制 | `60` |
| `request_timeout_ms` | 请求超时（毫秒） | `30000` |
> **注意：** 认证凭据现在统一存储在数据库中（`api_datasource_platform` 表），不再放在 `config.yml` 中。`config.yml` 中 `tencent.oauth.*` 仅作参考。
#### 3.1.1 `auth_config` 字段详解
```json
{
"token_in_query": true, // token 放在 URL 查询参数
"query_key": "access_token", // 参数名，默认 "access_token"
"token_in_query": true, // token 放在 URL 查询参数（如腾讯广告）
"query_key": "access_token", // 查询参数名，默认 "access_token"
"header_name": "Authorization", // token 放请求头时的头名
"header_format": "Bearer {token}",
"extra_query_params": { // 额外查询参数
"timestamp": "{timestamp}", // {timestamp} 自动替换为当前时间戳
"nonce": "{nonce}" // {nonce} 自动替换为随机字符串
}
"refresh_token": "xxxxx", // OAuth2 refresh_token
"extra_query_params": { // 额外查询参数
"timestamp": "{timestamp}", // {timestamp} → 当前 Unix 时间戳
"nonce": "{nonce}" // {nonce} → 随机字符串
},
"app_key": "xxx", // SIGN 认证的应用 Key
"app_secret": "xxx", // SIGN 认证的应用 Secret
"sign_algorithm": "md5", // 签名算法: md5 / md5_upper
"sign_secret": "xxx" // 签名专用密钥（可选，默认用 app_secret）
}
```
#### 3.1.2 认证类型说明
| 类型 | 说明 | token 传递方式 |
|------|------|------|
| `OAUTH2` | OAuth2 授权 | URL query 或 Header，支持自动刷新 |
| `TOKEN` | 静态 Token | `Authorization: Bearer <token>` |
| `API_KEY` | API Key | `X-API-Key <api_key>` |
| `SIGN` | 参数签名 | URL query + `sign` 参数 |
| `APP_SIGNATURE` | App-ID + 签名 | Header: `app-id` + `signature` |
### 3.2 接口管理 (`api_interface`)
| 字段 | 说明 | 示例 |
@@ -81,49 +110,72 @@ DELETE FROM api_datasource_platform;
| `name` / `code` | 接口名称 / 唯一编码 | `图片素材` / `image` |
| `url` | 接口地址（相对路径） | `/images/get` |
| `method` | 请求方法 | `GET` / `POST` |
| `request_config` | 请求配置(JSONB) | 详见 3.2.1 |
| `response_config` | 响应配置(JSONB) | 详见 3.2.2 |
| `table_definition` | 表结构定义(JSONB) | 详见 3.3 |
| `request_config` | 请求配置 (JSONB) | 详见 3.2.1 |
| `response_config` | 响应配置 (JSONB) | 详见 3.2.2 |
| `table_definition` | 表结构定义 (JSONB) | 详见 3.3 |
#### 3.2.1 `request_config` 字段详解
```json
{
"parameters_location": "query", // 参数位置: "query"(URL) / "body"(默认)
"parameters_location": "query", // 参数位置: "query"(URL) / "body"(默认POST)
"page": 1,
"page_size": 100,
"page_param": "page", // 分页参数名（自定义）
"page_param": "page", // 分页参数名（可自定义）
"page_size_param": "page_size",
"time_field": "last_modified_time", // 增量时间字段
"fields": ["field1", "field2"], // 请求字段（如音频的 fields）
"prefetch": { ... } // 预取配置（见下文）
"pagination_mode": "offset", // 分页模式: 默认页码 / "offset"（偏移量）
"time_field": "last_modified_time", // 增量时间字段名
"time_field_mode": "filtering", // 时间模式: "filtering"(腾讯) / "range"(快手)
"cursor_pagination": true, // 游标分页
"initial_cursor": "", // 初始游标值
"fields": ["field1", "field2"], // 请求字段列表
"body_wrapper_field": "param", // Body 包装字段（快手 API 的 param JSON）
"exclude_from_wrapper": ["method"], // 不包装的顶层字段
"row_inject": ["statisticsMonth"], // 将请求参数注入到响应行中
"prefetch": { ... }, // 预取配置（见下文）
"recursive": { ... }, // 递归配置（见下文）
"max_recursive_depth": 20, // 递归最大深度
"full_sync_start_time": 1700000000000 // 全量同步起始时间（毫秒时间戳）
}
```
**`parameters_location` 说明**：
- 未设置或 `"body"` → 参数放在 JSON body 中（POST 请求）
- `"query"` → 参数放在 URL 查询字符串中（GET 请求用）
- 未设置或 `"body"` → 参数放在请求体中（POST 请求）
- `"query"` → 参数放在 URL 查询字符串中（GET 请求）
- 当 `method=GET` 时，即使不设置也默认走 query
**预取（prefetch）**：某些接口需要"先拿列表→遍历每个元素查数据"（如先拉账户列表，再遍历拉图片）。
**支持 4 种分页模式：**
| 模式 | 配置 | 说明 |
|------|------|------|
| 普通分页 | 默认 | `page` 递增，支持 `total_page` 判断结束 |
| 偏移量分页 | `pagination_mode: "offset"` | `offset = (page-1) * pageSize` |
| 游标分页 | `cursor_pagination: true` | 从响应 `cursor_field` 取游标继续下一页 |
| hasMore 分页 | `response_config.has_more_field` | 从响应 `has_more` 字段判断是否还有下一页 |
**预取（prefetch）**：某些接口需要"先拿列表 → 遍历每个元素查数据"（如先拉账户列表，再遍历拉图片）。并发处理，并发数由 `sync.concurrency` 控制。
```json
"prefetch": {
"url": "/advertiser/get", // 预取接口地址
"url": "/advertiser/get", // 预取接口地址
"method": "GET",
"response_path": "data.list", // 从响应中取值路径
"target_param": "account_id", // 注入主请求的参数名
"value_field": "account_id" // 从预取结果取哪个字段
"response_path": "data.list", // 从响应中取值路径
"target_param": "account_id", // 注入主请求的参数名
"value_field": "account_id" // 从预取结果取哪个字段值
}
```
**并发处理**：有 prefetch 的接口会并发处理每个实体，并发数由 `config.yml` 的 `sync.concurrency` 控制。
**递归（recursive）**：用于树形结构接口（如钉钉部门树），先查根级 → 对每个子节点递归查下级。
**增量同步**：配置了 `time_field` 的接口，增量同步时会自动生成 `filtering` 参数：
```json
{"field": "last_modified_time", "operator": "GREATER_EQUALS", "values": ["<timestamp>"]}
"recursive": {
"key_field": "dept_id", // 递归键字段
"target_param": "dept_id" // 注入到请求的参数名
}
```
**时间分片（range mode）**：快手等平台使用 `beginTime/endTime` 做时间范围查询，全量同步时自动按 3 天分片循环拉取。
#### 3.2.2 `response_config` 字段详解
系统默认解析的响应格式：
@@ -132,9 +184,18 @@ DELETE FROM api_datasource_platform;
{ "code": 0, "message": "success", "data": { "list": [...], "page_info": { "total_page": N } } }
```
可通过 `response_config` 自定义数据路径：
可通过 `response_config` 自定义：
```json
{ "list_path": "data.list" }
{
"list_path": "data.list", // 数据列表路径
"success_field": "code", // 成功标识字段，默认 "code"
"success_value": 0, // 成功值，默认 0
"message_field": "message", // 错误消息字段
"cursor_field": "data.cursor", // 游标字段路径（游标分页用）
"has_more_field": "data.has_more", // hasMore 字段路径
"single_record": false // 是否为单条记录（自动包装为数组）
}
```
### 3.3 `table_definition` 字段详解
@@ -161,53 +222,69 @@ DELETE FROM api_datasource_platform;
| `updater` | VARCHAR(64) | 更新人 |
| `updated_at` | TIMESTAMPTZ | 更新时间 |
| `deleted_at` | TIMESTAMPTZ | 软删除 |
| `raw_data` | JSONB | 原始响应数据 |
| `raw_data` | JSONB | 原始响应数据（含展平的嵌套字段） |
---
## 四、API 接口
基础地址：`http://localhost:3002`
服务端口：`3013`（由 `config.yml` 中 `server.address` 配置）
### 4.1 平台管理
路由规则：系统使用 `RouteRegister` 自动注册，URL 路径由 **struct type name** 转换为 kebab-case 生成。
### 4.1 平台管理 CRUD
基础路径：`/datasource/platform/controller`
| 方法 | 路径 | 说明 |
|------|------|------|
| POST | `/api/datasourcePlatform/createDatasourcePlatform` | 创建平台 |
| GET | `/api/datasourcePlatform/listDatasourcePlatforms` | 列表 |
| GET | `/api/datasourcePlatform/getDatasourcePlatform` | 详情 |
| GET | `/api/datasourcePlatform/getPlatformByCode` | 按编码查 |
| PUT | `/api/datasourcePlatform/updateDatasourcePlatform` | 更新 |
| PUT | `/api/datasourcePlatform/updateDatasourcePlatformStatus` | 更新状态 |
| DELETE | `/api/datasourcePlatform/deleteDatasourcePlatform` | 删除 |
| GET | `/api/datasourcePlatform/getPlatformStatistics` | 统计 |
| POST | `/datasource/platform/controller/createDatasourcePlatform` | 创建平台 |
| GET | `/datasource/platform/controller/listDatasourcePlatforms` | 平台列表 |
| GET | `/datasource/platform/controller/getDatasourcePlatform` | 平台详情 |
| GET | `/datasource/platform/controller/getPlatformByCode` | 按平台编码查询 |
| PUT | `/datasource/platform/controller/updateDatasourcePlatform` | 更新平台 |
| PUT | `/datasource/platform/controller/updateDatasourcePlatformStatus` | 更新平台状态 |
| DELETE | `/datasource/platform/controller/deleteDatasourcePlatform` | 删除平台 |
| GET | `/datasource/platform/controller/getPlatformStatistics` | 平台统计 |
| POST | `/datasource/platform/controller/testPlatformConnection` | 测试平台连接 |
### 4.2 接口管理
### 4.2 接口管理 CRUD
基础路径：`/api/interface/controller`
| 方法 | 路径 | 说明 |
|------|------|------|
| POST | `/api/apiInterface/createApiInterface` | 创建接口 |
| GET | `/api/apiInterface/listApiInterfaces` | 列表 |
| GET | `/api/apiInterface/getApiInterface` | 详情 |
| PUT | `/api/apiInterface/updateApiInterface` | 更新 |
| PUT | `/api/apiInterface/updateApiInterfaceStatus` | 更新状态 |
| DELETE | `/api/apiInterface/deleteApiInterface` | 删除 |
| POST | `/api/interface/controller/createApiInterface` | 创建接口 |
| GET | `/api/interface/controller/listApiInterfaces` | 接口列表 |
| GET | `/api/interface/controller/getApiInterface` | 接口详情 |
| PUT | `/api/interface/controller/updateApiInterface` | 更新接口 |
| PUT | `/api/interface/controller/updateApiInterfaceStatus` | 更新接口状态 |
| DELETE | `/api/interface/controller/deleteApiInterface` | 删除接口 |
### 4.3 同步控制
基础路径：`/sync/ctrl`
| 方法 | 路径 | 说明 |
|------|------|------|
| POST | `/api/sync/ctrl/trigger` | 触发同步 |
| GET | `/api/sync/ctrl/config` | 查询配置 |
| **POST** | **`/sync/ctrl/triggerAll`** | **全平台同步（PPGo_Job 主任务）** |
| **POST** | **`/sync/ctrl/compensate`** | **补偿扫描（PPGo_Job 辅助任务）** |
| POST | `/sync/ctrl/trigger` | 触发单个接口同步 |
| GET | `/sync/ctrl/config` | 查询平台配置 |
**触发同步示例：**
**触发单个接口同步示例：**
```bash
curl -X POST http://localhost:3002/api/sync/ctrl/trigger \
curl -X POST http://localhost:3013/sync/ctrl/trigger \
-H 'Content-Type: application/json' \
-d '{"platformCode":"tencent","interfaceCode":"image","fullSync":true}'
```
参数说明：
- `platformCode`：平台编码（必填）
- `interfaceCode`：接口编码（必填）
- `fullSync`：`true`=全量拉取，`false`=增量（默认自动判断）
响应：
```json
{
@@ -219,89 +296,258 @@ curl -X POST http://localhost:3002/api/sync/ctrl/trigger \
}
```
**全平台同步示例：**
```bash
curl -X POST http://localhost:3013/sync/ctrl/triggerAll -v
```
响应：
```json
{
"success": true,
"message": "全平台同步完成",
"results": [
{ "platformCode": "tencent", "interfaceCode": "account_relation", "success": true },
{ "platformCode": "tencent", "interfaceCode": "image", "success": true }
]
}
```
### 4.4 报表引擎 API
基础路径：`/report`
| 方法 | 路径 | 说明 |
|------|------|------|
| **POST** | **`/report/extractAll`** | **全量抽取（PPGo_Job 报表任务）** |
| POST | `/report/extract` | 按天数据抽取 |
| POST | `/report/backfill` | 批量回填数据 |
| POST | `/report/query` | 用户选择查询 |
| POST | `/report/autoCreateTable` | 自动创建统计宽表 |
| POST | `/report/initTables` | 初始化系统表 |
| GET | `/report/businesses` | 业务列表 |
| GET/POST/DELETE | `/report/business[/save]` | 业务 CRUD |
| GET | `/report/reports` | 报表列表 |
| GET/POST/DELETE | `/report/report[/save][/saveWithFields]` | 报表 CRUD |
| GET | `/report/fields` | 报表字段列表 |
| GET/POST/DELETE | `/report/field[/save]` | 字段 CRUD |
| GET | `/report/extractConfigs` | 抽取配置列表 |
| GET/POST/DELETE | `/report/extractConfig[/save]` | 抽取配置 CRUD |
### 4.5 管理页面
| 路径 | 说明 |
|------|------|
| `GET /admin` | 数据引擎管理后台（平台/接口管理） |
| `GET /admin/report` | 报表引擎管理页面 |
---
## 五、配置文件 (`config.yml`)
```yaml
server:
address: ":3013" # 服务端口
# 报表引擎配置
report:
extract_enabled: false # 是否启用内部定时抽取（推荐 false，由 PPGo_Job 触发）
# 数据同步配置
sync:
page_size: 100 # 每次分页请求条数
concurrency: 5 # 并发处理数（prefetch 遍历实体时）
retry_count: 3 # 最大重试次数
sync_interval_minutes: 60 # 自动同步间隔（分钟）
compensation_interval_seconds: 300 # 补偿扫描间隔（秒）
auto_sync_enabled: true # 是否启用自动同步
tencent:
oauth:
client_id: "1112038234"
client_secret: "GxyjXFbZAs5dnsNQ"
access_token: "4bacfc7c9b0a31f70ec0eb4771f8b542"
refresh_token: "d15b37363a42449026d337708516e95e"
sync_interval_minutes: 60 # 自动同步间隔（内部调度模式用）
compensation_interval_seconds: 300 # 补偿扫描间隔（内部调度模式用）
auto_sync_enabled: false # 是否启用内部自动同步（推荐 false，由 PPGo_Job 触发）
sync_timeout_minutes: 120 # 单次同步超时（分钟）
default_lookback_days: 89 # 全量同步默认回溯天数
default_tenant_id: 1 # 自动同步使用的租户 ID
```
---
## 六、自动同步机制
## 六、调度策略（核心变更）
### 6.1 启动流程
系统支持 **两种调度模式**，通过配置开关切换：
1. 服务启动 → `InitAndStartAutoSync` 在 goroutine 中启动调度器
2. 自动扫描所有 ACTIVE 平台下有 `table_definition` 的接口
3. 无 `sync_tracker` 记录 → 全量拉取；有记录 → 增量拉取
### 6.1 推荐模式：PPGo_Job 外部调度
### 6.2 增量同步原理
```
config.yml:
report.extract_enabled: false
sync.auto_sync_enabled: false ← 默认值
```
- `sync_tracker` 表记录每个接口的最后同步时间
- 配置了 `time_field` 的接口，增量时生成 `filtering=[{"field":"last_modified_time","operator":"GREATER_EQUALS","values":["<时间戳>"]}]`
- 不支持时间过滤的接口（如 audio/advertiser）每次全量，`ON CONFLICT` 去重
内部调度器不启动，由 PPGo_Job 定时调用 HTTP 端点触发。详情见第八章。
### 6.2 备选模式：内部自动调度
```
config.yml:
report.extract_enabled: true
sync.auto_sync_enabled: true
```
- 同步引擎：进程启动后自动循环执行全平台同步，每次完成后等待 `sync_interval_minutes` 再继续
- 报表引擎：每天凌晨 2:00 执行全量抽取（gcron `0 0 2 * * *`）
- 补偿调度器：独立于 `auto_sync_enabled`，始终随服务启动（内部模式）
### 6.3 异常中断恢复
- 同步开始前写 `sync_tracker.sync_status='running'`
- 同步完成后写 `'success'`
- 重启检测到 `'running'` → 日志告警 → 重新全量
- 重启检测到 `'running'` → 日志告警 → 重新全量同步
---
## 七、补偿机制
## 七、Token 自动刷新机制
### 7.1 工作原理
1. 同步失败 → 自动写入 `sync_task_log`（status=failed）
2. 补偿调度器（随主服务自动启动）每 N 秒扫描 failed 记录
3. 对未达最大重试次数的任务，调用 `SyncByConfig` 重试
4. 重试间隔按退避策略递增：5min → 15min → 30min → 60min → 120min
5. 达最大次数 → 标记 `manual_review`，等待人工介入
系统支持 OAuth2 token 过期自动刷新，**不需要人工介入**：
### 7.2 配置
1. `ApiClient` 发送请求 → 收到 401 或 token 过期错误
2. 自动调用平台的 `RefreshToken` 回调函数（如 `RefreshTencentToken`）
3. 刷新成功后**立即重试**原请求（不消耗退避重试次数）
4. 每个 `doRequest` 最多刷新一次，避免死循环
5. 刷新后的 token 同时更新内存和数据库
```yaml
sync:
compensation_interval_seconds: 300 # 扫描间隔
retry_count: 3 # 最大重试次数
```
doRequest
├── execute() → HTTP 401 / token_expired
├── isTokenExpiredError() → true
├── RefreshTencentToken()
│ ├── POST https://api.e.qq.com/oauth/refresh_token
│ ├── 更新 platform.Token / platform.AccessToken
│ └── 更新数据库 api_datasource_platform.token + auth_config
└── continue（用新 token 重试，不消耗重试次数）
```
补偿调度器随主服务自动启动，无需手动运行。
### 7.2 token 过期错误检测
系统通过关键字匹配识别 token 过期错误：
```go
tokenExpiredKeywords := []string{
"TOKEN过期", "token过期", "token_expired", "Token过期",
"result=28", // 快手
"access_token", "token已失效", "token无效",
}
```
### 7.3 token 过期时的处理策略
| 场景 | 行为 |
|------|------|
| 自动同步中 | 跳过该平台剩余接口，继续处理下一个平台 |
| 补偿扫描中 | 标记为 `manual_review`，等待人工处理，不再重试 |
| 手动触发 | 自动刷新后重试 |
### 7.4 腾讯广告 OAuth2 刷新（当前已实现）
已支持腾讯广告 OAuth2 token 自动刷新，使用 `refresh_token` 换取新的 `access_token`：
- 刷新端点：`POST https://api.e.qq.com/oauth/refresh_token`
- 超时：30 秒
- 互斥锁：防止并发刷新（`refreshTokenMu sync.Mutex`）
---
## 八、当前已配置接口（腾讯广告）
## 八、PPGo_Job 定时任务配置
| 接口编码 | 名称 | 方法+路径 | 类型 | 增量 |
|------|------|------|------|------|
| `account_relation` | 账户列表 | `GET /advertiser/get` | 单接口分页 | 不支持 |
| `image` | 图片素材 | `GET /images/get` | prefetch 遍历账户 | ✅ `last_modified_time` |
| `video` | 视频素材 | `GET /videos/get` | prefetch 遍历账户 | ✅ `last_modified_time` |
| `audio` | 音频素材 | `POST /muse_audios/get` | 单接口 POST | 不支持 |
系统推荐由 **PPGo_Job** 外部调度，需在 PPGo_Job 中配置以下任务。
三个素材表自动包含 `verify_status DEFAULT 'PENDING'`、`verified_at`、`verified_by` 校验字段。
### 任务 1：全平台数据同步
### 图片/视频同步流程
| 字段 | 值 |
|------|-----|
| 任务名称 | `数据引擎 - 全平台同步` |
| 请求 URL | `http://<data-engine-host>:3013/sync/ctrl/triggerAll` |
| 请求方法 | `POST` |
| Header | `Content-Type: application/json` |
| Body | 空（不需要请求体） |
| Cron | `0 0 * * * ?`（每小时整点） |
| 超时 | `7200` 秒（2小时） |
| 说明 | 遍历所有 ACTIVE 平台下有 table_definition 的接口，自动判断全量/增量。所有平台的同步都在同一任务中完成 |
执行流程：
```
triggerAll
└── runAutoSync()
├── 查所有 ACTIVE 平台
├── 对每个平台 → 查所有有 table_definition 的接口
│ ├── 查 sync_tracker → 有记录=增量，无记录=全量
│ ├── SyncByConfig(platformCode, interfaceCode)
│ └── token 过期 → 跳过该平台剩余接口
└── 返回每个接口的执行结果
```
### 任务 2：失败任务补偿
| 字段 | 值 |
|------|-----|
| 任务名称 | `数据引擎 - 补偿扫描` |
| 请求 URL | `http://<data-engine-host>:3013/sync/ctrl/compensate` |
| 请求方法 | `POST` |
| Header | `Content-Type: application/json` |
| Body | 空 |
| Cron | `0 0/5 * * * ?`（每5分钟） |
| 超时 | `600` 秒（10分钟） |
| 说明 | 扫描 sync_task_log 中 status=failed 的任务，按退避策略重试 |
补偿退避策略：
```
第1次重试 → 等待 5分钟
第2次重试 → 等待 15分钟
第3次重试 → 等待 30分钟
第4次重试 → 等待 60分钟
第5次重试 → 等待 120分钟（封顶）
达最大次数 → 标记 manual_review，等待人工处理
```
### 任务 3：报表每日抽取（可选）
| 字段 | 值 |
|------|-----|
| 任务名称 | `数据引擎 - 每日抽取` |
| 请求 URL | `http://<data-engine-host>:3013/report/extractAll` |
| 请求方法 | `POST` |
| Header | `Content-Type: application/json` |
| Body | 空 |
| Cron | `0 0 2 * * ?`（每天凌晨2点） |
| 超时 | `3600` 秒（1小时） |
| 说明 | 遍历所有启用的业务+报表，抽取前一天数据到统计宽表 |
---
## 九、已配置平台
### 9.1 腾讯广告
| 项目 | 值 |
|------|-----|
| platform_code | `tencent` |
| API Base | `https://api.e.qq.com/v3.0` |
| 认证 | OAUTH2（access_token 在 URL query 中，支持自动刷新） |
**4 个接口：**
| 接口编码 | 名称 | 方法+路径 | 同步模式 | 增量 |
|---------|------|----------|---------|------|
| `account_relation` | 账户列表 | `GET /advertiser/get` | 普通分页 | 不支持（全量+去重） |
| `image` | 图片素材 | `GET /images/get` | **预取**（遍历账户） | ✅ `last_modified_time` |
| `video` | 视频素材 | `GET /videos/get` | **预取**（遍历账户） | ✅ `last_modified_time` |
| `audio` | 音频素材 | `POST /muse_audios/get` | 普通分页 POST | 不支持（全量+去重） |
**图片/视频同步流程：**
```
SyncByConfig("tencent", "image")
├── ① 预取: 分页拉取 /advertiser/get → 774 个 account_id
├── ① 预取: 分页拉取 /advertiser/get → 所有 account_id
│ 数据同时存入 tencent_account_relation 表
├── ② 并发遍历账户（config.yml concurrency=5）
│ 每个 account_id → GET /images/get?account_id=xxx&page=1&page_size=100
@@ -310,54 +556,102 @@ SyncByConfig("tencent", "image")
└── ③ 更新 sync_tracker 记录同步时间
```
### 音频同步流程
### 9.2 其他平台
```
单次 POST /muse_audios/get → 分页拉全量 → upsert 到 tencent_audio
| 平台 | seed 文件 | 接口数 |
|------|----------|--------|
| 钉钉 | `seed_data_dingtalk.sql` | ~10个 |
| 钉钉 HRM | `seed_data_dingtalk_hrm.sql` | ~5个 |
| 钉钉智能薪酬 | `seed_data_dingtalk_salary.sql` | ~3个 |
| 快手 | `seed_data_kuaishou.sql` | ~5个 |
---
## 十、补偿机制
### 10.1 工作原理
1. 同步失败 → 自动写入 `sync_task_log`（status=failed）
2. 补偿扫描（由 PPGo_Job 调用 `/sync/ctrl/compensate`）扫描 failed 记录
3. 对未达最大重试次数的任务，调用 `SyncByConfig` 重试
4. 重试间隔按退避策略递增：5min → 15min → 30min → 60min → 120min
5. 达最大次数 → 标记 `manual_review`，等待人工介入
6. Token 过期错误 → 直接标记 `manual_review`，不再重试
### 10.2 配置
```yaml
sync:
compensation_interval_seconds: 300 # 扫描间隔（内部调度用）
retry_count: 3 # 最大重试次数
```
---
## 九、快速开始
## 十一、快速开始
```bash
# 1. 建表
psql -h localhost -U postgres -d data-engine -f sql/init_core_tables.sql
# 1. 建核心表
PGPASSWORD='xxx' psql -h <host> -p 15432 -U postgres -d engine -f sql/init_core_tables.sql
# 2. 初始化数据
psql -h localhost -U postgres -d data-engine -f sql/seed_data.sql
# 2. 初始化腾讯广告数据
PGPASSWORD='xxx' psql -h <host> -p 15432 -U postgres -d engine -f sql/seed_data.sql
# 3. 启动
# 3. 配置 PPGo_Job（推荐）或启用内部调度
# 见第六章调度策略 和 第八章 PPGo_Job 配置
# 4. 启动服务
cd D:/WorkSpace/HDWL/data-engine
go run main.go
```
启动后自动同步和补偿调度器自动运行。也可手动触发：
```bash
# 触发图片全量同步
curl -X POST http://localhost:3002/api/sync/ctrl/trigger \
# 5. 手动测试同步
curl -X POST http://localhost:3013/sync/ctrl/trigger \
-H 'Content-Type: application/json' \
-d '{"platformCode":"tencent","interfaceCode":"image","fullSync":true}'
```
---
## 十、常见问题
## 十二、常见问题
**Q: 响应格式不符合 `{code: 0, data: {list: [...]}}` 怎么办？**
修改 `dynamic_sync.go` 的 `parseResp` 函数。
通过接口管理的 `response_config` 自定义 `list_path`、`success_field`、`success_value`。如需新的解析逻辑，修改 `dynamic_sync.go` 的 `parseRespExt` 函数。
**Q: 如何添加新平台？**
调用平台管理 API 创建平台 → 调用接口管理 API 创建接口（带 `table_definition`）→ 系统自动建表并同步。
1. 调用平台管理 API 创建平台配置
2. 调用接口管理 API 创建接口（含 `table_definition`）
3. 平台状态设为 `ACTIVE`，接口状态设为 `active`
4. 系统下次同步时自动建表并拉取数据
**Q: prefetch 的响应格式要求？**
必须是 JSON，`response_path` 指向一个数组。如 `response_path: "data.list"` 从 `data.list` 取值。
必须是 JSON，`response_path` 指向一个数组。如 `response_path: "data.list"` 从 `{"data":{"list":[...]}}` 取值。
**Q: 如何排查同步失败？**
1. `GET /api/sync/ctrl/config?platformCode=xxx` 查看配置
2. 查询 `sync_task_log` 表看失败记录
3. 补偿调度器会自动重试，日志会打印重试过程
1. 查询 `sync_task_log` 表看失败记录和错误信息
2. `GET /sync/ctrl/config?platformCode=tencent` 查看平台配置
3. 检查服务日志中 `equivalent curl` 输出，手动复现请求
4. PPGo_Job 补偿任务会自动重试，日志会打印重试过程
**Q: Token 过期了怎么办？**
如果配置了 OAuth2 且 `refresh_token` 有效，系统会自动刷新。手动强制刷新可用 `_update_token.go`：
```bash
# 先修改 _update_token.go 中 token 和 refresh_token 的值
cd D:/WorkSpace/HDWL/data-engine
go run _update_token.go
```
**Q: 内部调度和 PPGo_Job 调度有什么区别？**
| 对比项 | 内部调度 | PPGo_Job 调度（推荐） |
|-------|---------|-------------------|
| 依赖 | 无外部依赖 | 需要 PPGo_Job 服务 |
| 配置 | `auto_sync_enabled: true` | `auto_sync_enabled: false` |
| 控制 | 进程内循环，重启失效 | 统一调度中心，支持暂停/手动执行 |
| 多服务协调 | 不支持 | 支持 |
| 运维 | 需要登录服务器 | 通过 Web UI 管理所有任务 |
go.mod
+1
View File
@@ -40,6 +40,7 @@ require (
github.com/go-viper/mapstructure/v2 v2.5.0 // indirect
github.com/goccy/go-json v0.10.6 // indirect
github.com/gogf/gf/contrib/registry/consul/v2 v2.9.5 // indirect
github.com/gogf/gf/contrib/trace/otlphttp/v2 v2.9.5 // indirect
github.com/gogo/protobuf v1.3.2 // indirect
github.com/golang-jwt/jwt/v5 v5.3.1 // indirect
github.com/golang/glog v1.2.5 // indirect
go.sum
+8 -8
View File
@@ -5,8 +5,8 @@ github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03
github.com/BurntSushi/toml v1.6.0 h1:dRaEfpa2VI55EwlIW72hMRHdWouJeRF7TPYhI+AUQjk=
github.com/BurntSushi/toml v1.6.0/go.mod h1:ukJfTF/6rtPPRCnwkur4qwRxa8vTRFBF0uk2lLoLwho=
github.com/DataDog/datadog-go v3.2.0+incompatible/go.mod h1:LButxg5PwREeZtORoXG3tL4fMGNddJ+vMq1mwgfaqoQ=
github.com/Masterminds/semver/v3 v3.5.0 h1:kQceYJfbupGfZOKZQg0kou0DgAKhzDg2NZPAwZ/2OOE=
github.com/Masterminds/semver/v3 v3.5.0/go.mod h1:4V+yj/TJE1HU9XfppCwVMZq3I84lprf4nC11bSS5beM=
github.com/Masterminds/semver/v3 v3.4.0 h1:Zog+i5UMtVoCU8oKka5P7i9q9HgrJeGzI9SA1Xbatp0=
github.com/Masterminds/semver/v3 v3.4.0/go.mod h1:4V+yj/TJE1HU9XfppCwVMZq3I84lprf4nC11bSS5beM=
github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
@@ -196,8 +196,8 @@ github.com/hashicorp/go-uuid v1.0.0/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/b
github.com/hashicorp/go-uuid v1.0.1/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
github.com/hashicorp/go-uuid v1.0.3 h1:2gKiV6YVmrJ1i2CKKa9obLvRieoRGviZFL26PcT/Co8=
github.com/hashicorp/go-uuid v1.0.3/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
github.com/hashicorp/go-version v1.9.0 h1:CeOIz6k+LoN3qX9Z0tyQrPtiB1DFYRPfCIBtaXPSCnA=
github.com/hashicorp/go-version v1.9.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA=
github.com/hashicorp/go-version v1.2.1 h1:zEfKbn2+PDgroKdiOzqiE8rsmLqU2uwi5PB5pBJ3TkI=
github.com/hashicorp/go-version v1.2.1/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA=
github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
github.com/hashicorp/golang-lru v1.0.2 h1:dV3g9Z/unq5DpblPpw+Oqcv4dU/1omnb4Ok8iPY6p1c=
github.com/hashicorp/golang-lru v1.0.2/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
@@ -218,8 +218,8 @@ github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI
github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
github.com/klauspost/compress v1.18.6 h1:2jupLlAwFm95+YDR+NwD2MEfFO9d4z4Prjl1XXDjuao=
github.com/klauspost/compress v1.18.6/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
github.com/klauspost/cpuid/v2 v2.3.0 h1:S4CRMLnYUhGeDFDqkGriYKdfoFlDnMtqTiI/sFzhA9Y=
github.com/klauspost/cpuid/v2 v2.3.0/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0=
github.com/klauspost/cpuid/v2 v2.2.11 h1:0OwqZRYI2rFrjS4kvkDnqJkKHdHaRnCm68/DY4OxRzU=
github.com/klauspost/cpuid/v2 v2.2.11/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0=
github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc=
github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
@@ -256,8 +256,8 @@ github.com/mattn/go-runewidth v0.0.24/go.mod h1:XBkDxAl56ILZc9knddidhrOlY5R/pDhg
github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
github.com/miekg/dns v1.1.26/go.mod h1:bPDLeHnStXmXAq1m/Ch/hvfNHr14JKNPMBo3VZKjuso=
github.com/miekg/dns v1.1.41/go.mod h1:p6aan82bvRIyn+zDIv9xYNUpwa73JcSh9BKwknJysuI=
github.com/miekg/dns v1.1.72 h1:vhmr+TF2A3tuoGNkLDFK9zi36F2LS+hKTRW0Uf8kbzI=
github.com/miekg/dns v1.1.72/go.mod h1:+EuEPhdHOsfk6Wk5TT2CzssZdqkmFhf8r+aVyDEToIs=
github.com/miekg/dns v1.1.63 h1:8M5aAw6OMZfFXTT7K5V0Eu5YiiL8l7nUAkyN6C9YwaY=
github.com/miekg/dns v1.1.63/go.mod h1:6NGHfjhpmr5lt3XPLuyfDJi5AXbNIPM9PY6H6sF1Nfs=
github.com/mitchellh/cli v1.1.0/go.mod h1:xcISNoH86gajksDmfB23e/pu+B+GeFRMYmoHXxx3xhI=
github.com/mitchellh/go-homedir v1.1.0 h1:lukF9ziXFxDFPkA1vsr5zpc1XuPDn/wFntq5mG+4E0Y=
github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
main.go
+14 -5
View File
@@ -26,11 +26,20 @@ func main() {
ctx := gctx.New()
defer jaeger.ShutDown(ctx)
// 启动自动同步（后台循环执行，首次全量后续增量）
syncSvc.InitAndStartAutoSync(ctx)
// 启动每日自动抽取（每天凌晨2点抽取前一天数据）
report.StartDailyExtractJob()
// 内部调度开关（默认关闭，由 PPGo_Job 通过 HTTP 端点统一调度）
extractEnabled := g.Cfg().MustGet(ctx, "report.extract_enabled", false).Bool()
syncEnabled := g.Cfg().MustGet(ctx, "sync.auto_sync_enabled", false).Bool()
if extractEnabled || syncEnabled {
if syncEnabled {
syncSvc.InitAndStartAutoSync(ctx)
}
if extractEnabled {
report.StartDailyExtractJob()
}
logrus.Infof("[main] 内部调度已启用 extract=%v sync=%v", extractEnabled, syncEnabled)
} else {
logrus.Info("[main] 内部调度已禁用，由 PPGo_Job 外部调度")
}
http.RouteRegister([]interface{}{
// 接口管理
service/sync/api_client.go
+95 -22
View File
@@ -25,11 +25,27 @@ type ApiResult struct {
DurationMs int64
}
// TokenRefreshFunc token 刷新回调函数
// 由具体平台的 RefreshToken 函数实现，在检测到 401/token 过期时自动调用
type TokenRefreshFunc func(ctx context.Context, config *PlatformConfig) error
// ApiClient 通用 API 客户端
type ApiClient struct {
config *PlatformConfig
client *http.Client
rateLimiter *time.Ticker // 限流 ticker，可被 GC
config *PlatformConfig
client *http.Client
rateLimiter *time.Ticker // 限流 ticker，可被 GC
tokenRefreshFunc TokenRefreshFunc
bodyType string // 请求体序列化方式: ""=form-encoded, "json"=JSON
}
// SetBodyType 设置请求体序列化方式
func (c *ApiClient) SetBodyType(t string) {
c.bodyType = t
}
// SetTokenRefreshFunc 设置 token 刷新回调
func (c *ApiClient) SetTokenRefreshFunc(fn TokenRefreshFunc) {
c.tokenRefreshFunc = fn
}
// NewApiClient 创建客户端
@@ -91,11 +107,27 @@ func (c *ApiClient) doRequest(ctx context.Context, method, path string, body int
retryDelay = 1 * time.Second
}
refreshed := false // 标记是否已刷新过 token（每个 doRequest 最多刷新一次）
for attempt := 0; attempt <= maxRetries; attempt++ {
result, err = c.execute(ctx, method, path, body, paramsInQuery)
if err == nil {
return result, nil
}
// token 过期 → 刷新一次，立即重试（不占退避重试次数）
if isTokenExpiredError(err) && c.tokenRefreshFunc != nil && !refreshed {
logrus.Warnf("检测到 token 过期 [%s]，尝试自动刷新...", path)
if refreshErr := c.tokenRefreshFunc(ctx, c.config); refreshErr != nil {
logrus.Errorf("Token 刷新失败: %v", refreshErr)
} else {
refreshed = true
logrus.Info("Token 刷新成功，立即重试请求")
// 不递减 attempt，直接用新 token 重试
continue
}
}
logrus.Warnf("请求失败 (attempt %d/%d): %v", attempt+1, maxRetries+1, err)
if attempt < maxRetries {
time.Sleep(retryDelay * time.Duration(attempt+1))
@@ -122,28 +154,50 @@ func (c *ApiClient) execute(ctx context.Context, method, path string, body inter
fullURL = c.applyAuthURL(fullURL)
// 将 URL 认证参数注入 body 并清除 URL（避免重复参数）
// 注意：token_in_query=true 的认证方式（如腾讯广告），token 必须留在 URL 中，不移入 body
var reqBody io.Reader
var reqBodyBytes []byte
if body != nil && !paramsInQuery {
if paramsMap, ok := body.(map[string]interface{}); ok {
// 从 URL 注入认证参数到 body
if parsed, _ := url.Parse(fullURL); parsed != nil {
q := parsed.Query()
for k, vs := range q {
if len(vs) > 0 {
if _, exists := paramsMap[k]; !exists {
paramsMap[k] = vs[0]
// token_in_query=true 时认证参数必须留在 URL 中，不注入 body
tokenInQuery := false
if c.config.AuthConfig != nil {
if tiq, _ := c.config.AuthConfig["token_in_query"].(bool); tiq {
tokenInQuery = true
}
}
if !tokenInQuery {
if paramsMap, ok := body.(map[string]interface{}); ok {
// 从 URL 注入认证参数到 body
if parsed, _ := url.Parse(fullURL); parsed != nil {
q := parsed.Query()
for k, vs := range q {
if len(vs) > 0 {
if _, exists := paramsMap[k]; !exists {
paramsMap[k] = vs[0]
}
q.Del(k)
}
q.Del(k)
}
parsed.RawQuery = q.Encode()
fullURL = parsed.String()
}
}
}
if paramsMap, ok := body.(map[string]interface{}); ok {
if c.bodyType == "json" {
// JSON body（如腾讯广告 POST 接口）
b, err := json.Marshal(paramsMap)
if err != nil {
return nil, fmt.Errorf("JSON序列化请求体失败: %w", err)
}
parsed.RawQuery = q.Encode()
fullURL = parsed.String()
reqBodyBytes = b
reqBody = bytes.NewBuffer(b)
} else {
// Form body
formStr := c.buildFormBody(paramsMap)
reqBodyBytes = []byte(formStr)
reqBody = strings.NewReader(formStr)
}
// Form body
formStr := c.buildFormBody(paramsMap)
reqBodyBytes = []byte(formStr)
reqBody = strings.NewReader(formStr)
} else {
b, err := json.Marshal(body)
if err != nil {
@@ -181,9 +235,13 @@ func (c *ApiClient) execute(ctx context.Context, method, path string, body inter
// 打印等效 curl
curlCmd := fmt.Sprintf("curl -X %s '%s'", method, fullURL)
if reqBodyBytes != nil && len(reqBodyBytes) > 0 {
for _, pair := range strings.Split(string(reqBodyBytes), "&") {
if pair != "" {
curlCmd += fmt.Sprintf(" --data-urlencode '%s'", pair)
if c.bodyType == "json" {
curlCmd += fmt.Sprintf(" -H 'Content-Type: application/json' -d '%s'", string(reqBodyBytes))
} else {
for _, pair := range strings.Split(string(reqBodyBytes), "&") {
if pair != "" {
curlCmd += fmt.Sprintf(" --data-urlencode '%s'", pair)
}
}
}
}
@@ -198,7 +256,11 @@ func (c *ApiClient) execute(ctx context.Context, method, path string, body inter
req.Header.Set("User-Agent", "data-engine/1.0")
if body != nil && !paramsInQuery {
if _, ok := body.(map[string]interface{}); ok {
req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
if c.bodyType == "json" {
req.Header.Set("Content-Type", "application/json")
} else {
req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
}
} else {
req.Header.Set("Content-Type", "application/json")
}
@@ -219,6 +281,13 @@ func (c *ApiClient) execute(ctx context.Context, method, path string, body inter
if resp.StatusCode >= 400 {
return result, fmt.Errorf("HTTP %d: %s", resp.StatusCode, string(respBody))
}
// 检查腾讯API业务级token错误（HTTP 200但code=11002，access_token无效）
var apiResp struct {
Code int `json:"code"`
}
if err := json.Unmarshal(respBody, &apiResp); err == nil && apiResp.Code == 11002 {
return result, fmt.Errorf("token expired: code=11002, msg=%s", string(respBody))
}
return result, nil
}
@@ -283,6 +352,10 @@ func (c *ApiClient) buildFormBody(params map[string]interface{}) string {
}
case int, int8, int16, int32, int64:
q.Set(k, fmt.Sprintf("%d", val))
case []interface{}, map[string]interface{}:
// 数组或对象需要 JSON 序列化
b, _ := json.Marshal(v)
q.Set(k, string(b))
default:
q.Set(k, fmt.Sprintf("%v", v))
}
service/sync/compensation.go
+45 -4
View File
@@ -13,6 +13,22 @@ import (
"github.com/sirupsen/logrus"
)
// CompensateResult 单次补偿扫描结果
type CompensateResult struct {
TotalFailed int `json:"totalFailed"`
Retried int `json:"retried"`
Succeeded int `json:"succeeded"`
Failed int `json:"failed"`
MaxRetryReached int `json:"maxRetryReached"`
}
// TriggerCompensation 手动触发一次补偿扫描（等价于 runCompensation 的一次执行）
// 由 HTTP 端点调用，用于 PPGo_Job 调度
func TriggerCompensation(ctx context.Context) *CompensateResult {
ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1})
return runCompensation(ctx)
}
// StartCompensation 启动补偿调度器（在后台循环执行）
func StartCompensation(ctx context.Context) {
sec := g.Cfg().MustGet(ctx, "sync.compensation_interval_seconds", 300).Int()
@@ -38,7 +54,7 @@ func StartCompensation(ctx context.Context) {
}
}
func runCompensation(ctx context.Context) {
func runCompensation(ctx context.Context) *CompensateResult {
logrus.Info("=== 开始补偿扫描 ===")
tasks, err := dao.SyncTaskLog.QueryFailedTasks(ctx, &taskDto.QueryFailedTasksReq{
@@ -47,11 +63,16 @@ func runCompensation(ctx context.Context) {
})
if err != nil {
logrus.Errorf("查询失败任务异常: %v", err)
return
return &CompensateResult{}
}
result := &CompensateResult{
TotalFailed: len(tasks),
}
if len(tasks) == 0 {
logrus.Info("当前没有需要补偿的任务")
return
return result
}
logrus.Infof("发现 %d 个失败任务", len(tasks))
@@ -64,6 +85,7 @@ func runCompensation(ctx context.Context) {
Status: "manual_review",
ErrorMessage: fmt.Sprintf("已达最大重试次数 %d", task.MaxRetry),
})
result.MaxRetryReached++
continue
}
@@ -83,9 +105,25 @@ func runCompensation(ctx context.Context) {
RetryCount: &retryCount,
})
_, err := SyncByConfig(ctx, platformCode, interfaceCode, false)
result.Retried++
_, err := SyncByConfig(ctx, platformCode, interfaceCode, false) // 补偿用增量，数据已 upsert 落库，不需要全量重拉
if err != nil {
logrus.Errorf("补偿失败: %v", err)
// token 过期是平台级别问题，重试无意义，直接标记为 manual_review 等待人工处理
if isTokenExpiredError(err) {
logrus.Warnf("平台 [%s] token 过期，标记为 manual_review 等待人工处理", platformCode)
dao.SyncTaskLog.Update(ctx, &taskDto.UpdateSyncTaskLogReq{
ID: task.Id,
Status: "manual_review",
ErrorMessage: err.Error(),
ErrorCode: "TOKEN_EXPIRED",
})
result.Failed++
continue
}
backoff := []int{5, 15, 30, 60, 120}
waitMin := 5
if retryCount <= len(backoff) {
@@ -101,6 +139,7 @@ func runCompensation(ctx context.Context) {
ErrorCode: "COMPENSATION_FAILED",
NextRetryTime: nextRetry,
})
result.Failed++
} else {
logrus.Infof("补偿成功: %s/%s", platformCode, interfaceCode)
now := time.Now()
@@ -109,8 +148,10 @@ func runCompensation(ctx context.Context) {
Status: "success",
CompletedAt: now,
})
result.Succeeded++
}
}
logrus.Info("=== 补偿扫描完成 ===")
return result
}
service/sync/data_writer.go
+4
View File
@@ -4,6 +4,8 @@ import (
"context"
"time"
"dataengine/utils"
"gitea.redpowerfuture.com/red-future/common/db/gfdb"
"github.com/sirupsen/logrus"
)
@@ -15,12 +17,14 @@ func InsertRows(ctx context.Context, tableName string, conflictKeys []string, ro
}
now := time.Now()
tenantId := utils.GetCurrentTenantId(ctx)
for i := range rows {
if rows[i] == nil {
rows[i] = make(map[string]interface{})
}
// 始终覆盖 updated_at；不设置 created_at 让数据库维护首次值（upsert 时不会覆盖）
rows[i]["updated_at"] = now
rows[i]["tenant_id"] = tenantId
}
batchSize := 100
service/sync/dynamic_sync.go
+274 -92
View File
@@ -13,8 +13,10 @@ import (
dao "dataengine/dao/copydata"
taskDto "dataengine/model/dto/copydata"
entity "dataengine/model/entity/dict"
"dataengine/utils"
"gitea.redpowerfuture.com/red-future/common/db/gfdb"
"github.com/gogf/gf/v2/database/gdb"
"github.com/gogf/gf/v2/os/grpool"
"github.com/sirupsen/logrus"
)
@@ -46,6 +48,45 @@ type RecursiveConfig struct {
TargetParam string `json:"target_param"`
}
// isInterfaceDueForFullSync 检查单个接口是否该做全量同步了
// 即使调用方要求增量，如果距上次全量超过阈值，自动升级为全量
func isInterfaceDueForFullSync(ctx context.Context, platformCode, interfaceCode string) bool {
interval := GetFullSyncIntervalHours(ctx)
if interval <= 0 {
return false
}
var lastFullSync int64
// 使用 gfdb.Raw 确保 COALESCE 表达式不被框架改写
record, err := gfdb.DB(ctx).Model(ctx, consts.SyncTrackerTable).
Where("platform_code", platformCode).
Where("interface_code", interfaceCode).
Value("COALESCE(last_full_sync_time, 0)")
if err != nil {
logrus.Warnf("[%s/%s] 查询 last_full_sync_time 失败: %v，回退增量", platformCode, interfaceCode, err)
return false
}
if record != nil {
lastFullSync = record.Int64()
}
logrus.Infof("[%s/%s] 读取到 last_full_sync_time=%d", platformCode, interfaceCode, lastFullSync)
if lastFullSync == 0 {
logrus.Infof("[%s/%s] 从未全量，自动升级为全量同步", platformCode, interfaceCode)
return true
}
elapsed := time.Now().Unix() - lastFullSync
if elapsed > int64(interval)*3600 {
logrus.Infof("[%s/%s] 距上次全量 %d 小时（阈值 %d 小时），自动升级为全量同步",
platformCode, interfaceCode, elapsed/3600, interval)
return true
}
logrus.Debugf("[%s/%s] 距上次全量 %d 小时，未到阈值 %d 小时，执行增量", platformCode, interfaceCode, elapsed/3600, interval)
return false
}
// SyncByConfig 执行同步
func SyncByConfig(ctx context.Context, platformCode, interfaceCode string, isFullSync bool) (*SyncResult, error) {
// 创建超时 context 防止单次同步卡死
@@ -92,21 +133,38 @@ func SyncByConfig(ctx context.Context, platformCode, interfaceCode string, isFul
return nil, fmt.Errorf("建表失败: %w", err)
}
// 先读取已有的 last_sync_time，供 markSyncRunning 保留（防止全量覆盖为0）
existingLastSync := getLastSyncTime(ctx, platformCode, interfaceCode)
// 即使调用方要求增量，如果距上次全量超过阈值，自动升级为全量
if !isFullSync {
isFullSync = isInterfaceDueForFullSync(ctx, platformCode, interfaceCode)
}
// 检查上次同步状态（在标记 running 之前检查）
prevStatus := getSyncStatus(ctx, platformCode, interfaceCode)
lastSyncTime := int64(0)
if !isFullSync {
lastSyncTime = getLastSyncTime(ctx, platformCode, interfaceCode)
lastSyncTime = existingLastSync
}
if prevStatus == "running" {
logrus.Warnf("检测到上次同步异常中断 [%s/%s]，将重新全量同步", platformCode, interfaceCode)
lastSyncTime = 0
if lastSyncTime <= 0 {
logrus.Warnf("检测到上次同步异常中断 [%s/%s]，无有效同步点，执行全量", platformCode, interfaceCode)
lastSyncTime = 0
} else {
logrus.Infof("检测到上次同步异常中断 [%s/%s]，数据已落库，从同步点 %d 恢复增量", platformCode, interfaceCode, lastSyncTime)
}
}
// 标记同步开始（保留 last_sync_time 不变，状态设为 running）
markSyncRunning(ctx, platformCode, interfaceCode, lastSyncTime)
// 标记同步开始（不写入 last_sync_time，由 updateSyncTime 独家管理）
markSyncRunning(ctx, platformCode, interfaceCode)
api := NewApiClient(platform)
api.SetTokenRefreshFunc(RefreshTencentToken)
// 支持 body_type: "json" 配置（如腾讯广告 POST 接口需要 JSON body 而非 form-encoded）
if bt, ok := iface.RequestConfig["body_type"].(string); ok {
api.SetBodyType(bt)
}
defer api.Close()
prefetch := parsePrefetchConfig(iface.RequestConfig)
@@ -202,10 +260,10 @@ func syncSingleAPI(ctx context.Context, api *ApiClient, platform *PlatformConfig
inQuery := paramsInQuery(iface)
method := string(iface.Method)
// 游标参数名
// 游标参数名（与 page_param 分开，腾讯游标模式使用独立的 cursor 参数）
cursorParam := "cursor"
if p, ok := iface.RequestConfig["page_param"].(string); ok && p != "" {
cursorParam = p
if cp, ok := iface.RequestConfig["cursor_param"].(string); ok && cp != "" {
cursorParam = cp
}
cursorMode := isCursorPagination(iface)
@@ -314,11 +372,14 @@ func syncSingleAPI(ctx context.Context, api *ApiClient, platform *PlatformConfig
}
}
if globalMaxTime <= 0 {
globalMaxTime = time.Now().Unix()
if maxTime > globalMaxTime {
globalMaxTime = maxTime
}
}
updateSyncTime(ctx, platform.PlatformCode, iface.Code, globalMaxTime)
if globalMaxTime <= 0 {
globalMaxTime = time.Now().Unix()
}
updateSyncTime(ctx, platform.PlatformCode, iface.Code, globalMaxTime, taskType)
result.Duration = fmt.Sprintf("%.1fs", time.Since(start).Seconds())
logrus.Infof("同步完成 - 表:%s, %d条, 写入%d条, 耗时%s", td.TableName, result.TotalRows, result.InsertedRows, result.Duration)
@@ -401,7 +462,14 @@ func syncWithPrefetch(ctx context.Context, api *ApiClient, platform *PlatformCon
if cp, ok := prefetchIface.RequestConfig["cursor_pagination"].(bool); ok {
prefetchIsCursor = cp
}
if p, ok := prefetchIface.RequestConfig["page_param"].(string); ok && p != "" {
// 游标模式使用独立的 cursor_param，不与 page_param 共用
if prefetchIsCursor {
if cp, ok := prefetchIface.RequestConfig["cursor_param"].(string); ok && cp != "" {
prefetchPageParam = cp
} else {
prefetchPageParam = "cursor"
}
} else if p, ok := prefetchIface.RequestConfig["page_param"].(string); ok && p != "" {
prefetchPageParam = p
}
}
@@ -498,68 +566,94 @@ func syncWithPrefetch(ctx context.Context, api *ApiClient, platform *PlatformCon
time.Sleep(100 * time.Millisecond)
}
} else {
// ----- 常规分页预取 -----
firstExtra := make(map[string]interface{})
if prefetchIsCursor {
// 支持 initial_cursor 配置，如果没有则使用空字符串
if icv, ok := prefetchReqIface.RequestConfig["initial_cursor"]; ok {
firstExtra[prefetchPageParam] = icv
localLoaded := false
// 优先从本地表读取预取数据，避免每次都调用 API
localTable := ""
if prefetchIface != nil && prefetchIface.TableDefinition != nil {
if td, err := ParseTableDefinition(prefetchIface.TableDefinition); err == nil && td.TableName != "" {
localTable = td.TableName
}
}
if localTable != "" && prefetch.ValueField != "" {
result, err := gfdb.DB(ctx).Model(ctx, localTable).
Fields(prefetch.ValueField).
All()
if err != nil {
logrus.Warnf("从本地表 %s 读取预取数据失败: %v", localTable, err)
} else if result.Len() > 0 {
logrus.Infof("从本地表 %s 读取到 %d 个预取实体", localTable, result.Len())
for _, row := range result {
if v := row[prefetch.ValueField]; v != nil {
allEntities = append(allEntities, v.Val())
}
}
localLoaded = true
} else {
firstExtra[prefetchPageParam] = ""
logrus.Infof("本地表 %s 无数据，回退 API 预取", localTable)
}
}
body := buildReqBody(ctx, prefetchReqIface, 1, prefetchPageSize, lastSyncTime, firstExtra)
logrus.Debugf("预取请求 URL: %s, Method: %s, Body: %+v", prefetch.URL, prefetchMethod, body)
resp, err := api.Request(ctx, prefetchMethod, prefetch.URL, body, prefetchInQuery)
if err != nil {
recordFailure(ctx, platform.PlatformCode, iface.Code, taskType, fmt.Sprintf("预取第一页请求失败: %v", err))
return nil, fmt.Errorf("预取第一页失败: %w", err)
}
rows, prefetchTotalPages, _, nextCursor, err := parseRespExt(resp.Body, prefetchRespCfg)
if err != nil {
recordFailure(ctx, platform.PlatformCode, iface.Code, taskType, fmt.Sprintf("解析预取响应失败: %v", err))
return nil, fmt.Errorf("解析预取响应失败: %w", err)
}
collectPrefetchEntities(rows, prefetch, &allEntities, &allRows)
if prefetchIsCursor {
for nextCursor != "" && nextCursor != "nomore" {
body := buildReqBody(ctx, prefetchReqIface, 1, prefetchPageSize, lastSyncTime, map[string]interface{}{
prefetchPageParam: nextCursor,
})
resp, err := api.Request(ctx, prefetchMethod, prefetch.URL, body, prefetchInQuery)
if err != nil {
logrus.Errorf("预取游标 %s 请求失败: %v", nextCursor, err)
break
if !localLoaded {
// ----- 常规分页预取 -----
firstExtra := make(map[string]interface{})
if prefetchIsCursor {
if icv, ok := prefetchReqIface.RequestConfig["initial_cursor"]; ok {
firstExtra[prefetchPageParam] = icv
}
rows, _, _, nc, pe := parseRespExt(resp.Body, prefetchRespCfg)
if pe != nil {
logrus.Errorf("预取游标 %s 解析失败: %v", nextCursor, pe)
break
}
if len(rows) == 0 {
break
}
nextCursor = nc
collectPrefetchEntities(rows, prefetch, &allEntities, &allRows)
time.Sleep(100 * time.Millisecond)
// 没有 initial_cursor 时不覆盖 page 参数，保持默认 page=1
}
} else {
for page := 2; page <= prefetchTotalPages; page++ {
body := buildReqBody(ctx, prefetchReqIface, page, prefetchPageSize, lastSyncTime, nil)
resp, err := api.Request(ctx, prefetchMethod, prefetch.URL, body, prefetchInQuery)
if err != nil {
logrus.Errorf("预取第 %d 页请求失败: %v", page, err)
continue
body := buildReqBody(ctx, prefetchReqIface, 1, prefetchPageSize, lastSyncTime, firstExtra)
logrus.Debugf("预取请求 URL: %s, Method: %s, Body: %+v", prefetch.URL, prefetchMethod, body)
resp, err := api.Request(ctx, prefetchMethod, prefetch.URL, body, prefetchInQuery)
if err != nil {
recordFailure(ctx, platform.PlatformCode, iface.Code, taskType, fmt.Sprintf("预取第一页请求失败: %v", err))
return nil, fmt.Errorf("预取第一页失败: %w", err)
}
rows, prefetchTotalPages, _, nextCursor, err := parseRespExt(resp.Body, prefetchRespCfg)
if err != nil {
recordFailure(ctx, platform.PlatformCode, iface.Code, taskType, fmt.Sprintf("解析预取响应失败: %v", err))
return nil, fmt.Errorf("解析预取响应失败: %w", err)
}
collectPrefetchEntities(rows, prefetch, &allEntities, &allRows)
if prefetchIsCursor {
for nextCursor != "" && nextCursor != "nomore" {
body := buildReqBody(ctx, prefetchReqIface, 1, prefetchPageSize, lastSyncTime, map[string]interface{}{
prefetchPageParam: nextCursor,
})
resp, err := api.Request(ctx, prefetchMethod, prefetch.URL, body, prefetchInQuery)
if err != nil {
logrus.Errorf("预取游标 %s 请求失败: %v", nextCursor, err)
break
}
rows, _, _, nc, pe := parseRespExt(resp.Body, prefetchRespCfg)
if pe != nil {
logrus.Errorf("预取游标 %s 解析失败: %v", nextCursor, pe)
break
}
if len(rows) == 0 {
break
}
nextCursor = nc
collectPrefetchEntities(rows, prefetch, &allEntities, &allRows)
time.Sleep(100 * time.Millisecond)
}
rows, _, _, _, pe := parseRespExt(resp.Body, prefetchRespCfg)
if pe != nil {
logrus.Errorf("预取第 %d 页解析失败: %v", page, pe)
continue
} else {
for page := 2; page <= prefetchTotalPages; page++ {
body := buildReqBody(ctx, prefetchReqIface, page, prefetchPageSize, lastSyncTime, nil)
resp, err := api.Request(ctx, prefetchMethod, prefetch.URL, body, prefetchInQuery)
if err != nil {
logrus.Errorf("预取第 %d 页请求失败: %v", page, err)
continue
}
rows, _, _, _, pe := parseRespExt(resp.Body, prefetchRespCfg)
if pe != nil {
logrus.Errorf("预取第 %d 页解析失败: %v", page, pe)
continue
}
collectPrefetchEntities(rows, prefetch, &allEntities, &allRows)
time.Sleep(100 * time.Millisecond)
}
collectPrefetchEntities(rows, prefetch, &allEntities, &allRows)
time.Sleep(100 * time.Millisecond)
}
}
}
@@ -735,7 +829,7 @@ func syncWithPrefetch(ctx context.Context, api *ApiClient, platform *PlatformCon
if globalMaxTime <= 0 {
globalMaxTime = time.Now().Unix()
}
updateSyncTime(ctx, platform.PlatformCode, iface.Code, globalMaxTime)
updateSyncTime(ctx, platform.PlatformCode, iface.Code, globalMaxTime, taskType)
result.Duration = fmt.Sprintf("%.1fs", time.Since(start).Seconds())
logrus.Infof("同步完成 - 表:%s, %d条, 写入%d条, 耗时%s", td.TableName, result.TotalRows, result.InsertedRows, result.Duration)
@@ -814,7 +908,7 @@ func syncRecursive(ctx context.Context, api *ApiClient, platform *PlatformConfig
}
inserted, _ := savePage(ctx, td, allRows)
updateSyncTime(ctx, platform.PlatformCode, iface.Code, time.Now().Unix())
updateSyncTime(ctx, platform.PlatformCode, iface.Code, time.Now().Unix(), "full")
result := &SyncResult{
TableName: td.TableName,
@@ -910,7 +1004,7 @@ func buildPrefetchParams(iface *entity.ApiInterface) map[string]interface{} {
k == "cursor_pagination" || k == "time_field_mode" ||
k == "recursive" || k == "max_recursive_depth" ||
k == "initial_cursor" || k == "pagination_mode" ||
k == "full_sync_start_time" || k == "row_inject" {
k == "body_type" || k == "full_sync_start_time" || k == "row_inject" {
continue
}
if k == pageParam || k == psParam {
@@ -1058,7 +1152,7 @@ func buildReqBody(ctx context.Context, iface *entity.ApiInterface, page, pageSiz
k == "body_wrapper_field" || k == "exclude_from_wrapper" ||
k == "top_level_params" || k == "recursive" ||
k == "max_recursive_depth" || k == "initial_cursor" ||
k == "pagination_mode" || k == "full_sync_start_time" ||
k == "body_type" || k == "full_sync_start_time" ||
k == "row_inject" {
continue
}
@@ -1109,6 +1203,9 @@ func buildReqBody(ctx context.Context, iface *entity.ApiInterface, page, pageSiz
} else {
timeMs = time.Now().Add(-time.Duration(GetDefaultLookbackDays(ctx)) * 24 * time.Hour).UnixMilli()
}
} else if timeMs < 1000000000000 {
// lastSyncTime 存的是秒（已归一化），快手 API 需要毫秒
timeMs = timeMs * 1000
}
// 仅在配置未指定 queryType 时设默认值，尊重配置
if _, exists := body["queryType"]; !exists {
@@ -1145,6 +1242,21 @@ func buildReqBody(ctx context.Context, iface *entity.ApiInterface, page, pageSiz
} else {
body["filtering"] = []interface{}{timeFilter}
}
} else {
// 全量且没有 full_sync_start_time：用 default_lookback_days 兜底
lookbackDays := GetDefaultLookbackDays(ctx)
defaultStart := time.Now().Add(-time.Duration(lookbackDays) * 24 * time.Hour).Unix()
logrus.Infof("全量同步使用默认回溯 %d 天，过滤时间戳: %d", lookbackDays, defaultStart)
timeFilter := map[string]interface{}{
"field": tf,
"operator": "GREATER_EQUALS",
"values": []interface{}{fmt.Sprintf("%d", defaultStart)},
}
if existing, ok := body["filtering"].([]interface{}); ok {
body["filtering"] = append(existing, timeFilter)
} else {
body["filtering"] = []interface{}{timeFilter}
}
}
}
}
@@ -1386,8 +1498,11 @@ func savePage(ctx context.Context, td *TableDefinition, rows []map[string]interf
return 0, nil
}
colSet := make(map[string]bool)
for _, c := range td.Columns {
colDefs := make(map[string]*ColumnDef)
for i := range td.Columns {
c := &td.Columns[i]
colSet[c.Name] = true
colDefs[c.Name] = c
}
var clean []map[string]interface{}
for _, row := range rows {
@@ -1397,6 +1512,14 @@ func savePage(ctx context.Context, td *TableDefinition, rows []map[string]interf
c[k] = v
}
}
// 填充有默认值的字段（API 响应中缺失时使用）
for name, col := range colDefs {
if col.DefaultValue != "" {
if _, exists := c[name]; !exists {
c[name] = col.DefaultValue
}
}
}
if r, ok := row["raw_data"]; ok {
c["raw_data"] = r
}
@@ -1406,46 +1529,85 @@ func savePage(ctx context.Context, td *TableDefinition, rows []map[string]interf
}
func getLastSyncTime(ctx context.Context, platformCode, interfaceCode string) int64 {
var t int64
gfdb.DB(ctx).Model(ctx, consts.SyncTrackerTable).
Fields("last_sync_time").
v, err := gfdb.DB(ctx).Model(ctx, consts.SyncTrackerTable).
Where("platform_code", platformCode).
Where("interface_code", interfaceCode).
Scan(&t)
return t
Value("last_sync_time")
if err != nil || v == nil {
return 0
}
return v.Int64()
}
func getSyncStatus(ctx context.Context, platformCode, interfaceCode string) string {
var s string
gfdb.DB(ctx).Model(ctx, consts.SyncTrackerTable).
Fields("sync_status").
v, err := gfdb.DB(ctx).Model(ctx, consts.SyncTrackerTable).
Where("platform_code", platformCode).
Where("interface_code", interfaceCode).
Scan(&s)
return s
Value("sync_status")
if err != nil || v == nil {
return ""
}
return v.String()
}
func markSyncRunning(ctx context.Context, platformCode, interfaceCode string, lastSyncTime int64) {
func getSyncCount(ctx context.Context, platformCode, interfaceCode string) int64 {
v, err := gfdb.DB(ctx).Model(ctx, consts.SyncTrackerTable).
Where("platform_code", platformCode).
Where("interface_code", interfaceCode).
Value("COALESCE(sync_count, 0)")
if err != nil || v == nil {
return 0
}
return v.Int64()
}
func markSyncRunning(ctx context.Context, platformCode, interfaceCode string) {
tenantId := utils.GetCurrentTenantId(ctx)
gfdb.DB(ctx).Model(ctx, consts.SyncTrackerTable).
Data(map[string]interface{}{
"platform_code": platformCode,
"interface_code": interfaceCode,
"last_sync_time": lastSyncTime,
"sync_status": "running",
"tenant_id": tenantId,
}).
OnConflict("platform_code", "interface_code").
Save()
}
func updateSyncTime(ctx context.Context, platformCode, interfaceCode string, t int64) {
// normalizeSyncTimestamp 将同步时间戳统一归一化到秒
// 外部 API 响应中的时间戳可能是秒或毫秒，统一存为秒
func normalizeSyncTimestamp(ts int64) int64 {
if ts > 1000000000000 { // > 1e12 显然是毫秒
return ts / 1000
}
return ts
}
// updateSyncTime 更新同步跟踪记录
// syncType: "full" / "incremental"
func updateSyncTime(ctx context.Context, platformCode, interfaceCode string, t int64, syncType string) {
t = normalizeSyncTimestamp(t)
tenantId := utils.GetCurrentTenantId(ctx)
// 先读取当前的 sync_count
currentCount := getSyncCount(ctx, platformCode, interfaceCode)
data := gdb.Map{
"platform_code": platformCode,
"interface_code": interfaceCode,
"last_sync_time": t,
"last_sync_at": time.Now(),
"sync_status": "success",
"last_sync_type": syncType,
"sync_count": currentCount + 1,
"tenant_id": tenantId,
}
if syncType == "full" {
data["last_full_sync_time"] = time.Now().Unix() // 用当前时间，不是数据时间戳
}
gfdb.DB(ctx).Model(ctx, consts.SyncTrackerTable).
Data(map[string]interface{}{
"platform_code": platformCode,
"interface_code": interfaceCode,
"last_sync_time": t,
"last_sync_at": time.Now(),
"sync_status": "success",
}).
Data(data).
OnConflict("platform_code", "interface_code").
Save()
}
@@ -1465,6 +1627,26 @@ func recordFailure(ctx context.Context, platformCode, interfaceCode, taskType, e
})
}
// isTokenExpiredError 判断错误是否因 token 过期导致
func isTokenExpiredError(err error) bool {
if err == nil {
return false
}
msg := err.Error()
// 常见 token 过期标识（兼容各平台）
tokenExpiredKeywords := []string{
"TOKEN过期", "token过期", "token_expired", "Token过期",
"result=28", // 快手
"access_token", "token已失效", "token无效",
}
for _, kw := range tokenExpiredKeywords {
if strings.Contains(msg, kw) {
return true
}
}
return false
}
// findInterfaceByURL 在所有接口中查找匹配 URL 的接口
func findInterfaceByURL(ifaces []entity.ApiInterface, url string) *entity.ApiInterface {
for i := range ifaces {
service/sync/helpers.go
+9
View File
@@ -51,6 +51,15 @@ func GetSyncTimeout(ctx context.Context) int {
return t
}
// GetFullSyncIntervalHours 获取全量同步间隔（小时；0=禁用自动全量，首次全量后只走增量）
func GetFullSyncIntervalHours(ctx context.Context) int {
h := g.Cfg().MustGet(ctx, "sync.full_sync_interval_hours", 0).Int()
if h < 0 {
return 0
}
return h
}
// GetDefaultLookbackDays 获取全量同步默认回溯天数（默认90）
func GetDefaultLookbackDays(ctx context.Context) int {
d := g.Cfg().MustGet(ctx, "sync.default_lookback_days", 90).Int()
service/sync/oauth.go
+133
View File
@@ -0,0 +1,133 @@
package sync
import (
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"net/url"
"sync"
"time"
consts "dataengine/consts/public"
"dataengine/utils"
"gitea.redpowerfuture.com/red-future/common/db/gfdb"
"github.com/gogf/gf/v2/frame/g"
"github.com/gogf/gf/v2/os/gtime"
"github.com/sirupsen/logrus"
)
// refreshTokenMu 防止并发刷新 token（多个接口同时检测到过期时串行化）
var refreshTokenMu sync.Mutex
// tencentTokenResponse 腾讯广告 OAuth 刷新 token 响应
type tencentTokenResponse struct {
Code int `json:"code"`
Message string `json:"message"`
Data *struct {
AccessToken string `json:"access_token"`
RefreshToken string `json:"refresh_token"`
ExpiresIn int `json:"access_token_expires_in"`
RefreshExpiresIn int `json:"refresh_token_expires_in"`
} `json:"data"`
}
// RefreshTencentToken 刷新腾讯广告 OAuth2 token
// 由 ApiClient 在检测到 401/token 过期时自动调用
func RefreshTencentToken(ctx context.Context, platform *PlatformConfig) error {
refreshTokenMu.Lock()
defer refreshTokenMu.Unlock()
if platform.AuthConfig == nil {
return fmt.Errorf("平台 [%s] 未配置 auth_config", platform.PlatformCode)
}
clientID := platform.ClientId
clientSecret := platform.ClientSecret
refreshToken, _ := platform.AuthConfig["refresh_token"].(string)
if clientID == "" || clientSecret == "" || refreshToken == "" {
return fmt.Errorf("平台 [%s] OAuth2 配置不完整: client_id / client_secret / refresh_token 缺失",
platform.PlatformCode)
}
logrus.Infof("正在刷新腾讯广告 token [platform=%s]", platform.PlatformCode)
// 调用腾讯 OAuth 刷新端点（使用短超时防止阻塞）
refreshCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
// 腾讯 OAuth refresh_token 接口使用 GET 方法，参数在查询字符串中
refreshURL := fmt.Sprintf("https://api.e.qq.com/oauth/refresh_token?client_id=%s&client_secret=%s&refresh_token=%s",
url.QueryEscape(clientID), url.QueryEscape(clientSecret), url.QueryEscape(refreshToken))
req, err := http.NewRequestWithContext(refreshCtx, "GET", refreshURL, nil)
if err != nil {
return fmt.Errorf("创建刷新请求失败: %w", err)
}
resp, err := http.DefaultClient.Do(req)
if err != nil {
return fmt.Errorf("请求刷新 token 失败: %w", err)
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
if err != nil {
return fmt.Errorf("读取刷新响应失败: %w", err)
}
var tokenResp tencentTokenResponse
if err := json.Unmarshal(body, &tokenResp); err != nil {
return fmt.Errorf("解析刷新响应失败: %w", err)
}
if tokenResp.Code != 0 {
return fmt.Errorf("腾讯 OAuth 刷新失败: code=%d, msg=%s", tokenResp.Code, tokenResp.Message)
}
if tokenResp.Data == nil {
return fmt.Errorf("腾讯 OAuth 刷新响应缺少 data 字段")
}
newAccessToken := tokenResp.Data.AccessToken
newRefreshToken := tokenResp.Data.RefreshToken
if newAccessToken == "" {
return fmt.Errorf("腾讯 OAuth 刷新返回的 access_token 为空")
}
// --- 更新内存中的配置 ---
// token 字段（存明文 token）
platform.Token = newAccessToken
// AccessToken 字段（运行时使用的 token）
platform.AccessToken = newAccessToken
// auth_config 中的 refresh_token
platform.AuthConfig["refresh_token"] = newRefreshToken
// --- 更新数据库 ---
tenantID := utils.GetCurrentTenantId(ctx)
if tenantID == 0 {
tenantID = 1 // 保底默认租户
}
// 使用 gfdb 直接更新 token 和 auth_config（避免 DAO 层的 OmitEmpty 吞掉空值）
_, err = gfdb.DB(ctx).Model(ctx, consts.DatasourcePlatformTable).
Data(g.Map{
"token": newAccessToken,
"auth_config": platform.AuthConfig,
"updated_at": gtime.Now(),
}).
Where("platform_code", platform.PlatformCode).
Where("tenant_id", tenantID).
Update()
if err != nil {
return fmt.Errorf("更新数据库 token 失败: %w", err)
}
logrus.Infof("腾讯广告 Token 刷新成功 [platform=%s] (access_token有效期:%ds, refresh_token有效期:%ds)",
platform.PlatformCode, tokenResp.Data.ExpiresIn, tokenResp.Data.RefreshExpiresIn)
return nil
}
service/sync/sync_scheduler.go
+122 -6
View File
@@ -4,29 +4,78 @@ import (
"context"
"time"
consts "dataengine/consts/public"
dao "dataengine/dao/dict"
dto "dataengine/model/dto/dict"
"gitea.redpowerfuture.com/red-future/common/beans"
"gitea.redpowerfuture.com/red-future/common/db/gfdb"
"github.com/gogf/gf/v2/frame/g"
"github.com/sirupsen/logrus"
)
// SyncRunItemResult 单次同步执行结果
type SyncRunItemResult struct {
PlatformCode string `json:"platformCode"`
InterfaceCode string `json:"interfaceCode"`
Success bool `json:"success"`
Error string `json:"error,omitempty"`
}
// StartAutoSync 启动自动同步（独立 goroutine，每次完成后等待 interval 再执行下一次）
func StartAutoSync(ctx context.Context) {
interval := GetSyncInterval(ctx)
logrus.Infof("自动同步调度器启动，间隔: %d 分钟（完成一次后开始计时）", interval)
for {
runAutoSync(ctx)
runAutoSync(ctx, false) // 后台调度不关心返回结果，错误已内部打日志
logrus.Infof("自动同步完成，等待 %d 分钟后执行下一次", interval)
time.Sleep(time.Duration(interval) * time.Minute)
}
}
func runAutoSync(ctx context.Context) {
// determineSyncMode 判断本次同步应该增量还是全量
// 返回 true=全量, false=增量
func determineSyncMode(ctx context.Context) bool {
interval := GetFullSyncIntervalHours(ctx)
if interval <= 0 {
return false // 自动全量关闭，永远增量
}
// 查询 sync_tracker 中最早的全量同步时间
v, err := gfdb.DB(ctx).Model(ctx, consts.SyncTrackerTable).
Value("COALESCE(MIN(last_full_sync_time), 0)")
if err != nil {
logrus.Warnf("查询全量同步时间失败，回退增量模式: %v", err)
return false
}
minFullSync := int64(0)
if v != nil {
minFullSync = v.Int64()
}
if minFullSync == 0 {
logrus.Info("检测到从未全量同步的接口，本次执行全量同步")
return true
}
elapsed := time.Now().Unix() - minFullSync
if elapsed > int64(interval)*3600 {
logrus.Infof("距上次全量已过 %d 小时（阈值 %d 小时），本次执行全量同步", elapsed/3600, interval)
return true
}
logrus.Debugf("距上次全量 %d 小时，未到阈值 %d 小时，执行增量同步", elapsed/3600, interval)
return false
}
func runAutoSync(ctx context.Context, forceFull bool) []SyncRunItemResult {
logrus.Info("=== 开始自动同步 ===")
// 判断本次同步模式（增量 / 全量）
isFullSync := forceFull || determineSyncMode(ctx)
logrus.Infof("本次同步模式: %s", map[bool]string{true: "全量", false: "增量"}[isFullSync])
// 从配置读取同步租户 ID（运维部署时配置）
tenantId := g.Cfg().MustGet(ctx, "sync.default_tenant_id", 1).Uint64()
@@ -39,9 +88,11 @@ func runAutoSync(ctx context.Context) {
}, tenantId)
if err != nil {
logrus.Errorf("查询平台列表失败: %v", err)
return
return nil
}
var results []SyncRunItemResult
for _, p := range platforms {
// 查询该平台下有 table_definition 的接口
interfaces, _, err := dao.ApiInterface.List(ctx, &dto.ListApiInterfaceReq{
@@ -59,20 +110,85 @@ func runAutoSync(ctx context.Context) {
}
logrus.Infof("自动同步: %s / %s", p.PlatformCode, iface.Code)
// isFullSync=false 表示去查 sync_tracker：
// 有记录 → 增量，无记录 → lastSyncTime=0 → 全量
_, err := SyncByConfig(ctx, p.PlatformCode, iface.Code, false)
_, err := SyncByConfig(ctx, p.PlatformCode, iface.Code, isFullSync)
item := SyncRunItemResult{
PlatformCode: p.PlatformCode,
InterfaceCode: iface.Code,
}
if err != nil {
logrus.Errorf("自动同步失败 [%s/%s]: %v", p.PlatformCode, iface.Code, err)
item.Success = false
item.Error = err.Error()
results = append(results, item)
// token 过期是平台级别问题，该平台剩余接口继续请求只会重复失败，直接跳过
if isTokenExpiredError(err) {
logrus.Warnf("平台 [%s] token 已过期，跳过该平台剩余接口", p.PlatformCode)
break
}
} else {
item.Success = true
results = append(results, item)
}
}
}
logrus.Info("=== 自动同步完成 ===")
return results
}
// TriggerAllSync 手动触发全量同步（等价于 runAutoSync 的一次执行）
// 由 HTTP 端点调用，用于 PPGo_Job 调度。
// forceFull=true 强制全量，false 则由自动策略判断。
// 返回每个平台接口的执行结果列表。
func TriggerAllSync(ctx context.Context, forceFull bool) []SyncRunItemResult {
ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: g.Cfg().MustGet(ctx, "sync.default_tenant_id", 1).Uint64()})
return runAutoSync(ctx, forceFull)
}
// RecoverInterruptedSyncs 服务启动时恢复异常中断的同步
// 扫描 sync_tracker 中 sync_status="running" 的接口，自动重新全量同步
// 适用于服务进程崩溃后重启的场景，不等 PPGo_Job 下轮调度
func RecoverInterruptedSyncs(ctx context.Context) {
type trackerItem struct {
PlatformCode string
InterfaceCode string
}
var items []trackerItem
err := gfdb.DB(ctx).Model(ctx, consts.SyncTrackerTable).
Fields("platform_code", "interface_code").
Where("sync_status", "running").
Scan(&items)
if err != nil {
logrus.Warnf("查询中断同步任务失败: %v", err)
return
}
if len(items) == 0 {
logrus.Info("启动恢复扫描：没有发现异常中断的同步任务")
return
}
tenantId := g.Cfg().MustGet(ctx, "sync.default_tenant_id", 1).Uint64()
ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: tenantId})
logrus.Infof("启动恢复扫描：发现 %d 个异常中断的同步任务，开始恢复...", len(items))
for _, item := range items {
logrus.Infof("正在恢复中断的同步 [%s/%s]...", item.PlatformCode, item.InterfaceCode)
_, err := SyncByConfig(ctx, item.PlatformCode, item.InterfaceCode, true)
if err != nil {
logrus.Errorf("恢复同步失败 [%s/%s]: %v", item.PlatformCode, item.InterfaceCode, err)
} else {
logrus.Infof("恢复同步成功 [%s/%s]", item.PlatformCode, item.InterfaceCode)
}
}
}
// InitAndStartAutoSync 在 main 中调用：初始化配置后启动自动同步和补偿
func InitAndStartAutoSync(ctx context.Context) {
// 服务启动时恢复异常中断的同步（不等 PPGo_Job 调度）
RecoverInterruptedSyncs(ctx)
// 读取配置中的同步开关
enabled := g.Cfg().MustGet(ctx, "sync.auto_sync_enabled", false).Bool()
if enabled {
service/sync/table_manager.go
+11 -5
View File
@@ -11,9 +11,10 @@ import (
// ColumnDef 列定义
type ColumnDef struct {
Name string `json:"name"`
Type string `json:"type"`
Comment string `json:"comment,omitempty"`
Name string `json:"name"`
Type string `json:"type"`
Comment string `json:"comment,omitempty"`
DefaultValue string `json:"default_value,omitempty"`
}
// TableDefinition 表结构定义
@@ -41,10 +42,11 @@ func ParseTableDefinition(raw map[string]interface{}) (*TableDefinition, error)
n, _ := cm["name"].(string)
t, _ := cm["type"].(string)
comment, _ := cm["comment"].(string)
defaultVal, _ := cm["default_value"].(string)
if n == "" || t == "" {
continue
}
td.Columns = append(td.Columns, ColumnDef{Name: n, Type: t, Comment: comment})
td.Columns = append(td.Columns, ColumnDef{Name: n, Type: t, Comment: comment, DefaultValue: defaultVal})
}
if keys, _ := raw["conflict_keys"].([]interface{}); keys != nil {
for _, k := range keys {
@@ -82,7 +84,11 @@ func buildCreateSQL(td *TableDefinition) string {
"deleted_at TIMESTAMP WITH TIME ZONE",
}
for _, c := range td.Columns {
cols = append(cols, fmt.Sprintf("%s %s", c.Name, c.Type))
colSQL := fmt.Sprintf("%s %s", c.Name, c.Type)
if c.DefaultValue != "" {
colSQL += fmt.Sprintf(" DEFAULT '%s'", strings.ReplaceAll(c.DefaultValue, "'", "''"))
}
cols = append(cols, colSQL)
}
cols = append(cols, "raw_data JSONB DEFAULT '{}'")
sql/init_core_tables.sql
+6
View File
@@ -205,6 +205,9 @@ CREATE TABLE IF NOT EXISTS sync_tracker (
last_sync_time BIGINT NOT NULL DEFAULT 0,
last_sync_at TIMESTAMP WITH TIME ZONE,
sync_status VARCHAR(32) DEFAULT 'pending',
last_full_sync_time BIGINT NOT NULL DEFAULT 0,
sync_count BIGINT NOT NULL DEFAULT 0,
last_sync_type VARCHAR(16) NOT NULL DEFAULT '',
PRIMARY KEY (id)
);
@@ -223,3 +226,6 @@ COMMENT ON COLUMN sync_tracker.interface_code IS '接口编码';
COMMENT ON COLUMN sync_tracker.last_sync_time IS '最后同步时间(Unix时间戳)';
COMMENT ON COLUMN sync_tracker.last_sync_at IS '最后同步时间点';
COMMENT ON COLUMN sync_tracker.sync_status IS '同步状态: pending/success/running/failed';
COMMENT ON COLUMN sync_tracker.last_full_sync_time IS '最后全量同步时间(Unix时间戳秒)，用于决定何时需要再次全量';
COMMENT ON COLUMN sync_tracker.sync_count IS '累计同步次数';
COMMENT ON COLUMN sync_tracker.last_sync_type IS '最后一次同步类型: full/incremental';
sql/seed_data.sql
+10 -5
View File
@@ -57,14 +57,16 @@ INSERT INTO api_interface (
"page_size": 100,
"page_param": "page",
"page_size_param": "page_size",
"pagination_mode": "PAGINATION_MODE_NORMAL",
"pagination_mode": "PAGINATION_MODE_CURSOR",
"cursor_pagination": true,
"fields": ["account_id", "corporation_name", "is_adx", "is_bid", "is_mp"]
}'::jsonb,
'{
"success_field": "code",
"success_value": 0,
"message_field": "message",
"list_path": "data.list"
"list_path": "data.list",
"cursor_field": "data.cursor_page_info.cursor"
}'::jsonb,
'{
"table_name": "tencent_account_relation",
@@ -110,7 +112,8 @@ INSERT INTO api_interface (
"success_field": "code",
"success_value": 0,
"message_field": "message",
"list_path": "data.list"
"list_path": "data.list",
"cursor_field": "data.cursor_page_info.cursor"
}'::jsonb,
'{
"table_name": "tencent_image",
@@ -175,7 +178,8 @@ INSERT INTO api_interface (
"success_field": "code",
"success_value": 0,
"message_field": "message",
"list_path": "data.list"
"list_path": "data.list",
"cursor_field": "data.cursor_page_info.cursor"
}'::jsonb,
'{
"table_name": "tencent_video",
@@ -238,7 +242,8 @@ INSERT INTO api_interface (
"success_field": "code",
"success_value": 0,
"message_field": "message",
"list_path": "data.list"
"list_path": "data.list",
"cursor_field": "data.cursor_page_info.cursor"
}'::jsonb,
'{
"table_name": "tencent_audio",