using System.Text.Json;
using Admin.NET.Plugin.AiDOP.DataPlatform.Executors;
namespace Admin.NET.Plugin.AiDOP.Manufacturing;
///
/// S6 报工双模式入站:执行器 → mdp_stg_s6_report → mdp_std_s6_report。
/// 源优先 T8 Cj_Bg_Head_Rep(实体 S6_REPORT);API 对偶 S6_REPORT_API。
///
public class ReportWorkMdpSyncService : ITransient
{
private const string InboundEntityCode = "S6_REPORT";
private readonly ISqlSugarClient _db;
private readonly MdpSourcePullDispatcher _pullDispatcher;
public ReportWorkMdpSyncService(ISqlSugarClient db, MdpSourcePullDispatcher pullDispatcher)
{
_db = db;
_pullDispatcher = pullDispatcher;
}
public async Task RunInboundAsync(
long tenantId = 0,
bool fullRefresh = false,
string? entityCode = null,
CancellationToken cancellationToken = default)
{
cancellationToken.ThrowIfCancellationRequested();
await EnsureTablesAsync();
var code = string.IsNullOrWhiteSpace(entityCode) ? InboundEntityCode : entityCode.Trim();
var now = DateTime.Now;
var pullCtx = new MdpPullContext
{
TenantId = tenantId,
FullRefresh = fullRefresh,
TaskCode = "S6_REPORT_INBOUND",
BatchId = $"S6_RPT_IN_{now:yyyyMMddHHmmss}"
};
var pull = await _pullDispatcher.PullByEntityCodeAsync(code, pullCtx, cancellationToken);
var transformBatch = $"{pullCtx.BatchId}_STD";
var stdRows = await TransformStandardAsync(tenantId, transformBatch, now);
return new ReportWorkInboundResult
{
PullBatchId = pullCtx.BatchId,
RowsPulled = pull.RowsWritten,
RowsWrittenStg = pull.RowsWritten,
TransformBatchId = transformBatch,
StandardRows = stdRows
};
}
private async Task TransformStandardAsync(long tenantId, string batchId, DateTime now)
{
// 将 PENDING stg 投影到 std(JSON 字段兼容大小写)
const string sql = """
INSERT INTO mdp_std_s6_report
(tenant_id, factory_id, source_system, work_order_no, report_date, report_qty, ztid,
source_row_id, source_biz_key, sync_batch_id, sync_time)
SELECT
IFNULL(s.tenant_id, @tid),
1,
IFNULL(NULLIF(s.source_system,''), 'T8'),
IFNULL(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(s.raw_data, '$.noid')), 'null'),
IFNULL(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(s.raw_data, '$.NOID')), 'null'), s.source_biz_key)),
COALESCE(
STR_TO_DATE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(s.raw_data, '$.kgdate')), 'null'), ''), '%Y-%m-%d %H:%i:%s'),
STR_TO_DATE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(s.raw_data, '$.KGDATE')), 'null'), ''), '%Y-%m-%d %H:%i:%s'),
NULL),
CAST(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(s.raw_data, '$.sl')), 'null'), '') AS DECIMAL(18,6)),
NULLIF(JSON_UNQUOTE(JSON_EXTRACT(s.raw_data, '$.ztid')), 'null'),
IFNULL(NULLIF(s.source_row_id,''), s.source_biz_key),
s.source_biz_key,
@batch,
@now
FROM mdp_stg_s6_report s
WHERE s.process_status = 'PENDING'
AND s.source_biz_key IS NOT NULL
AND s.source_biz_key <> ''
ON DUPLICATE KEY UPDATE
work_order_no = VALUES(work_order_no),
report_date = VALUES(report_date),
report_qty = VALUES(report_qty),
ztid = VALUES(ztid),
source_row_id = VALUES(source_row_id),
sync_batch_id = VALUES(sync_batch_id),
sync_time = VALUES(sync_time),
update_time = CURRENT_TIMESTAMP
""";
var affected = await _db.Ado.ExecuteCommandAsync(sql,
new SugarParameter("@tid", tenantId),
new SugarParameter("@batch", batchId),
new SugarParameter("@now", now));
await _db.Ado.ExecuteCommandAsync(
"UPDATE mdp_stg_s6_report SET process_status='DONE', update_time=NOW() WHERE process_status='PENDING'");
return affected;
}
private async Task EnsureTablesAsync()
{
await _db.Ado.ExecuteCommandAsync("""
CREATE TABLE IF NOT EXISTS mdp_stg_s6_report (
id bigint NOT NULL AUTO_INCREMENT,
tenant_id bigint NOT NULL DEFAULT 0,
source_system varchar(50) DEFAULT NULL,
source_table varchar(200) DEFAULT NULL,
source_row_id varchar(200) DEFAULT NULL,
source_biz_key varchar(300) DEFAULT NULL,
raw_data json DEFAULT NULL,
sync_batch_id varchar(100) DEFAULT NULL,
create_time datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,
sync_time datetime DEFAULT CURRENT_TIMESTAMP,
process_status varchar(20) NOT NULL DEFAULT 'PENDING',
process_message varchar(500) DEFAULT NULL,
update_time datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (id),
UNIQUE KEY uk_source_key (source_system, source_table, source_biz_key)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
""");
await _db.Ado.ExecuteCommandAsync("""
CREATE TABLE IF NOT EXISTS mdp_std_s6_report (
id bigint NOT NULL AUTO_INCREMENT,
tenant_id bigint NOT NULL DEFAULT 0,
factory_id bigint DEFAULT 1,
source_system varchar(50) NOT NULL DEFAULT 'T8',
work_order_no varchar(100) NOT NULL,
report_date datetime DEFAULT NULL,
report_qty decimal(18,6) DEFAULT NULL,
ztid varchar(50) DEFAULT NULL,
source_row_id varchar(100) NOT NULL,
source_biz_key varchar(200) NOT NULL,
sync_batch_id varchar(100) NOT NULL,
sync_time datetime NOT NULL,
update_time datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (id),
UNIQUE KEY uk_mdp_std_s6_report (tenant_id, source_system, source_biz_key)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
""");
}
}
public class ReportWorkInboundResult
{
public string PullBatchId { get; set; } = string.Empty;
public int RowsPulled { get; set; }
public int RowsWrittenStg { get; set; }
public string TransformBatchId { get; set; } = string.Empty;
public int StandardRows { get; set; }
}