|
|
@@ -101,6 +101,123 @@ public class FqcMdpSyncService : ITransient
|
|
|
/// <summary>FQC 全量刷新并发锁键（同库同时只跑一个）。</summary>
|
|
|
private const string FqcRefreshLockKey = "aidop:s7:fqc:refresh";
|
|
|
|
|
|
+ /// <summary>本库 MySQL 贴源 source_system（与 MdpDbPullExecutor 落 stg 的取值一致，决定 uk_source_key 归属）。</summary>
|
|
|
+ private const string NativeSourceSystem = "AIDOPDEV_MYSQL";
|
|
|
+
|
|
|
+ /// <summary>targeted sync 的单据作用域（租户 + 检验单 id + 报检单号）。</summary>
|
|
|
+ private sealed class FqcBillScope
|
|
|
+ {
|
|
|
+ public long TenantId { get; init; }
|
|
|
+ public long BillId { get; init; }
|
|
|
+ /// <summary>qms_qcpp_inspbill.lydjbh = qms_fqcbj.FBILLNO；为空则跳过任务表同步。</summary>
|
|
|
+ public string? SourceBillNo { get; init; }
|
|
|
+ }
|
|
|
+
|
|
|
+ /// <summary>把作用域参数补进 transform 的参数表（未启用作用域时不加，保持全量 SQL 原样）。</summary>
|
|
|
+ private static void AddScopePars(List<SugarParameter> pars, FqcBillScope? scope)
|
|
|
+ {
|
|
|
+ if (scope == null) return;
|
|
|
+ pars.Add(new SugarParameter("@ScopeTenant", scope.TenantId));
|
|
|
+ pars.Add(new SugarParameter("@ScopeBillId", scope.BillId.ToString()));
|
|
|
+ pars.Add(new SugarParameter("@ScopeBillNo", scope.SourceBillNo));
|
|
|
+ }
|
|
|
+
|
|
|
+ /// <summary>
|
|
|
+ /// 单据级 targeted 同步（业务写入后立即可见，替代把 12s 全量 <see cref="RunFullAsync"/> 挂到高频提交路径）。
|
|
|
+ ///
|
|
|
+ /// 只处理 billId 这一张检验单：源 → stg（贴源 upsert）→ std 三表（复用全量同一套 transform，字段映射零分叉）。
|
|
|
+ /// qms_qcpp_inspbill (id=billId) → mdp_std_fqc_result
|
|
|
+ /// qms_qcpp_inspbillst(glid=billId) → mdp_std_fqc_result_detail
|
|
|
+ /// qms_fqcbj (FBILLNO=inspbill.lydjbh) → mdp_std_fqc_task（submit-result/退回会改 FINSPECTSTATUS 等进度列）
|
|
|
+ ///
|
|
|
+ /// 边界：只读业务源、只写 mdp_stg_fqc_pull / mdp_std_fqc_*；全程 tenant_id 强约束（源查询与 transform 双重）；
|
|
|
+ /// 纯 upsert，不 DELETE、不跑 orphan purge、不抢全量 GET_LOCK、不产生 S7_FQC_FULL 批次。
|
|
|
+ /// </summary>
|
|
|
+ public async Task<FqcMdpSyncResult> SyncBillAsync(
|
|
|
+ long billId, long tenantId, string triggerType = "BILL", CancellationToken cancellationToken = default)
|
|
|
+ {
|
|
|
+ cancellationToken.ThrowIfCancellationRequested();
|
|
|
+ if (billId <= 0 || tenantId <= 0) return new FqcMdpSyncResult { BatchId = "SKIPPED_INVALID_SCOPE", Skipped = true };
|
|
|
+
|
|
|
+ var now = DateTime.Now;
|
|
|
+ var batchId = $"S7_FQC_BILL_{now:yyyyMMddHHmmssfff}_{billId}";
|
|
|
+ var runLogId = await InsertRunLogAsync(batchId, now, triggerType);
|
|
|
+ var result = new FqcMdpSyncResult { BatchId = batchId, RunLogId = runLogId };
|
|
|
+
|
|
|
+ try
|
|
|
+ {
|
|
|
+ // 来源单号：决定是否需要同步任务表；同时用租户约束再证一次单据归属
|
|
|
+ var sourceBillNo = await _db.Ado.SqlQuerySingleAsync<string>(
|
|
|
+ "SELECT lydjbh FROM qms_qcpp_inspbill WHERE id=@id AND tenant_id=@tid LIMIT 1",
|
|
|
+ new SugarParameter("@id", billId), new SugarParameter("@tid", tenantId));
|
|
|
+ var scope = new FqcBillScope { TenantId = tenantId, BillId = billId, SourceBillNo = sourceBillNo };
|
|
|
+
|
|
|
+ // ① 贴源：把该单据的源行刷进 stg（transform 一律读 stg，故必须先刷，否则同步的还是旧快照）
|
|
|
+ result.StageRows = await UpsertStgFromSourceAsync(
|
|
|
+ "qms_qcpp_inspbill", "s.id=@id AND s.tenant_id=@tid",
|
|
|
+ new List<SugarParameter> { new("@id", billId), new("@tid", tenantId) }, batchId, now);
|
|
|
+ result.StageRows += await UpsertStgFromSourceAsync(
|
|
|
+ "qms_qcpp_inspbillst", "s.glid=@id AND s.tenant_id=@tid",
|
|
|
+ new List<SugarParameter> { new("@id", billId), new("@tid", tenantId) }, batchId, now);
|
|
|
+ if (!string.IsNullOrWhiteSpace(sourceBillNo))
|
|
|
+ result.StageRows += await UpsertStgFromSourceAsync(
|
|
|
+ "qms_fqcbj", "s.FBILLNO=@bjbh AND s.tenant_id=@tid",
|
|
|
+ new List<SugarParameter> { new("@bjbh", sourceBillNo), new("@tid", tenantId) }, batchId, now);
|
|
|
+
|
|
|
+ // ② 标准层：复用全量 transform，仅加单据作用域。result 必须先于 detail（detail 回填 std_result_id）
|
|
|
+ result.ResultRows = await TransformResultStandardAsync(batchId, now, null, scope);
|
|
|
+ result.DetailRows = await TransformResultDetailStandardAsync(batchId, now, null, scope);
|
|
|
+ if (!string.IsNullOrWhiteSpace(sourceBillNo))
|
|
|
+ result.TaskRows = await TransformTaskStandardAsync(batchId, now, null, scope);
|
|
|
+
|
|
|
+ await MarkRunSuccessAsync(runLogId, now, result);
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+ catch (Exception ex)
|
|
|
+ {
|
|
|
+ await MarkRunFailedAsync(runLogId, now, ex.Message);
|
|
|
+ throw;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /// <summary>
|
|
|
+ /// 单表贴源 upsert：源行 → mdp_stg_fqc_pull，raw_data 按源表<b>全列</b>动态生成 JSON_OBJECT，
|
|
|
+ /// 与 MdpDbPullExecutor 的信封形状一致（source_row_id=source_biz_key=主键，JSON 键=源列名），
|
|
|
+ /// 因此绝不会因为 transform 未来新增字段而漂移。命中既有 uk_source_key 则原地更新，不产生重复 stg 行。
|
|
|
+ /// </summary>
|
|
|
+ private async Task<int> UpsertStgFromSourceAsync(
|
|
|
+ string sourceTable, string whereSql, List<SugarParameter> pars, 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=@t ORDER BY ORDINAL_POSITION",
|
|
|
+ new List<SugarParameter> { new("@t", sourceTable) });
|
|
|
+ // 列名来自 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}`")) + ")";
|
|
|
+ var allPars = new List<SugarParameter>(pars)
|
|
|
+ {
|
|
|
+ new("@BatchId", batchId),
|
|
|
+ new("@Now", now),
|
|
|
+ new("@SrcSys", NativeSourceSystem),
|
|
|
+ new("@SrcTab", sourceTable),
|
|
|
+ };
|
|
|
+
|
|
|
+ return await _db.Ado.ExecuteCommandAsync(
|
|
|
+ $"""
|
|
|
+ INSERT INTO mdp_stg_fqc_pull
|
|
|
+ (tenant_id, source_system, source_table, source_row_id, source_biz_key, raw_data, sync_batch_id, sync_time, process_status)
|
|
|
+ SELECT s.tenant_id, @SrcSys, @SrcTab, CAST(s.id AS CHAR), CAST(s.id AS CHAR), {jsonObj}, @BatchId, @Now, 'PENDING'
|
|
|
+ FROM `{sourceTable}` s
|
|
|
+ WHERE {whereSql}
|
|
|
+ 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), update_time=CURRENT_TIMESTAMP
|
|
|
+ """,
|
|
|
+ allPars);
|
|
|
+ }
|
|
|
+
|
|
|
/// <summary>
|
|
|
/// 全量刷新后清理本库(AIDOP)贴源 orphan：源单据（qms_qcpp_inspbill / qms_fqcbj / qms_qcpp_inspbillst）
|
|
|
/// 已删除但 std 层仍残留的行。仅清 source_system='AIDOP'（含历史 NULL/空）的行，SQLSERVER 双源行不受影响；
|
|
|
@@ -458,14 +575,18 @@ public class FqcMdpSyncService : ITransient
|
|
|
}
|
|
|
|
|
|
/// <summary>标准化任务：pull stg(qms_fqcbj) → mdp_std_fqc_task。返回处理行数。</summary>
|
|
|
- private async Task<int> TransformTaskStandardAsync(string batchId, DateTime now, string? sourceSystem = null)
|
|
|
+ private async Task<int> TransformTaskStandardAsync(string batchId, DateTime now, string? sourceSystem = null, FqcBillScope? scope = null)
|
|
|
{
|
|
|
var srcClause = sourceSystem == null ? "" : " AND t.source_system=@Src";
|
|
|
var tTenant = MdpJsonSql.TenantFromStgJsonCol("t", MdpJsonSql.Int("t", "tenant_id"));
|
|
|
var where = $"t.source_table='qms_fqcbj'{srcClause} AND {MdpJsonSql.TenantGuard(tTenant)}";
|
|
|
+ // 单据级作用域（targeted sync）：任务经报检单号关联（qms_fqcbj.FBILLNO = inspbill.lydjbh），并强制租户相等
|
|
|
+ if (scope != null)
|
|
|
+ where += $" AND ({tTenant})=@ScopeTenant AND {MdpJsonSql.Str("t", "FBILLNO")}=@ScopeBillNo";
|
|
|
|
|
|
var countPars = new List<SugarParameter>();
|
|
|
if (sourceSystem != null) countPars.Add(new SugarParameter("@Src", sourceSystem));
|
|
|
+ AddScopePars(countPars, scope);
|
|
|
var rows = await _db.Ado.GetIntAsync(
|
|
|
$"SELECT COUNT(1) FROM mdp_stg_fqc_pull t WHERE {where}", countPars);
|
|
|
|
|
|
@@ -496,20 +617,29 @@ public class FqcMdpSyncService : ITransient
|
|
|
""";
|
|
|
var insPars = new List<SugarParameter> { new("@BatchId", batchId), new("@Now", now) };
|
|
|
if (sourceSystem != null) insPars.Add(new SugarParameter("@Src", sourceSystem));
|
|
|
+ AddScopePars(insPars, scope);
|
|
|
await _db.Ado.ExecuteCommandAsync(insertSql, insPars);
|
|
|
return rows;
|
|
|
}
|
|
|
|
|
|
/// <summary>标准化结果：pull stg(qms_qcpp_inspbill) + WorkOrdMaster(本库富化 SAP 工单号) → mdp_std_fqc_result。</summary>
|
|
|
- private async Task<int> TransformResultStandardAsync(string batchId, DateTime now, string? sourceSystem = null)
|
|
|
+ private async Task<int> TransformResultStandardAsync(string batchId, DateTime now, string? sourceSystem = null, FqcBillScope? scope = null)
|
|
|
{
|
|
|
var srcClause = sourceSystem == null ? "" : " AND a.source_system=@Src";
|
|
|
var aTenant = MdpJsonSql.TenantFromStgJsonCol("a", MdpJsonSql.Int("a", "tenant_id"));
|
|
|
var where = $"a.source_table='qms_qcpp_inspbill'{srcClause} AND {MdpJsonSql.TenantGuard(aTenant)}";
|
|
|
+ // 单据级作用域（targeted sync）：按 stg 的 source_row_id（=inspbill.id）收敛，并强制租户相等，杜绝跨租户命中
|
|
|
+ if (scope != null)
|
|
|
+ where += $" AND ({aTenant})=@ScopeTenant AND IFNULL(a.source_row_id, {MdpJsonSql.Str("a", "id")})=@ScopeBillId";
|
|
|
var sapSub = $"(SELECT MIN(wo.WorkOrd) FROM WorkOrdMaster wo WHERE wo.Batch = {MdpJsonSql.Str("a", "sczld")})";
|
|
|
+ // 判定码表归一（FIX-A2）：源 pd 存两套编码——业务写入 0/1（FqcInspBillFlowService 校验口径 0=合格/1=不合格），
|
|
|
+ // 历史归口数据直存中文。std 面向展示统一为中文；非 0/1 的历史值 ELSE 原样透传，绝不破坏既有数据。
|
|
|
+ // 仅映射整单 pd；明细行 pd（qms_qcpp_inspbillst.pd）源本身即中文，不映射。源表 pd 保持 0/1 不动。
|
|
|
+ var judgmentExpr = $"CASE {MdpJsonSql.Str("a", "pd")} WHEN '0' THEN '合格' WHEN '1' THEN '不合格' ELSE {MdpJsonSql.Str("a", "pd")} END";
|
|
|
|
|
|
var countPars = new List<SugarParameter>();
|
|
|
if (sourceSystem != null) countPars.Add(new SugarParameter("@Src", sourceSystem));
|
|
|
+ AddScopePars(countPars, scope);
|
|
|
var rows = await _db.Ado.GetIntAsync(
|
|
|
$"SELECT COUNT(1) FROM mdp_stg_fqc_pull a WHERE {where}", countPars);
|
|
|
|
|
|
@@ -523,7 +653,7 @@ public class FqcMdpSyncService : ITransient
|
|
|
SELECT
|
|
|
{aTenant}, IFNULL(NULLIF(a.source_system,''), 'AIDOP'),
|
|
|
{MdpJsonSql.Str("a", "FBILLNO")}, {MdpJsonSql.DateTimeSec("a", "FINSPEENDDATE")}, {MdpJsonSql.Str("a", "scph")}, {MdpJsonSql.Str("a", "FMATERIALCFG")}, {MdpJsonSql.Str("a", "wlmc")}, {MdpJsonSql.Str("a", "ggxh")},
|
|
|
- {MdpJsonSql.Str("a", "jyr")}, {MdpJsonSql.Dec("a", "jysj", 18, 6)}, {MdpJsonSql.Str("a", "lydjbh")}, {MdpJsonSql.Str("a", "sczld")}, {MdpJsonSql.Dec("a", "sczldsl", 18, 6)}, {sapSub}, {MdpJsonSql.Str("a", "pd")},
|
|
|
+ {MdpJsonSql.Str("a", "jyr")}, {MdpJsonSql.Dec("a", "jysj", 18, 6)}, {MdpJsonSql.Str("a", "lydjbh")}, {MdpJsonSql.Str("a", "sczld")}, {MdpJsonSql.Dec("a", "sczldsl", 18, 6)}, {sapSub}, {judgmentExpr},
|
|
|
{MdpJsonSql.Dec("a", "hgsl", 18, 6)}, {MdpJsonSql.Dec("a", "bhgsl", 18, 6)}, {MdpJsonSql.Str("a", "jfbh")}, {MdpJsonSql.Str("a", "jgbh")}, {MdpJsonSql.Str("a", "jgbb")},
|
|
|
IFNULL(a.source_row_id, {MdpJsonSql.Str("a", "id")}), IFNULL(NULLIF(a.source_biz_key,''), {MdpJsonSql.Str("a", "FBILLNO")}),
|
|
|
@BatchId, @Now
|
|
|
@@ -540,26 +670,32 @@ public class FqcMdpSyncService : ITransient
|
|
|
""";
|
|
|
var insPars = new List<SugarParameter> { new("@BatchId", batchId), new("@Now", now) };
|
|
|
if (sourceSystem != null) insPars.Add(new SugarParameter("@Src", sourceSystem));
|
|
|
+ AddScopePars(insPars, scope);
|
|
|
await _db.Ado.ExecuteCommandAsync(insertSql, insPars);
|
|
|
return rows;
|
|
|
}
|
|
|
|
|
|
/// <summary>标准化明细：pull stg(qms_qcpp_inspbillst) → mdp_std_fqc_result_detail（回填 std_result_id，j1..j50 逐列）。</summary>
|
|
|
- private async Task<int> TransformResultDetailStandardAsync(string batchId, DateTime now, string? sourceSystem = null)
|
|
|
+ private async Task<int> TransformResultDetailStandardAsync(string batchId, DateTime now, string? sourceSystem = null, FqcBillScope? scope = null)
|
|
|
{
|
|
|
var srcClause = sourceSystem == null ? "" : " AND d.source_system=@Src";
|
|
|
var dTenant = MdpJsonSql.TenantFromStgJsonCol("d", MdpJsonSql.Int("d", "tenant_id"));
|
|
|
+ // 单据级作用域（targeted sync）：明细按 glid（=inspbill.id）收敛，并强制租户相等
|
|
|
+ var scopeClause = scope == null
|
|
|
+ ? ""
|
|
|
+ : $"\n AND ({dTenant})=@ScopeTenant AND {MdpJsonSql.Str("d", "glid")}=@ScopeBillId";
|
|
|
var fromWhere =
|
|
|
$"""
|
|
|
FROM mdp_stg_fqc_pull d
|
|
|
LEFT JOIN mdp_std_fqc_result r
|
|
|
ON r.tenant_id = {dTenant} AND r.source_row_id = {MdpJsonSql.Str("d", "glid")}
|
|
|
WHERE d.source_table='qms_qcpp_inspbillst'{srcClause}
|
|
|
- AND {MdpJsonSql.TenantGuard(dTenant)}
|
|
|
+ AND {MdpJsonSql.TenantGuard(dTenant)}{scopeClause}
|
|
|
""";
|
|
|
|
|
|
var countPars = new List<SugarParameter>();
|
|
|
if (sourceSystem != null) countPars.Add(new SugarParameter("@Src", sourceSystem));
|
|
|
+ AddScopePars(countPars, scope);
|
|
|
var rows = await _db.Ado.GetIntAsync($"SELECT COUNT(1) {fromWhere}", countPars);
|
|
|
|
|
|
var jCols = string.Join(", ", Enumerable.Range(1, 50).Select(i => $"j{i}"));
|
|
|
@@ -592,6 +728,7 @@ public class FqcMdpSyncService : ITransient
|
|
|
""";
|
|
|
var insPars = new List<SugarParameter> { new("@BatchId", batchId), new("@Now", now) };
|
|
|
if (sourceSystem != null) insPars.Add(new SugarParameter("@Src", sourceSystem));
|
|
|
+ AddScopePars(insPars, scope);
|
|
|
await _db.Ado.ExecuteCommandAsync(insertSql, insPars);
|
|
|
return rows;
|
|
|
}
|
YY968XX