Explorar o código

feat(mdp): S5/S6/S7 列表双源 dopdemorq SQL Server 只读入站与切源 FULL Replace(server 1.0.268)

- 新增 MdpStdFullReplace:切源单事务 DELETE(按tenant)+source过滤INSERT,消除旧源std残留
- Purchase/ProductionReceipt/IpqcInspection 转换服务加可选 sourceSystem 过滤 + RunSourceSwitchFullAsync(PullAll+FullRefresh→FULL Replace,默认参数不改既有本库/API路径行为)
- 1.0.268.sql 登记 DOPDEMORQ_SQLSERVER 源 + 5个 _SQLSERVER 实体,全 status=0(就位不启用)/全 FULL;密码NULL带外注入;source_id按source_code动态查
- 无 Job/Controller 自动调用;SQLSERVER 路径默认 disabled
YY968XX hai 2 meses
pai
achega
2165304140

+ 6 - 3
server/Admin.NET.Web.Entry/Admin.NET.Web.Entry.csproj

@@ -11,9 +11,9 @@
     <GenerateSatelliteAssembliesForCore>true</GenerateSatelliteAssembliesForCore>
     <Copyright>Admin.NET</Copyright>
     <Description>Admin.NET ͨ��Ȩ�޿���ƽ̨</Description>
-    <AssemblyVersion>1.0.267</AssemblyVersion>
-    <FileVersion>1.0.267</FileVersion>
-    <Version>1.0.267</Version>
+    <AssemblyVersion>1.0.268</AssemblyVersion>
+    <FileVersion>1.0.268</FileVersion>
+    <Version>1.0.268</Version>
   </PropertyGroup>
 
   <ItemGroup>
@@ -226,6 +226,9 @@
     <None Update="UpdateScripts\1.0.267.sql">
       <CopyToOutputDirectory>Always</CopyToOutputDirectory>
     </None>
+    <None Update="UpdateScripts\1.0.268.sql">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </None>
   </ItemGroup>
 
   <ItemGroup>

+ 71 - 0
server/Admin.NET.Web.Entry/UpdateScripts/1.0.268.sql

@@ -0,0 +1,71 @@
+-- 1.0.268:S5/S6/S7 列表双源(dopdemorq SQL Server)源/实体登记(默认 disabled)
+--
+-- 用途:注册第二源 DOPDEMORQ_SQLSERVER + S5 采购收货 / S5 生产收货 / S6 IPQC Head 的 _SQLSERVER 入站实体。
+-- 说明:全部 source/entity status=0 就位不启用,合入后不改变现有 MYSQL/API 运行行为;经 P0/Phase1/Phase2 验证。
+--
+-- 硬约束落地:
+--   1) 全部 source / entity status=0(就位不启用):合入后不改变现有 MYSQL/API 运行行为。
+--   2) SQLSERVER 实体一律 sync_mode='FULL':Phase 1 只走 PullAll + fullRefresh=true bootstrap,
+--      规避 MdpDbPullExecutor 严格 `UpdateTime > @cursor` 的跨批同值静默丢行风险(P0 已确认,不改核心)。
+--   3) 密码不入 git:db_password_enc 留 NULL;正式启用前由运维用 CryptogramUtil.Encrypt 密文经带外 UPDATE 注入;
+--      dopsa 为 db_owner(非只读),正式启用前须替换为 SELECT-only 账号(Phase 2 门禁)。
+--   4) source_table_name 与本库版一致(dopdemorq 表名字段名与本库一致;tenant_id 由 DOP 入站上下文赋值,源无此列)。
+-- 幂等:WHERE NOT EXISTS 守卫,可重复执行。
+
+-- ========== (1) 第二源:dopdemorq SQL Server(disabled)==========
+INSERT INTO mdp_source
+  (tenant_id, source_code, source_name, source_type, status,
+   db_type, db_host, db_port, db_name, db_user, db_password_enc, remark, create_time, update_time)
+SELECT 0, 'DOPDEMORQ_SQLSERVER', 'dopdemorq 源库(SQL Server) S5-S7列表第二源', 'DB', 0,
+       'SQLSERVER', '123.60.180.165', 1433, 'dopdemorq', 'dopsa', NULL,
+       'Phase1双源:默认disabled;密码带外注入;正式启用前替换SELECT-only账号', NOW(), NOW()
+WHERE NOT EXISTS (SELECT 1 FROM mdp_source WHERE source_code='DOPDEMORQ_SQLSERVER');
+
+-- ========== (2) S5 采购收货 SQLSERVER 入站实体(disabled, FULL)==========
+INSERT INTO mdp_entity
+  (tenant_id, source_id, entity_code, entity_name, entity_type, source_table_name, target_table_name,
+   sync_mode, batch_size, biz_key_expr, status, remark, create_time, update_time)
+SELECT 0, (SELECT id FROM mdp_source WHERE source_code='DOPDEMORQ_SQLSERVER'),
+       'S5_PURCHASE_RECEIPT_MASTER_SQLSERVER', 'S5采购收货主入站(SQLServer)', 'TABLE',
+       'PurOrdRctMaster', 'mdp_stg_purchase_receipt',
+       'FULL', 5000, 'Domain,Receiver', 0, 'Phase1双源 disabled', NOW(), NOW()
+WHERE NOT EXISTS (SELECT 1 FROM mdp_entity WHERE entity_code='S5_PURCHASE_RECEIPT_MASTER_SQLSERVER');
+
+INSERT INTO mdp_entity
+  (tenant_id, source_id, entity_code, entity_name, entity_type, source_table_name, target_table_name,
+   sync_mode, batch_size, biz_key_expr, status, remark, create_time, update_time)
+SELECT 0, (SELECT id FROM mdp_source WHERE source_code='DOPDEMORQ_SQLSERVER'),
+       'S5_PURCHASE_RECEIPT_DETAIL_SQLSERVER', 'S5采购收货明细入站(SQLServer)', 'TABLE',
+       'PurOrdRctDetail', 'mdp_stg_purchase_receipt',
+       'FULL', 5000, 'Domain,Receiver,Line', 0, 'Phase1双源 disabled(FULL 规避增量边界)', NOW(), NOW()
+WHERE NOT EXISTS (SELECT 1 FROM mdp_entity WHERE entity_code='S5_PURCHASE_RECEIPT_DETAIL_SQLSERVER');
+
+-- ========== (3) S5 生产收货 SQLSERVER 入站实体(沿用 S7_NBR_* 命名,disabled, FULL)==========
+INSERT INTO mdp_entity
+  (tenant_id, source_id, entity_code, entity_name, entity_type, source_table_name, target_table_name,
+   sync_mode, batch_size, biz_key_expr, status, remark, create_time, update_time)
+SELECT 0, (SELECT id FROM mdp_source WHERE source_code='DOPDEMORQ_SQLSERVER'),
+       'S7_NBR_MASTER_SQLSERVER', 'S5生产入库主入站(SQLServer)', 'TABLE',
+       'NbrMaster', 'mdp_stg_production_receipt',
+       'FULL', 5000, 'Domain,Nbr', 0, 'Phase1双源 disabled(FULL 规避增量边界)', NOW(), NOW()
+WHERE NOT EXISTS (SELECT 1 FROM mdp_entity WHERE entity_code='S7_NBR_MASTER_SQLSERVER');
+
+INSERT INTO mdp_entity
+  (tenant_id, source_id, entity_code, entity_name, entity_type, source_table_name, target_table_name,
+   sync_mode, batch_size, biz_key_expr, status, remark, create_time, update_time)
+SELECT 0, (SELECT id FROM mdp_source WHERE source_code='DOPDEMORQ_SQLSERVER'),
+       'S7_NBR_DETAIL_SQLSERVER', 'S5生产入库明细入站(SQLServer)', 'TABLE',
+       'NbrDetail', 'mdp_stg_production_receipt',
+       'FULL', 5000, 'Domain,Nbr,Line', 0, 'Phase1双源 disabled', NOW(), NOW()
+WHERE NOT EXISTS (SELECT 1 FROM mdp_entity WHERE entity_code='S7_NBR_DETAIL_SQLSERVER');
+
+-- ========== (4) S6 IPQC Head SQLSERVER 入站实体(disabled, FULL;qms_gcjyd 无 UpdateTime)==========
+-- 注:仅 Head。IPQC Detail(qms_gcjydzb) 属 bespoke,Phase 2 处理,本文件不登记其实体。
+INSERT INTO mdp_entity
+  (tenant_id, source_id, entity_code, entity_name, entity_type, source_table_name, target_table_name,
+   sync_mode, batch_size, biz_key_expr, status, remark, create_time, update_time)
+SELECT 0, (SELECT id FROM mdp_source WHERE source_code='DOPDEMORQ_SQLSERVER'),
+       'S6_IPQC_INSPECTION_HEAD_SQLSERVER', 'S6过程检验头入站(SQLServer)', 'TABLE',
+       'qms_gcjyd', 'mdp_stg_ipqc_pull',
+       'FULL', 1000, 'djbh', 0, 'Phase1双源 disabled(仅Head)', NOW(), NOW()
+WHERE NOT EXISTS (SELECT 1 FROM mdp_entity WHERE entity_code='S6_IPQC_INSPECTION_HEAD_SQLSERVER');

+ 67 - 0
server/Plugins/Admin.NET.Plugin.AiDOP/DataPlatform/MdpStdFullReplace.cs

@@ -0,0 +1,67 @@
+using System.Text.RegularExpressions;
+using SqlSugar;
+
+namespace Admin.NET.Plugin.AiDOP.DataPlatform;
+
+/// <summary>
+/// 标准层「当前 active 源 FULL Replace」局部工具(S5/S6/S7 列表双源 Phase 1 用)。
+///
+/// 用途:切源/FULL 刷新时,保证 std = 当前 active source 的完整结果集,消除旧源独有业务键残留。
+/// 语义:<b>先由调用方 PullAll 成功灌 stg,再</b>在单事务内 DELETE 当前 tenant(+可选业务条件) 的 std,
+///       随后执行「仅当前 source_system」的 INSERT(由调用方以委托传入其对象专属 transform SQL)。
+///
+/// 安全边界:
+///   - 仅本工具删除 std;不 TRUNCATE 整表、不跨 tenant、不跨业务类型(extraWhere 由调用方精确限定)。
+///   - DELETE + INSERT 同一本库事务;任一失败回滚,旧 std 保持不变。
+///   - 必须在 Pull 成功之后调用;Pull 失败应让异常在调用方先抛出,绝不先删 std。
+///   - 非数据平台核心(不改 Factory/Executor/Writer),仅一层薄事务封装。
+/// </summary>
+internal static class MdpStdFullReplace
+{
+    // ⚠️ 适用边界(Phase 2 防御性复核确认):本工具的 DELETE 仅按 tenant_id(+可选 extraWhere) 限定,
+    //    因此**仅适用于 tenant 独占的业务专表 std**(如 mdp_std_purchase_receipt / _production_receipt / _ipqc_inspection,
+    //    实测各仅其转换服务写入)。若某 std 与其它业务共享,必须由调用方传入精确 extraWhere 业务分区条件,否则会误删他业务行。
+    private static readonly Regex IdentifierRe = new(@"^[A-Za-z0-9_]+$", RegexOptions.Compiled);
+
+    /// <summary>
+    /// 单事务内 FULL Replace 当前 tenant 的 std:先 DELETE,再执行 source 过滤的 INSERT 委托。
+    /// </summary>
+    /// <param name="db">调用方自身注入的 ISqlSugarClient(同一 scope,事务覆盖委托内的写入)。</param>
+    /// <param name="stdTable">标准层表名(标识符白名单校验)。</param>
+    /// <param name="tenantId">精确隔离的租户 id。</param>
+    /// <param name="extraWhere">可选附加删除条件(如业务类型列),已由调用方保证安全,不含用户输入。</param>
+    /// <param name="insertScopedAsync">仅写入当前 source_system 结果的 INSERT 委托,返回写入行数。</param>
+    public static async Task<int> ReplaceAsync(
+        ISqlSugarClient db,
+        string stdTable,
+        long tenantId,
+        string? extraWhere,
+        Func<Task<int>> insertScopedAsync,
+        CancellationToken cancellationToken = default)
+    {
+        if (string.IsNullOrWhiteSpace(stdTable) || !IdentifierRe.IsMatch(stdTable))
+            throw new InvalidOperationException($"非法 std 表名:{stdTable}");
+
+        cancellationToken.ThrowIfCancellationRequested();
+
+        var where = string.IsNullOrWhiteSpace(extraWhere)
+            ? "tenant_id=@__tenant"
+            : $"tenant_id=@__tenant AND {extraWhere}";
+
+        await db.Ado.BeginTranAsync();
+        try
+        {
+            await db.Ado.ExecuteCommandAsync(
+                $"DELETE FROM `{stdTable}` WHERE {where}",
+                new SugarParameter("@__tenant", tenantId));
+            var rows = await insertScopedAsync();
+            await db.Ado.CommitTranAsync();
+            return rows;
+        }
+        catch
+        {
+            await db.Ado.RollbackTranAsync();
+            throw;
+        }
+    }
+}

+ 89 - 8
server/Plugins/Admin.NET.Plugin.AiDOP/Manufacturing/IpqcInspectionMdpSyncService.cs

@@ -1,3 +1,4 @@
+using Admin.NET.Plugin.AiDOP.DataPlatform;
 using Admin.NET.Plugin.AiDOP.DataPlatform.Executors;
 
 namespace Admin.NET.Plugin.AiDOP.Manufacturing;
@@ -29,6 +30,10 @@ public class IpqcInspectionMdpSyncService : ITransient
     private const string JobCode = "S6_IPQC_INSPECTION_MDP_SYNC";
     private const string InboundEntityCode = "S6_IPQC_INSPECTION_HEAD";
 
+    // 双源(dopdemorq SQL Server)Phase 1:仅 Head(qms_gcjyd 无 UpdateTime → FULL)。实体默认 status=0(就位不启用)。Detail(qms_gcjydzb) 属 bespoke,留 Phase 2 不碰。
+    private const string SqlServerSourceCode = "DOPDEMORQ_SQLSERVER";
+    private const string SqlServerHeadEntityCode = "S6_IPQC_INSPECTION_HEAD_SQLSERVER";
+
     private readonly ISqlSugarClient _db;
     private readonly MdpSourcePullDispatcher _pullDispatcher;
 
@@ -132,6 +137,67 @@ public class IpqcInspectionMdpSyncService : ITransient
         };
     }
 
+    /// <summary>
+    /// 双源切换 FULL Replace(Phase 1,仅 Head):从指定 DB 源(默认 DOPDEMORQ_SQLSERVER)PullAll 全量灌 pull-stg,
+    /// 成功后在单事务内以「仅当前源」结果 FULL 重建 mdp_std_ipqc_inspection(按 tenant 精确隔离),消除旧源独有业务键残留。
+    /// qms_gcjyd 无 UpdateTime → FULL;用 PullAllByEntityCodeAsync + FullRefresh=true。
+    /// <b>只处理 Head</b>:不调用 SyncHeadStagingAsync(vestigial 观测写入)/ SyncDetailStagingAsync / TransformDetailStandardAsync;
+    /// IPQC Detail(qms_gcjydzb) 属 bespoke,留 Phase 2。
+    /// SQLSERVER 实体默认 status=0(就位不启用);dopdemorq 该表当前空,Phase 1 不实际执行 destructive 切换。
+    /// </summary>
+    public async Task<IpqcInspectionInboundResult> RunHeadSourceSwitchFullAsync(
+        string sourceCode = SqlServerSourceCode,
+        string headEntityCode = SqlServerHeadEntityCode,
+        long tenantId = 0,
+        CancellationToken cancellationToken = default)
+    {
+        cancellationToken.ThrowIfCancellationRequested();
+        await EnsureTablesAsync();
+        await EnsurePullStgTableAsync();
+
+        var now = DateTime.Now;
+        // 1) PullAll 抽尽 + fullRefresh=true。Pull 失败会抛异常,std 未动。
+        var pullCtx = new MdpPullContext
+        {
+            TenantId = tenantId,
+            FullRefresh = true,
+            TaskCode = "S6_IPQC_INSPECTION_INBOUND",
+            BatchId = $"S6_IPQC_SW_{now:yyyyMMddHHmmss}"
+        };
+        var pull = await _pullDispatcher.PullAllByEntityCodeAsync(headEntityCode, pullCtx, cancellationToken);
+
+        // 2) FULL Replace(仅头 std):事务内 DELETE 当前 tenant,再 INSERT 仅当前 source_system 的头结果。
+        var batchId = $"S6_IPQC_SWSTD_{now:yyyyMMddHHmmss}";
+        var runLogId = await InsertRunLogAsync(batchId, now, "SOURCE_SWITCH");
+        var result = new IpqcInspectionMdpSyncResult { BatchId = batchId, RunLogId = runLogId, HeadStageRows = pull.RowsWritten };
+        try
+        {
+            result.HeadStandardRows = await MdpStdFullReplace.ReplaceAsync(
+                _db, "mdp_std_ipqc_inspection", tenantId, extraWhere: null,
+                insertScopedAsync: () => TransformHeadStandardAsync(batchId, now, sourceCode),
+                cancellationToken);
+            await MarkRunSuccessAsync(runLogId, now, result);
+        }
+        catch (Exception ex)
+        {
+            await MarkRunFailedAsync(runLogId, now, ex.Message);
+            throw;
+        }
+
+        return new IpqcInspectionInboundResult
+        {
+            PullBatchId = pullCtx.BatchId,
+            RowsPulled = pull.RowsPulled,
+            RowsWrittenStg = pull.RowsWritten,
+            NewCursor = pull.NewCursor,
+            TransformBatchId = batchId,
+            HeadStageRows = result.HeadStageRows,
+            HeadStandardRows = result.HeadStandardRows,
+            DetailStageRows = 0,
+            DetailStandardRows = 0
+        };
+    }
+
     private async Task EnsurePullStgTableAsync()
     {
         await _db.Ado.ExecuteCommandAsync(
@@ -378,12 +444,22 @@ 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)
+    private async Task<int> TransformHeadStandardAsync(string batchId, DateTime now, string? sourceSystem = null)
     {
+        // 双源:sourceSystem 非空时仅统计/转换当前 source(切源 FULL Replace 用);为空时保持既有全源行为不变。
+        var countPars = new List<SugarParameter>();
+        var countSrc = "";
+        var insertSrc = "";
+        if (sourceSystem != null)
+        {
+            countSrc = " AND source_system=@Src";
+            insertSrc = " AND m.source_system=@Src";
+            countPars.Add(new SugarParameter("@Src", sourceSystem));
+        }
         var rows = await _db.Ado.GetIntAsync(
-            "SELECT COUNT(1) FROM mdp_stg_ipqc_pull WHERE source_table='qms_gcjyd'");
-        await _db.Ado.ExecuteCommandAsync(
-            """
+            $"SELECT COUNT(1) FROM mdp_stg_ipqc_pull WHERE source_table='qms_gcjyd'{countSrc}", countPars);
+        var insertSql =
+            $"""
             INSERT INTO mdp_std_ipqc_inspection
             (tenant_id, factory_id, source_system, bill_no, product_model, production_batch_no, production_work_order,
              result_judgement, attachment, remark, inspector, process_code, process_name, production_person,
@@ -416,7 +492,7 @@ public class IpqcInspectionMdpSyncService : ITransient
                 IFNULL(NULLIF(m.source_biz_key,''), JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.djbh'))),
                 @BatchId, @Now
             FROM mdp_stg_ipqc_pull m
-            WHERE m.source_table='qms_gcjyd'
+            WHERE m.source_table='qms_gcjyd'{insertSrc}
             ON DUPLICATE KEY UPDATE
                 bill_no=VALUES(bill_no), product_model=VALUES(product_model), production_batch_no=VALUES(production_batch_no),
                 production_work_order=VALUES(production_work_order), result_judgement=VALUES(result_judgement),
@@ -427,9 +503,14 @@ public class IpqcInspectionMdpSyncService : ITransient
                 inspec_standard_version=VALUES(inspec_standard_version), inspec_standard_code=VALUES(inspec_standard_code),
                 inspection_status=VALUES(inspection_status),
                 sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
-            """,
-            new SugarParameter("@BatchId", batchId),
-            new SugarParameter("@Now", now));
+            """;
+        var insPars = new List<SugarParameter>
+        {
+            new("@BatchId", batchId),
+            new("@Now", now)
+        };
+        if (sourceSystem != null) insPars.Add(new SugarParameter("@Src", sourceSystem));
+        await _db.Ado.ExecuteCommandAsync(insertSql, insPars);
         return rows;
     }
 

+ 83 - 10
server/Plugins/Admin.NET.Plugin.AiDOP/MaterialWarehouse/ProductionReceiptMdpSyncService.cs

@@ -1,3 +1,4 @@
+using Admin.NET.Plugin.AiDOP.DataPlatform;
 using Admin.NET.Plugin.AiDOP.DataPlatform.Executors;
 
 namespace Admin.NET.Plugin.AiDOP.MaterialWarehouse;
@@ -23,6 +24,11 @@ public class ProductionReceiptMdpSyncService : ITransient
     private const string InboundEntityCode = "S7_NBR_MASTER";
     private const string InboundDetailEntityCode = "S7_NBR_DETAIL";
 
+    // 双源(dopdemorq SQL Server)Phase 1:沿用现有 S7_NBR_* 命名语义(业务=生产入库),加 _SQLSERVER 第二源实体。实体默认 status=0(就位不启用)。
+    private const string SqlServerSourceCode = "DOPDEMORQ_SQLSERVER";
+    private const string SqlServerMasterEntityCode = "S7_NBR_MASTER_SQLSERVER";
+    private const string SqlServerDetailEntityCode = "S7_NBR_DETAIL_SQLSERVER";
+
     private readonly ISqlSugarClient _db;
     private readonly MdpSourcePullDispatcher _pullDispatcher;
 
@@ -112,6 +118,64 @@ public class ProductionReceiptMdpSyncService : ITransient
         };
     }
 
+    /// <summary>
+    /// 双源切换 FULL Replace(Phase 1):从指定 DB 源(默认 DOPDEMORQ_SQLSERVER)PullAll 全量灌 stg,
+    /// 成功后在单事务内以「仅当前源」结果 FULL 重建 mdp_std_production_receipt(按 tenant 精确隔离),消除旧源独有业务键残留。
+    /// 用 PullAllByEntityCodeAsync 抽尽 + FullRefresh=true;transform 仅当前 source_system;std DELETE+INSERT 同事务,Pull 成功后才进入。
+    /// SQLSERVER 实体默认 status=0(就位不启用);dopdemorq 六表当前空,Phase 1 不实际执行 destructive 切换。
+    /// </summary>
+    public async Task<ProductionReceiptInboundResult> RunSourceSwitchFullAsync(
+        string sourceCode = SqlServerSourceCode,
+        string masterEntityCode = SqlServerMasterEntityCode,
+        string detailEntityCode = SqlServerDetailEntityCode,
+        long tenantId = 0,
+        CancellationToken cancellationToken = default)
+    {
+        cancellationToken.ThrowIfCancellationRequested();
+        await EnsureTablesAsync();
+        await EnsureStgTableAsync();
+
+        var now = DateTime.Now;
+        // 1) PullAll 抽尽 + fullRefresh=true。任一 Pull 失败会抛异常,std 未动。
+        var pullCtx = new MdpPullContext
+        {
+            TenantId = tenantId,
+            FullRefresh = true,
+            TaskCode = "S7_PRODUCTION_RECEIPT_INBOUND",
+            BatchId = $"S7_NBR_SW_{now:yyyyMMddHHmmss}"
+        };
+        var master = await _pullDispatcher.PullAllByEntityCodeAsync(masterEntityCode, pullCtx, cancellationToken);
+        var detail = await _pullDispatcher.PullAllByEntityCodeAsync(detailEntityCode, pullCtx, cancellationToken);
+
+        // 2) FULL Replace:事务内 DELETE 当前 tenant 的 std,再 INSERT 仅当前 source_system 的结果。
+        var batchId = $"S7_NBR_SWSTD_{now:yyyyMMddHHmmss}";
+        var runLogId = await InsertRunLogAsync(batchId, now, "SOURCE_SWITCH");
+        var result = new ProductionReceiptMdpSyncResult { BatchId = batchId, RunLogId = runLogId };
+        try
+        {
+            result.Rows = await MdpStdFullReplace.ReplaceAsync(
+                _db, "mdp_std_production_receipt", tenantId, extraWhere: null,
+                insertScopedAsync: () => TransformStandardAsync(batchId, now, sourceCode),
+                cancellationToken);
+            await MarkRunSuccessAsync(runLogId, now, result);
+        }
+        catch (Exception ex)
+        {
+            await MarkRunFailedAsync(runLogId, now, ex.Message);
+            throw;
+        }
+
+        return new ProductionReceiptInboundResult
+        {
+            PullBatchId = pullCtx.BatchId,
+            RowsPulled = master.RowsPulled + detail.RowsPulled,
+            RowsWrittenStg = master.RowsWritten + detail.RowsWritten,
+            NewCursor = detail.NewCursor ?? master.NewCursor,
+            TransformBatchId = batchId,
+            StdRows = result.Rows
+        };
+    }
+
     private async Task<(int RowsPulled, int RowsWritten, string? NewCursor)> PopulateStgAsync(
         MdpPullContext pullCtx, CancellationToken cancellationToken)
     {
@@ -206,10 +270,14 @@ public class ProductionReceiptMdpSyncService : ITransient
     /// <summary>
     /// 标准化:stg(NbrMaster/NbrDetail) + 维表(本库) → mdp_std_production_receipt。
     /// </summary>
-    private async Task<int> TransformStandardAsync(string batchId, DateTime now)
+    private async Task<int> TransformStandardAsync(string batchId, DateTime now, string? sourceSystem = null)
     {
+        // 双源:sourceSystem 非空时仅统计/转换当前 source(切源 FULL Replace 用);为空时保持既有全源行为不变。
+        var srcClause = sourceSystem == null ? "" : " AND n.source_system=@Src AND d.source_system=@Src";
+        var countPars = new List<SugarParameter>();
+        if (sourceSystem != null) countPars.Add(new SugarParameter("@Src", sourceSystem));
         var rows = await _db.Ado.GetIntAsync(
-            """
+            $"""
             SELECT COUNT(1)
             FROM mdp_stg_production_receipt n
             INNER JOIN mdp_stg_production_receipt d
@@ -219,10 +287,10 @@ public class ProductionReceiptMdpSyncService : ITransient
             WHERE n.source_table='NbrMaster'
               AND JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.Type'))='WOI'
               AND CAST(JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.IsActive')) AS SIGNED)=1
-              AND JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.RecID')) IS NOT NULL
-            """);
-        await _db.Ado.ExecuteCommandAsync(
-            """
+              AND JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.RecID')) IS NOT NULL{srcClause}
+            """, countPars);
+        var insertSql =
+            $"""
             INSERT INTO mdp_std_production_receipt
             (tenant_id, factory_id, source_system, domain, master_rec_id, detail_rec_id, nbr, line,
              receipt_date, status, status_desc, remark, prod_line, work_ord, erp_work_ord,
@@ -276,7 +344,7 @@ public class ProductionReceiptMdpSyncService : ITransient
             WHERE n.source_table='NbrMaster'
               AND JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.Type'))='WOI'
               AND CAST(JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.IsActive')) AS SIGNED)=1
-              AND JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.RecID')) IS NOT NULL
+              AND JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.RecID')) IS NOT NULL{srcClause}
             ON DUPLICATE KEY UPDATE
                 factory_id=VALUES(factory_id), master_rec_id=VALUES(master_rec_id), nbr=VALUES(nbr), line=VALUES(line),
                 receipt_date=VALUES(receipt_date), status=VALUES(status), status_desc=VALUES(status_desc), remark=VALUES(remark),
@@ -287,9 +355,14 @@ public class ProductionReceiptMdpSyncService : ITransient
                 qty_rec=VALUES(qty_rec), qty_to=VALUES(qty_to), location_from=VALUES(location_from),
                 location_from_desc=VALUES(location_from_desc), ord_nbr=VALUES(ord_nbr),
                 sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
-            """,
-            new SugarParameter("@BatchId", batchId),
-            new SugarParameter("@Now", now));
+            """;
+        var insPars = new List<SugarParameter>
+        {
+            new("@BatchId", batchId),
+            new("@Now", now)
+        };
+        if (sourceSystem != null) insPars.Add(new SugarParameter("@Src", sourceSystem));
+        await _db.Ado.ExecuteCommandAsync(insertSql, insPars);
         return rows;
     }
 

+ 88 - 10
server/Plugins/Admin.NET.Plugin.AiDOP/MaterialWarehouse/PurchaseReceiptMdpSyncService.cs

@@ -1,3 +1,4 @@
+using Admin.NET.Plugin.AiDOP.DataPlatform;
 using Admin.NET.Plugin.AiDOP.DataPlatform.Executors;
 
 namespace Admin.NET.Plugin.AiDOP.MaterialWarehouse;
@@ -23,6 +24,11 @@ public class PurchaseReceiptMdpSyncService : ITransient
     private const string InboundEntityCode = "S5_PURCHASE_RECEIPT_DETAIL";
     private const string InboundMasterEntityCode = "S5_PURCHASE_RECEIPT_MASTER";
 
+    // 双源(dopdemorq SQL Server)Phase 1:默认第二源与其入站实体。实体默认 status=0,未启用前 RunSourceSwitchFullAsync 会因找不到启用实体而抛错(就位不启用)。
+    private const string SqlServerSourceCode = "DOPDEMORQ_SQLSERVER";
+    private const string SqlServerMasterEntityCode = "S5_PURCHASE_RECEIPT_MASTER_SQLSERVER";
+    private const string SqlServerDetailEntityCode = "S5_PURCHASE_RECEIPT_DETAIL_SQLSERVER";
+
     private readonly ISqlSugarClient _db;
     private readonly MdpSourcePullDispatcher _pullDispatcher;
 
@@ -113,6 +119,69 @@ public class PurchaseReceiptMdpSyncService : ITransient
         };
     }
 
+    /// <summary>
+    /// 双源切换 FULL Replace(Phase 1):从指定 DB 源(默认 DOPDEMORQ_SQLSERVER)PullAll 全量灌 stg,
+    /// 成功后在单事务内以「仅当前源」结果 FULL 重建 mdp_std_purchase_receipt(按 tenant 精确隔离),消除旧源独有业务键残留。
+    ///
+    /// 与既有 RunFull/RunInbound 的区别:① 用 PullAllByEntityCodeAsync 抽尽(非单批 PullByEntityCode);
+    /// ② FullRefresh=true 显式全量 bootstrap;③ transform 仅当前 source_system;④ std DELETE+INSERT 同事务,Pull 成功后才进入。
+    ///
+    /// 切源语义为「部署级单活源」。SQLSERVER 实体默认 status=0,未启用前本方法运行时会因 PullByEntityCode 找不到启用实体抛错——符合 Phase 1「就位不启用」。
+    /// dopdemorq 当前六表为空:本方法 FULL Replace 的理论结果是把当前 tenant 对应 std 置空;Phase 1 不实际执行 destructive 切换。
+    /// </summary>
+    public async Task<PurchaseReceiptInboundResult> RunSourceSwitchFullAsync(
+        string sourceCode = SqlServerSourceCode,
+        string masterEntityCode = SqlServerMasterEntityCode,
+        string detailEntityCode = SqlServerDetailEntityCode,
+        long tenantId = 0,
+        CancellationToken cancellationToken = default)
+    {
+        cancellationToken.ThrowIfCancellationRequested();
+        await EnsureTablesAsync();
+        await EnsureStgTableAsync();
+
+        var now = DateTime.Now;
+        // 1) PullAll 抽尽 + fullRefresh=true(不用单批 PullByEntityCode)。任一 Pull 失败会抛异常,std 未动。
+        var pullCtx = new MdpPullContext
+        {
+            TenantId = tenantId,
+            FullRefresh = true,
+            TaskCode = "S5_PURCHASE_RECEIPT_INBOUND",
+            BatchId = $"S5_PUR_RCT_SW_{now:yyyyMMddHHmmss}"
+        };
+        var master = await _pullDispatcher.PullAllByEntityCodeAsync(masterEntityCode, pullCtx, cancellationToken);
+        var detail = await _pullDispatcher.PullAllByEntityCodeAsync(detailEntityCode, pullCtx, cancellationToken);
+
+        // 2) FULL Replace:事务内 DELETE 当前 tenant 的 std,再 INSERT 仅当前 source_system 的结果。
+        var batchId = $"S5_PUR_RCT_SWSTD_{now:yyyyMMddHHmmss}";
+        var runLogId = await InsertRunLogAsync(batchId, now, "SOURCE_SWITCH");
+        var result = new PurchaseReceiptMdpSyncResult { BatchId = batchId, RunLogId = runLogId };
+        try
+        {
+            result.StdRows = await MdpStdFullReplace.ReplaceAsync(
+                _db, "mdp_std_purchase_receipt", tenantId, extraWhere: null,
+                insertScopedAsync: () => TransformStandardAsync(batchId, now, sourceCode),
+                cancellationToken);
+            await MarkRunSuccessAsync(runLogId, now, result);
+        }
+        catch (Exception ex)
+        {
+            await MarkRunFailedAsync(runLogId, now, ex.Message);
+            throw;
+        }
+
+        return new PurchaseReceiptInboundResult
+        {
+            PullBatchId = pullCtx.BatchId,
+            RowsPulled = master.RowsPulled + detail.RowsPulled,
+            RowsWrittenStg = master.RowsWritten + detail.RowsWritten,
+            NewCursor = detail.NewCursor ?? master.NewCursor,
+            PullMessage = $"{master.Message}; {detail.Message}",
+            TransformBatchId = batchId,
+            StdRows = result.StdRows
+        };
+    }
+
     private async Task<(int RowsPulled, int RowsWritten, string? NewCursor, string? Message)> PopulateStgAsync(
         MdpPullContext pullCtx, CancellationToken cancellationToken)
     {
@@ -205,10 +274,14 @@ public class PurchaseReceiptMdpSyncService : ITransient
     /// <summary>
     /// 标准化:stg(PurOrdRctDetail/Master) + 维表(本库) → mdp_std_purchase_receipt。
     /// </summary>
-    private async Task<int> TransformStandardAsync(string batchId, DateTime now)
+    private async Task<int> TransformStandardAsync(string batchId, DateTime now, string? sourceSystem = null)
     {
+        // 双源:sourceSystem 非空时仅统计/转换当前 source(切源 FULL Replace 用);为空时保持既有全源行为不变。
+        var srcClause = sourceSystem == null ? "" : " AND p.source_system=@Src AND d.source_system=@Src";
+        var countPars = new List<SugarParameter>();
+        if (sourceSystem != null) countPars.Add(new SugarParameter("@Src", sourceSystem));
         var rows = await _db.Ado.GetIntAsync(
-            """
+            $"""
             SELECT COUNT(1)
             FROM mdp_stg_purchase_receipt p
             INNER JOIN mdp_stg_purchase_receipt d
@@ -216,10 +289,10 @@ public class PurchaseReceiptMdpSyncService : ITransient
              AND JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.Domain')) = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Domain'))
              AND JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.Receiver')) = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Receiver'))
             WHERE p.source_table='PurOrdRctDetail'
-              AND JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.RctType'))='rc'
-            """);
-        await _db.Ado.ExecuteCommandAsync(
-            """
+              AND JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.RctType'))='rc'{srcClause}
+            """, countPars);
+        var insertSql =
+            $"""
             INSERT INTO mdp_std_purchase_receipt
             (tenant_id, factory_id, source_system, domain, receiver, line, rct_date, supp, sort_name,
              item_num, item_name, item_spec, um, qty_ordered, qty_received, lot_serial, location,
@@ -287,7 +360,7 @@ public class PurchaseReceiptMdpSyncService : ITransient
              AND dr.SAP_pr_billno = sd.Req
              AND IFNULL(dr.SAP_pr_billno,'')<>''
             WHERE p.source_table='PurOrdRctDetail'
-              AND JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.RctType'))='rc'
+              AND JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.RctType'))='rc'{srcClause}
             ON DUPLICATE KEY UPDATE
                 factory_id=VALUES(factory_id), rct_date=VALUES(rct_date), supp=VALUES(supp), sort_name=VALUES(sort_name),
                 item_num=VALUES(item_num), item_name=VALUES(item_name), item_spec=VALUES(item_spec), um=VALUES(um),
@@ -296,9 +369,14 @@ public class PurchaseReceiptMdpSyncService : ITransient
                 pur_ord=VALUES(pur_ord), pur_line=VALUES(pur_line), sales_job=VALUES(sales_job), address1=VALUES(address1),
                 req=VALUES(req), req_line=VALUES(req_line), dop_req=VALUES(dop_req),
                 sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
-            """,
-            new SugarParameter("@BatchId", batchId),
-            new SugarParameter("@Now", now));
+            """;
+        var insPars = new List<SugarParameter>
+        {
+            new("@BatchId", batchId),
+            new("@Now", now)
+        };
+        if (sourceSystem != null) insPars.Add(new SugarParameter("@Src", sourceSystem));
+        await _db.Ado.ExecuteCommandAsync(insertSql, insPars);
         return rows;
     }