T8BaseInboundMdpSyncService.cs 19 KB

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