From b45c4a2316e9294963a8443b8596834c5362de41 Mon Sep 17 00:00:00 2001
From: lmk <1095689763@qq.com>
Date: Thu, 16 Jul 2026 14:34:06 +0800
Subject: [PATCH] =?UTF-8?q?=E6=95=B0=E6=8D=AE=E5=BC=95=E6=93=8E=E9=87=8D?=
 =?UTF-8?q?=E6=9E=84?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 common/report/cron.go                       |  83 +--
 config.yml                                  |  16 +-
 controller/report/report_controller.go      |  20 +
 controller/sync/platform_sync_controller.go | 111 ++++
 dao/copydata/sync_task_log_dao.go           |   2 +
 docs/USAGE.md                               | 528 +++++++++++++++-----
 go.mod                                      |   1 +
 go.sum                                      |  16 +-
 main.go                                     |  19 +-
 service/sync/api_client.go                  | 117 ++++-
 service/sync/compensation.go                |  49 +-
 service/sync/data_writer.go                 |   4 +
 service/sync/dynamic_sync.go                | 366 ++++++++++----
 service/sync/helpers.go                     |   9 +
 service/sync/oauth.go                       | 133 +++++
 service/sync/sync_scheduler.go              | 128 ++++-
 service/sync/table_manager.go               |  16 +-
 sql/init_core_tables.sql                    |   6 +
 sql/seed_data.sql                           |  15 +-
 19 files changed, 1332 insertions(+), 307 deletions(-)
 create mode 100644 service/sync/oauth.go

diff --git a/common/report/cron.go b/common/report/cron.go
index a364f3b..d386360 100644
--- a/common/report/cron.go
+++ b/common/report/cron.go
@@ -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 {
diff --git a/config.yml b/config.yml
index a1fa4d9..72cfb2c 100644
--- a/config.yml
+++ b/config.yml
@@ -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"
diff --git a/controller/report/report_controller.go b/controller/report/report_controller.go
index 5b255a6..dd77ed6 100644
--- a/controller/report/report_controller.go
+++ b/controller/report/report_controller.go
@@ -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)
diff --git a/controller/sync/platform_sync_controller.go b/controller/sync/platform_sync_controller.go
index d84821d..3a20732 100644
--- a/controller/sync/platform_sync_controller.go
+++ b/controller/sync/platform_sync_controller.go
@@ -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 {
diff --git a/dao/copydata/sync_task_log_dao.go b/dao/copydata/sync_task_log_dao.go
index 40fed64..40ce3cf 100644
--- a/dao/copydata/sync_task_log_dao.go
+++ b/dao/copydata/sync_task_log_dao.go
@@ -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()
diff --git a/docs/USAGE.md b/docs/USAGE.md
index 426ac64..428e5d7 100644
--- a/docs/USAGE.md
+++ b/docs/USAGE.md
@@ -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 管理所有任务 |
diff --git a/go.mod b/go.mod
index 7a3605c..c4ec5af 100644
--- a/go.mod
+++ b/go.mod
@@ -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
diff --git a/go.sum b/go.sum
index d9ceff7..1f74126 100644
--- a/go.sum
+++ b/go.sum
@@ -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=
diff --git a/main.go b/main.go
index 229ac11..aa3e63a 100644
--- a/main.go
+++ b/main.go
@@ -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{}{
 		// 接口管理
diff --git a/service/sync/api_client.go b/service/sync/api_client.go
index 3c4528a..39b3595 100644
--- a/service/sync/api_client.go
+++ b/service/sync/api_client.go
@@ -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))
 		}
diff --git a/service/sync/compensation.go b/service/sync/compensation.go
index 7580c4f..5a2dd4b 100644
--- a/service/sync/compensation.go
+++ b/service/sync/compensation.go
@@ -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
 }
diff --git a/service/sync/data_writer.go b/service/sync/data_writer.go
index 6644fa9..05bbad4 100644
--- a/service/sync/data_writer.go
+++ b/service/sync/data_writer.go
@@ -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
diff --git a/service/sync/dynamic_sync.go b/service/sync/dynamic_sync.go
index 8b01048..ff757c9 100644
--- a/service/sync/dynamic_sync.go
+++ b/service/sync/dynamic_sync.go
@@ -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 {
diff --git a/service/sync/helpers.go b/service/sync/helpers.go
index 12527e9..52bd5cb 100644
--- a/service/sync/helpers.go
+++ b/service/sync/helpers.go
@@ -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()
diff --git a/service/sync/oauth.go b/service/sync/oauth.go
new file mode 100644
index 0000000..5362221
--- /dev/null
+++ b/service/sync/oauth.go
@@ -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
+}
diff --git a/service/sync/sync_scheduler.go b/service/sync/sync_scheduler.go
index 355ee0d..7b35376 100644
--- a/service/sync/sync_scheduler.go
+++ b/service/sync/sync_scheduler.go
@@ -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 {
diff --git a/service/sync/table_manager.go b/service/sync/table_manager.go
index 0f1450b..84dcf81 100644
--- a/service/sync/table_manager.go
+++ b/service/sync/table_manager.go
@@ -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 '{}'")
 
diff --git a/sql/init_core_tables.sql b/sql/init_core_tables.sql
index e8d13e8..2683e22 100644
--- a/sql/init_core_tables.sql
+++ b/sql/init_core_tables.sql
@@ -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';
diff --git a/sql/seed_data.sql b/sql/seed_data.sql
index 2b0dbba..1442057 100644
--- a/sql/seed_data.sql
+++ b/sql/seed_data.sql
@@ -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",