|
|
@@ -95,6 +95,103 @@ public class IpqcInspectionMdpSyncService : ITransient
|
|
|
}
|
|
|
}
|
|
|
|
|
|
+ /// <summary>本库 MySQL 贴源 source_system(与 MdpDbPullExecutor 落 stg 的取值一致,决定 uk_source_key 归属)。</summary>
|
|
|
+ private const string NativeSourceSystem = "AIDOPDEV_MYSQL";
|
|
|
+
|
|
|
+ /// <summary>targeted sync 的单据作用域(租户 + 检验单 id)。</summary>
|
|
|
+ private sealed class IpqcBillScope
|
|
|
+ {
|
|
|
+ public long TenantId { get; init; }
|
|
|
+ public long InspectionId { get; init; }
|
|
|
+ }
|
|
|
+
|
|
|
+ /// <summary>
|
|
|
+ /// 单据级 targeted 同步(业务写入后立即可见),替代把 Full/Inbound 全表扫描挂到高频提交路径。
|
|
|
+ ///
|
|
|
+ /// 只处理 inspectionId 这一张过程检验单的<b>头表</b>:
|
|
|
+ /// qms_gcjyd (id=inspectionId AND tenant_id=tenantId) → mdp_stg_ipqc_pull → mdp_std_ipqc_inspection
|
|
|
+ ///
|
|
|
+ /// <b>刻意不同步明细</b>(qms_gcjydzb → mdp_std_ipqc_inspection_detail):该表全仓无任何应用
|
|
|
+ /// INSERT/UPDATE/DELETE,源不变则标准层不可能因业务写入而滞后,纳入即为无依据的额外开销。
|
|
|
+ ///
|
|
|
+ /// 边界:只读业务源、只写 mdp_stg_ipqc_pull / mdp_std_ipqc_inspection / run_log;
|
|
|
+ /// 租户双重收口(源查询 + transform 均按 tenantId);纯 upsert,无 DELETE / 无 orphan purge /
|
|
|
+ /// 无 MdpStdFullReplace / 不复用 RunFullAsync 的 pbxfxp 账套映射租户。
|
|
|
+ /// </summary>
|
|
|
+ public async Task<IpqcInspectionMdpSyncResult> SyncInspectionAsync(
|
|
|
+ long inspectionId, long tenantId, string triggerType = "BILL", CancellationToken cancellationToken = default)
|
|
|
+ {
|
|
|
+ cancellationToken.ThrowIfCancellationRequested();
|
|
|
+ if (inspectionId <= 0 || tenantId <= 0)
|
|
|
+ return new IpqcInspectionMdpSyncResult { BatchId = "SKIPPED_INVALID_SCOPE" };
|
|
|
+
|
|
|
+ await EnsureTablesAsync();
|
|
|
+ await EnsurePullStgTableAsync();
|
|
|
+
|
|
|
+ var now = DateTime.Now;
|
|
|
+ var batchId = $"S6_IPQC_BILL_{now:yyyyMMddHHmmssfff}_{inspectionId}";
|
|
|
+ var runLogId = await InsertRunLogAsync(batchId, now, triggerType);
|
|
|
+ var result = new IpqcInspectionMdpSyncResult { BatchId = batchId, RunLogId = runLogId };
|
|
|
+ _tenantSkipWarnings.Clear();
|
|
|
+
|
|
|
+ try
|
|
|
+ {
|
|
|
+ var scope = new IpqcBillScope { TenantId = tenantId, InspectionId = inspectionId };
|
|
|
+ // ① 贴源:把该单据的源行刷进 pull stg(transform 一律读 stg,不先刷就还是旧快照)
|
|
|
+ result.HeadStageRows = await UpsertHeadStgFromSourceAsync(scope, batchId, now);
|
|
|
+ // ② 标准层:复用全量同一套 transform,仅加单据作用域,字段映射零分叉
|
|
|
+ result.HeadStandardRows = await TransformHeadStandardAsync(batchId, now, null, scope);
|
|
|
+ result.TenantSkipWarnings = _tenantSkipWarnings.ToList();
|
|
|
+ await MarkRunSuccessAsync(runLogId, now, result);
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+ catch (Exception ex)
|
|
|
+ {
|
|
|
+ await MarkRunFailedAsync(runLogId, now, ex.Message);
|
|
|
+ throw;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /// <summary>
|
|
|
+ /// 单据级头表贴源 upsert:qms_gcjyd 源行 → mdp_stg_ipqc_pull。
|
|
|
+ /// raw_data 按源表<b>全列</b>动态生成 JSON_OBJECT,与 MdpDbPullExecutor/MdpStagingWriter 的信封形状一致
|
|
|
+ /// (JSON 键 = 源列名;source_row_id = id;source_biz_key = djbh,空则回落 id,与 MdpStagingWriter.BuildBizKey
|
|
|
+ /// 的 "biz_key_expr=djbh,取不到则回落 sourceRowId" 语义一致),因此命中既有 uk_source_key 行原地更新,
|
|
|
+ /// 不产生重复 stg 行,也不会因 transform 将来新增字段而漂移。
|
|
|
+ /// </summary>
|
|
|
+ private async Task<int> UpsertHeadStgFromSourceAsync(IpqcBillScope scope, string batchId, DateTime now)
|
|
|
+ {
|
|
|
+ var cols = await _db.Ado.SqlQueryAsync<string>(
|
|
|
+ "SELECT COLUMN_NAME FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='qms_gcjyd' ORDER BY ORDINAL_POSITION");
|
|
|
+ // 列名取自 information_schema 且强制标识符白名单,非用户输入,无注入面
|
|
|
+ cols = cols.Where(c => !string.IsNullOrWhiteSpace(c) && System.Text.RegularExpressions.Regex.IsMatch(c, "^[A-Za-z0-9_]+$")).ToList();
|
|
|
+ if (cols.Count == 0) return 0;
|
|
|
+
|
|
|
+ var jsonObj = "JSON_OBJECT(" + string.Join(", ", cols.Select(c => $"'{c}', s.`{c}`")) + ")";
|
|
|
+ return await _db.Ado.ExecuteCommandAsync(
|
|
|
+ $"""
|
|
|
+ INSERT INTO mdp_stg_ipqc_pull
|
|
|
+ (tenant_id, source_system, source_table, source_row_id, source_biz_key, raw_data, sync_batch_id, sync_time, process_status, create_time)
|
|
|
+ SELECT s.tenant_id, @SrcSys, 'qms_gcjyd', CAST(s.id AS CHAR),
|
|
|
+ IFNULL(NULLIF(s.djbh,''), CAST(s.id AS CHAR)), {jsonObj},
|
|
|
+ @BatchId, @Now, 'PENDING', @Now
|
|
|
+ FROM qms_gcjyd s
|
|
|
+ WHERE s.id=@Id AND s.tenant_id=@TenantId
|
|
|
+ ON DUPLICATE KEY UPDATE
|
|
|
+ tenant_id=VALUES(tenant_id), source_row_id=VALUES(source_row_id), raw_data=VALUES(raw_data),
|
|
|
+ sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time),
|
|
|
+ process_status='PENDING', update_time=CURRENT_TIMESTAMP
|
|
|
+ """,
|
|
|
+ new List<SugarParameter>
|
|
|
+ {
|
|
|
+ new("@SrcSys", NativeSourceSystem),
|
|
|
+ new("@BatchId", batchId),
|
|
|
+ new("@Now", now),
|
|
|
+ new("@Id", scope.InspectionId),
|
|
|
+ new("@TenantId", scope.TenantId),
|
|
|
+ });
|
|
|
+ }
|
|
|
+
|
|
|
/// <summary>
|
|
|
/// 双模式入站:执行器抽数落 pull-stg,再跑标准层(头读 pull stg)。
|
|
|
/// </summary>
|
|
|
@@ -532,7 +629,7 @@ public class IpqcInspectionMdpSyncService : ITransient
|
|
|
}
|
|
|
|
|
|
/// <summary>标准化头:mdp_stg_ipqc_pull(qms_gcjyd) → mdp_std_ipqc_inspection。</summary>
|
|
|
- private async Task<int> TransformHeadStandardAsync(string batchId, DateTime now, string? sourceSystem = null)
|
|
|
+ private async Task<int> TransformHeadStandardAsync(string batchId, DateTime now, string? sourceSystem = null, IpqcBillScope? scope = null)
|
|
|
{
|
|
|
// 双源:sourceSystem 非空时仅统计/转换当前 source(切源 FULL Replace 用);为空时保持既有全源行为不变。
|
|
|
var countPars = new List<SugarParameter>();
|
|
|
@@ -545,6 +642,16 @@ public class IpqcInspectionMdpSyncService : ITransient
|
|
|
countPars.Add(new SugarParameter("@Src", sourceSystem));
|
|
|
}
|
|
|
var mTenant = MdpJsonSql.TenantFromStg("m");
|
|
|
+ // 单据作用域(targeted sync):按 source_row_id 收敛到单张检验单,并强制租户相等。
|
|
|
+ // 租户显式来自业务单,不依赖 RunFullAsync 的 pbxfxp 账套映射上下文。scope 为 null 时全量行为逐字不变。
|
|
|
+ if (scope != null)
|
|
|
+ {
|
|
|
+ var scopeSql = $" AND m.source_row_id=@ScopeRowId AND ({mTenant})=@ScopeTenant";
|
|
|
+ countSrc += scopeSql;
|
|
|
+ insertSrc += scopeSql;
|
|
|
+ countPars.Add(new SugarParameter("@ScopeRowId", scope.InspectionId.ToString()));
|
|
|
+ countPars.Add(new SugarParameter("@ScopeTenant", scope.TenantId));
|
|
|
+ }
|
|
|
var rows = await _db.Ado.GetIntAsync(
|
|
|
$"SELECT COUNT(1) FROM mdp_stg_ipqc_pull m WHERE m.source_table='qms_gcjyd'{countSrc} AND {MdpJsonSql.TenantGuard(mTenant)}", countPars);
|
|
|
var skipped = await _db.Ado.GetIntAsync(
|
|
|
@@ -603,6 +710,11 @@ public class IpqcInspectionMdpSyncService : ITransient
|
|
|
new("@Now", now)
|
|
|
};
|
|
|
if (sourceSystem != null) insPars.Add(new SugarParameter("@Src", sourceSystem));
|
|
|
+ if (scope != null)
|
|
|
+ {
|
|
|
+ insPars.Add(new SugarParameter("@ScopeRowId", scope.InspectionId.ToString()));
|
|
|
+ insPars.Add(new SugarParameter("@ScopeTenant", scope.TenantId));
|
|
|
+ }
|
|
|
await _db.Ado.ExecuteCommandAsync(insertSql, insPars);
|
|
|
return rows;
|
|
|
}
|