feat(mdp): S5生产领料 bespoke→B-①通用管线 + dopdemorq 双源只读入站（server 1.0.270）

- ProductionIssueMdpSyncService 由 bespoke 直读 NbrMaster 迁通用管线：源→mdp_stg_production_issue→transform→mdp_std_production_issue；加 RunInbound/RunSourceSwitchFullAsync（PullAll+FullRefresh→tenant scoped FULL Replace，决策1A/2A 单active源、read-only、无回写）
- 新增 MdpJsonSql：stg→std 转换层跨源类型兼容 SQL 表达式生成器（datetime ISO-T/bit true-false/JSON null 归一），不改 MDP 核心
- transform 从 raw_data 保留源 tenant_id（本库 std 租户不回归）；DepartmentMaster 本库 LEFT JOIN（mode-A 同构）
- 1.0.270.sql 登记 S5_PRODUCTION_ISSUE_MASTER(本库 status=1，RunFull 迁 stg 依赖) + _SQLSERVER(status=0 就位不启用)，均 FULL
- 真实 dopdemorq 数据验证 源→stg→std→API→页面全通过；本库等价零回归

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.269</AssemblyVersion>
-    <FileVersion>1.0.269</FileVersion>
-    <Version>1.0.269</Version>
+    <AssemblyVersion>1.0.270</AssemblyVersion>
+    <FileVersion>1.0.270</FileVersion>
+    <Version>1.0.270</Version>
   </PropertyGroup>
 
   <ItemGroup>
@@ -229,6 +229,9 @@
     <None Update="UpdateScripts\1.0.268.sql">
       <CopyToOutputDirectory>Always</CopyToOutputDirectory>
     </None>
+    <None Update="UpdateScripts\1.0.270.sql">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </None>
   </ItemGroup>
 
   <ItemGroup>

server/Admin.NET.Web.Entry/UpdateScripts/1.0.270.sql

@@ -0,0 +1,32 @@
+-- 1.0.270：S5 生产领料 B-① 迁通用管线 + dopdemorq 双源入站实体登记（本库 status=1 / SQLSERVER status=0）
+--
+-- 背景：ProductionIssue 由 bespoke 直读 NbrMaster 迁入通用管线（源→mdp_stg_production_issue→transform→mdp_std_production_issue）。
+--       登记两个 NbrMaster 入站实体（本库 + dopdemorq SQL Server），均落 mdp_stg_production_issue、biz_key=Domain,Nbr、FULL。
+-- 状态：WIP 占位（不占正式版本号、未注册 csproj Copy、不随 AutoVersionUpdate 自动执行）；Phase 3 提交时 rename 为 1.0.<next>.sql 并注册 Copy。
+--
+-- 硬约束：
+--   1) 本库实体 S5_PRODUCTION_ISSUE_MASTER status=1：RunFullAsync 迁 stg 管线后依赖它（否则现有 s5-production-issue-mdp/refresh 端点断）。
+--   2) SQLSERVER 实体 status=0：就位不启用，合入后不改变现有运行行为；启用需带外注密码+受控切源。
+--   3) 两实体均 sync_mode='FULL'（本批 FULL-only，规避增量边界；不启自动增量）。
+--   4) source_table_name='NbrMaster'（两库同名）；transform 按 Type='SM' AND IsActive 过滤（领料）。tenant_id 由入站上下文赋值，源无此列。
+-- 幂等：WHERE NOT EXISTS 守卫，可重复执行。
+
+-- 本库入站实体（AIDOPDEV_MYSQL, status=1）
+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='AIDOPDEV_MYSQL'),
+       'S5_PRODUCTION_ISSUE_MASTER', 'S5生产领料主入站(本库)', 'TABLE',
+       'NbrMaster', 'mdp_stg_production_issue',
+       'FULL', 5000, 'Domain,Nbr', 1, 'S5生产领料 B-① 本库入站', NOW(), NOW()
+WHERE NOT EXISTS (SELECT 1 FROM mdp_entity WHERE entity_code='S5_PRODUCTION_ISSUE_MASTER');
+
+-- dopdemorq SQL Server 入站实体（DOPDEMORQ_SQLSERVER, status=0 就位不启用）
+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_PRODUCTION_ISSUE_MASTER_SQLSERVER', 'S5生产领料主入站(SQLServer)', 'TABLE',
+       'NbrMaster', 'mdp_stg_production_issue',
+       'FULL', 5000, 'Domain,Nbr', 0, 'S5生产领料 B-① 双源 disabled', NOW(), NOW()
+WHERE NOT EXISTS (SELECT 1 FROM mdp_entity WHERE entity_code='S5_PRODUCTION_ISSUE_MASTER_SQLSERVER');

server/Plugins/Admin.NET.Plugin.AiDOP/DataPlatform/MdpJsonSql.cs

@@ -0,0 +1,42 @@
+namespace Admin.NET.Plugin.AiDOP.DataPlatform;
+
+/// <summary>
+/// 数据中台 stg → std transform 层的 JSON 取值 SQL 表达式生成器（跨源类型兼容）。
+///
+/// 背景：SQL Server 源经 <c>MdpDbPullExecutor</c> 的 <c>System.Text.Json</c> 序列化后，raw_data JSON 与 MySQL 源不一致：
+///   datetime → ISO-8601 <c>2026-07-29T00:00:00</c>（T 分隔）；bit → JSON boolean <c>true/false</c>；
+///   JSON null → <c>JSON_UNQUOTE</c> 后为字符串 <c>'null'</c>。MySQL STRICT 模式下裸 CAST/STR_TO_DATE 会 500。
+///
+/// 本类<b>只生成 MySQL transform SQL 表达式字符串</b>，供调用方内插进 INSERT...SELECT；
+/// <b>不改 raw_data、不改 MdpDbPullExecutor/MdpStagingWriter/MdpSourceScopeFactory 核心、不改已运行 entity 的入站行为</b>。
+/// 属业务转换辅助，非核心 MDP 修改。字段名 field 为编译期常量，非用户输入，无注入风险。
+/// </summary>
+internal static class MdpJsonSql
+{
+    /// <summary>裸取值：JSON_UNQUOTE(JSON_EXTRACT(alias.raw_data,'$.field'))。</summary>
+    private static string Ext(string alias, string field) =>
+        $"JSON_UNQUOTE(JSON_EXTRACT({alias}.raw_data,'$.{field}'))";
+
+    /// <summary>可空字符串：把 JSON null 的 'null' 字符串归一为 SQL NULL。</summary>
+    public static string Str(string alias, string field) =>
+        $"NULLIF({Ext(alias, field)},'null')";
+
+    /// <summary>秒级 datetime：兼容空格/ISO-T/带毫秒，NULL/空/'null' → NULL。</summary>
+    public static string DateTimeSec(string alias, string field) =>
+        $"STR_TO_DATE(REPLACE(LEFT(NULLIF(NULLIF({Ext(alias, field)},'null'),''),19),'T',' '),'%Y-%m-%d %H:%i:%s')";
+
+    /// <summary>可空 DECIMAL：'null'/空 → NULL，避免 CAST('null' AS DECIMAL) STRICT 500。</summary>
+    public static string Dec(string alias, string field, int p, int s) =>
+        $"CAST(NULLIF(NULLIF({Ext(alias, field)},'null'),'') AS DECIMAL({p},{s}))";
+
+    /// <summary>可空整数：'null'/空 → NULL。</summary>
+    public static string Int(string alias, string field) =>
+        $"CAST(NULLIF(NULLIF({Ext(alias, field)},'null'),'') AS SIGNED)";
+
+    /// <summary>布尔判真：兼容 1/0 与 SQL Server bit 序列化的 true/false。</summary>
+    public static string BoolTrue(string alias, string field) =>
+        $"LOWER({Ext(alias, field)}) IN ('1','true')";
+
+    /// <summary>原样取值（不做 'null' 归一）：仅用于确知非空/仅展示的场景。</summary>
+    public static string Raw(string alias, string field) => Ext(alias, field);
+}

server/Plugins/Admin.NET.Plugin.AiDOP/MaterialWarehouse/ProductionIssueMdpSyncService.cs

@@ -1,36 +1,47 @@
+using Admin.NET.Plugin.AiDOP.DataPlatform;
+using Admin.NET.Plugin.AiDOP.DataPlatform.Executors;
+
 namespace Admin.NET.Plugin.AiDOP.MaterialWarehouse;
 
 /// <summary>
-/// S5 生产领料单 数据中台只读同步转换服务（DOP 内部，头级单表，独立于 S5MdpSyncTransformService 的 KPI 管线）。
+/// S5 生产领料单 数据中台只读同步转换服务（DOP 内部，头级单表）。
 ///
-/// 源：aidopdev NbrMaster(m)，业务类型 Type='SM'；维表 DepartmentMaster(d) LEFT JOIN（Domain+Department）。
-/// 链路：NbrMaster(SM) 只读 → mdp_std_production_issue(typed)。
-///     不经 stg、不建明细、不建 dwd（只读列表）；SM=领料单业务类型有 WorkOrderPickBillService 代码实证（工单下达 Type=SM）。
+/// B-① 双源通用管线（2026-07-29 由 bespoke 直读 NbrMaster 迁入）：
+///   源(NbrMaster) --MdpDbPullExecutor--> mdp_stg_production_issue(raw_data JSON) --transform--> mdp_std_production_issue。
+///   入站实体：S5_PRODUCTION_ISSUE_MASTER(源=AIDOPDEV_MYSQL) / S5_PRODUCTION_ISSUE_MASTER_SQLSERVER(源=DOPDEMORQ_SQLSERVER)。
+///   维表 DepartmentMaster 仍本库 LEFT JOIN（与 mode-A 采购/生产收货同构；department_desc 富化，部门 code 由事实表保留）。
 ///
 /// 约束：
-///   - 只读源表，仅写 mdp_std_production_issue；绝不 INSERT/UPDATE/DELETE NbrMaster；不碰 WorkOrderPickBillService 写路径。
-///   - 头过滤 m.Type='SM' AND m.IsActive=1（领料无 IsReturn 语义）。
-///   - status_desc 用旧系统领料 SQL 口径：PretreatmentState 非空取其值，Status='C'→'已下架'，'A'→'备料中'，否则空。
-///   - 领料人 issue_user 取 m.User1（列表 SQL 口径），原样落库。
-///   - trans_type 展示：TransType='Z61'→'补料'，否则'正常'；trans_type_text：TransType='PrevProcess'→'需要前处理'，否则空（旧系统口径）。
-///   - 不做 danjia/jiage/库存事务/状态流转；SM 为 0 行时转换成功完成、处理数为 0，不报错。
+///   - 只读源/贴源，仅写 mdp_stg_production_issue / mdp_std_production_issue；绝不写 NbrMaster；不碰 WorkOrderPickBillService 写路径。
+///   - 头过滤 Type='SM' AND IsActive（领料无 IsReturn 语义）；SM=领料单业务类型（Type SM/WOI/WOD/CA 互斥）。
+///   - 跨源类型兼容经 MdpJsonSql（datetime ISO-T / bit true-false / JSON null 归一），不改 MDP 核心。
+///   - 切源单 active source + FULL Replace（决策1A）；不做 merge、不做回写/outbox（决策2A）。
+///   - SM 为 0 行时转换成功完成、处理数为 0，不报错。
 /// </summary>
 public class ProductionIssueMdpSyncService : ITransient
 {
     private const string JobCode = "S5_PRODUCTION_ISSUE_MDP_SYNC";
+    private const string InboundEntityCode = "S5_PRODUCTION_ISSUE_MASTER";
+
+    // 双源（dopdemorq SQL Server）：第二源与其入站实体。实体默认 status=0（就位不启用）。
+    private const string SqlServerSourceCode = "DOPDEMORQ_SQLSERVER";
+    private const string SqlServerEntityCode = "S5_PRODUCTION_ISSUE_MASTER_SQLSERVER";
 
     private readonly ISqlSugarClient _db;
+    private readonly MdpSourcePullDispatcher _pullDispatcher;
 
-    public ProductionIssueMdpSyncService(ISqlSugarClient db)
+    public ProductionIssueMdpSyncService(ISqlSugarClient db, MdpSourcePullDispatcher pullDispatcher)
     {
         _db = db;
+        _pullDispatcher = pullDispatcher;
     }
 
-    /// <summary>全量同步转换：源(SM 头) → 标准层(头)。</summary>
+    /// <summary>全量：本库 NbrMaster 执行器灌 stg → 标准层（读 stg）。</summary>
     public async Task<ProductionIssueMdpSyncResult> RunFullAsync(CancellationToken cancellationToken = default, string triggerType = "AUTO")
     {
         cancellationToken.ThrowIfCancellationRequested();
         await EnsureTablesAsync();
+        await EnsureStgTableAsync();
 
         var now = DateTime.Now;
         var batchId = $"S5_PROD_ISSUE_FULL_{now:yyyyMMddHHmmss}";
@@ -39,6 +50,14 @@ public class ProductionIssueMdpSyncService : ITransient
 
         try
         {
+            var pullCtx = new MdpPullContext
+            {
+                TenantId = 0,
+                FullRefresh = true,
+                TaskCode = "S5_PRODUCTION_ISSUE_INBOUND",
+                BatchId = $"{batchId}_PULL"
+            };
+            await PopulateStgAsync(InboundEntityCode, pullCtx, cancellationToken);
             result.HeadRows = await TransformHeadStandardAsync(batchId, now);
             await MarkRunSuccessAsync(runLogId, now, result);
             return result;
@@ -50,6 +69,136 @@ public class ProductionIssueMdpSyncService : ITransient
         }
     }
 
+    /// <summary>双模式入站：执行器抽头落 stg，再跑标准层（读 stg）。</summary>
+    public async Task<ProductionIssueInboundResult> RunInboundAsync(
+        long tenantId = 0, bool fullRefresh = false, CancellationToken cancellationToken = default)
+    {
+        cancellationToken.ThrowIfCancellationRequested();
+        await EnsureTablesAsync();
+        await EnsureStgTableAsync();
+
+        var now = DateTime.Now;
+        var pullCtx = new MdpPullContext
+        {
+            TenantId = tenantId,
+            FullRefresh = fullRefresh,
+            TaskCode = "S5_PRODUCTION_ISSUE_INBOUND",
+            BatchId = $"S5_PROD_ISSUE_IN_{now:yyyyMMddHHmmss}"
+        };
+        var pull = await PopulateStgAsync(InboundEntityCode, pullCtx, cancellationToken);
+
+        var batchId = $"S5_PROD_ISSUE_STD_{now:yyyyMMddHHmmss}";
+        var runLogId = await InsertRunLogAsync(batchId, now, "INBOUND");
+        var result = new ProductionIssueMdpSyncResult { BatchId = batchId, RunLogId = runLogId };
+        try
+        {
+            result.HeadRows = await TransformHeadStandardAsync(batchId, now);
+            await MarkRunSuccessAsync(runLogId, now, result);
+        }
+        catch (Exception ex)
+        {
+            await MarkRunFailedAsync(runLogId, now, ex.Message);
+            throw;
+        }
+
+        return new ProductionIssueInboundResult
+        {
+            PullBatchId = pullCtx.BatchId,
+            RowsPulled = pull.RowsPulled,
+            RowsWrittenStg = pull.RowsWritten,
+            NewCursor = pull.NewCursor,
+            TransformBatchId = result.BatchId,
+            StdRows = result.HeadRows
+        };
+    }
+
+    /// <summary>
+    /// 双源切换 FULL Replace（Phase 1）：从指定 DB 源（默认 DOPDEMORQ_SQLSERVER）PullAll 全量灌 stg，
+    /// 成功后在单事务内以「仅当前源」结果 FULL 重建 mdp_std_production_issue（按 tenant 精确隔离），消除旧源独有业务键残留。
+    /// PullAllByEntityCodeAsync 抽尽 + FullRefresh=true；transform 仅当前 source_system；Pull 成功后才进入 destructive replace。
+    /// SQLSERVER 实体默认 status=0（就位不启用）；本批 FULL-only，不启自动增量。
+    /// </summary>
+    public async Task<ProductionIssueInboundResult> RunSourceSwitchFullAsync(
+        string sourceCode = SqlServerSourceCode,
+        string entityCode = SqlServerEntityCode,
+        long tenantId = 0,
+        CancellationToken cancellationToken = default)
+    {
+        cancellationToken.ThrowIfCancellationRequested();
+        await EnsureTablesAsync();
+        await EnsureStgTableAsync();
+
+        var now = DateTime.Now;
+        var pullCtx = new MdpPullContext
+        {
+            TenantId = tenantId,
+            FullRefresh = true,
+            TaskCode = "S5_PRODUCTION_ISSUE_INBOUND",
+            BatchId = $"S5_PROD_ISSUE_SW_{now:yyyyMMddHHmmss}"
+        };
+        var pull = await _pullDispatcher.PullAllByEntityCodeAsync(entityCode, pullCtx, cancellationToken);
+
+        var batchId = $"S5_PROD_ISSUE_SWSTD_{now:yyyyMMddHHmmss}";
+        var runLogId = await InsertRunLogAsync(batchId, now, "SOURCE_SWITCH");
+        var result = new ProductionIssueMdpSyncResult { BatchId = batchId, RunLogId = runLogId };
+        try
+        {
+            result.HeadRows = await MdpStdFullReplace.ReplaceAsync(
+                _db, "mdp_std_production_issue", 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 ProductionIssueInboundResult
+        {
+            PullBatchId = pullCtx.BatchId,
+            RowsPulled = pull.RowsPulled,
+            RowsWrittenStg = pull.RowsWritten,
+            NewCursor = pull.NewCursor,
+            TransformBatchId = batchId,
+            StdRows = result.HeadRows
+        };
+    }
+
+    private async Task<(int RowsPulled, int RowsWritten, string? NewCursor)> PopulateStgAsync(
+        string entityCode, MdpPullContext pullCtx, CancellationToken cancellationToken)
+    {
+        var r = await _pullDispatcher.PullByEntityCodeAsync(entityCode, pullCtx, cancellationToken);
+        return (r.RowsPulled, r.RowsWritten, r.NewCursor);
+    }
+
+    private async Task EnsureStgTableAsync()
+    {
+        await _db.Ado.ExecuteCommandAsync(
+            """
+            CREATE TABLE IF NOT EXISTS mdp_stg_production_issue (
+                id BIGINT PRIMARY KEY AUTO_INCREMENT,
+                tenant_id BIGINT NOT NULL,
+                source_system VARCHAR(50) NULL,
+                source_table VARCHAR(200),
+                source_row_id VARCHAR(200),
+                source_biz_key VARCHAR(300) NULL,
+                raw_data JSON,
+                sync_batch_id VARCHAR(100),
+                sync_time DATETIME NULL DEFAULT CURRENT_TIMESTAMP,
+                process_status VARCHAR(20) NOT NULL DEFAULT 'PENDING',
+                process_message VARCHAR(500) NULL,
+                create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+                update_time DATETIME NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+                UNIQUE KEY uk_source_key (source_system, source_table, source_biz_key),
+                KEY idx_batch (sync_batch_id),
+                KEY idx_src (source_table, source_row_id),
+                KEY idx_tenant (tenant_id)
+            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='S5生产领料执行器贴源层'
+            """);
+    }
+
     /// <summary>防御式建表（与 UpdateScripts DDL 同构，幂等）。</summary>
     private async Task EnsureTablesAsync()
     {
@@ -99,43 +248,63 @@ public class ProductionIssueMdpSyncService : ITransient
     }
 
     /// <summary>
-    /// 标准化头：NbrMaster(SM) + DepartmentMaster -> mdp_std_production_issue。返回处理头行数。
-    /// SQL 由旧系统领料头 SQL 翻译（SQL Server -&gt; MySQL）：IsNull-&gt;IFNULL、upper() 保留、
-    /// rtrim(a+' '+b)-&gt;TRIM(CONCAT(...))、with(nolock) 去除、convert(varchar,EffDate,23)-&gt; DATE(EffDate)。
+    /// 标准化头：stg(NbrMaster,SM) + DepartmentMaster(本库) → mdp_std_production_issue。
+    /// sourceSystem 非空时仅转当前源（切源 FULL Replace 用）；为空时全源（本库全量）。跨源类型经 MdpJsonSql 兼容。
     /// </summary>
-    private async Task<int> TransformHeadStandardAsync(string batchId, DateTime now)
+    private async Task<int> TransformHeadStandardAsync(string batchId, DateTime now, string? sourceSystem = null)
     {
+        var srcClause = sourceSystem == null ? "" : " AND m.source_system=@Src";
+        // 表达式片段（跨源类型兼容）
+        var domainE = MdpJsonSql.Str("m", "Domain");
+        var deptE = MdpJsonSql.Str("m", "Department");
+        var statusUpper = $"UPPER(IFNULL({MdpJsonSql.Str("m", "Status")},''))";
+        var pretreatE = MdpJsonSql.Str("m", "PretreatmentState");
+        var transTypeE = MdpJsonSql.Str("m", "TransType");
+
+        var countPars = new List<SugarParameter>();
+        if (sourceSystem != null) countPars.Add(new SugarParameter("@Src", sourceSystem));
         var rows = await _db.Ado.GetIntAsync(
-            "SELECT COUNT(1) FROM NbrMaster m WHERE m.Type='SM' AND m.IsActive=1");
-        await _db.Ado.ExecuteCommandAsync(
-            """
+            $"SELECT COUNT(1) FROM mdp_stg_production_issue m WHERE m.source_table='NbrMaster' " +
+            $"AND {MdpJsonSql.Str("m", "Type")}='SM' AND {MdpJsonSql.BoolTrue("m", "IsActive")}{srcClause}",
+            countPars);
+
+        var insertSql =
+            $"""
             INSERT INTO mdp_std_production_issue
             (tenant_id, factory_id, source_system, domain, rec_id, nbr, issue_date, status, status_desc,
              work_ord, department, department_desc, qty_ord, prod_line, applicant_name, issue_user,
              user1, remark, create_user, source_create_time, pretreatment_state, trans_type, trans_type_text,
              eff_date, address, source_biz_key, sync_batch_id, sync_time)
             SELECT
-                IFNULL(m.tenant_id, 0), 1, 'AIDOP', m.Domain, m.RecID, m.Nbr, m.Date,
-                UPPER(IFNULL(m.Status, '')),
+                IFNULL({MdpJsonSql.Int("m", "tenant_id")}, IFNULL(m.tenant_id, 0)), 1, IFNULL(NULLIF(m.source_system,''), 'AIDOP'),
+                {domainE}, {MdpJsonSql.Int("m", "RecID")}, {MdpJsonSql.Str("m", "Nbr")},
+                {MdpJsonSql.DateTimeSec("m", "Date")},
+                {statusUpper},
                 CASE
-                    WHEN IFNULL(m.PretreatmentState,'')<>'' THEN m.PretreatmentState
-                    WHEN UPPER(IFNULL(m.Status,''))='C' THEN '已下架'
-                    WHEN UPPER(IFNULL(m.Status,''))='A' THEN '备料中'
+                    WHEN IFNULL({pretreatE},'')<>'' THEN {pretreatE}
+                    WHEN {statusUpper}='C' THEN '已下架'
+                    WHEN {statusUpper}='A' THEN '备料中'
                     ELSE ''
                 END,
-                m.WorkOrd, m.Department, TRIM(CONCAT(m.Department, ' ', IFNULL(d.Descr, ''))),
-                m.QtyOrd, m.ProdLine, m.Name, CAST(m.User1 AS CHAR),
-                CAST(m.User1 AS CHAR), m.Remark, m.CreateUser, m.CreateTime,
-                IFNULL(m.PretreatmentState, ''),
-                CASE WHEN m.TransType='Z61' THEN '补料' ELSE '正常' END,
-                CASE WHEN m.TransType='PrevProcess' THEN '需要前处理' ELSE '' END,
-                m.EffDate, m.Address,
-                CONCAT(m.Domain, ':', m.RecID), @BatchId, @Now
-            FROM NbrMaster m
-            LEFT JOIN DepartmentMaster d ON d.Domain = m.Domain AND d.Department = m.Department
-            WHERE m.Type='SM' AND m.IsActive=1
+                {MdpJsonSql.Str("m", "WorkOrd")}, {deptE},
+                TRIM(CONCAT(IFNULL({deptE},''), ' ', IFNULL(d.Descr, ''))),
+                {MdpJsonSql.Dec("m", "QtyOrd", 18, 5)}, {MdpJsonSql.Str("m", "ProdLine")}, {MdpJsonSql.Str("m", "Name")},
+                {MdpJsonSql.Str("m", "User1")},
+                {MdpJsonSql.Str("m", "User1")}, {MdpJsonSql.Str("m", "Remark")}, {MdpJsonSql.Str("m", "CreateUser")},
+                {MdpJsonSql.DateTimeSec("m", "CreateTime")},
+                IFNULL({pretreatE}, ''),
+                CASE WHEN {transTypeE}='Z61' THEN '补料' ELSE '正常' END,
+                CASE WHEN {transTypeE}='PrevProcess' THEN '需要前处理' ELSE '' END,
+                {MdpJsonSql.DateTimeSec("m", "EffDate")}, {MdpJsonSql.Str("m", "Address")},
+                IFNULL(NULLIF(m.source_biz_key,''), CONCAT(IFNULL({domainE},''), ':', IFNULL({MdpJsonSql.Str("m", "RecID")},''))),
+                @BatchId, @Now
+            FROM mdp_stg_production_issue m
+            LEFT JOIN DepartmentMaster d ON d.Domain = {domainE} AND d.Department = {deptE}
+            WHERE m.source_table='NbrMaster'
+              AND {MdpJsonSql.Str("m", "Type")}='SM'
+              AND {MdpJsonSql.BoolTrue("m", "IsActive")}{srcClause}
             ON DUPLICATE KEY UPDATE
-                factory_id=VALUES(factory_id), nbr=VALUES(nbr), issue_date=VALUES(issue_date),
+                factory_id=VALUES(factory_id), source_system=VALUES(source_system), nbr=VALUES(nbr), issue_date=VALUES(issue_date),
                 status=VALUES(status), status_desc=VALUES(status_desc), work_ord=VALUES(work_ord),
                 department=VALUES(department), department_desc=VALUES(department_desc), qty_ord=VALUES(qty_ord),
                 prod_line=VALUES(prod_line), applicant_name=VALUES(applicant_name), issue_user=VALUES(issue_user),
@@ -144,9 +313,14 @@ public class ProductionIssueMdpSyncService : ITransient
                 trans_type=VALUES(trans_type), trans_type_text=VALUES(trans_type_text),
                 eff_date=VALUES(eff_date), address=VALUES(address),
                 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;
     }
 
@@ -210,3 +384,14 @@ public sealed class ProductionIssueMdpSyncResult
     public string BatchId { get; set; } = string.Empty;
     public int HeadRows { get; set; }
 }
+
+/// <summary>生产领料双源入站结果（stg 抽数 + std 转换）。</summary>
+public sealed class ProductionIssueInboundResult
+{
+    public string PullBatchId { get; set; } = string.Empty;
+    public int RowsPulled { get; set; }
+    public int RowsWrittenStg { get; set; }
+    public string? NewCursor { get; set; }
+    public string TransformBatchId { get; set; } = string.Empty;
+    public int StdRows { get; set; }
+}