MdpDbSyncRules.cs 8.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192
  1. using System.Text.Json;
  2. using System.Text.RegularExpressions;
  3. using Admin.NET.Plugin.AiDOP.Entity.DataPlatform;
  4. namespace Admin.NET.Plugin.AiDOP.DataPlatform;
  5. public sealed record MdpConfigIssue(string Code, string Field, string Message);
  6. public sealed record MdpMappedRow(Dictionary<string, object?> Values, IReadOnlyList<string> Errors);
  7. public static class MdpDbSyncRules
  8. {
  9. public const long SourceMenuId = 1320990000403L;
  10. public const long SyncTaskMenuId = 1320990000404L;
  11. public const long SyncLogMenuId = 1320990000405L;
  12. public static readonly string[] StagingColumns =
  13. [
  14. "tenant_id", "source_system", "source_table", "source_row_id", "source_biz_key",
  15. "raw_data", "sync_batch_id", "sync_time", "process_status", "create_time", "update_time"
  16. ];
  17. private static readonly Regex Identifier = new("^[A-Za-z_][A-Za-z0-9_]*$", RegexOptions.Compiled);
  18. private static readonly Regex SourceTable = new("^[A-Za-z0-9_\\.\\[\\]]+$", RegexOptions.Compiled);
  19. private static readonly Regex DotPath = new("^[A-Za-z_][A-Za-z0-9_]*(\\.[A-Za-z_][A-Za-z0-9_]*)*$", RegexOptions.Compiled);
  20. public static bool IsSupportedDbType(string? dbType)
  21. {
  22. var value = (dbType ?? "").Trim().ToUpperInvariant();
  23. return value is "SQLSERVER" or "MSSQL" or "MYSQL";
  24. }
  25. public static string? RequireIdentifier(string? value, string field)
  26. {
  27. var text = (value ?? "").Trim();
  28. if (!Identifier.IsMatch(text))
  29. throw new InvalidOperationException($"非法 {field}:{value}");
  30. return text;
  31. }
  32. public static string RequireSourceTable(string? value)
  33. {
  34. var text = (value ?? "").Trim();
  35. if (!SourceTable.IsMatch(text))
  36. throw new InvalidOperationException($"非法 source_table_name:{value}");
  37. return text;
  38. }
  39. public static IReadOnlyList<MdpConfigIssue> ValidateDbEntity(MdpEntity entity, string? sourceType)
  40. {
  41. var issues = new List<MdpConfigIssue>();
  42. if (!string.Equals(sourceType, "DB", StringComparison.OrdinalIgnoreCase))
  43. return issues;
  44. void Add(string code, string field, string message) => issues.Add(new MdpConfigIssue(code, field, message));
  45. if (string.IsNullOrWhiteSpace(entity.SourceTableName))
  46. Add("SOURCE_TABLE_MISSING", "sourceTableName", "源表或视图不能为空");
  47. else if (!SourceTable.IsMatch(entity.SourceTableName.Trim()))
  48. Add("SOURCE_TABLE_INVALID", "sourceTableName", "源表名含非法字符");
  49. if (!string.IsNullOrWhiteSpace(entity.TransformSourceTable)
  50. && !SourceTable.IsMatch(entity.TransformSourceTable.Trim()))
  51. Add("TRANSFORM_SOURCE_TABLE_INVALID", "transformSourceTable", "转换逻辑表名含非法字符");
  52. if (string.IsNullOrWhiteSpace(entity.TargetTableName))
  53. Add("TARGET_TABLE_MISSING", "targetTableName", "目标贴源表不能为空");
  54. else if (!Identifier.IsMatch(entity.TargetTableName.Trim()))
  55. Add("TARGET_TABLE_INVALID", "targetTableName", "目标贴源表名含非法字符");
  56. var mode = (entity.SyncMode ?? "").Trim().ToUpperInvariant();
  57. if (mode is not ("FULL" or "INCR" or "ROLLING"))
  58. Add("SYNC_MODE_INVALID", "syncMode", "同步模式只能是 FULL、INCR 或 ROLLING");
  59. if (string.IsNullOrWhiteSpace(entity.BizKeyExpr))
  60. Add("BIZ_KEY_MISSING", "bizKeyExpr", "业务键不能为空");
  61. if (entity.BatchSize is < 1 or > 10000)
  62. Add("BATCH_SIZE_INVALID", "batchSize", "批大小必须在 1 到 10000 之间");
  63. if (string.IsNullOrWhiteSpace(entity.TieBreakerColumn) || !Identifier.IsMatch(entity.TieBreakerColumn.Trim()))
  64. Add("TIE_BREAKER_MISSING", "tieBreakerColumn", "稳定排序列不能为空且只能是字段名");
  65. if (mode is "INCR" or "ROLLING" && (string.IsNullOrWhiteSpace(entity.IncrColumn) || !Identifier.IsMatch(entity.IncrColumn.Trim())))
  66. Add("INCR_COLUMN_MISSING", "incrColumn", "增量同步必须填写增量列");
  67. return issues;
  68. }
  69. public static MdpMappedRow Apply(IDictionary<string, object?> source, IReadOnlyList<MdpFieldMapping> mappings)
  70. {
  71. var values = new Dictionary<string, object?>(source, StringComparer.OrdinalIgnoreCase);
  72. var errors = new List<string>();
  73. foreach (var map in mappings.OrderBy(x => x.SortOrder).ThenBy(x => x.Id))
  74. {
  75. var type = (map.FieldType ?? "DIRECT").Trim().ToUpperInvariant();
  76. object? value = null;
  77. switch (type)
  78. {
  79. case "DIRECT":
  80. values.TryGetValue(map.SourceField, out value);
  81. break;
  82. case "CONST":
  83. value = map.ConstValue;
  84. break;
  85. case "JSONPATH":
  86. if (!values.TryGetValue(map.SourceField, out var raw) || raw == null)
  87. {
  88. errors.Add($"{map.TargetField} 缺少 JSON 源字段");
  89. continue;
  90. }
  91. var path = string.IsNullOrWhiteSpace(map.TransformScript) ? map.SourceField : map.TransformScript.Trim();
  92. if (!DotPath.IsMatch(path))
  93. {
  94. errors.Add($"{map.TargetField} 的 JSON 路径不合法");
  95. continue;
  96. }
  97. value = ReadDotPath(raw.ToString(), path, out var pathError);
  98. if (pathError != null)
  99. {
  100. errors.Add($"{map.TargetField} {pathError}");
  101. continue;
  102. }
  103. break;
  104. case "SCRIPT":
  105. case "LOOKUP":
  106. errors.Add($"{map.TargetField} 使用了数据库同步不支持的映射类型 {type}");
  107. continue;
  108. default:
  109. errors.Add($"{map.TargetField} 使用了未知映射类型 {type}");
  110. continue;
  111. }
  112. if (IsEmpty(value) && !string.IsNullOrWhiteSpace(map.DefaultValue))
  113. value = map.DefaultValue;
  114. values[map.TargetField] = value;
  115. if (map.IsRequired == 1 && IsEmpty(value))
  116. errors.Add($"{map.TargetField} 为必填");
  117. }
  118. return new MdpMappedRow(values, errors);
  119. }
  120. public static string Redact(string? text)
  121. {
  122. if (string.IsNullOrEmpty(text)) return "";
  123. var value = Regex.Replace(text, @"\b1[3-9]\d{9}\b", "***");
  124. value = Regex.Replace(value, @"[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}", "***");
  125. value = Regex.Replace(value, @"\b\d{17}[\dXx]\b", "***");
  126. return value.Length > 300 ? value[..300] : value;
  127. }
  128. public static string? BuildJobId(long tenantId, string taskCode)
  129. {
  130. var token = Regex.Replace((taskCode ?? "").Trim().ToUpperInvariant(), "[^A-Z0-9]", "_");
  131. var jobId = $"job_mdp_dbsync_{tenantId}_{token}";
  132. return jobId.Length <= 64 ? jobId : null;
  133. }
  134. public static string? BuildTriggerId(long tenantId, string taskCode)
  135. {
  136. var jobId = BuildJobId(tenantId, taskCode);
  137. if (jobId == null) return null;
  138. var triggerId = "trigger_" + jobId["job_".Length..];
  139. return triggerId.Length <= 64 ? triggerId : null;
  140. }
  141. private static object? ReadDotPath(string? json, string path, out string? error)
  142. {
  143. error = null;
  144. try
  145. {
  146. using var doc = JsonDocument.Parse(string.IsNullOrWhiteSpace(json) ? "null" : json);
  147. JsonElement current = doc.RootElement;
  148. foreach (var part in path.Split('.'))
  149. {
  150. if (current.ValueKind != JsonValueKind.Object || !current.TryGetProperty(part, out current))
  151. {
  152. error = "JSON 路径不存在";
  153. return null;
  154. }
  155. }
  156. return current.ValueKind switch
  157. {
  158. JsonValueKind.String => current.GetString(),
  159. JsonValueKind.Number => current.TryGetInt64(out var n) ? n : current.GetDecimal(),
  160. JsonValueKind.True => true,
  161. JsonValueKind.False => false,
  162. JsonValueKind.Null => null,
  163. _ => current.GetRawText()
  164. };
  165. }
  166. catch (JsonException)
  167. {
  168. error = "源字段不是合法 JSON";
  169. return null;
  170. }
  171. }
  172. private static bool IsEmpty(object? value) =>
  173. value == null || value == DBNull.Value || string.IsNullOrWhiteSpace(value.ToString());
  174. }