|
|
@@ -0,0 +1,281 @@
|
|
|
+namespace Admin.NET.Plugin.AiDOP.Manufacturing;
|
|
|
+
|
|
|
+/// <summary>
|
|
|
+/// S6 过程检验单(IPQC)数据中台只读同步转换服务(DOP 内部,独立于 S6MdpSyncTransformService 的 KPI 管线)。
|
|
|
+///
|
|
|
+/// 源:aidopdev.qms_gcjyd(过程检验单头,归口自旧系统 dopdemo,31 列)。
|
|
|
+/// 链路:qms_gcjyd 只读 → mdp_stg_ipqc_inspection(raw_data JSON) → mdp_std_ipqc_inspection(typed)。
|
|
|
+///
|
|
|
+/// 归口说明(CTO 拍板方案 A):
|
|
|
+/// qms_gcjyd 原仅存于旧系统 dopdemo(0 行)、aidopdev 无此表,且 dopdemo 非运行时连接。
|
|
|
+/// 本服务 EnsureTablesAsync 在 aidopdev 建同构空表 qms_gcjyd(IF NOT EXISTS),
|
|
|
+/// 作为归口源;后续由上游一次性抽数填充,届时无需改代码即随刷新流通。
|
|
|
+///
|
|
|
+/// 约束:
|
|
|
+/// - 只读 qms_gcjyd,仅写 mdp_stg_*/mdp_std_*/run_log;绝不 INSERT/UPDATE/DELETE 业务源单据表。
|
|
|
+/// - qms_gcjyd 无 tenant/org/factory 列 → 标准层按常量 tenant_id=0, factory_id=1 落库。
|
|
|
+/// - 本批只读列表:只做头表转换,不碰明细 qms_gcjydzb(留待 DETAIL 批)。
|
|
|
+/// - result_judgement(jgpd)为字典码,原样透出,语义解码后置。
|
|
|
+/// - 源为 0 行时转换成功完成、处理数为 0,不报错。
|
|
|
+/// </summary>
|
|
|
+public class IpqcInspectionMdpSyncService : ITransient
|
|
|
+{
|
|
|
+ private const string JobCode = "S6_IPQC_INSPECTION_MDP_SYNC";
|
|
|
+
|
|
|
+ private readonly ISqlSugarClient _db;
|
|
|
+
|
|
|
+ public IpqcInspectionMdpSyncService(ISqlSugarClient db)
|
|
|
+ {
|
|
|
+ _db = db;
|
|
|
+ }
|
|
|
+
|
|
|
+ /// <summary>
|
|
|
+ /// 全量同步转换:贴源(头) → 标准化(头)。
|
|
|
+ /// </summary>
|
|
|
+ public async Task<IpqcInspectionMdpSyncResult> RunFullAsync(CancellationToken cancellationToken = default, string triggerType = "AUTO")
|
|
|
+ {
|
|
|
+ cancellationToken.ThrowIfCancellationRequested();
|
|
|
+ await EnsureTablesAsync();
|
|
|
+
|
|
|
+ var now = DateTime.Now;
|
|
|
+ var batchId = $"S6_IPQC_INSP_FULL_{now:yyyyMMddHHmmss}";
|
|
|
+ var runLogId = await InsertRunLogAsync(batchId, now, triggerType);
|
|
|
+ var result = new IpqcInspectionMdpSyncResult { BatchId = batchId, RunLogId = runLogId };
|
|
|
+
|
|
|
+ try
|
|
|
+ {
|
|
|
+ result.HeadStageRows = await SyncHeadStagingAsync(batchId, now);
|
|
|
+ result.HeadStandardRows = await TransformHeadStandardAsync(batchId, now);
|
|
|
+ await MarkRunSuccessAsync(runLogId, now, result);
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+ catch (Exception ex)
|
|
|
+ {
|
|
|
+ await MarkRunFailedAsync(runLogId, now, ex.Message);
|
|
|
+ throw;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /// <summary>防御式建表(幂等):归口源空表 qms_gcjyd + 贴源层 + 标准层。</summary>
|
|
|
+ private async Task EnsureTablesAsync()
|
|
|
+ {
|
|
|
+ // 归口源表(方案 A):同构自旧系统 dopdemo.qms_gcjyd 头表,空表,待上游抽数填充。
|
|
|
+ // id 不设 AUTO_INCREMENT:抽数将携带旧系统主键。
|
|
|
+ await _db.Ado.ExecuteCommandAsync(
|
|
|
+ """
|
|
|
+ CREATE TABLE IF NOT EXISTS qms_gcjyd (
|
|
|
+ id BIGINT NOT NULL PRIMARY KEY COMMENT '主表id',
|
|
|
+ djbh VARCHAR(255) NULL COMMENT '单据编号',
|
|
|
+ cplx VARCHAR(255) NULL COMMENT '产品型号',
|
|
|
+ scph VARCHAR(255) NULL COMMENT '生产批号',
|
|
|
+ jyrq VARCHAR(255) NULL COMMENT '检验日期',
|
|
|
+ lydjbh VARCHAR(255) NULL COMMENT '来源单据编号',
|
|
|
+ jgpd BIGINT NULL COMMENT '结果判定',
|
|
|
+ jyhgsl DECIMAL(10,0) NULL COMMENT '检验合格数量',
|
|
|
+ jybhgsl DECIMAL(10,0) NULL COMMENT '检验不合格数量',
|
|
|
+ fj TEXT NULL COMMENT '附件',
|
|
|
+ bz TEXT NULL COMMENT '备注',
|
|
|
+ jyr VARCHAR(255) NULL COMMENT '检验人',
|
|
|
+ gxbm VARCHAR(255) NULL COMMENT '工序编码',
|
|
|
+ gxmc VARCHAR(255) NULL COMMENT '工序名称',
|
|
|
+ sczyry VARCHAR(255) NULL COMMENT '生产作业人员',
|
|
|
+ ybl DECIMAL(10,0) NULL COMMENT '样本量',
|
|
|
+ bdbh VARCHAR(255) NULL COMMENT '表单编号',
|
|
|
+ bbh VARCHAR(255) NULL COMMENT '版本号',
|
|
|
+ sxrq DATETIME NULL COMMENT '生效日期',
|
|
|
+ wlbm VARCHAR(255) NULL COMMENT '物料编码',
|
|
|
+ wlmc VARCHAR(255) NULL COMMENT '物料名称',
|
|
|
+ lysl DECIMAL(10,0) NULL COMMENT '留样数量',
|
|
|
+ phsl DECIMAL(10,0) NULL COMMENT '破坏数量',
|
|
|
+ jygfid VARCHAR(255) NULL COMMENT '检验规范',
|
|
|
+ jygfbbm VARCHAR(255) NULL COMMENT '没有用',
|
|
|
+ jgbb VARCHAR(255) NULL COMMENT '检规版本',
|
|
|
+ jgbh VARCHAR(255) NULL COMMENT '检规编号',
|
|
|
+ title VARCHAR(255) NULL,
|
|
|
+ jyr1 VARCHAR(255) NULL,
|
|
|
+ ChangeStatus INT NULL,
|
|
|
+ status VARCHAR(255) NULL COMMENT '检验状态'
|
|
|
+ ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='过程检验单(头) 归口自旧系统dopdemo,空表待上游抽数'
|
|
|
+ """);
|
|
|
+ await _db.Ado.ExecuteCommandAsync(
|
|
|
+ """
|
|
|
+ CREATE TABLE IF NOT EXISTS mdp_stg_ipqc_inspection (
|
|
|
+ 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',
|
|
|
+ source_table VARCHAR(100) NOT NULL,
|
|
|
+ source_row_id VARCHAR(100) NOT NULL,
|
|
|
+ source_biz_key VARCHAR(200) NULL,
|
|
|
+ sync_batch_id VARCHAR(100) NOT NULL,
|
|
|
+ sync_time DATETIME NOT NULL,
|
|
|
+ process_status VARCHAR(20) NOT NULL DEFAULT 'PENDING',
|
|
|
+ raw_data JSON NOT NULL,
|
|
|
+ update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
|
|
+ UNIQUE KEY uk_mdp_stg_ipqc_insp (tenant_id, source_table, source_row_id)
|
|
|
+ ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='S6过程检验单头贴源层'
|
|
|
+ """);
|
|
|
+ await _db.Ado.ExecuteCommandAsync(
|
|
|
+ """
|
|
|
+ CREATE TABLE IF NOT EXISTS mdp_std_ipqc_inspection (
|
|
|
+ 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',
|
|
|
+ bill_no VARCHAR(64) NOT NULL,
|
|
|
+ product_model VARCHAR(255) NULL,
|
|
|
+ production_batch_no VARCHAR(255) NULL,
|
|
|
+ production_work_order VARCHAR(255) NULL,
|
|
|
+ result_judgement VARCHAR(32) NULL,
|
|
|
+ attachment TEXT NULL,
|
|
|
+ remark TEXT NULL,
|
|
|
+ inspector VARCHAR(255) NULL,
|
|
|
+ process_code VARCHAR(255) NULL,
|
|
|
+ process_name VARCHAR(255) NULL,
|
|
|
+ production_person VARCHAR(255) NULL,
|
|
|
+ sample_qty DECIMAL(18,6) NULL,
|
|
|
+ form_no VARCHAR(255) NULL,
|
|
|
+ version_no VARCHAR(255) NULL,
|
|
|
+ effective_date DATETIME NULL,
|
|
|
+ material_code VARCHAR(255) NULL,
|
|
|
+ material_name VARCHAR(255) NULL,
|
|
|
+ inspec_standard_version VARCHAR(255) NULL,
|
|
|
+ inspec_standard_code VARCHAR(255) NULL,
|
|
|
+ inspection_status VARCHAR(255) NULL,
|
|
|
+ source_row_id VARCHAR(100) NOT NULL,
|
|
|
+ source_biz_key VARCHAR(200) NULL,
|
|
|
+ sync_batch_id VARCHAR(100) NOT NULL,
|
|
|
+ sync_time DATETIME NOT NULL,
|
|
|
+ update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
|
|
+ UNIQUE KEY uk_mdp_std_ipqc_insp (tenant_id, source_row_id),
|
|
|
+ KEY idx_mdp_std_ipqc_insp_bill (tenant_id, bill_no),
|
|
|
+ KEY idx_mdp_std_ipqc_insp_effdate (tenant_id, effective_date)
|
|
|
+ ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='S6过程检验单头标准层'
|
|
|
+ """);
|
|
|
+ }
|
|
|
+
|
|
|
+ /// <summary>贴源头:qms_gcjyd -> mdp_stg_ipqc_inspection(raw_data JSON 快照)。返回源头行数。</summary>
|
|
|
+ private async Task<int> SyncHeadStagingAsync(string batchId, DateTime now)
|
|
|
+ {
|
|
|
+ var rows = await _db.Ado.GetIntAsync("SELECT COUNT(1) FROM qms_gcjyd");
|
|
|
+ await _db.Ado.ExecuteCommandAsync(
|
|
|
+ """
|
|
|
+ INSERT INTO mdp_stg_ipqc_inspection
|
|
|
+ (tenant_id, factory_id, source_system, source_table, source_row_id, source_biz_key, sync_batch_id, sync_time, process_status, raw_data)
|
|
|
+ SELECT
|
|
|
+ 0, 1, 'AIDOP', 'qms_gcjyd',
|
|
|
+ CAST(m.id AS CHAR), CAST(m.djbh AS CHAR), @BatchId, @Now, 'PENDING',
|
|
|
+ JSON_OBJECT(
|
|
|
+ 'id', m.id, 'djbh', m.djbh, 'cplx', m.cplx, 'scph', m.scph, 'lydjbh', m.lydjbh,
|
|
|
+ 'jgpd', m.jgpd, 'fj', CAST(m.fj AS CHAR), 'bz', CAST(m.bz AS CHAR), 'jyr', m.jyr,
|
|
|
+ 'gxbm', m.gxbm, 'gxmc', m.gxmc, 'sczyry', m.sczyry, 'ybl', m.ybl,
|
|
|
+ 'bdbh', m.bdbh, 'bbh', m.bbh, 'sxrq', m.sxrq, 'wlbm', m.wlbm, 'wlmc', m.wlmc,
|
|
|
+ 'jgbb', m.jgbb, 'jgbh', m.jgbh, 'status', m.status
|
|
|
+ )
|
|
|
+ FROM qms_gcjyd m
|
|
|
+ ON DUPLICATE KEY UPDATE
|
|
|
+ source_biz_key=VALUES(source_biz_key), sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time),
|
|
|
+ process_status=VALUES(process_status), raw_data=VALUES(raw_data), update_time=CURRENT_TIMESTAMP
|
|
|
+ """,
|
|
|
+ new SugarParameter("@BatchId", batchId),
|
|
|
+ new SugarParameter("@Now", now));
|
|
|
+ return rows;
|
|
|
+ }
|
|
|
+
|
|
|
+ /// <summary>标准化头:qms_gcjyd -> mdp_std_ipqc_inspection(typed)。返回处理头行数。</summary>
|
|
|
+ private async Task<int> TransformHeadStandardAsync(string batchId, DateTime now)
|
|
|
+ {
|
|
|
+ var rows = await _db.Ado.GetIntAsync("SELECT COUNT(1) FROM qms_gcjyd");
|
|
|
+ await _db.Ado.ExecuteCommandAsync(
|
|
|
+ """
|
|
|
+ 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,
|
|
|
+ sample_qty, form_no, version_no, effective_date, material_code, material_name,
|
|
|
+ inspec_standard_version, inspec_standard_code, inspection_status,
|
|
|
+ source_row_id, source_biz_key, sync_batch_id, sync_time)
|
|
|
+ SELECT
|
|
|
+ 0, 1, 'AIDOP', IFNULL(m.djbh, CAST(m.id AS CHAR)), m.cplx, m.scph, m.lydjbh,
|
|
|
+ CAST(m.jgpd AS CHAR), CAST(m.fj AS CHAR), CAST(m.bz AS CHAR), m.jyr, m.gxbm, m.gxmc, m.sczyry,
|
|
|
+ m.ybl, m.bdbh, m.bbh, m.sxrq, m.wlbm, m.wlmc,
|
|
|
+ m.jgbb, m.jgbh, m.status,
|
|
|
+ CAST(m.id AS CHAR), CAST(m.djbh AS CHAR), @BatchId, @Now
|
|
|
+ FROM qms_gcjyd m
|
|
|
+ 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),
|
|
|
+ attachment=VALUES(attachment), remark=VALUES(remark), inspector=VALUES(inspector),
|
|
|
+ process_code=VALUES(process_code), process_name=VALUES(process_name), production_person=VALUES(production_person),
|
|
|
+ sample_qty=VALUES(sample_qty), form_no=VALUES(form_no), version_no=VALUES(version_no),
|
|
|
+ effective_date=VALUES(effective_date), material_code=VALUES(material_code), material_name=VALUES(material_name),
|
|
|
+ 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));
|
|
|
+ return rows;
|
|
|
+ }
|
|
|
+
|
|
|
+ private async Task<long> 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, 'S6过程检验单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<SugarParameter> { new("@BatchId", batchId) });
|
|
|
+ }
|
|
|
+
|
|
|
+ private async Task MarkRunSuccessAsync(long runLogId, DateTime startedAt, IpqcInspectionMdpSyncResult 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=@StageRows, 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("@StageRows", result.HeadStageRows),
|
|
|
+ new SugarParameter("@StandardRows", result.HeadStandardRows),
|
|
|
+ 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();
|
|
|
+}
|
|
|
+
|
|
|
+/// <summary>过程检验单 MDP 同步转换结果。</summary>
|
|
|
+public sealed class IpqcInspectionMdpSyncResult
|
|
|
+{
|
|
|
+ public long RunLogId { get; set; }
|
|
|
+ public string BatchId { get; set; } = string.Empty;
|
|
|
+ public int HeadStageRows { get; set; }
|
|
|
+ public int HeadStandardRows { get; set; }
|
|
|
+}
|