T8BaseInboundMdpSyncService.cs 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339
  1. using Admin.NET.Plugin.AiDOP.DataPlatform.Executors;
  2. using SqlSugar;
  3. namespace Admin.NET.Plugin.AiDOP.DataPlatform;
  4. /// <summary>
  5. /// T8 基表双模式入站(S5/S6/S7 KPI 共用贴源+标准层)。
  6. /// 照抄 S6_REPORT 范式:执行器 → mdp_stg_t8_* → transform → mdp_std_t8_*。
  7. /// 源:mdp_source=T8_V5_SQLSERVER(MdpSourceScopeFactory 复用 Database.json 的 t8_v5 只读连接)。
  8. /// 实体:mdp_entity T8_KC_TZ_HEAD / T8_KC_TZ_LIST / T8_KC_DD_HEAD / T8_KC_DD_LIST / T8_KC_ZJ_LIST / T8_SYS_PELIST
  9. /// (种子把 target_table_name 指向对应 mdp_stg_t8_*、sync_mode=FULL、biz_key_expr=Id)。
  10. /// 只读源、只写 mdp;不写 T8;FULL 全量(源表极小、re-upsert 刷新可带回字段/作废标志更新)。
  11. /// 下游 S5/S6/S7 KPI 已改读 mdp_std_t8_*;故 KPI 跑批前须先跑一次本入站。
  12. /// 说明:S5_L1_002 已中台化——其依赖的 Cj_Bg_Head_Rep(报工头,取min kgdate) 与 kc_dd_list_cllist(订单物料明细)
  13. /// 随本服务贴源到 mdp_std_t8_cj_bg_head_rep / mdp_std_t8_kc_dd_list_cllist(biz_key=Id,实测唯一)。
  14. /// 仅 S5_L1_004(Rep_总账_存货_V3 TVF,报表聚合结果无逐行主键)仍走 legacy 直连 T8,其实体 T8_REP_ZONGZHANG_CUNHUO_V3 不贴源。
  15. /// </summary>
  16. public sealed class T8BaseInboundMdpSyncService : ITransient
  17. {
  18. private readonly ISqlSugarClient _db;
  19. private readonly MdpSourcePullDispatcher _pullDispatcher;
  20. // 入站实体码(对应 6 张 T8 基表)→ 目标 std 表;stg 表名由 mdp_entity.target_table_name 决定。
  21. private static readonly (string EntityCode, string StgTable, string StdTable)[] Entities =
  22. {
  23. ("T8_KC_TZ_HEAD", "mdp_stg_t8_kc_tz_head", "mdp_std_t8_kc_tz_head"),
  24. ("T8_KC_TZ_LIST", "mdp_stg_t8_kc_tz_list", "mdp_std_t8_kc_tz_list"),
  25. ("T8_KC_DD_HEAD", "mdp_stg_t8_kc_dd_head", "mdp_std_t8_kc_dd_head"),
  26. ("T8_KC_DD_LIST", "mdp_stg_t8_kc_dd_list", "mdp_std_t8_kc_dd_list"),
  27. ("T8_KC_ZJ_LIST", "mdp_stg_t8_kc_zj_list", "mdp_std_t8_kc_zj_list"),
  28. ("T8_SYS_PELIST", "mdp_stg_t8_sys_pelist", "mdp_std_t8_sys_pelist"),
  29. // S5_L1_002 中台化新增:报工头(取 min kgdate 开工时间) 与 订单物料明细(分母行数)。
  30. ("T8_CJ_BG_HEAD_REP", "mdp_stg_t8_cj_bg_head_rep", "mdp_std_t8_cj_bg_head_rep"),
  31. ("T8_KC_DD_LIST_CLLIST", "mdp_stg_t8_kc_dd_list_cllist", "mdp_std_t8_kc_dd_list_cllist"),
  32. };
  33. public T8BaseInboundMdpSyncService(ISqlSugarClient db, MdpSourcePullDispatcher pullDispatcher)
  34. {
  35. _db = db;
  36. _pullDispatcher = pullDispatcher;
  37. }
  38. public async Task<T8BaseInboundResult> RunInboundAsync(
  39. long tenantId = 0,
  40. bool fullRefresh = true,
  41. string? entityCode = null,
  42. CancellationToken cancellationToken = default)
  43. {
  44. cancellationToken.ThrowIfCancellationRequested();
  45. await EnsureTablesAsync();
  46. var now = DateTime.Now;
  47. var batchId = $"T8_BASE_IN_{now:yyyyMMddHHmmss}";
  48. var result = new T8BaseInboundResult { BatchId = batchId };
  49. var targets = string.IsNullOrWhiteSpace(entityCode)
  50. ? Entities
  51. : Entities.Where(e => string.Equals(e.EntityCode, entityCode.Trim(), StringComparison.OrdinalIgnoreCase)).ToArray();
  52. foreach (var (code, stgTable, stdTable) in targets)
  53. {
  54. cancellationToken.ThrowIfCancellationRequested();
  55. var pullCtx = new MdpPullContext
  56. {
  57. TenantId = tenantId,
  58. FullRefresh = fullRefresh,
  59. TaskCode = "T8_BASE_MDP_INBOUND",
  60. BatchId = $"{batchId}_{code}"
  61. };
  62. var pull = await _pullDispatcher.PullByEntityCodeAsync(code, pullCtx, cancellationToken);
  63. var transformBatch = $"{pullCtx.BatchId}_STD";
  64. var stdRows = await TransformStandardAsync(code, stgTable, stdTable, tenantId, transformBatch, now);
  65. result.Tables.Add(new T8BaseInboundTableResult
  66. {
  67. EntityCode = code,
  68. StgRows = pull.RowsWritten,
  69. StdRows = stdRows
  70. });
  71. }
  72. return result;
  73. }
  74. /// <summary>把 PENDING 贴源行按各表 typed 契约投影到 std(JSON key 区分大小写:源标识列为 Id/id,其余小写)。</summary>
  75. private async Task<int> TransformStandardAsync(
  76. string entityCode, string stgTable, string stdTable, long tenantId, string batchId, DateTime now)
  77. {
  78. var sql = entityCode switch
  79. {
  80. "T8_KC_TZ_HEAD" => BuildTzHeadTransform(),
  81. "T8_KC_TZ_LIST" => BuildTzListTransform(),
  82. "T8_KC_DD_HEAD" => BuildDdHeadTransform(),
  83. "T8_KC_DD_LIST" => BuildDdListTransform(),
  84. "T8_KC_ZJ_LIST" => BuildZjListTransform(),
  85. "T8_SYS_PELIST" => BuildPelistTransform(),
  86. "T8_CJ_BG_HEAD_REP" => BuildCjBgHeadRepTransform(),
  87. "T8_KC_DD_LIST_CLLIST" => BuildDdListCllistTransform(),
  88. _ => throw new InvalidOperationException($"未支持的 T8 入站实体:{entityCode}")
  89. };
  90. var affected = await _db.Ado.ExecuteCommandAsync(sql,
  91. new SugarParameter("@tid", tenantId),
  92. new SugarParameter("@batch", batchId),
  93. new SugarParameter("@now", now));
  94. await _db.Ado.ExecuteCommandAsync(
  95. $"UPDATE {stgTable} SET process_status='DONE', update_time=NOW() WHERE process_status='PENDING'");
  96. return affected;
  97. }
  98. // JSON 取值片段:数字/整数经 NULLIF 去 'null';datetime 把 ISO 'T' 换空格再 STR_TO_DATE(尾部小数位被忽略)。
  99. private static string JStr(string k) => $"NULLIF(JSON_UNQUOTE(JSON_EXTRACT(s.raw_data,'$.{k}')),'null')";
  100. private static string JNum(string k) => $"CAST({JStr(k)} AS DECIMAL(18,6))";
  101. private static string JInt(string k) => $"CAST({JStr(k)} AS SIGNED)";
  102. // ISO 时间戳(System.Text.Json 输出如 2026-05-08T13:45:16.92):'T'→空格、去亚秒再 STR_TO_DATE,
  103. // 否则 MySQL 严格模式对残留 '.92' 报 Truncated incorrect datetime value。KPI 只用日级,丢亚秒无影响。
  104. private static string JDate(string k) => $"STR_TO_DATE(SUBSTRING_INDEX(REPLACE({JStr(k)},'T',' '),'.',1),'%Y-%m-%d %H:%i:%s')";
  105. private static string SrcId => $"CAST(COALESCE({JStr("Id")},{JStr("id")}) AS SIGNED)";
  106. private static string StgHead(string table) => $@"
  107. FROM {table} s
  108. WHERE s.process_status='PENDING' AND s.source_biz_key IS NOT NULL AND s.source_biz_key<>''";
  109. private static string BuildTzHeadTransform() => $@"
  110. INSERT INTO mdp_std_t8_kc_tz_head
  111. (tenant_id, source_system, src_id, ztid, lbs, lynoid, hzyn, zfyn, shyn, shtime, date0,
  112. source_row_id, source_biz_key, sync_batch_id, sync_time)
  113. SELECT IFNULL(s.tenant_id,@tid), IFNULL(NULLIF(s.source_system,''),'T8'),
  114. {SrcId}, {JStr("ztid")}, {JStr("lbs")}, {JStr("lynoid")}, {JInt("hzyn")}, {JInt("zfyn")}, {JInt("shyn")},
  115. {JDate("shtime")}, {JDate("date0")},
  116. IFNULL(NULLIF(s.source_row_id,''),s.source_biz_key), s.source_biz_key, @batch, @now
  117. {StgHead("mdp_stg_t8_kc_tz_head")}
  118. ON DUPLICATE KEY UPDATE
  119. src_id=VALUES(src_id), ztid=VALUES(ztid), lbs=VALUES(lbs), lynoid=VALUES(lynoid),
  120. hzyn=VALUES(hzyn), zfyn=VALUES(zfyn), shyn=VALUES(shyn),
  121. shtime=VALUES(shtime), date0=VALUES(date0),
  122. source_row_id=VALUES(source_row_id), sync_batch_id=VALUES(sync_batch_id),
  123. sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP";
  124. private static string BuildTzListTransform() => $@"
  125. INSERT INTO mdp_std_t8_kc_tz_list
  126. (tenant_id, source_system, src_id, idid, code, lynoid, slzx, sl, gdyn, gdtime, rwnoid, jhdate, addtime,
  127. source_row_id, source_biz_key, sync_batch_id, sync_time)
  128. SELECT IFNULL(s.tenant_id,@tid), IFNULL(NULLIF(s.source_system,''),'T8'),
  129. {SrcId}, {JInt("idid")}, {JStr("code")}, {JStr("lynoid")}, {JNum("slzx")}, {JNum("sl")},
  130. {JInt("gdyn")}, {JDate("gdtime")}, {JStr("rwnoid")}, {JDate("jhdate")}, {JDate("addtime")},
  131. IFNULL(NULLIF(s.source_row_id,''),s.source_biz_key), s.source_biz_key, @batch, @now
  132. {StgHead("mdp_stg_t8_kc_tz_list")}
  133. ON DUPLICATE KEY UPDATE
  134. src_id=VALUES(src_id), idid=VALUES(idid), code=VALUES(code), lynoid=VALUES(lynoid),
  135. slzx=VALUES(slzx), sl=VALUES(sl), gdyn=VALUES(gdyn), gdtime=VALUES(gdtime),
  136. rwnoid=VALUES(rwnoid), jhdate=VALUES(jhdate), addtime=VALUES(addtime),
  137. source_row_id=VALUES(source_row_id), sync_batch_id=VALUES(sync_batch_id),
  138. sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP";
  139. private static string BuildDdHeadTransform() => $@"
  140. INSERT INTO mdp_std_t8_kc_dd_head
  141. (tenant_id, source_system, src_id, ztid, lbs, zf, shyn, noid,
  142. source_row_id, source_biz_key, sync_batch_id, sync_time)
  143. SELECT IFNULL(s.tenant_id,@tid), IFNULL(NULLIF(s.source_system,''),'T8'),
  144. {SrcId}, {JStr("ztid")}, {JStr("lbs")}, {JInt("zf")}, {JInt("shyn")}, {JStr("noid")},
  145. IFNULL(NULLIF(s.source_row_id,''),s.source_biz_key), s.source_biz_key, @batch, @now
  146. {StgHead("mdp_stg_t8_kc_dd_head")}
  147. ON DUPLICATE KEY UPDATE
  148. src_id=VALUES(src_id), ztid=VALUES(ztid), lbs=VALUES(lbs),
  149. zf=VALUES(zf), shyn=VALUES(shyn), noid=VALUES(noid),
  150. source_row_id=VALUES(source_row_id), sync_batch_id=VALUES(sync_batch_id),
  151. sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP";
  152. private static string BuildDdListTransform() => $@"
  153. INSERT INTO mdp_std_t8_kc_dd_list
  154. (tenant_id, source_system, src_id, idid, rwnoid, code, sl, slzx, jhdate, gdyn, gdtime, addtime,
  155. source_row_id, source_biz_key, sync_batch_id, sync_time)
  156. SELECT IFNULL(s.tenant_id,@tid), IFNULL(NULLIF(s.source_system,''),'T8'),
  157. {SrcId}, {JInt("idid")}, {JStr("rwnoid")}, {JStr("code")}, {JNum("sl")}, {JNum("slzx")},
  158. {JDate("jhdate")}, {JInt("gdyn")}, {JDate("gdtime")}, {JDate("addtime")},
  159. IFNULL(NULLIF(s.source_row_id,''),s.source_biz_key), s.source_biz_key, @batch, @now
  160. {StgHead("mdp_stg_t8_kc_dd_list")}
  161. ON DUPLICATE KEY UPDATE
  162. src_id=VALUES(src_id), idid=VALUES(idid), rwnoid=VALUES(rwnoid), code=VALUES(code),
  163. sl=VALUES(sl), slzx=VALUES(slzx), jhdate=VALUES(jhdate), gdyn=VALUES(gdyn), gdtime=VALUES(gdtime),
  164. addtime=VALUES(addtime),
  165. source_row_id=VALUES(source_row_id), sync_batch_id=VALUES(sync_batch_id),
  166. sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP";
  167. private static string BuildZjListTransform() => $@"
  168. INSERT INTO mdp_std_t8_kc_zj_list
  169. (tenant_id, source_system, src_id, ztid, lyid, zjyn, shdate,
  170. source_row_id, source_biz_key, sync_batch_id, sync_time)
  171. SELECT IFNULL(s.tenant_id,@tid), IFNULL(NULLIF(s.source_system,''),'T8'),
  172. {SrcId}, {JStr("ztid")}, {JInt("lyid")}, {JInt("zjyn")}, {JDate("shdate")},
  173. IFNULL(NULLIF(s.source_row_id,''),s.source_biz_key), s.source_biz_key, @batch, @now
  174. {StgHead("mdp_stg_t8_kc_zj_list")}
  175. ON DUPLICATE KEY UPDATE
  176. src_id=VALUES(src_id), ztid=VALUES(ztid), lyid=VALUES(lyid), zjyn=VALUES(zjyn), shdate=VALUES(shdate),
  177. source_row_id=VALUES(source_row_id), sync_batch_id=VALUES(sync_batch_id),
  178. sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP";
  179. private static string BuildPelistTransform() => $@"
  180. INSERT INTO mdp_std_t8_sys_pelist
  181. (tenant_id, source_system, src_id, ztid, zzzt, gw,
  182. source_row_id, source_biz_key, sync_batch_id, sync_time)
  183. SELECT IFNULL(s.tenant_id,@tid), IFNULL(NULLIF(s.source_system,''),'T8'),
  184. {SrcId}, {JStr("ztid")}, {JStr("zzzt")}, {JStr("gw")},
  185. IFNULL(NULLIF(s.source_row_id,''),s.source_biz_key), s.source_biz_key, @batch, @now
  186. {StgHead("mdp_stg_t8_sys_pelist")}
  187. ON DUPLICATE KEY UPDATE
  188. src_id=VALUES(src_id), ztid=VALUES(ztid), zzzt=VALUES(zzzt), gw=VALUES(gw),
  189. source_row_id=VALUES(source_row_id), sync_batch_id=VALUES(sync_batch_id),
  190. sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP";
  191. // S5_L1_002 分子:报工头,按 noid 取 min(kgdate)=开工时间(下游 KPI 侧 GROUP BY noid 重算 MIN,保留逐行 kgdate)。
  192. private static string BuildCjBgHeadRepTransform() => $@"
  193. INSERT INTO mdp_std_t8_cj_bg_head_rep
  194. (tenant_id, source_system, src_id, noid, kgdate, ztid,
  195. source_row_id, source_biz_key, sync_batch_id, sync_time)
  196. SELECT IFNULL(s.tenant_id,@tid), IFNULL(NULLIF(s.source_system,''),'T8'),
  197. {SrcId}, {JStr("noid")}, {JDate("kgdate")}, {JStr("ztid")},
  198. IFNULL(NULLIF(s.source_row_id,''),s.source_biz_key), s.source_biz_key, @batch, @now
  199. {StgHead("mdp_stg_t8_cj_bg_head_rep")}
  200. ON DUPLICATE KEY UPDATE
  201. src_id=VALUES(src_id), noid=VALUES(noid), kgdate=VALUES(kgdate), ztid=VALUES(ztid),
  202. source_row_id=VALUES(source_row_id), sync_batch_id=VALUES(sync_batch_id),
  203. sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP";
  204. // S5_L1_002 分母:订单物料明细,count 行数 per 工单(关联 kc_dd_head.src_id = cllist.idid)。
  205. private static string BuildDdListCllistTransform() => $@"
  206. INSERT INTO mdp_std_t8_kc_dd_list_cllist
  207. (tenant_id, source_system, src_id, idid,
  208. source_row_id, source_biz_key, sync_batch_id, sync_time)
  209. SELECT IFNULL(s.tenant_id,@tid), IFNULL(NULLIF(s.source_system,''),'T8'),
  210. {SrcId}, {JInt("idid")},
  211. IFNULL(NULLIF(s.source_row_id,''),s.source_biz_key), s.source_biz_key, @batch, @now
  212. {StgHead("mdp_stg_t8_kc_dd_list_cllist")}
  213. ON DUPLICATE KEY UPDATE
  214. src_id=VALUES(src_id), idid=VALUES(idid),
  215. source_row_id=VALUES(source_row_id), sync_batch_id=VALUES(sync_batch_id),
  216. sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP";
  217. /// <summary>建 8 stg(通用信封契约)+ 8 std(KPI 实际引用的 typed 列)。幂等 IF NOT EXISTS。</summary>
  218. private async Task EnsureTablesAsync()
  219. {
  220. foreach (var (_, stgTable, _) in Entities)
  221. await _db.Ado.ExecuteCommandAsync(StgDdl(stgTable));
  222. await _db.Ado.ExecuteCommandAsync(StdDdl("mdp_std_t8_kc_tz_head",
  223. "src_id bigint DEFAULT NULL, ztid varchar(100) DEFAULT NULL, lbs varchar(100) DEFAULT NULL, " +
  224. "lynoid varchar(400) DEFAULT NULL, " +
  225. "hzyn int DEFAULT NULL, zfyn int DEFAULT NULL, shyn int DEFAULT NULL, " +
  226. "shtime datetime DEFAULT NULL, date0 date DEFAULT NULL"));
  227. await _db.Ado.ExecuteCommandAsync(StdDdl("mdp_std_t8_kc_tz_list",
  228. "src_id bigint DEFAULT NULL, idid bigint DEFAULT NULL, code varchar(400) DEFAULT NULL, " +
  229. "lynoid varchar(400) DEFAULT NULL, slzx decimal(18,6) DEFAULT NULL, sl decimal(18,6) DEFAULT NULL, " +
  230. "gdyn bigint DEFAULT NULL, gdtime datetime DEFAULT NULL, rwnoid varchar(400) DEFAULT NULL, " +
  231. "jhdate date DEFAULT NULL, addtime datetime DEFAULT NULL"));
  232. await _db.Ado.ExecuteCommandAsync(StdDdl("mdp_std_t8_kc_dd_head",
  233. "src_id bigint DEFAULT NULL, ztid varchar(100) DEFAULT NULL, lbs varchar(100) DEFAULT NULL, " +
  234. "zf int DEFAULT NULL, shyn int DEFAULT NULL, noid varchar(400) DEFAULT NULL"));
  235. await _db.Ado.ExecuteCommandAsync(StdDdl("mdp_std_t8_kc_dd_list",
  236. "src_id bigint DEFAULT NULL, idid bigint DEFAULT NULL, rwnoid varchar(400) DEFAULT NULL, " +
  237. "code varchar(400) DEFAULT NULL, sl decimal(18,6) DEFAULT NULL, slzx decimal(18,6) DEFAULT NULL, " +
  238. "jhdate date DEFAULT NULL, gdyn int DEFAULT NULL, gdtime datetime DEFAULT NULL, addtime datetime DEFAULT NULL"));
  239. await _db.Ado.ExecuteCommandAsync(StdDdl("mdp_std_t8_kc_zj_list",
  240. "src_id bigint DEFAULT NULL, ztid varchar(100) DEFAULT NULL, lyid bigint DEFAULT NULL, " +
  241. "zjyn int DEFAULT NULL, shdate datetime DEFAULT NULL"));
  242. await _db.Ado.ExecuteCommandAsync(StdDdl("mdp_std_t8_sys_pelist",
  243. "src_id bigint DEFAULT NULL, ztid varchar(100) DEFAULT NULL, zzzt varchar(100) DEFAULT NULL, " +
  244. "gw varchar(400) DEFAULT NULL"));
  245. // S5_L1_002 中台化新增 2 张 std(stg 由上方 Entities 循环建)。
  246. await _db.Ado.ExecuteCommandAsync(StdDdl("mdp_std_t8_cj_bg_head_rep",
  247. "src_id bigint DEFAULT NULL, noid varchar(400) DEFAULT NULL, kgdate date DEFAULT NULL, " +
  248. "ztid varchar(100) DEFAULT NULL"));
  249. await _db.Ado.ExecuteCommandAsync(StdDdl("mdp_std_t8_kc_dd_list_cllist",
  250. "src_id bigint DEFAULT NULL, idid bigint DEFAULT NULL"));
  251. // 既有 mdp_std_t8_kc_tz_head(CREATE IF NOT EXISTS 不会补列)幂等补 lynoid(S5_L1_002 分子按工单号分组用)。
  252. await EnsureColumnAsync("mdp_std_t8_kc_tz_head", "lynoid", "varchar(400) DEFAULT NULL");
  253. }
  254. /// <summary>幂等补列:目标表缺该列时 ADD COLUMN;已有则跳过。用于向既有 std 表安全追加字段。</summary>
  255. private async Task EnsureColumnAsync(string table, string column, string columnDef)
  256. {
  257. var exists = await _db.Ado.GetIntAsync(
  258. "SELECT COUNT(*) FROM information_schema.COLUMNS " +
  259. "WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=@t AND COLUMN_NAME=@c",
  260. new SugarParameter("@t", table), new SugarParameter("@c", column));
  261. if (exists == 0)
  262. await _db.Ado.ExecuteCommandAsync($"ALTER TABLE {table} ADD COLUMN {column} {columnDef}");
  263. }
  264. private static string StgDdl(string table) => $@"
  265. CREATE TABLE IF NOT EXISTS {table} (
  266. id bigint NOT NULL AUTO_INCREMENT,
  267. tenant_id bigint NOT NULL DEFAULT 0,
  268. source_system varchar(50) DEFAULT NULL,
  269. source_table varchar(200) DEFAULT NULL,
  270. source_row_id varchar(200) DEFAULT NULL,
  271. source_biz_key varchar(300) DEFAULT NULL,
  272. raw_data json DEFAULT NULL,
  273. sync_batch_id varchar(100) DEFAULT NULL,
  274. create_time datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,
  275. sync_time datetime DEFAULT CURRENT_TIMESTAMP,
  276. process_status varchar(20) NOT NULL DEFAULT 'PENDING',
  277. process_message varchar(500) DEFAULT NULL,
  278. update_time datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  279. PRIMARY KEY (id),
  280. UNIQUE KEY uk_source_key (source_system, source_table, source_biz_key),
  281. KEY idx_batch (sync_batch_id)
  282. ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='T8 基表贴源层(双模式入站)'";
  283. private static string StdDdl(string table, string bizCols) => $@"
  284. CREATE TABLE IF NOT EXISTS {table} (
  285. id bigint NOT NULL AUTO_INCREMENT,
  286. tenant_id bigint NOT NULL DEFAULT 0,
  287. source_system varchar(50) NOT NULL DEFAULT 'T8',
  288. {bizCols},
  289. source_row_id varchar(200) NOT NULL,
  290. source_biz_key varchar(300) NOT NULL,
  291. sync_batch_id varchar(100) NOT NULL,
  292. sync_time datetime NOT NULL,
  293. update_time datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  294. PRIMARY KEY (id),
  295. UNIQUE KEY uk_{table} (tenant_id, source_system, source_biz_key),
  296. KEY idx_{table}_batch (sync_batch_id)
  297. ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='T8 基表标准层(双模式入站)'";
  298. }
  299. /// <summary>T8 基表入站结果。</summary>
  300. public sealed class T8BaseInboundResult
  301. {
  302. public string BatchId { get; set; } = string.Empty;
  303. public List<T8BaseInboundTableResult> Tables { get; } = new();
  304. }
  305. public sealed class T8BaseInboundTableResult
  306. {
  307. public string EntityCode { get; set; } = string.Empty;
  308. public int StgRows { get; set; }
  309. public int StdRows { get; set; }
  310. }