|
|
@@ -0,0 +1,668 @@
|
|
|
+using Microsoft.Extensions.Hosting;
|
|
|
+using Microsoft.Extensions.Logging;
|
|
|
+using Microsoft.Extensions.Options;
|
|
|
+using SqlSugar;
|
|
|
+using System.Data;
|
|
|
+using System.Text;
|
|
|
+
|
|
|
+namespace Admin.NET.Plugin.AiDOP.Supply;
|
|
|
+
|
|
|
+/// <summary>
|
|
|
+/// S8 Stage-3 采购订单完成态标准事实层投影服务。
|
|
|
+///
|
|
|
+/// <para>链路：Source B（WMS/MES 的 <c>PurOrdDetail</c>）→ <c>mdp_std_purchase_order_completion</c>。
|
|
|
+/// 目的是让 S8 Stage-3 能在【不覆盖】<c>mdp_std_purchase_order.status</c> 的前提下拿到
|
|
|
+/// 「这条采购行收货是否已关闭」这一 WMS 侧事实。</para>
|
|
|
+///
|
|
|
+/// <para><b>Grain = 一条采购行</b>（<c>tenant_id + po_no + po_line</c>）。
|
|
|
+/// <c>domain</c> / <c>potype</c> / <c>source_row_id</c> / <c>source_id</c> 是 provenance，不进唯一键。</para>
|
|
|
+///
|
|
|
+/// <para><b>当前状态，不是事件</b>：上游收货过程的写法是
|
|
|
+/// <c>Status = (case when 收满 then 'C' else '' end)</c>，退货会把 <c>'C'</c> 打回空串。
|
|
|
+/// 因此本服务的 UPSERT 必须无条件刷新结论，允许 COMPLETED → NOT_COMPLETED 回退。</para>
|
|
|
+///
|
|
|
+/// <para><b>Full Reconciliation 是正确性主路径</b>：HotWatch 只覆盖活动中的采购单，
|
|
|
+/// 实测 5 个 <c>PUR_ORDER</c> 关注里 3 个已提前终止（且无一是因采购完成而终止），
|
|
|
+/// 98 条源行里 88 条根本没有对应关注。所以单靠增量必然漏判，周期性全量对账不可省。</para>
|
|
|
+///
|
|
|
+/// <para><b>刻意不做</b>：不读 IQC、不算 ObjectCompleted、不做订单级聚合、
|
|
|
+/// 不用数量重新推导完成（唯一 Authority 是源侧 Status）、不写任何源表、不改 Source A 的任何列。</para>
|
|
|
+/// </summary>
|
|
|
+public class PurchaseOrderCompletionMdpSyncService : ITransient
|
|
|
+{
|
|
|
+ private const string JobCode = "S3_PO_COMPLETION_MDP_SYNC";
|
|
|
+ private const string JobName = "S3采购订单完成态标准事实层投影";
|
|
|
+
|
|
|
+ /// <summary>完成结论取值域。</summary>
|
|
|
+ public const string StatusCompleted = "COMPLETED";
|
|
|
+ public const string StatusNotCompleted = "NOT_COMPLETED";
|
|
|
+ public const string StatusUnknown = "UNKNOWN";
|
|
|
+
|
|
|
+ private readonly ISqlSugarClient _db;
|
|
|
+ private readonly DataPlatform.MdpSourceScopeFactory _scopeFactory;
|
|
|
+ private readonly AidopPoCompletionOptions _opt;
|
|
|
+ private readonly IHostEnvironment _env;
|
|
|
+ private readonly ILogger<PurchaseOrderCompletionMdpSyncService> _logger;
|
|
|
+
|
|
|
+ public PurchaseOrderCompletionMdpSyncService(
|
|
|
+ ISqlSugarClient db,
|
|
|
+ DataPlatform.MdpSourceScopeFactory scopeFactory,
|
|
|
+ IOptions<AidopPoCompletionOptions> opt,
|
|
|
+ IHostEnvironment env,
|
|
|
+ ILogger<PurchaseOrderCompletionMdpSyncService> logger)
|
|
|
+ {
|
|
|
+ _db = db;
|
|
|
+ _scopeFactory = scopeFactory;
|
|
|
+ _opt = opt.Value;
|
|
|
+ _env = env;
|
|
|
+ _logger = logger;
|
|
|
+ }
|
|
|
+
|
|
|
+ /// <summary>
|
|
|
+ /// 完成态映射。<b>这是全链路唯一的状态映射入口</b>——HotWatch 增量若接入，必须复用本方法，
|
|
|
+ /// 不得另写一套。
|
|
|
+ ///
|
|
|
+ /// <para><c>'C'</c> → COMPLETED；空串 → NOT_COMPLETED；
|
|
|
+ /// <c>null</c> 与任何其它非空值 → UNKNOWN。</para>
|
|
|
+ ///
|
|
|
+ /// <para>注意这里刻意<b>不</b>写成 <c>IFNULL(status,'') == ""</c>：
|
|
|
+ /// NULL 表示「源侧没给值」，未知值表示「源侧给了我们不认识的值」，
|
|
|
+ /// 两者都不等于「已确认尚未完成」，滑成 NOT_COMPLETED 会让下游把未知当成确定结论。</para>
|
|
|
+ /// </summary>
|
|
|
+ public static string MapCompletionStatus(string? rawStatus)
|
|
|
+ {
|
|
|
+ if (rawStatus is null) return StatusUnknown;
|
|
|
+ var s = rawStatus.Trim();
|
|
|
+ if (s.Length == 0) return StatusNotCompleted;
|
|
|
+ return string.Equals(s, "C", StringComparison.OrdinalIgnoreCase)
|
|
|
+ ? StatusCompleted
|
|
|
+ : StatusUnknown;
|
|
|
+ }
|
|
|
+
|
|
|
+ /// <summary>写入本表时使用的 raw_status 归一：保留 NULL，其余 TRIM。</summary>
|
|
|
+ public static string? NormalizeRawStatus(string? rawStatus) => rawStatus?.Trim();
|
|
|
+
|
|
|
+ /// <summary>
|
|
|
+ /// 全量对账（Full Reconciliation）。
|
|
|
+ /// <paramref name="tenantId"/> 为 0 表示全租户；>0 时只写/只淘汰该租户，绝不触碰其它租户。
|
|
|
+ /// </summary>
|
|
|
+ public async Task<PurchaseOrderCompletionSyncResult> RunFullAsync(
|
|
|
+ long tenantId = 0,
|
|
|
+ string triggerType = "AUTO",
|
|
|
+ CancellationToken cancellationToken = default)
|
|
|
+ {
|
|
|
+ cancellationToken.ThrowIfCancellationRequested();
|
|
|
+
|
|
|
+ var now = DateTime.Now;
|
|
|
+ var batchId = $"S3_PO_COMPL_{(tenantId > 0 ? tenantId + "_" : "")}{now:yyyyMMddHHmmss}";
|
|
|
+ var runLogId = await InsertRunLogAsync(batchId, now, triggerType, tenantId);
|
|
|
+ var result = new PurchaseOrderCompletionSyncResult { BatchId = batchId, RunLogId = runLogId };
|
|
|
+
|
|
|
+ try
|
|
|
+ {
|
|
|
+ // ① 源绑定：非开发环境必须显式绑定真实 WMS 源，绝不静默回落
|
|
|
+ var binding = await ResolveSourceBindingAsync(cancellationToken);
|
|
|
+ result.SourceCode = binding.SourceCode;
|
|
|
+
|
|
|
+ var remote = await _scopeFactory.GetScopeAsync(binding.SourceCode, cancellationToken);
|
|
|
+
|
|
|
+ // 作用域工厂对「本库样板源」与「未配账号的源」会直接返回主库连接。
|
|
|
+ // 那会让本服务把 Source A 的本地表当成 WMS 执行态读进来，结论会整体错误，
|
|
|
+ // 所以这里必须结构性挡住，而不是依赖配置写对。
|
|
|
+ if (ReferenceEquals(remote, _db))
|
|
|
+ throw new InvalidOperationException(
|
|
|
+ $"源 {binding.SourceCode} 解析结果是主库连接，说明它是本库样板源或未配置独立账号，"
|
|
|
+ + "不能承担 S8_PO_COMPLETION_SOURCE 角色");
|
|
|
+
|
|
|
+ // ② 读源：只取本角色范围内的采购行
|
|
|
+ var sourceRows = await ReadSourceRowsAsync(remote, cancellationToken);
|
|
|
+ result.SourceRows = sourceRows.Count;
|
|
|
+
|
|
|
+ // ③ 归属与校验
|
|
|
+ var tenantIndex = await LoadLocalTenantIndexAsync(cancellationToken);
|
|
|
+ var poLineIndex = await LoadLocalPoLineIndexAsync(cancellationToken);
|
|
|
+ var pending = new List<CompletionRow>(sourceRows.Count);
|
|
|
+
|
|
|
+ foreach (var row in sourceRows)
|
|
|
+ {
|
|
|
+ cancellationToken.ThrowIfCancellationRequested();
|
|
|
+
|
|
|
+ if (string.IsNullOrWhiteSpace(row.PoNo) || string.IsNullOrWhiteSpace(row.PoLine))
|
|
|
+ {
|
|
|
+ result.SkippedRows++;
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ var resolved = ResolveTenant(tenantIndex, row.PoNo, row.Domain);
|
|
|
+ if (resolved <= 0)
|
|
|
+ {
|
|
|
+ result.SkippedRows++;
|
|
|
+ result.TenantUnresolvedRows++;
|
|
|
+ _logger.LogWarning(
|
|
|
+ "[PoCompletion] 跳过：采购单 {PurOrd} 行 {Line} 无法唯一解析租户（候选数 {Count}）",
|
|
|
+ row.PoNo, row.PoLine, CountTenantCandidates(tenantIndex, row.PoNo, row.Domain));
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ // 单租户刷新：其它租户的源行本轮不参与，也因此不能被本轮淘汰
|
|
|
+ if (tenantId > 0 && resolved != tenantId)
|
|
|
+ continue;
|
|
|
+
|
|
|
+ var matches = poLineIndex.TryGetValue(LineKey(resolved, row.PoNo, row.PoLine), out var c) ? c : 0;
|
|
|
+ if (matches > 1)
|
|
|
+ throw new InvalidOperationException(
|
|
|
+ $"数据契约破坏：租户 {resolved} 的采购行 {row.PoNo}#{row.PoLine} 在本库命中 {matches} 条，"
|
|
|
+ + "唯一键 uk_po_line 应当保证唯一");
|
|
|
+ if (matches == 0)
|
|
|
+ {
|
|
|
+ result.SkippedRows++;
|
|
|
+ result.UnmatchedRows++;
|
|
|
+ _logger.LogWarning(
|
|
|
+ "[PoCompletion] 跳过：源行 {PurOrd}#{Line} 在本库租户 {Tenant} 下无对应采购行",
|
|
|
+ row.PoNo, row.PoLine, resolved);
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ result.MatchedRows++;
|
|
|
+ var raw = NormalizeRawStatus(row.RawStatus);
|
|
|
+ var completion = MapCompletionStatus(row.RawStatus);
|
|
|
+ if (completion == StatusCompleted) result.CompletedRows++;
|
|
|
+ else if (completion == StatusNotCompleted) result.NotCompletedRows++;
|
|
|
+ else result.UnknownStatusRows++;
|
|
|
+
|
|
|
+ pending.Add(new CompletionRow
|
|
|
+ {
|
|
|
+ TenantId = resolved,
|
|
|
+ PoNo = row.PoNo.Trim(),
|
|
|
+ PoLine = row.PoLine.Trim(),
|
|
|
+ Domain = string.IsNullOrWhiteSpace(row.Domain) ? null : row.Domain.Trim(),
|
|
|
+ Potype = string.IsNullOrWhiteSpace(row.Potype) ? null : row.Potype.Trim(),
|
|
|
+ RawStatus = raw,
|
|
|
+ CompletionStatus = completion,
|
|
|
+ SourceSystem = binding.SourceCode,
|
|
|
+ SourceId = binding.SourceId,
|
|
|
+ SourceRowId = string.IsNullOrWhiteSpace(row.SourceRowId) ? null : row.SourceRowId.Trim(),
|
|
|
+ SourceUpdateTime = row.SourceUpdateTime,
|
|
|
+ SourceUpdateUser = string.IsNullOrWhiteSpace(row.SourceUpdateUser) ? null : row.SourceUpdateUser.Trim(),
|
|
|
+ });
|
|
|
+ }
|
|
|
+
|
|
|
+ // ④ UPSERT：无条件刷新结论，支持 COMPLETED → NOT_COMPLETED 回退
|
|
|
+ result.WrittenRows = await UpsertAsync(pending, batchId, now, cancellationToken);
|
|
|
+
|
|
|
+ // ⑤ 淘汰：源侧已不存在的行回到 NOT_OBSERVED（本表的表达方式就是「没有这一行」）
|
|
|
+ // 只在本轮源侧读取成功后执行；任何异常都会在上面抛出并跳过这里。
|
|
|
+ result.StaleRemoved = await RetireStaleAsync(batchId, tenantId, cancellationToken);
|
|
|
+
|
|
|
+ await CompleteRunLogAsync(runLogId, result, now);
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+ catch (Exception ex)
|
|
|
+ {
|
|
|
+ _logger.LogError(ex, "[PoCompletion] 批次 {BatchId} 失败", batchId);
|
|
|
+ await FailRunLogAsync(runLogId, ex.Message);
|
|
|
+ throw;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ // ── 源绑定与 fail-closed ────────────────────────────────────────────────────────
|
|
|
+
|
|
|
+ /// <summary>
|
|
|
+ /// 解析并校验源绑定。非开发环境下，未配置、配置为空、或仍指向 DEV/UAT 默认源，
|
|
|
+ /// 一律直接失败（fail-closed），不得静默继续。
|
|
|
+ /// </summary>
|
|
|
+ private async Task<SourceBinding> ResolveSourceBindingAsync(CancellationToken ct)
|
|
|
+ {
|
|
|
+ var code = (_opt.SourceCode ?? string.Empty).Trim();
|
|
|
+ var isDev = _env.IsDevelopment();
|
|
|
+
|
|
|
+ if (code.Length == 0)
|
|
|
+ throw new InvalidOperationException(
|
|
|
+ "AiDOP:PoCompletion:SourceCode 未配置，无法确定承担 S8_PO_COMPLETION_SOURCE 角色的数据源");
|
|
|
+
|
|
|
+ if (!isDev && (_opt.DevOnlySourceCodes ?? Array.Empty<string>())
|
|
|
+ .Any(x => string.Equals((x ?? string.Empty).Trim(), code, StringComparison.OrdinalIgnoreCase)))
|
|
|
+ throw new InvalidOperationException(
|
|
|
+ $"当前环境 {_env.EnvironmentName} 非开发环境，禁止使用 DEV/UAT 数据源 {code}；"
|
|
|
+ + "请在 AiDOP:PoCompletion:SourceCode 显式绑定真实 WMS/MES 源");
|
|
|
+
|
|
|
+ var src = await _db.Ado.SqlQuerySingleAsync<SourceBindingRow>(
|
|
|
+ """
|
|
|
+ SELECT id AS Id, source_type AS SourceType, IFNULL(db_user,'') AS DbUser
|
|
|
+ FROM mdp_source
|
|
|
+ WHERE source_code=@Code AND status=1
|
|
|
+ LIMIT 1
|
|
|
+ """,
|
|
|
+ new List<SugarParameter> { new("@Code", code) });
|
|
|
+
|
|
|
+ if (src == null)
|
|
|
+ throw new InvalidOperationException($"mdp_source 未找到启用源：{code}");
|
|
|
+ if (!string.Equals(src.SourceType, "DB", StringComparison.OrdinalIgnoreCase))
|
|
|
+ throw new InvalidOperationException($"源 {code} 的 source_type={src.SourceType}，不是 DB");
|
|
|
+ if (string.IsNullOrWhiteSpace(src.DbUser))
|
|
|
+ throw new InvalidOperationException(
|
|
|
+ $"源 {code} 未配置数据库账号，连接会回落到主库，不能承担 S8_PO_COMPLETION_SOURCE 角色");
|
|
|
+
|
|
|
+ return new SourceBinding { SourceCode = code, SourceId = src.Id };
|
|
|
+ }
|
|
|
+
|
|
|
+ // ── 源侧读取 ────────────────────────────────────────────────────────────────────
|
|
|
+
|
|
|
+ /// <summary>
|
|
|
+ /// 读 Source B 的采购行。范围 = 与自建单推送口径一致的 Potype，
|
|
|
+ /// 不做无差别全表扫，也不顺带拉无关列。
|
|
|
+ /// </summary>
|
|
|
+ private async Task<List<SourceRow>> ReadSourceRowsAsync(ISqlSugarClient remote, CancellationToken ct)
|
|
|
+ {
|
|
|
+ var top = _opt.MaxSourceRows > 0 ? _opt.MaxSourceRows : 50000;
|
|
|
+ var potype = (_opt.SourcePotype ?? string.Empty).Trim();
|
|
|
+
|
|
|
+ var sql =
|
|
|
+ $"""
|
|
|
+ SELECT TOP {top}
|
|
|
+ RecID, Domain, Potype, PurOrd, Line, Status, UpdateTime, UpdateUser
|
|
|
+ FROM PurOrdDetail
|
|
|
+ WHERE (@Potype = '' OR LOWER(LTRIM(RTRIM(ISNULL(Potype,'')))) = LOWER(@Potype))
|
|
|
+ """;
|
|
|
+
|
|
|
+ var table = await remote.Ado.GetDataTableAsync(sql, new List<SugarParameter> { new("@Potype", potype) });
|
|
|
+ var list = new List<SourceRow>(table.Rows.Count);
|
|
|
+ foreach (DataRow r in table.Rows)
|
|
|
+ {
|
|
|
+ ct.ThrowIfCancellationRequested();
|
|
|
+ list.Add(new SourceRow
|
|
|
+ {
|
|
|
+ SourceRowId = Str(r, "RecID"),
|
|
|
+ Domain = Str(r, "Domain"),
|
|
|
+ Potype = Str(r, "Potype"),
|
|
|
+ PoNo = Str(r, "PurOrd"),
|
|
|
+ PoLine = Str(r, "Line"),
|
|
|
+ RawStatus = Str(r, "Status"),
|
|
|
+ SourceUpdateTime = Dt(r, "UpdateTime"),
|
|
|
+ SourceUpdateUser = Str(r, "UpdateUser"),
|
|
|
+ });
|
|
|
+ }
|
|
|
+ return list;
|
|
|
+ }
|
|
|
+
|
|
|
+ // ── 租户归属 ────────────────────────────────────────────────────────────────────
|
|
|
+
|
|
|
+ /// <summary>
|
|
|
+ /// 本地采购单 → 租户候选集。Domain 只用于放宽匹配（自建单本库不填 Domain），
|
|
|
+ /// <b>绝不</b>用 Domain 反推租户 —— 实测同一个 Domain 8010 横跨多个 Ai-DOP 租户。
|
|
|
+ /// </summary>
|
|
|
+ private async Task<Dictionary<string, List<PoTenantCandidate>>> LoadLocalTenantIndexAsync(CancellationToken ct)
|
|
|
+ {
|
|
|
+ var rows = await _db.Ado.SqlQueryAsync<PoTenantCandidate>(
|
|
|
+ """
|
|
|
+ SELECT PurOrd AS PoNo, IFNULL(Domain,'') AS Domain, tenant_id AS TenantId
|
|
|
+ FROM PurOrdDetail
|
|
|
+ WHERE IFNULL(tenant_id,0) > 0
|
|
|
+ GROUP BY PurOrd, IFNULL(Domain,''), tenant_id
|
|
|
+ """);
|
|
|
+
|
|
|
+ var map = new Dictionary<string, List<PoTenantCandidate>>(StringComparer.OrdinalIgnoreCase);
|
|
|
+ foreach (var r in rows)
|
|
|
+ {
|
|
|
+ if (string.IsNullOrWhiteSpace(r.PoNo)) continue;
|
|
|
+ var key = r.PoNo.Trim();
|
|
|
+ if (!map.TryGetValue(key, out var bucket))
|
|
|
+ map[key] = bucket = new List<PoTenantCandidate>();
|
|
|
+ bucket.Add(r);
|
|
|
+ }
|
|
|
+ return map;
|
|
|
+ }
|
|
|
+
|
|
|
+ /// <summary>
|
|
|
+ /// 候选租户集合。Domain 只用于<b>放宽</b>匹配（自建单本库不填 Domain，空 Domain 视为通配），
|
|
|
+ /// <b>绝不</b>反过来用 Domain 推租户 —— 实测同一个 Domain 8010 横跨多个 Ai-DOP 租户。
|
|
|
+ /// </summary>
|
|
|
+ public static IEnumerable<long> TenantCandidates(
|
|
|
+ IEnumerable<PoTenantCandidate> candidates, string poNo, string? domain)
|
|
|
+ {
|
|
|
+ var po = (poNo ?? string.Empty).Trim();
|
|
|
+ var dom = (domain ?? string.Empty).Trim();
|
|
|
+ return candidates
|
|
|
+ .Where(x => string.Equals((x.PoNo ?? string.Empty).Trim(), po, StringComparison.OrdinalIgnoreCase))
|
|
|
+ .Where(x => (x.Domain ?? string.Empty).Trim().Length == 0
|
|
|
+ || string.Equals((x.Domain ?? string.Empty).Trim(), dom, StringComparison.OrdinalIgnoreCase))
|
|
|
+ .Select(x => x.TenantId)
|
|
|
+ .Distinct();
|
|
|
+ }
|
|
|
+
|
|
|
+ /// <summary>
|
|
|
+ /// 唯一才用：0 个（UNRESOLVED）或 >1 个（AMBIGUOUS）一律返回 0，交由调用方跳过。
|
|
|
+ /// <b>禁止</b>退化成「取第一个」—— 那会在多租户重名时把事实写到别人名下。
|
|
|
+ /// </summary>
|
|
|
+ public static long ResolveTenantId(
|
|
|
+ IEnumerable<PoTenantCandidate> candidates, string poNo, string? domain)
|
|
|
+ {
|
|
|
+ var hits = TenantCandidates(candidates, poNo, domain).Take(2).ToList();
|
|
|
+ return hits.Count == 1 ? hits[0] : 0;
|
|
|
+ }
|
|
|
+
|
|
|
+ private static long ResolveTenant(
|
|
|
+ Dictionary<string, List<PoTenantCandidate>> index, string poNo, string? domain)
|
|
|
+ => index.TryGetValue(poNo.Trim(), out var bucket) ? ResolveTenantId(bucket, poNo, domain) : 0;
|
|
|
+
|
|
|
+ private static int CountTenantCandidates(
|
|
|
+ Dictionary<string, List<PoTenantCandidate>> index, string poNo, string? domain)
|
|
|
+ => index.TryGetValue(poNo.Trim(), out var bucket)
|
|
|
+ ? TenantCandidates(bucket, poNo, domain).Count()
|
|
|
+ : 0;
|
|
|
+
|
|
|
+ // ── 本地采购行校验 ──────────────────────────────────────────────────────────────
|
|
|
+
|
|
|
+ private async Task<Dictionary<string, int>> LoadLocalPoLineIndexAsync(CancellationToken ct)
|
|
|
+ {
|
|
|
+ var rows = await _db.Ado.SqlQueryAsync<PoLineKeyRow>(
|
|
|
+ "SELECT tenant_id AS TenantId, po_no AS PoNo, po_line AS PoLine FROM mdp_std_purchase_order");
|
|
|
+
|
|
|
+ var map = new Dictionary<string, int>(StringComparer.OrdinalIgnoreCase);
|
|
|
+ foreach (var r in rows)
|
|
|
+ {
|
|
|
+ var key = LineKey(r.TenantId, r.PoNo ?? string.Empty, r.PoLine ?? string.Empty);
|
|
|
+ map[key] = map.TryGetValue(key, out var n) ? n + 1 : 1;
|
|
|
+ }
|
|
|
+ return map;
|
|
|
+ }
|
|
|
+
|
|
|
+ private static string LineKey(long tenantId, string poNo, string poLine)
|
|
|
+ => $"{tenantId}|{poNo.Trim()}|{poLine.Trim()}";
|
|
|
+
|
|
|
+ // ── 写入 ────────────────────────────────────────────────────────────────────────
|
|
|
+
|
|
|
+ /// <summary>
|
|
|
+ /// 幂等 UPSERT。结论列一律 <c>VALUES(...)</c> 无条件覆盖——这是支持
|
|
|
+ /// COMPLETED → NOT_COMPLETED 回退的前提，绝不能改成「仅当新值为 C 才更新」。
|
|
|
+ /// </summary>
|
|
|
+ private async Task<int> UpsertAsync(
|
|
|
+ List<CompletionRow> rows, string batchId, DateTime now, CancellationToken ct)
|
|
|
+ {
|
|
|
+ if (rows.Count == 0) return 0;
|
|
|
+
|
|
|
+ var size = _opt.UpsertBatchSize > 0 ? _opt.UpsertBatchSize : 200;
|
|
|
+ var written = 0;
|
|
|
+
|
|
|
+ foreach (var chunk in Chunk(rows, size))
|
|
|
+ {
|
|
|
+ ct.ThrowIfCancellationRequested();
|
|
|
+
|
|
|
+ var sb = new StringBuilder();
|
|
|
+ sb.Append("""
|
|
|
+ INSERT INTO mdp_std_purchase_order_completion
|
|
|
+ (tenant_id, factory_id, po_no, po_line, domain, potype,
|
|
|
+ raw_status, completion_status,
|
|
|
+ source_system, source_id, source_row_id, source_update_time, source_update_user,
|
|
|
+ observed_at, sync_batch_id, sync_time)
|
|
|
+ VALUES
|
|
|
+ """);
|
|
|
+
|
|
|
+ var pars = new List<SugarParameter> { new("@Observed", now), new("@Batch", batchId), new("@Now", now) };
|
|
|
+ for (var i = 0; i < chunk.Count; i++)
|
|
|
+ {
|
|
|
+ var r = chunk[i];
|
|
|
+ if (i > 0) sb.Append(',');
|
|
|
+ sb.Append($"(@t{i},1,@p{i},@l{i},@d{i},@y{i},@r{i},@c{i},@s{i},@i{i},@x{i},@u{i},@w{i},@Observed,@Batch,@Now)");
|
|
|
+ pars.Add(new SugarParameter($"@t{i}", r.TenantId));
|
|
|
+ pars.Add(new SugarParameter($"@p{i}", r.PoNo));
|
|
|
+ pars.Add(new SugarParameter($"@l{i}", r.PoLine));
|
|
|
+ pars.Add(new SugarParameter($"@d{i}", (object?)r.Domain ?? DBNull.Value));
|
|
|
+ pars.Add(new SugarParameter($"@y{i}", (object?)r.Potype ?? DBNull.Value));
|
|
|
+ pars.Add(new SugarParameter($"@r{i}", (object?)r.RawStatus ?? DBNull.Value));
|
|
|
+ pars.Add(new SugarParameter($"@c{i}", r.CompletionStatus));
|
|
|
+ pars.Add(new SugarParameter($"@s{i}", r.SourceSystem));
|
|
|
+ pars.Add(new SugarParameter($"@i{i}", (object?)r.SourceId ?? DBNull.Value));
|
|
|
+ pars.Add(new SugarParameter($"@x{i}", (object?)r.SourceRowId ?? DBNull.Value));
|
|
|
+ pars.Add(new SugarParameter($"@u{i}", (object?)r.SourceUpdateTime ?? DBNull.Value));
|
|
|
+ pars.Add(new SugarParameter($"@w{i}", (object?)r.SourceUpdateUser ?? DBNull.Value));
|
|
|
+ }
|
|
|
+
|
|
|
+ sb.Append("""
|
|
|
+ ON DUPLICATE KEY UPDATE
|
|
|
+ factory_id=VALUES(factory_id), domain=VALUES(domain), potype=VALUES(potype),
|
|
|
+ raw_status=VALUES(raw_status), completion_status=VALUES(completion_status),
|
|
|
+ source_system=VALUES(source_system), source_id=VALUES(source_id),
|
|
|
+ source_row_id=VALUES(source_row_id),
|
|
|
+ source_update_time=VALUES(source_update_time),
|
|
|
+ source_update_user=VALUES(source_update_user),
|
|
|
+ observed_at=VALUES(observed_at), sync_batch_id=VALUES(sync_batch_id),
|
|
|
+ sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
|
|
|
+ """);
|
|
|
+
|
|
|
+ await _db.Ado.ExecuteCommandAsync(sb.ToString(), pars);
|
|
|
+ written += chunk.Count;
|
|
|
+ }
|
|
|
+
|
|
|
+ return written;
|
|
|
+ }
|
|
|
+
|
|
|
+ /// <summary>
|
|
|
+ /// 淘汰本轮未覆盖的行 —— 它们在源侧已不存在，应回到 NOT_OBSERVED（即本表无此行）。
|
|
|
+ ///
|
|
|
+ /// <para>只在本方法被调用时执行，而调用点在源侧读取与写入全部成功之后；
|
|
|
+ /// 读取失败会在更早处抛出，因此「源侧抓取失败却把旧事实删掉」不可能发生。</para>
|
|
|
+ ///
|
|
|
+ /// <para>单租户刷新只淘汰该租户 —— 租户 A 的刷新绝不允许删掉租户 B 的事实。</para>
|
|
|
+ /// </summary>
|
|
|
+ private async Task<int> RetireStaleAsync(string batchId, long tenantId, CancellationToken ct)
|
|
|
+ {
|
|
|
+ ct.ThrowIfCancellationRequested();
|
|
|
+
|
|
|
+ if (tenantId > 0)
|
|
|
+ {
|
|
|
+ return await _db.Ado.ExecuteCommandAsync(
|
|
|
+ """
|
|
|
+ DELETE FROM mdp_std_purchase_order_completion
|
|
|
+ WHERE tenant_id=@TenantId AND IFNULL(sync_batch_id,'') <> @BatchId
|
|
|
+ """,
|
|
|
+ new List<SugarParameter> { new("@TenantId", tenantId), new("@BatchId", batchId) });
|
|
|
+ }
|
|
|
+
|
|
|
+ return await _db.Ado.ExecuteCommandAsync(
|
|
|
+ """
|
|
|
+ DELETE FROM mdp_std_purchase_order_completion
|
|
|
+ WHERE tenant_id > 0 AND IFNULL(sync_batch_id,'') <> @BatchId
|
|
|
+ """,
|
|
|
+ new List<SugarParameter> { new("@BatchId", batchId) });
|
|
|
+ }
|
|
|
+
|
|
|
+ // ── run log ─────────────────────────────────────────────────────────────────────
|
|
|
+
|
|
|
+ private async Task<long> InsertRunLogAsync(string batchId, DateTime startedAt, string triggerType, long tenantId)
|
|
|
+ {
|
|
|
+ await _db.Ado.ExecuteCommandAsync(
|
|
|
+ """
|
|
|
+ INSERT INTO mdp_transform_run_log
|
|
|
+ (tenant_id, job_code, job_name, trigger_type, batch_id, status, start_time)
|
|
|
+ VALUES (@TenantId, @JobCode, @JobName, @TriggerType, @BatchId, 'RUNNING', @StartTime)
|
|
|
+ """,
|
|
|
+ new SugarParameter("@TenantId", tenantId),
|
|
|
+ new SugarParameter("@JobCode", JobCode),
|
|
|
+ new SugarParameter("@JobName", JobName),
|
|
|
+ new SugarParameter("@TriggerType", NormalizeTriggerType(triggerType)),
|
|
|
+ new SugarParameter("@BatchId", batchId),
|
|
|
+ new SugarParameter("@StartTime", startedAt));
|
|
|
+
|
|
|
+ return await _db.Ado.GetLongAsync(
|
|
|
+ "SELECT id FROM mdp_transform_run_log WHERE batch_id=@BatchId ORDER BY id DESC LIMIT 1",
|
|
|
+ new List<SugarParameter> { new("@BatchId", batchId) });
|
|
|
+ }
|
|
|
+
|
|
|
+ private async Task CompleteRunLogAsync(long runLogId, PurchaseOrderCompletionSyncResult r, DateTime startedAt)
|
|
|
+ {
|
|
|
+ var endedAt = DateTime.Now;
|
|
|
+
|
|
|
+ // summary_json 是 JSON 列，必须写合法 JSON —— 纯文本会被 MySQL 直接拒绝。
|
|
|
+ // 这里只放计数与源编码，不放任何连接信息。
|
|
|
+ var summary = System.Text.Json.JsonSerializer.Serialize(new
|
|
|
+ {
|
|
|
+ sourceCode = r.SourceCode,
|
|
|
+ sourceRows = r.SourceRows,
|
|
|
+ matchedRows = r.MatchedRows,
|
|
|
+ writtenRows = r.WrittenRows,
|
|
|
+ completedRows = r.CompletedRows,
|
|
|
+ notCompletedRows = r.NotCompletedRows,
|
|
|
+ unknownStatusRows = r.UnknownStatusRows,
|
|
|
+ skippedRows = r.SkippedRows,
|
|
|
+ tenantUnresolvedRows = r.TenantUnresolvedRows,
|
|
|
+ unmatchedRows = r.UnmatchedRows,
|
|
|
+ staleRemoved = r.StaleRemoved,
|
|
|
+ });
|
|
|
+
|
|
|
+ await _db.Ado.ExecuteCommandAsync(
|
|
|
+ """
|
|
|
+ UPDATE mdp_transform_run_log
|
|
|
+ SET status='SUCCESS', stage_rows=@SrcRows, standard_rows=@StdRows, end_time=@EndTime,
|
|
|
+ duration_ms=@Duration, summary_json=@Summary, update_time=CURRENT_TIMESTAMP
|
|
|
+ WHERE id=@Id
|
|
|
+ """,
|
|
|
+ new SugarParameter("@SrcRows", r.SourceRows),
|
|
|
+ new SugarParameter("@StdRows", r.WrittenRows),
|
|
|
+ new SugarParameter("@EndTime", endedAt),
|
|
|
+ new SugarParameter("@Duration", (int)(endedAt - startedAt).TotalMilliseconds),
|
|
|
+ // 刻意不截断：这是固定形状的计数 JSON（长度有界），截断会产生非法 JSON 并让整轮失败
|
|
|
+ new SugarParameter("@Summary", summary),
|
|
|
+ new SugarParameter("@Id", runLogId));
|
|
|
+ }
|
|
|
+
|
|
|
+ private async Task FailRunLogAsync(long runLogId, string message)
|
|
|
+ {
|
|
|
+ await _db.Ado.ExecuteCommandAsync(
|
|
|
+ """
|
|
|
+ UPDATE mdp_transform_run_log
|
|
|
+ SET status='FAILED', end_time=@EndTime, error_message=@Msg, update_time=CURRENT_TIMESTAMP
|
|
|
+ WHERE id=@Id
|
|
|
+ """,
|
|
|
+ new SugarParameter("@EndTime", DateTime.Now),
|
|
|
+ new SugarParameter("@Msg", Truncate(message, 900)),
|
|
|
+ new SugarParameter("@Id", runLogId));
|
|
|
+ }
|
|
|
+
|
|
|
+ // ── 工具 ────────────────────────────────────────────────────────────────────────
|
|
|
+
|
|
|
+ private static string NormalizeTriggerType(string triggerType)
|
|
|
+ => string.IsNullOrWhiteSpace(triggerType) ? "AUTO" : triggerType.Trim().ToUpperInvariant();
|
|
|
+
|
|
|
+ private static string Truncate(string? s, int max)
|
|
|
+ => string.IsNullOrEmpty(s) ? string.Empty : (s.Length <= max ? s : s[..max]);
|
|
|
+
|
|
|
+ private static string? Str(DataRow row, string col)
|
|
|
+ {
|
|
|
+ if (!row.Table.Columns.Contains(col)) return null;
|
|
|
+ var v = row[col];
|
|
|
+ return v == null || v == DBNull.Value ? null : Convert.ToString(v);
|
|
|
+ }
|
|
|
+
|
|
|
+ private static DateTime? Dt(DataRow row, string col)
|
|
|
+ {
|
|
|
+ if (!row.Table.Columns.Contains(col)) return null;
|
|
|
+ var v = row[col];
|
|
|
+ if (v == null || v == DBNull.Value) return null;
|
|
|
+ return Convert.ToDateTime(v);
|
|
|
+ }
|
|
|
+
|
|
|
+ private static IEnumerable<List<T>> Chunk<T>(List<T> source, int size)
|
|
|
+ {
|
|
|
+ for (var i = 0; i < source.Count; i += size)
|
|
|
+ yield return source.GetRange(i, Math.Min(size, source.Count - i));
|
|
|
+ }
|
|
|
+
|
|
|
+ // ── 内部模型 ────────────────────────────────────────────────────────────────────
|
|
|
+
|
|
|
+ private sealed class SourceRow
|
|
|
+ {
|
|
|
+ public string? SourceRowId { get; set; }
|
|
|
+ public string? Domain { get; set; }
|
|
|
+ public string? Potype { get; set; }
|
|
|
+ public string PoNo { get; set; } = string.Empty;
|
|
|
+ public string PoLine { get; set; } = string.Empty;
|
|
|
+ public string? RawStatus { get; set; }
|
|
|
+ public DateTime? SourceUpdateTime { get; set; }
|
|
|
+ public string? SourceUpdateUser { get; set; }
|
|
|
+ }
|
|
|
+
|
|
|
+ private sealed class CompletionRow
|
|
|
+ {
|
|
|
+ public long TenantId { get; set; }
|
|
|
+ public string PoNo { get; set; } = string.Empty;
|
|
|
+ public string PoLine { get; set; } = string.Empty;
|
|
|
+ public string? Domain { get; set; }
|
|
|
+ public string? Potype { get; set; }
|
|
|
+ public string? RawStatus { get; set; }
|
|
|
+ public string CompletionStatus { get; set; } = StatusUnknown;
|
|
|
+ public string SourceSystem { get; set; } = string.Empty;
|
|
|
+ public long? SourceId { get; set; }
|
|
|
+ public string? SourceRowId { get; set; }
|
|
|
+ public DateTime? SourceUpdateTime { get; set; }
|
|
|
+ public string? SourceUpdateUser { get; set; }
|
|
|
+ }
|
|
|
+
|
|
|
+ private sealed class PoLineKeyRow
|
|
|
+ {
|
|
|
+ public long TenantId { get; set; }
|
|
|
+ public string? PoNo { get; set; }
|
|
|
+ public string? PoLine { get; set; }
|
|
|
+ }
|
|
|
+
|
|
|
+ private sealed class SourceBinding
|
|
|
+ {
|
|
|
+ public string SourceCode { get; set; } = string.Empty;
|
|
|
+ public long? SourceId { get; set; }
|
|
|
+ }
|
|
|
+
|
|
|
+ private sealed class SourceBindingRow
|
|
|
+ {
|
|
|
+ public long Id { get; set; }
|
|
|
+ public string? SourceType { get; set; }
|
|
|
+ public string? DbUser { get; set; }
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+/// <summary>
|
|
|
+/// 本地采购单 → 租户候选。<c>Domain</c> 为空表示自建单（本库不填 165 账套），视为通配。
|
|
|
+/// 仅用于把 Source B 的行归到某个 Ai-DOP 租户，<b>不是</b>租户边界本身。
|
|
|
+/// </summary>
|
|
|
+public sealed class PoTenantCandidate
|
|
|
+{
|
|
|
+ public string PoNo { get; set; } = string.Empty;
|
|
|
+ public string Domain { get; set; } = string.Empty;
|
|
|
+ public long TenantId { get; set; }
|
|
|
+}
|
|
|
+
|
|
|
+/// <summary>S3 采购完成态投影结果。</summary>
|
|
|
+public sealed class PurchaseOrderCompletionSyncResult
|
|
|
+{
|
|
|
+ public string BatchId { get; set; } = string.Empty;
|
|
|
+ public long RunLogId { get; set; }
|
|
|
+
|
|
|
+ /// <summary>承担本次角色的数据源编码（不含任何连接信息）。</summary>
|
|
|
+ public string SourceCode { get; set; } = string.Empty;
|
|
|
+
|
|
|
+ /// <summary>源侧读到的行数。</summary>
|
|
|
+ public int SourceRows { get; set; }
|
|
|
+
|
|
|
+ /// <summary>唯一命中本地采购行的行数。</summary>
|
|
|
+ public int MatchedRows { get; set; }
|
|
|
+
|
|
|
+ /// <summary>实际写入（新增或刷新）的行数。</summary>
|
|
|
+ public int WrittenRows { get; set; }
|
|
|
+
|
|
|
+ /// <summary>被跳过的源行数（租户不可解析 + 本库无对应采购行 + 缺关键字段）。</summary>
|
|
|
+ public int SkippedRows { get; set; }
|
|
|
+
|
|
|
+ /// <summary>其中：租户无法唯一解析。</summary>
|
|
|
+ public int TenantUnresolvedRows { get; set; }
|
|
|
+
|
|
|
+ /// <summary>其中：本库无对应采购行。</summary>
|
|
|
+ public int UnmatchedRows { get; set; }
|
|
|
+
|
|
|
+ public int CompletedRows { get; set; }
|
|
|
+ public int NotCompletedRows { get; set; }
|
|
|
+
|
|
|
+ /// <summary>源侧给了 NULL 或无法识别的状态值。</summary>
|
|
|
+ public int UnknownStatusRows { get; set; }
|
|
|
+
|
|
|
+ /// <summary>本轮源侧已不存在、被回收为 NOT_OBSERVED 的行数。</summary>
|
|
|
+ public int StaleRemoved { get; set; }
|
|
|
+}
|
YY968XX