using Admin.NET.Plugin.AiDOP.DataPlatform;
using Admin.NET.Plugin.AiDOP.DataPlatform.Executors;
using Admin.NET.Plugin.AiDOP.Entity.DataPlatform;
namespace Admin.NET.Plugin.AiDOP.MaterialWarehouse;
///
/// S5 生产领料单 数据中台只读同步转换服务(DOP 内部,头级单表)。
///
/// 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_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,不报错。
///
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, MdpSourcePullDispatcher pullDispatcher)
{
_db = db;
_pullDispatcher = pullDispatcher;
}
/// 全量:按当前启用入站实体灌 stg → 标准层(WP9 S5 后默认 165)。
public async Task RunFullAsync(CancellationToken cancellationToken = default, string triggerType = "AUTO")
{
cancellationToken.ThrowIfCancellationRequested();
await EnsureTablesAsync();
await EnsureStgTableAsync();
var active = await ResolveActiveInboundAsync(cancellationToken);
var now = DateTime.Now;
var batchId = $"S5_PROD_ISSUE_FULL_{now:yyyyMMddHHmmss}";
var runLogId = await InsertRunLogAsync(batchId, now, triggerType);
var result = new ProductionIssueMdpSyncResult { BatchId = batchId, RunLogId = runLogId };
try
{
var pullCtx = new MdpPullContext
{
TenantId = 0,
FullRefresh = true,
TaskCode = "S5_PRODUCTION_ISSUE_INBOUND",
BatchId = $"{batchId}_PULL"
};
await PopulateStgAsync(active.EntityCode, pullCtx, cancellationToken);
if (active.UseSourceFilter)
{
result.HeadRows = await MdpStdFullReplace.ReplaceAsync(
_db, "mdp_std_production_issue", 0, extraWhere: null,
insertScopedAsync: () => TransformHeadStandardAsync(batchId, now, active.SourceCode),
cancellationToken);
}
else
{
result.HeadRows = await TransformHeadStandardAsync(batchId, now);
}
await MarkRunSuccessAsync(runLogId, now, result);
return result;
}
catch (Exception ex)
{
await MarkRunFailedAsync(runLogId, now, ex.Message);
throw;
}
}
/// 双模式入站:执行器抽头落 stg,再跑标准层(读 stg)。
public async Task RunInboundAsync(
long tenantId = 0, bool fullRefresh = false, CancellationToken cancellationToken = default)
{
cancellationToken.ThrowIfCancellationRequested();
await EnsureTablesAsync();
await EnsureStgTableAsync();
var active = await ResolveActiveInboundAsync(cancellationToken);
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(active.EntityCode, 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
{
if (active.UseSourceFilter)
{
result.HeadRows = await MdpStdFullReplace.ReplaceAsync(
_db, "mdp_std_production_issue", tenantId, extraWhere: null,
insertScopedAsync: () => TransformHeadStandardAsync(batchId, now, active.SourceCode),
cancellationToken);
}
else
{
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
};
}
///
/// WP9 S5:切到 165 权威源并 FULL Replace 标准层;同时确保本库历史 SM 已打 AIDOP_LEGACY。
///
public async Task ActivateSqlServerSourceAndSwitchAsync(
long tenantId = 0,
CancellationToken cancellationToken = default)
{
cancellationToken.ThrowIfCancellationRequested();
await EnsureNbrMasterSourceSystemColumnAsync();
await MarkLocalSmLegacyAsync();
await FlipInboundEntityStatusAsync();
return await RunSourceSwitchFullAsync(
SqlServerSourceCode, SqlServerEntityCode, tenantId, cancellationToken);
}
private async Task<(string EntityCode, string SourceCode, bool UseSourceFilter)> ResolveActiveInboundAsync(
CancellationToken cancellationToken)
{
var sqlOn = await _db.Queryable()
.Where(x => x.EntityCode == SqlServerEntityCode && x.Status == 1)
.AnyAsync(cancellationToken);
if (sqlOn)
return (SqlServerEntityCode, SqlServerSourceCode, true);
var mysqlOn = await _db.Queryable()
.Where(x => x.EntityCode == InboundEntityCode && x.Status == 1)
.AnyAsync(cancellationToken);
if (mysqlOn)
return (InboundEntityCode, "AIDOPDEV_MYSQL", false);
throw Oops.Oh("生产领料同步失败:未启用任何 S5 入站实体,请执行 WP9 S5 切源或检查 mdp_entity。");
}
private async Task FlipInboundEntityStatusAsync()
{
await _db.Ado.ExecuteCommandAsync(
"""
UPDATE mdp_entity
SET status = 0, update_time = NOW()
WHERE entity_code = @Mysql AND status <> 0
""",
new SugarParameter("@Mysql", InboundEntityCode));
await _db.Ado.ExecuteCommandAsync(
"""
UPDATE mdp_entity
SET status = 1, update_time = NOW()
WHERE entity_code = @Sql AND status <> 1
""",
new SugarParameter("@Sql", SqlServerEntityCode));
}
private async Task EnsureNbrMasterSourceSystemColumnAsync()
{
var exists = await _db.Ado.GetIntAsync(
"""
SELECT COUNT(1) FROM information_schema.COLUMNS
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'NbrMaster' AND COLUMN_NAME = 'source_system'
""");
if (exists == 0)
{
await _db.Ado.ExecuteCommandAsync(
"ALTER TABLE NbrMaster ADD COLUMN source_system VARCHAR(50) NULL DEFAULT NULL AFTER Remark");
}
}
private async Task MarkLocalSmLegacyAsync()
{
await _db.Ado.ExecuteCommandAsync(
"""
UPDATE NbrMaster
SET source_system = 'AIDOP_LEGACY'
WHERE Type = 'SM'
AND IFNULL(source_system, '') IN ('', 'AIDOP')
""");
}
///
/// 双源切换 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,不启自动增量。
///
public async Task 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生产领料执行器贴源层'
""");
}
/// 防御式建表(与 UpdateScripts DDL 同构,幂等)。
private async Task EnsureTablesAsync()
{
await _db.Ado.ExecuteCommandAsync(
"""
CREATE TABLE IF NOT EXISTS mdp_std_production_issue (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
tenant_id BIGINT NOT NULL DEFAULT 0,
factory_id BIGINT NULL DEFAULT 1,
source_system VARCHAR(50) NOT NULL DEFAULT 'AIDOP',
domain VARCHAR(80) NOT NULL,
rec_id INT NOT NULL,
nbr VARCHAR(24) NULL,
issue_date DATETIME NULL,
status VARCHAR(8) NULL,
status_desc VARCHAR(20) NULL,
work_ord VARCHAR(64) NULL,
department VARCHAR(8) NULL,
department_desc VARCHAR(255) NULL,
qty_ord DECIMAL(18,5) NULL DEFAULT 0,
prod_line VARCHAR(8) NULL,
applicant_name VARCHAR(12) NULL,
issue_user VARCHAR(255) NULL,
user1 TEXT NULL,
remark VARCHAR(200) NULL,
create_user VARCHAR(24) NULL,
source_create_time DATETIME NULL,
pretreatment_state VARCHAR(50) NULL,
trans_type VARCHAR(24) NULL,
trans_type_text VARCHAR(50) NULL,
eff_date DATETIME NULL,
address VARCHAR(120) NULL,
source_biz_key VARCHAR(200) NULL,
sync_batch_id VARCHAR(100) NOT NULL,
sync_time DATETIME NOT NULL,
create_time DATETIME DEFAULT CURRENT_TIMESTAMP,
update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
UNIQUE KEY uk_mdp_std_prod_issue (tenant_id, domain, rec_id),
KEY idx_mdp_std_prod_issue_nbr (tenant_id, nbr),
KEY idx_mdp_std_prod_issue_date (tenant_id, issue_date),
KEY idx_mdp_std_prod_issue_status (tenant_id, status),
KEY idx_mdp_std_prod_issue_workord (tenant_id, work_ord),
KEY idx_mdp_std_prod_issue_prodline (tenant_id, prod_line),
KEY idx_mdp_std_prod_issue_dept (tenant_id, department)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='S5生产领料单头标准层'
""");
}
///
/// 标准化头:stg(NbrMaster,SM) + DepartmentMaster(本库) → mdp_std_production_issue。
/// sourceSystem 非空时仅转当前源(切源 FULL Replace 用);为空时全源(本库全量)。跨源类型经 MdpJsonSql 兼容。
///
private async Task 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();
if (sourceSystem != null) countPars.Add(new SugarParameter("@Src", sourceSystem));
var rows = await _db.Ado.GetIntAsync(
$"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({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({pretreatE},'')<>'' THEN {pretreatE}
WHEN {statusUpper}='C' THEN '已下架'
WHEN {statusUpper}='A' THEN '备料中'
ELSE ''
END,
{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), 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),
user1=VALUES(user1), remark=VALUES(remark), create_user=VALUES(create_user),
source_create_time=VALUES(source_create_time), pretreatment_state=VALUES(pretreatment_state),
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
""";
var insPars = new List
{
new("@BatchId", batchId),
new("@Now", now)
};
if (sourceSystem != null) insPars.Add(new SugarParameter("@Src", sourceSystem));
await _db.Ado.ExecuteCommandAsync(insertSql, insPars);
return rows;
}
private async Task InsertRunLogAsync(string batchId, DateTime startedAt, string triggerType)
{
await _db.Ado.ExecuteCommandAsync(
"""
INSERT INTO mdp_transform_run_log
(tenant_id, job_code, job_name, trigger_type, batch_id, status, start_time)
VALUES (0, @JobCode, 'S5生产领料单MDP同步与标准化转换', @TriggerType, @BatchId, 'RUNNING', @StartTime)
""",
new SugarParameter("@JobCode", JobCode),
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 { new("@BatchId", batchId) });
}
private async Task MarkRunSuccessAsync(long runLogId, DateTime startedAt, ProductionIssueMdpSyncResult result)
{
var finishedAt = DateTime.Now;
await _db.Ado.ExecuteCommandAsync(
"""
UPDATE mdp_transform_run_log
SET status='SUCCESS', end_time=@EndTime, duration_ms=@DurationMs,
stage_rows=0, standard_rows=@StandardRows, dwd_rows=0, update_time=CURRENT_TIMESTAMP
WHERE id=@Id
""",
new SugarParameter("@EndTime", finishedAt),
new SugarParameter("@DurationMs", (int)(finishedAt - startedAt).TotalMilliseconds),
new SugarParameter("@StandardRows", result.HeadRows),
new SugarParameter("@Id", runLogId));
}
private async Task MarkRunFailedAsync(long runLogId, DateTime startedAt, string message)
{
var finishedAt = DateTime.Now;
await _db.Ado.ExecuteCommandAsync(
"""
UPDATE mdp_transform_run_log
SET status='FAILED', end_time=@EndTime, duration_ms=@DurationMs,
error_message=@ErrorMessage, update_time=CURRENT_TIMESTAMP
WHERE id=@Id
""",
new SugarParameter("@EndTime", finishedAt),
new SugarParameter("@DurationMs", (int)(finishedAt - startedAt).TotalMilliseconds),
new SugarParameter("@ErrorMessage", message.Length > 2000 ? message[..2000] : message),
new SugarParameter("@Id", runLogId));
}
private static string NormalizeTriggerType(string? triggerType)
=> string.IsNullOrWhiteSpace(triggerType) ? "AUTO" : triggerType.Trim().ToUpperInvariant();
}
/// 生产领料单 MDP 同步转换结果。
public sealed class ProductionIssueMdpSyncResult
{
public long RunLogId { get; set; }
public string BatchId { get; set; } = string.Empty;
public int HeadRows { get; set; }
}
/// 生产领料双源入站结果(stg 抽数 + std 转换)。
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; }
}