|
|
@@ -17,6 +17,42 @@ public class MdpMonitorService : IDynamicApiController, ITransient
|
|
|
["S4"] = "S4_MDP_SYNC_TRANSFORM"
|
|
|
};
|
|
|
|
|
|
+ private static readonly Dictionary<string, MdpJobCatalogItem> JobCatalog = new(StringComparer.OrdinalIgnoreCase)
|
|
|
+ {
|
|
|
+ ["S1_MDP_SYNC_TRANSFORM"] = new(
|
|
|
+ "S1_MDP_SYNC_TRANSFORM",
|
|
|
+ "order_delivery",
|
|
|
+ "订单交付域",
|
|
|
+ ["S1", "S2", "S3", "S4", "S7", "S9"],
|
|
|
+ "GLOBAL_DOMAIN",
|
|
|
+ "订单交付域 MDP 同步",
|
|
|
+ "job_s1_mdp_sync_transform"),
|
|
|
+ ["S2_MDP_SYNC_TRANSFORM"] = new(
|
|
|
+ "S2_MDP_SYNC_TRANSFORM",
|
|
|
+ "work_schedule",
|
|
|
+ "工单排程域",
|
|
|
+ ["S2", "S3", "S5", "S6", "S8", "S9"],
|
|
|
+ "GLOBAL_DOMAIN",
|
|
|
+ "工单排程域 MDP 同步",
|
|
|
+ "job_s2_mdp_sync_transform"),
|
|
|
+ ["S3_MDP_SYNC_TRANSFORM"] = new(
|
|
|
+ "S3_MDP_SYNC_TRANSFORM",
|
|
|
+ "supply_purchase",
|
|
|
+ "供应采购域",
|
|
|
+ ["S3", "S4", "S5", "S8", "S9"],
|
|
|
+ "GLOBAL_DOMAIN",
|
|
|
+ "供应采购域 MDP 同步",
|
|
|
+ "job_s3_mdp_sync_transform"),
|
|
|
+ ["S4_MDP_SYNC_TRANSFORM"] = new(
|
|
|
+ "S4_MDP_SYNC_TRANSFORM",
|
|
|
+ "purchase_execution",
|
|
|
+ "采购执行域",
|
|
|
+ ["S4", "S5", "S8", "S9"],
|
|
|
+ "GLOBAL_DOMAIN",
|
|
|
+ "采购执行域 MDP 同步",
|
|
|
+ "job_s4_mdp_sync_transform")
|
|
|
+ };
|
|
|
+
|
|
|
private readonly ISqlSugarClient _db;
|
|
|
private readonly UserManager _userManager;
|
|
|
|
|
|
@@ -28,13 +64,11 @@ public class MdpMonitorService : IDynamicApiController, ITransient
|
|
|
|
|
|
[DisplayName("MDP模块选项")]
|
|
|
[HttpGet("mdp-monitor/modules")]
|
|
|
- public object GetModules()
|
|
|
- {
|
|
|
- return ModuleJobCodes
|
|
|
- .OrderBy(u => u.Key)
|
|
|
- .Select(u => new { moduleCode = u.Key, jobCode = u.Value })
|
|
|
- .ToList();
|
|
|
- }
|
|
|
+ public object GetModules() => BuildCatalogResponse();
|
|
|
+
|
|
|
+ [DisplayName("MDP任务目录")]
|
|
|
+ [HttpGet("mdp-monitor/catalog")]
|
|
|
+ public object GetCatalog() => BuildCatalogResponse();
|
|
|
|
|
|
[DisplayName("MDP最近运行状态")]
|
|
|
[HttpGet("mdp-monitor/latest")]
|
|
|
@@ -42,7 +76,7 @@ public class MdpMonitorService : IDynamicApiController, ITransient
|
|
|
{
|
|
|
var tenantId = _userManager.TenantId;
|
|
|
var (whereSql, pars) = BuildWhere(input, tenantId);
|
|
|
- return await _db.Ado.SqlQuerySingleAsync<MdpMonitorRunLogRow>(
|
|
|
+ var row = await _db.Ado.SqlQuerySingleAsync<MdpMonitorRunLogRow>(
|
|
|
$"""
|
|
|
{SelectColumnsSql()}
|
|
|
FROM mdp_transform_run_log
|
|
|
@@ -52,6 +86,8 @@ public class MdpMonitorService : IDynamicApiController, ITransient
|
|
|
""",
|
|
|
pars)
|
|
|
?? new MdpMonitorRunLogRow();
|
|
|
+ AttachCatalogInfo(row);
|
|
|
+ return row;
|
|
|
}
|
|
|
|
|
|
[DisplayName("MDP运行日志列表")]
|
|
|
@@ -75,6 +111,9 @@ public class MdpMonitorService : IDynamicApiController, ITransient
|
|
|
""",
|
|
|
pars);
|
|
|
|
|
|
+ foreach (var row in list)
|
|
|
+ AttachCatalogInfo(row);
|
|
|
+
|
|
|
return new { total, page, pageSize, list };
|
|
|
}
|
|
|
|
|
|
@@ -94,7 +133,10 @@ public class MdpMonitorService : IDynamicApiController, ITransient
|
|
|
LIMIT 1
|
|
|
""",
|
|
|
pars);
|
|
|
- return row ?? throw Oops.Oh("运行日志不存在");
|
|
|
+ if (row == null)
|
|
|
+ throw Oops.Oh("运行日志不存在");
|
|
|
+ AttachCatalogInfo(row);
|
|
|
+ return row;
|
|
|
}
|
|
|
|
|
|
[DisplayName("MDP同步链路详情")]
|
|
|
@@ -123,7 +165,8 @@ public class MdpMonitorService : IDynamicApiController, ITransient
|
|
|
e.incr_column AS IncrColumn, e.status AS Status
|
|
|
FROM mdp_entity e
|
|
|
LEFT JOIN mdp_source s ON s.id = e.source_id
|
|
|
- WHERE e.entity_code LIKE @EntityPrefix AND e.tenant_id = @TenantId
|
|
|
+ WHERE e.entity_code LIKE @EntityPrefix
|
|
|
+ AND (e.tenant_id = @TenantId OR e.tenant_id = 0)
|
|
|
ORDER BY e.entity_code
|
|
|
""",
|
|
|
new SugarParameter("@EntityPrefix", entityPrefix),
|
|
|
@@ -131,13 +174,16 @@ public class MdpMonitorService : IDynamicApiController, ITransient
|
|
|
|
|
|
if (entities.Count == 0)
|
|
|
{
|
|
|
- return new MdpLineageOutput
|
|
|
+ var emptyOutput = new MdpLineageOutput
|
|
|
{
|
|
|
ModuleCode = moduleCode,
|
|
|
JobCode = jobCode,
|
|
|
- Stages = BuildStageDescriptions(moduleCode),
|
|
|
+ BatchId = input.BatchId,
|
|
|
+ Stages = BuildStageDescriptions(jobCode, moduleCode),
|
|
|
Entities = new List<MdpLineageEntityRow>()
|
|
|
};
|
|
|
+ AttachLineageCatalogInfo(emptyOutput);
|
|
|
+ return emptyOutput;
|
|
|
}
|
|
|
|
|
|
var entityIds = string.Join(",", entities.Select(u => u.Id));
|
|
|
@@ -165,9 +211,11 @@ public class MdpMonitorService : IDynamicApiController, ITransient
|
|
|
error_msg AS ErrorMsg
|
|
|
FROM mdp_sync_log
|
|
|
WHERE sync_batch_id = @BatchId
|
|
|
- AND tenant_id = @TenantId
|
|
|
+ AND (tenant_id = @TenantId OR tenant_id = 0)
|
|
|
AND entity_id IN (
|
|
|
- SELECT id FROM mdp_entity WHERE entity_code LIKE @EntityPrefix
|
|
|
+ SELECT id FROM mdp_entity
|
|
|
+ WHERE entity_code LIKE @EntityPrefix
|
|
|
+ AND (tenant_id = @TenantId OR tenant_id = 0)
|
|
|
)
|
|
|
ORDER BY sync_start, id
|
|
|
""",
|
|
|
@@ -179,41 +227,59 @@ public class MdpMonitorService : IDynamicApiController, ITransient
|
|
|
.ToDictionary(u => u.Key, u => u.OrderByDescending(x => x.SyncStart).First());
|
|
|
}
|
|
|
|
|
|
+ var batchId = input.BatchId?.Trim();
|
|
|
foreach (var entity in entities)
|
|
|
{
|
|
|
entity.SourceFullName = BuildObjectFullName(entity.SourceDbType, entity.SourceDbHost, entity.SourceDbPort, entity.SourceDbName, entity.SourceTableName ?? entity.SourceApiPath);
|
|
|
entity.TargetFullName = BuildObjectFullName(entity.TargetDbType, entity.TargetDbHost, entity.TargetDbPort, entity.TargetDbName, entity.TargetTableName);
|
|
|
|
|
|
- if (mappingsByEntity.TryGetValue(entity.Id, out var entityMappings))
|
|
|
+ if (mappingsByEntity.TryGetValue(entity.Id, out var entityMappings) && entityMappings.Count > 0)
|
|
|
{
|
|
|
+ foreach (var mapping in entityMappings)
|
|
|
+ {
|
|
|
+ mapping.MappingSource = "CONFIG";
|
|
|
+ mapping.IsFallback = false;
|
|
|
+ }
|
|
|
+
|
|
|
entity.FieldMappings = entityMappings;
|
|
|
entity.FieldMappingCount = entityMappings.Count;
|
|
|
}
|
|
|
+ else
|
|
|
+ {
|
|
|
+ entity.FieldMappings = BuildFallbackFieldMappings(entity, batchId);
|
|
|
+ entity.FieldMappingCount = entity.FieldMappings.Count;
|
|
|
+ }
|
|
|
+
|
|
|
if (syncLogsByEntity.TryGetValue(entity.Id, out var syncLog))
|
|
|
entity.SyncLog = syncLog;
|
|
|
}
|
|
|
|
|
|
- return new MdpLineageOutput
|
|
|
+ var output = new MdpLineageOutput
|
|
|
{
|
|
|
ModuleCode = moduleCode,
|
|
|
JobCode = jobCode,
|
|
|
BatchId = input.BatchId,
|
|
|
- Stages = BuildStageDescriptions(moduleCode),
|
|
|
+ Stages = BuildStageDescriptions(jobCode, moduleCode),
|
|
|
Entities = entities
|
|
|
};
|
|
|
+ AttachLineageCatalogInfo(output);
|
|
|
+ return output;
|
|
|
}
|
|
|
|
|
|
private static (string WhereSql, List<SugarParameter> Parameters) BuildWhere(MdpMonitorQueryInput input, long tenantId)
|
|
|
{
|
|
|
- var where = new List<string> { "IFNULL(job_code, '') LIKE '%MDP%'", "tenant_id = @TenantId" };
|
|
|
+ var where = new List<string> { BuildMdpRunLogTenantWhere(tenantId) };
|
|
|
var pars = new List<SugarParameter> { new("@TenantId", tenantId) };
|
|
|
|
|
|
- var jobCode = ResolveJobCode(input.ModuleCode, input.JobCode);
|
|
|
- if (!string.IsNullOrWhiteSpace(jobCode))
|
|
|
+ var (jobFilterSql, jobFilterPars, noMatch) = BuildJobCodeFilter(input);
|
|
|
+ if (noMatch)
|
|
|
+ where.Add("1=0");
|
|
|
+ else
|
|
|
{
|
|
|
- where.Add("job_code=@JobCode");
|
|
|
- pars.Add(new SugarParameter("@JobCode", jobCode));
|
|
|
+ where.Add(jobFilterSql);
|
|
|
+ pars.AddRange(jobFilterPars);
|
|
|
}
|
|
|
+
|
|
|
if (!string.IsNullOrWhiteSpace(input.BatchId))
|
|
|
{
|
|
|
where.Add("batch_id LIKE @BatchId");
|
|
|
@@ -238,6 +304,283 @@ public class MdpMonitorService : IDynamicApiController, ITransient
|
|
|
return (string.Join(" AND ", where), pars);
|
|
|
}
|
|
|
|
|
|
+ /// <summary>
|
|
|
+ /// MDP 转换任务当前以 tenant_id=0 写入运行日志；登录租户查询时需兼容这类全局任务记录。
|
|
|
+ /// </summary>
|
|
|
+ internal static string BuildMdpRunLogTenantWhere(long tenantId) =>
|
|
|
+ tenantId > 0 ? "(tenant_id = @TenantId OR tenant_id = 0)" : "tenant_id = 0";
|
|
|
+
|
|
|
+ private static object BuildCatalogResponse()
|
|
|
+ {
|
|
|
+ return JobCatalog.Values
|
|
|
+ .OrderBy(u => u.JobCode, StringComparer.OrdinalIgnoreCase)
|
|
|
+ .Select(u => new
|
|
|
+ {
|
|
|
+ jobCode = u.JobCode,
|
|
|
+ displayName = u.DisplayName,
|
|
|
+ businessDomainCode = u.BusinessDomainCode,
|
|
|
+ businessDomainName = u.BusinessDomainName,
|
|
|
+ consumerModules = u.ConsumerModules,
|
|
|
+ scopeType = u.ScopeType,
|
|
|
+ moduleCode = ResolveModuleCodeFromJobCode(u.JobCode),
|
|
|
+ scheduleJobId = u.ScheduleJobId
|
|
|
+ })
|
|
|
+ .ToList();
|
|
|
+ }
|
|
|
+
|
|
|
+ private static (string Sql, List<SugarParameter> Parameters, bool NoMatch) BuildJobCodeFilter(MdpMonitorQueryInput input)
|
|
|
+ {
|
|
|
+ var pars = new List<SugarParameter>();
|
|
|
+
|
|
|
+ if (!string.IsNullOrWhiteSpace(input.JobCode))
|
|
|
+ {
|
|
|
+ pars.Add(new SugarParameter("@JobCode", input.JobCode.Trim().ToUpperInvariant()));
|
|
|
+ return ("job_code=@JobCode", pars, false);
|
|
|
+ }
|
|
|
+
|
|
|
+ var hasDomainFilter = !string.IsNullOrWhiteSpace(input.BusinessDomainCode);
|
|
|
+ var consumerModule = !string.IsNullOrWhiteSpace(input.ConsumerModule)
|
|
|
+ ? input.ConsumerModule.Trim()
|
|
|
+ : !string.IsNullOrWhiteSpace(input.ModuleCode) ? input.ModuleCode.Trim() : null;
|
|
|
+ var hasConsumerFilter = !string.IsNullOrWhiteSpace(consumerModule);
|
|
|
+
|
|
|
+ if (!hasDomainFilter && !hasConsumerFilter)
|
|
|
+ return ("IFNULL(job_code, '') LIKE '%MDP%'", pars, false);
|
|
|
+
|
|
|
+ var allowed = ResolveAllowedJobCodes(input.BusinessDomainCode, consumerModule);
|
|
|
+ if (allowed.Count == 0)
|
|
|
+ return (string.Empty, pars, true);
|
|
|
+
|
|
|
+ if (allowed.Count == 1)
|
|
|
+ {
|
|
|
+ pars.Add(new SugarParameter("@JobCode", allowed.First()));
|
|
|
+ return ("job_code=@JobCode", pars, false);
|
|
|
+ }
|
|
|
+
|
|
|
+ var inParts = new List<string>();
|
|
|
+ var index = 0;
|
|
|
+ foreach (var code in allowed.OrderBy(u => u, StringComparer.OrdinalIgnoreCase))
|
|
|
+ {
|
|
|
+ var paramName = $"@JobCode{index++}";
|
|
|
+ inParts.Add(paramName);
|
|
|
+ pars.Add(new SugarParameter(paramName, code));
|
|
|
+ }
|
|
|
+
|
|
|
+ return ($"job_code IN ({string.Join(", ", inParts)})", pars, false);
|
|
|
+ }
|
|
|
+
|
|
|
+ private static HashSet<string> ResolveAllowedJobCodes(string? businessDomainCode, string? consumerModule)
|
|
|
+ {
|
|
|
+ IEnumerable<MdpJobCatalogItem> items = JobCatalog.Values;
|
|
|
+
|
|
|
+ if (!string.IsNullOrWhiteSpace(businessDomainCode))
|
|
|
+ {
|
|
|
+ var domain = businessDomainCode.Trim();
|
|
|
+ items = items.Where(u => string.Equals(u.BusinessDomainCode, domain, StringComparison.OrdinalIgnoreCase));
|
|
|
+ }
|
|
|
+
|
|
|
+ if (!string.IsNullOrWhiteSpace(consumerModule))
|
|
|
+ {
|
|
|
+ var module = consumerModule.Trim();
|
|
|
+ items = items.Where(u => u.ConsumerModules.Contains(module, StringComparer.OrdinalIgnoreCase));
|
|
|
+ }
|
|
|
+
|
|
|
+ return items.Select(u => u.JobCode).ToHashSet(StringComparer.OrdinalIgnoreCase);
|
|
|
+ }
|
|
|
+
|
|
|
+ private static void AttachCatalogInfo(MdpMonitorRunLogRow row)
|
|
|
+ {
|
|
|
+ if (row == null || string.IsNullOrWhiteSpace(row.JobCode))
|
|
|
+ return;
|
|
|
+
|
|
|
+ if (!JobCatalog.TryGetValue(row.JobCode.Trim(), out var item))
|
|
|
+ return;
|
|
|
+
|
|
|
+ row.BusinessDomainCode = item.BusinessDomainCode;
|
|
|
+ row.BusinessDomainName = item.BusinessDomainName;
|
|
|
+ row.ConsumerModules = string.Join(",", item.ConsumerModules);
|
|
|
+ row.ScopeType = item.ScopeType;
|
|
|
+ row.DisplayName = item.DisplayName;
|
|
|
+ row.ScheduleJobId = item.ScheduleJobId;
|
|
|
+ }
|
|
|
+
|
|
|
+ private static void AttachLineageCatalogInfo(MdpLineageOutput output)
|
|
|
+ {
|
|
|
+ if (output == null || string.IsNullOrWhiteSpace(output.JobCode))
|
|
|
+ return;
|
|
|
+
|
|
|
+ if (!JobCatalog.TryGetValue(output.JobCode.Trim(), out var item))
|
|
|
+ return;
|
|
|
+
|
|
|
+ output.BusinessDomainCode = item.BusinessDomainCode;
|
|
|
+ output.BusinessDomainName = item.BusinessDomainName;
|
|
|
+ output.ConsumerModules = string.Join(",", item.ConsumerModules);
|
|
|
+ output.ScopeType = item.ScopeType;
|
|
|
+ output.DisplayName = item.DisplayName;
|
|
|
+ output.ScheduleJobId = item.ScheduleJobId;
|
|
|
+ }
|
|
|
+
|
|
|
+ private static string FormatConsumerModulesLabel(MdpJobCatalogItem item) =>
|
|
|
+ string.Join("、", item.ConsumerModules);
|
|
|
+
|
|
|
+ private static List<MdpLineageStageRow> BuildStageDescriptions(string? jobCode, string? moduleCode)
|
|
|
+ {
|
|
|
+ if (!string.IsNullOrWhiteSpace(jobCode) && JobCatalog.TryGetValue(jobCode.Trim(), out var catalogItem))
|
|
|
+ {
|
|
|
+ return catalogItem.BusinessDomainCode switch
|
|
|
+ {
|
|
|
+ "order_delivery" => BuildOrderDeliveryStages(catalogItem),
|
|
|
+ "work_schedule" => BuildWorkScheduleStages(catalogItem),
|
|
|
+ "supply_purchase" => BuildSupplyPurchaseStages(catalogItem),
|
|
|
+ "purchase_execution" => BuildPurchaseExecutionStages(catalogItem),
|
|
|
+ _ => BuildGenericStages(moduleCode)
|
|
|
+ };
|
|
|
+ }
|
|
|
+
|
|
|
+ return BuildGenericStages(moduleCode);
|
|
|
+ }
|
|
|
+
|
|
|
+ private static List<MdpLineageStageRow> BuildOrderDeliveryStages(MdpJobCatalogItem item)
|
|
|
+ {
|
|
|
+ var consumers = FormatConsumerModulesLabel(item);
|
|
|
+ return new List<MdpLineageStageRow>
|
|
|
+ {
|
|
|
+ new() { StageCode = "STAGING", StageName = "订单交付域 · 贴源同步", Layer = "mdp_stg", Description = $"按 mdp_entity 登记订单交付域源对象抽取数据，保留 raw_data JSON 便于追溯；产出供 {consumers} 消费。", InputObjects = "旧系统 / 当前库源对象", OutputObjects = "mdp_stg_so, mdp_stg_ship_trans", Execution = "S1MdpSyncTransformService.SyncStagingAsync" },
|
|
|
+ new() { StageCode = "STANDARD", StageName = "订单交付域 · 标准层转换", Layer = "mdp_std", Description = "解析贴源 raw_data，做字段标准化、租户兜底和幂等写入。", InputObjects = "mdp_stg_so, mdp_stg_ship_trans", OutputObjects = "mdp_std_so, mdp_std_ship_trans", Execution = "S1MdpSyncTransformService.BuildStandardCommands" },
|
|
|
+ new() { StageCode = "DWD", StageName = "订单交付域 · DWD宽表", Layer = "dwd", Description = $"沉淀订单交付事实，供 {consumers} 看板与诊断读取。", InputObjects = "mdp_std_so, mdp_std_ship_trans", OutputObjects = "dwd_ship_trans", Execution = "S1MdpSyncTransformService.BuildDwdAsync" },
|
|
|
+ new() { StageCode = "KPI", StageName = "指标写入", Layer = "ado_s9", Description = $"计算订单交付域 L1 指标并写入统一指标值表，供 {consumers} 消费。", InputObjects = "mdp_std_so, dwd_ship_trans", OutputObjects = "ado_s9_kpi_value_l1_day", Execution = "S1MdpSyncTransformService.BuildS1KpiValuesAsync" }
|
|
|
+ };
|
|
|
+ }
|
|
|
+
|
|
|
+ private static List<MdpLineageStageRow> BuildWorkScheduleStages(MdpJobCatalogItem item)
|
|
|
+ {
|
|
|
+ var consumers = FormatConsumerModulesLabel(item);
|
|
|
+ return new List<MdpLineageStageRow>
|
|
|
+ {
|
|
|
+ new() { StageCode = "STAGING", StageName = "工单排程域 · 贴源同步", Layer = "mdp_stg", Description = $"按 mdp_entity 登记工单排程域源对象抽取数据；产出供 {consumers} 消费。", InputObjects = "工单 / 工序 / 排程源对象", OutputObjects = "mdp_stg_*", Execution = "S2MdpSyncTransformService.SyncStagingAsync" },
|
|
|
+ new() { StageCode = "STANDARD", StageName = "工单排程域 · 标准层转换", Layer = "mdp_std", Description = "将工单、工序、排程等对象标准化。", InputObjects = "mdp_stg_*", OutputObjects = "mdp_std_*", Execution = "S2MdpSyncTransformService.BuildStandardCommands" },
|
|
|
+ new() { StageCode = "DWD", StageName = "工单排程域 · DWD宽表", Layer = "dwd", Description = $"生成制造执行与排程分析宽表，供 {consumers} 读取。", InputObjects = "mdp_std_*", OutputObjects = "dwd_*", Execution = "S2MdpSyncTransformService.BuildDwdAsync" },
|
|
|
+ new() { StageCode = "KPI", StageName = "指标写入", Layer = "ado_s9", Description = $"写入工单排程域 KPI，供 {consumers} 消费。", InputObjects = "mdp_std_* / dwd_*", OutputObjects = "ado_s9_kpi_value_*", Execution = "S2MdpSyncTransformService.BuildS2KpiValuesAsync" }
|
|
|
+ };
|
|
|
+ }
|
|
|
+
|
|
|
+ private static List<MdpLineageStageRow> BuildSupplyPurchaseStages(MdpJobCatalogItem item)
|
|
|
+ {
|
|
|
+ var consumers = FormatConsumerModulesLabel(item);
|
|
|
+ return new List<MdpLineageStageRow>
|
|
|
+ {
|
|
|
+ new() { StageCode = "STAGING", StageName = "供应采购域 · 贴源同步", Layer = "mdp_stg", Description = $"按 mdp_entity 登记供应采购域源对象抽取数据；产出供 {consumers} 消费。", InputObjects = "供应 / 物料 / 采购源对象", OutputObjects = "mdp_stg_*", Execution = "S3MdpSyncTransformService.SyncStagingAsync" },
|
|
|
+ new() { StageCode = "STANDARD", StageName = "供应采购域 · 标准层转换", Layer = "mdp_std", Description = "将供应、物料、采购、交货计划等对象标准化。", InputObjects = "mdp_stg_*", OutputObjects = "mdp_std_*", Execution = "S3MdpSyncTransformService.BuildStandardCommands" },
|
|
|
+ new() { StageCode = "DWD", StageName = "供应采购域 · DWD宽表", Layer = "dwd", Description = $"生成供应交付、齐套、风险等分析宽表，供 {consumers} 读取。", InputObjects = "mdp_std_*", OutputObjects = "dwd_supplier_delivery / dwd_material_readiness 等", Execution = "S3MdpSyncTransformService.BuildDwdAsync" },
|
|
|
+ new() { StageCode = "KPI", StageName = "指标写入", Layer = "ado_s9", Description = $"写入供应采购域指标，供 {consumers} 消费。", InputObjects = "mdp_std_* / dwd_*", OutputObjects = "ado_s9_kpi_value_*", Execution = "S3MdpSyncTransformService.BuildS3KpiValuesAsync" }
|
|
|
+ };
|
|
|
+ }
|
|
|
+
|
|
|
+ private static List<MdpLineageStageRow> BuildPurchaseExecutionStages(MdpJobCatalogItem item)
|
|
|
+ {
|
|
|
+ var consumers = FormatConsumerModulesLabel(item);
|
|
|
+ return new List<MdpLineageStageRow>
|
|
|
+ {
|
|
|
+ new() { StageCode = "STAGING", StageName = "采购执行域 · 贴源同步", Layer = "mdp_stg_s4_*", Description = $"同步采购执行域 IQC/发货/退货/欠料事实，供 {consumers} 消费；共享采购主链仍由供应采购域维护。", InputObjects = "PurOrdRctDetail / scm_shdzb / srm_polist_ds / dwd_material_shortage", OutputObjects = "mdp_stg_s4_iqc / mdp_stg_s4_shipment / mdp_stg_s4_return / mdp_stg_s4_shortage", Execution = "S4MdpSyncTransformService.SyncStagingAsync" },
|
|
|
+ new() { StageCode = "STANDARD", StageName = "采购执行域 · 标准层转换", Layer = "mdp_std_s4_*", Description = "将采购执行域贴源对象标准化，并统计供应采购域共享标准层行数。", InputObjects = "mdp_stg_s4_*", OutputObjects = "mdp_std_s4_iqc / mdp_std_s4_shipment / mdp_std_s4_return / mdp_std_s4_shortage", Execution = "S4MdpSyncTransformService.BuildStandardCommands" },
|
|
|
+ new() { StageCode = "DWD", StageName = "采购执行域 · DWD宽表", Layer = "dwd", Description = $"写入采购执行分析宽表，供 {consumers} 读取。", InputObjects = "dwd_supplier_delivery / mdp_std_s4_*", OutputObjects = "dwd_s4_purchase_execution / dwd_po_trans / dwd_qc_trans", Execution = "S4MdpSyncTransformService.BuildDwdAsync" },
|
|
|
+ new() { StageCode = "KPI", StageName = "指标写入", Layer = "ado_s9", Description = $"写入采购执行域 L1/L2/L3 指标，供 {consumers} 消费。", InputObjects = "mdp_std_delivery_schedule / dwd_supplier_delivery / dwd_s4_purchase_execution", OutputObjects = "ado_s9_kpi_value_l1/l2/l3_day", Execution = "S4MdpSyncTransformService.BuildS4KpiValuesAsync" }
|
|
|
+ };
|
|
|
+ }
|
|
|
+
|
|
|
+ /// <summary>
|
|
|
+ /// 贴源同步在代码中维护、但未写入 mdp_field_mapping 的实体主键/业务键提示（仅监控展示兜底）。
|
|
|
+ /// </summary>
|
|
|
+ private static readonly Dictionary<string, (string SourceRowId, string SourceBizKeyExpr)> KnownStagingEntityKeys =
|
|
|
+ new(StringComparer.OrdinalIgnoreCase)
|
|
|
+ {
|
|
|
+ ["S4_IQC_RECEIPT"] = ("RecID", "CONCAT(IFNULL(s.`Domain`,''), ':', IFNULL(s.`Receiver`,''), ':', IFNULL(s.`Line`,''))"),
|
|
|
+ ["S4_SHIPMENT_EXEC"] = ("id", "CONCAT(IFNULL(s.`glid`,''), ':', IFNULL(s.`id`,''))"),
|
|
|
+ ["S4_RETURN_EXEC"] = ("Id", "s.`dsnum`"),
|
|
|
+ ["S4_SHORTAGE_EXEC"] = ("id", "CONCAT(IFNULL(s.`work_order`,''), ':', IFNULL(s.`component_item_code`,''))")
|
|
|
+ };
|
|
|
+
|
|
|
+ private static List<MdpLineageFieldMappingRow> BuildFallbackFieldMappings(MdpLineageEntityRow entity, string? batchId)
|
|
|
+ {
|
|
|
+ var sourceRowId = entity.IncrColumn;
|
|
|
+ string? sourceBizKeyExpr = null;
|
|
|
+ if (!string.IsNullOrWhiteSpace(entity.EntityCode) &&
|
|
|
+ KnownStagingEntityKeys.TryGetValue(entity.EntityCode, out var known))
|
|
|
+ {
|
|
|
+ sourceRowId ??= known.SourceRowId;
|
|
|
+ sourceBizKeyExpr = known.SourceBizKeyExpr;
|
|
|
+ }
|
|
|
+
|
|
|
+ var fallbackNote = "当前实体未配置逐字段映射，贴源同步保留源行 raw_data";
|
|
|
+ var mappings = new List<MdpLineageFieldMappingRow>();
|
|
|
+ var sort = 10;
|
|
|
+
|
|
|
+ void Add(string sourceField, string targetField, string fieldType, string? transformScript, string? constValue = null, bool isRequired = false)
|
|
|
+ {
|
|
|
+ mappings.Add(new MdpLineageFieldMappingRow
|
|
|
+ {
|
|
|
+ EntityId = entity.Id,
|
|
|
+ SourceField = sourceField,
|
|
|
+ TargetField = targetField,
|
|
|
+ FieldType = fieldType,
|
|
|
+ TransformScript = transformScript,
|
|
|
+ ConstValue = constValue,
|
|
|
+ IsRequired = isRequired,
|
|
|
+ SortOrder = sort,
|
|
|
+ MappingSource = "FALLBACK",
|
|
|
+ IsFallback = true
|
|
|
+ });
|
|
|
+ sort += 10;
|
|
|
+ }
|
|
|
+
|
|
|
+ Add("tenant_id", "tenant_id", "DIRECT", "当前租户或全局租户兜底");
|
|
|
+ Add($"CONST:{entity.SourceCode ?? "AIDOP"}", "source_system", "CONST", "来源数据源编码", entity.SourceCode ?? "AIDOP", isRequired: true);
|
|
|
+ Add($"CONST:{entity.SourceTableName ?? entity.SourceApiPath ?? "--"}", "source_table", "CONST", "源表名", entity.SourceTableName ?? entity.SourceApiPath, isRequired: true);
|
|
|
+
|
|
|
+ if (!string.IsNullOrWhiteSpace(sourceRowId))
|
|
|
+ Add(sourceRowId, "source_row_id", "DIRECT", "来自 mdp_entity.incr_column 或实体主键配置");
|
|
|
+ else
|
|
|
+ Add("--", "source_row_id", "DIRECT", fallbackNote);
|
|
|
+
|
|
|
+ if (!string.IsNullOrWhiteSpace(sourceBizKeyExpr))
|
|
|
+ Add(sourceBizKeyExpr, "source_biz_key", "EXPR", "来自实体业务键表达式", isRequired: true);
|
|
|
+ else
|
|
|
+ Add("--", "source_biz_key", "EXPR", fallbackNote);
|
|
|
+
|
|
|
+ Add("*", "raw_data", "JSON", "源行整行 JSON");
|
|
|
+ Add("CONST", "sync_batch_id", "CONST", "当前同步批次", string.IsNullOrWhiteSpace(batchId) ? "当前同步批次" : batchId, isRequired: true);
|
|
|
+ Add("NOW()", "sync_time", "CONST", "同步时间", isRequired: true);
|
|
|
+
|
|
|
+ return mappings;
|
|
|
+ }
|
|
|
+
|
|
|
+ private static List<MdpLineageStageRow> BuildGenericStages(string? moduleCode)
|
|
|
+ {
|
|
|
+ if (string.Equals(moduleCode, "S1", StringComparison.OrdinalIgnoreCase))
|
|
|
+ return BuildOrderDeliveryStages(JobCatalog["S1_MDP_SYNC_TRANSFORM"]);
|
|
|
+ if (string.Equals(moduleCode, "S2", StringComparison.OrdinalIgnoreCase))
|
|
|
+ return BuildWorkScheduleStages(JobCatalog["S2_MDP_SYNC_TRANSFORM"]);
|
|
|
+ if (string.Equals(moduleCode, "S3", StringComparison.OrdinalIgnoreCase))
|
|
|
+ return BuildSupplyPurchaseStages(JobCatalog["S3_MDP_SYNC_TRANSFORM"]);
|
|
|
+ if (string.Equals(moduleCode, "S4", StringComparison.OrdinalIgnoreCase))
|
|
|
+ return BuildPurchaseExecutionStages(JobCatalog["S4_MDP_SYNC_TRANSFORM"]);
|
|
|
+
|
|
|
+ return new List<MdpLineageStageRow>
|
|
|
+ {
|
|
|
+ new() { StageCode = "STAGING", StageName = "贴源同步", Layer = "mdp_stg", Description = "按 mdp_entity 登记源对象抽取数据。", InputObjects = "源对象", OutputObjects = "mdp_stg_*", Execution = "MDP 同步服务" },
|
|
|
+ new() { StageCode = "STANDARD", StageName = "标准层转换", Layer = "mdp_std", Description = "标准层/DWD/KPI 当前由后端 Service 承载。", InputObjects = "mdp_stg_*", OutputObjects = "mdp_std_* / dwd_* / 指标表", Execution = "MDP 转换服务" }
|
|
|
+ };
|
|
|
+ }
|
|
|
+
|
|
|
+ private static string? ResolveModuleCodeFromJobCode(string? jobCode)
|
|
|
+ {
|
|
|
+ if (string.IsNullOrWhiteSpace(jobCode))
|
|
|
+ return null;
|
|
|
+
|
|
|
+ return ModuleJobCodes.FirstOrDefault(u =>
|
|
|
+ string.Equals(u.Value, jobCode.Trim(), StringComparison.OrdinalIgnoreCase)).Key;
|
|
|
+ }
|
|
|
+
|
|
|
private static string? ResolveJobCode(string? moduleCode, string? jobCode)
|
|
|
{
|
|
|
if (!string.IsNullOrWhiteSpace(jobCode))
|
|
|
@@ -271,48 +614,6 @@ public class MdpMonitorService : IDynamicApiController, ITransient
|
|
|
return string.Join(" / ", new[] { dbType, hostPart, databaseObject }.Where(u => !string.IsNullOrWhiteSpace(u)));
|
|
|
}
|
|
|
|
|
|
- private static List<MdpLineageStageRow> BuildStageDescriptions(string moduleCode)
|
|
|
- {
|
|
|
- if (string.Equals(moduleCode, "S1", StringComparison.OrdinalIgnoreCase))
|
|
|
- {
|
|
|
- return new List<MdpLineageStageRow>
|
|
|
- {
|
|
|
- new() { StageCode = "STAGING", StageName = "贴源同步", Layer = "mdp_stg", Description = "按 mdp_entity 登记的 S1 源对象抽取数据，保留 raw_data JSON 便于追溯。", InputObjects = "旧系统 / 当前库源对象", OutputObjects = "mdp_stg_so, mdp_stg_ship_trans", Execution = "S1MdpSyncTransformService.SyncStagingAsync" },
|
|
|
- new() { StageCode = "STANDARD", StageName = "标准层转换", Layer = "mdp_std", Description = "解析贴源 raw_data，做字段标准化、租户兜底和幂等写入。", InputObjects = "mdp_stg_so, mdp_stg_ship_trans", OutputObjects = "mdp_std_so, mdp_std_ship_trans", Execution = "S1MdpSyncTransformService.BuildStandardCommands" },
|
|
|
- new() { StageCode = "DWD", StageName = "DWD宽表", Layer = "dwd", Description = "沉淀 S1 订单交付事实，供订单交付、看板和诊断读取。", InputObjects = "mdp_std_so, mdp_std_ship_trans", OutputObjects = "dwd_ship_trans", Execution = "S1MdpSyncTransformService.BuildDwdAsync" },
|
|
|
- new() { StageCode = "KPI", StageName = "指标写入", Layer = "ado_s9", Description = "计算 S1 L1 指标并写入统一指标值表。", InputObjects = "mdp_std_so, dwd_ship_trans", OutputObjects = "ado_s9_kpi_value_l1_day", Execution = "S1MdpSyncTransformService.BuildS1KpiValuesAsync" }
|
|
|
- };
|
|
|
- }
|
|
|
-
|
|
|
- if (string.Equals(moduleCode, "S3", StringComparison.OrdinalIgnoreCase))
|
|
|
- {
|
|
|
- return new List<MdpLineageStageRow>
|
|
|
- {
|
|
|
- new() { StageCode = "STAGING", StageName = "贴源同步", Layer = "mdp_stg", Description = "按 mdp_entity 登记的 S3 源对象抽取数据，保留 raw_data JSON 便于追溯。", InputObjects = "S3 源对象", OutputObjects = "mdp_stg_*", Execution = "S3MdpSyncTransformService.SyncStagingAsync" },
|
|
|
- new() { StageCode = "STANDARD", StageName = "标准层转换", Layer = "mdp_std", Description = "将供应、物料、采购、交付等对象标准化。", InputObjects = "mdp_stg_*", OutputObjects = "mdp_std_*", Execution = "S3MdpSyncTransformService.BuildStandardCommands" },
|
|
|
- new() { StageCode = "DWD", StageName = "DWD宽表", Layer = "dwd", Description = "生成供应交付、齐套、风险等分析宽表。", InputObjects = "mdp_std_*", OutputObjects = "dwd_supplier_delivery / dwd_material_readiness 等", Execution = "S3MdpSyncTransformService.BuildDwdAsync" },
|
|
|
- new() { StageCode = "KPI", StageName = "指标写入", Layer = "ado_s9", Description = "写入 S3 供应协同指标。", InputObjects = "mdp_std_* / dwd_*", OutputObjects = "ado_s9_kpi_value_*", Execution = "S3MdpSyncTransformService.BuildS3KpiValuesAsync" }
|
|
|
- };
|
|
|
- }
|
|
|
-
|
|
|
- if (string.Equals(moduleCode, "S4", StringComparison.OrdinalIgnoreCase))
|
|
|
- {
|
|
|
- return new List<MdpLineageStageRow>
|
|
|
- {
|
|
|
- new() { StageCode = "STAGING", StageName = "S4 贴源同步", Layer = "mdp_stg_s4_*", Description = "S4 专属 IQC/发货/退货/欠料贴源；共享采购主链仍由 S3 维护，S4 同时统计共享贴源行数。", InputObjects = "PurOrdRctDetail / scm_shdzb / srm_polist_ds / dwd_material_shortage", OutputObjects = "mdp_stg_s4_iqc / mdp_stg_s4_shipment / mdp_stg_s4_return / mdp_stg_s4_shortage", Execution = "S4MdpSyncTransformService.SyncStagingAsync" },
|
|
|
- new() { StageCode = "STANDARD", StageName = "S4 标准层转换", Layer = "mdp_std_s4_*", Description = "将 S4 贴源对象标准化，并统计 S3 共享标准层行数。", InputObjects = "mdp_stg_s4_*", OutputObjects = "mdp_std_s4_iqc / mdp_std_s4_shipment / mdp_std_s4_return / mdp_std_s4_shortage", Execution = "S4MdpSyncTransformService.BuildStandardCommands" },
|
|
|
- new() { StageCode = "DWD", StageName = "采购执行宽表", Layer = "dwd", Description = "写入 dwd_s4_purchase_execution、dwd_po_trans（扩展字段）、dwd_qc_trans。", InputObjects = "dwd_supplier_delivery / mdp_std_s4_*", OutputObjects = "dwd_s4_purchase_execution / dwd_po_trans / dwd_qc_trans", Execution = "S4MdpSyncTransformService.BuildDwdAsync" },
|
|
|
- new() { StageCode = "KPI", StageName = "指标写入", Layer = "ado_s9", Description = "写入 S4 采购执行 L1/L2/L3 指标（S4_L1_001~004、L2/L3 全量）。", InputObjects = "mdp_std_delivery_schedule / dwd_supplier_delivery / dwd_s4_purchase_execution", OutputObjects = "ado_s9_kpi_value_l1/l2/l3_day", Execution = "S4MdpSyncTransformService.BuildS4KpiValuesAsync" }
|
|
|
- };
|
|
|
- }
|
|
|
-
|
|
|
- return new List<MdpLineageStageRow>
|
|
|
- {
|
|
|
- new() { StageCode = "STAGING", StageName = "贴源同步", Layer = "mdp_stg", Description = "按 mdp_entity 登记源对象抽取数据。", InputObjects = "源对象", OutputObjects = "mdp_stg_*", Execution = "模块 MDP 同步服务" },
|
|
|
- new() { StageCode = "STANDARD", StageName = "标准层转换", Layer = "mdp_std", Description = "标准层/DWD/KPI 当前由模块后端 Service 承载。", InputObjects = "mdp_stg_*", OutputObjects = "mdp_std_* / dwd_* / 指标表", Execution = "模块 MDP 转换服务" }
|
|
|
- };
|
|
|
- }
|
|
|
-
|
|
|
private static string SelectColumnsSql()
|
|
|
{
|
|
|
return """
|
|
|
@@ -324,8 +625,19 @@ public class MdpMonitorService : IDynamicApiController, ITransient
|
|
|
}
|
|
|
}
|
|
|
|
|
|
+public sealed record MdpJobCatalogItem(
|
|
|
+ string JobCode,
|
|
|
+ string BusinessDomainCode,
|
|
|
+ string BusinessDomainName,
|
|
|
+ string[] ConsumerModules,
|
|
|
+ string ScopeType,
|
|
|
+ string DisplayName,
|
|
|
+ string? ScheduleJobId);
|
|
|
+
|
|
|
public class MdpMonitorQueryInput
|
|
|
{
|
|
|
+ public string? BusinessDomainCode { get; set; }
|
|
|
+ public string? ConsumerModule { get; set; }
|
|
|
public string? ModuleCode { get; set; }
|
|
|
public string? JobCode { get; set; }
|
|
|
public string? BatchId { get; set; }
|
|
|
@@ -363,6 +675,12 @@ public sealed class MdpMonitorRunLogRow
|
|
|
public string? SummaryJson { get; set; }
|
|
|
public DateTime? CreateTime { get; set; }
|
|
|
public DateTime? UpdateTime { get; set; }
|
|
|
+ public string? BusinessDomainCode { get; set; }
|
|
|
+ public string? BusinessDomainName { get; set; }
|
|
|
+ public string? ConsumerModules { get; set; }
|
|
|
+ public string? ScopeType { get; set; }
|
|
|
+ public string? DisplayName { get; set; }
|
|
|
+ public string? ScheduleJobId { get; set; }
|
|
|
}
|
|
|
|
|
|
public sealed class MdpLineageOutput
|
|
|
@@ -370,6 +688,12 @@ public sealed class MdpLineageOutput
|
|
|
public string? ModuleCode { get; set; }
|
|
|
public string? JobCode { get; set; }
|
|
|
public string? BatchId { get; set; }
|
|
|
+ public string? BusinessDomainCode { get; set; }
|
|
|
+ public string? BusinessDomainName { get; set; }
|
|
|
+ public string? ConsumerModules { get; set; }
|
|
|
+ public string? ScopeType { get; set; }
|
|
|
+ public string? DisplayName { get; set; }
|
|
|
+ public string? ScheduleJobId { get; set; }
|
|
|
public List<MdpLineageStageRow> Stages { get; set; } = new();
|
|
|
public List<MdpLineageEntityRow> Entities { get; set; } = new();
|
|
|
}
|
|
|
@@ -427,6 +751,8 @@ public sealed class MdpLineageFieldMappingRow
|
|
|
public bool IsRequired { get; set; }
|
|
|
public string? DefaultValue { get; set; }
|
|
|
public int SortOrder { get; set; }
|
|
|
+ public string? MappingSource { get; set; }
|
|
|
+ public bool IsFallback { get; set; }
|
|
|
}
|
|
|
|
|
|
public sealed class MdpLineageSyncLogRow
|
skygu