T8BaseInboundMdpSyncService.cs 19 KB

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