MdpRebuildContracts.cs 8.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276
  1. using System.Text.RegularExpressions;
  2. namespace Admin.NET.Plugin.AiDOP.DataPlatform.MdpRebuild;
  3. public sealed record MdpRebuildScope(string ModuleCode, long TenantId, long FactoryId)
  4. {
  5. public static readonly HashSet<string> RebuildModules = new(StringComparer.OrdinalIgnoreCase)
  6. {
  7. "S1", "S2", "S3", "S4", "S5", "S6", "S7"
  8. };
  9. public static readonly HashSet<string> SnapshotModules = new(StringComparer.OrdinalIgnoreCase)
  10. {
  11. "S1", "S2", "S3", "S4", "S5", "S6", "S7", "S9"
  12. };
  13. public static MdpRebuildScope Create(string moduleCode, long tenantId, long factoryId)
  14. {
  15. var mc = NormalizeModule(moduleCode);
  16. if (!RebuildModules.Contains(mc))
  17. throw new InvalidOperationException($"模块 {mc} 不支持全量数据重算");
  18. if (tenantId <= 0)
  19. throw new InvalidOperationException($"{mc} 全量重算必须指定有效 tenantId,禁止 tenantId=0");
  20. if (factoryId <= 0)
  21. throw new InvalidOperationException($"{mc} 全量重算必须指定有效 factoryId");
  22. if (factoryId == tenantId)
  23. throw new InvalidOperationException($"{mc} 全量重算的 factoryId 不能等于 tenantId");
  24. return new MdpRebuildScope(mc, tenantId, factoryId);
  25. }
  26. public static string NormalizeModule(string? moduleCode)
  27. {
  28. var mc = (moduleCode ?? string.Empty).Trim().ToUpperInvariant();
  29. if (mc.Length == 0) throw new InvalidOperationException("moduleCode 不能为空");
  30. return mc;
  31. }
  32. public string LockName => $"{ModuleCode}_MDP_FULL:{TenantId}:{FactoryId}";
  33. public string ScopeKey => $"{ModuleCode}:{TenantId}:{FactoryId}";
  34. }
  35. public static class ModuleRebuildStatus
  36. {
  37. public const string Queued = "QUEUED";
  38. public const string Running = "RUNNING";
  39. public const string Success = "SUCCESS";
  40. public const string Failed = "FAILED";
  41. public const string Cancelled = "CANCELLED";
  42. }
  43. public static class ModuleRebuildStages
  44. {
  45. public const string Queued = "QUEUED";
  46. public const string AcquiringLock = "ACQUIRING_LOCK";
  47. public const string Preparing = "PREPARING";
  48. public const string Staging = "STAGING";
  49. public const string Standard = "STANDARD";
  50. public const string Dwd = "DWD";
  51. public const string Kpi = "KPI";
  52. public const string Atomic = "ATOMIC";
  53. public const string T8Inbound = "T8_INBOUND";
  54. public const string KpiPreparing = "KPI_PREPARING";
  55. public const string KpiCalculating = "KPI_CALCULATING";
  56. public const string Finalizing = "FINALIZING";
  57. public const string Success = "SUCCESS";
  58. public const string Failed = "FAILED";
  59. public const string Cancelled = "CANCELLED";
  60. public static int StageTotal(string moduleCode) => Normalize(moduleCode) switch
  61. {
  62. "S4" => 4,
  63. "S5" or "S6" or "S7" => 4,
  64. _ => 5
  65. };
  66. public static int ToStageIndex(string moduleCode, string stage)
  67. {
  68. var mc = Normalize(moduleCode);
  69. if (mc is "S5" or "S6" or "S7")
  70. {
  71. return stage switch
  72. {
  73. T8Inbound => 1,
  74. KpiPreparing => 2,
  75. KpiCalculating => 3,
  76. Atomic or Finalizing or Success => 4,
  77. _ => 0
  78. };
  79. }
  80. if (mc == "S4")
  81. {
  82. return stage switch
  83. {
  84. Staging => 1,
  85. Standard => 2,
  86. Dwd => 3,
  87. Kpi or Finalizing or Success => 4,
  88. _ => 0
  89. };
  90. }
  91. return stage switch
  92. {
  93. Staging => 1,
  94. Standard => 2,
  95. Dwd => 3,
  96. Kpi => 4,
  97. Atomic or Finalizing or Success => 5,
  98. _ => 0
  99. };
  100. }
  101. public static string ToChinese(string? stage) => stage switch
  102. {
  103. Queued => "排队中",
  104. AcquiringLock => "等待资源",
  105. Preparing => "准备运行环境",
  106. Staging => "拉取源数据",
  107. Standard => "标准化数据",
  108. Dwd => "生成 DWD 明细",
  109. Kpi => "重算 KPI",
  110. Atomic => "重建原子数据",
  111. T8Inbound => "T8 基表入站",
  112. KpiPreparing => "准备 KPI 计算",
  113. KpiCalculating => "计算 KPI",
  114. Finalizing => "收尾",
  115. Success => "已完成",
  116. Failed => "失败",
  117. Cancelled => "已取消",
  118. _ => stage ?? ""
  119. };
  120. public static string ConfirmTitle(string moduleCode) => $"确认重算 {Normalize(moduleCode)} 数据?";
  121. private static string Normalize(string? moduleCode) =>
  122. string.IsNullOrWhiteSpace(moduleCode) ? "" : moduleCode.Trim().ToUpperInvariant();
  123. }
  124. public sealed record ModuleProgressUpdate(
  125. string Stage,
  126. int StageIndex,
  127. int ProgressPercent,
  128. string Message,
  129. int? Rows = null,
  130. string? CompletedStage = null,
  131. string? DetailJson = null);
  132. public sealed class ModuleRebuildResult
  133. {
  134. public string BatchId { get; set; } = string.Empty;
  135. public long RunLogId { get; set; }
  136. public int StageRows { get; set; }
  137. public int StandardRows { get; set; }
  138. public int DwdRows { get; set; }
  139. public int KpiRows { get; set; }
  140. public int AtomicRows { get; set; }
  141. public string? DetailJson { get; set; }
  142. }
  143. public sealed class ModuleRebuildAlreadyRunningException : Exception
  144. {
  145. public ModuleRebuildAlreadyRunningException(string moduleCode)
  146. : base($"{moduleCode} 数据重算正在执行,请勿重复提交")
  147. {
  148. ModuleCode = moduleCode;
  149. }
  150. public string ModuleCode { get; }
  151. }
  152. public static class MdpSqlScope
  153. {
  154. private static readonly Regex FromAlias = new(
  155. @"\bFROM\s+[`.\w]+\s+(?:AS\s+)?([`\w]+)",
  156. RegexOptions.IgnoreCase | RegexOptions.Compiled);
  157. public static string InjectTenantFactory(string sql)
  158. => Inject(sql, includeFactory: true);
  159. public static string InjectTenantOnly(string sql)
  160. => Inject(sql, includeFactory: false);
  161. private static string Inject(string sql, bool includeFactory)
  162. {
  163. if (string.IsNullOrWhiteSpace(sql)) return sql;
  164. return Regex.Replace(sql, @"\bWHERE\b", m =>
  165. {
  166. var start = m.Index;
  167. var window = sql.Substring(start, Math.Min(240, sql.Length - start));
  168. if (window.Contains("@TenantId", StringComparison.Ordinal))
  169. return m.Value;
  170. var prefix = ResolveSourcePrefix(sql, start);
  171. var factoryFilter = includeFactory
  172. ? $"COALESCE(NULLIF({prefix}factory_id,0),1)=@FactoryId AND "
  173. : string.Empty;
  174. return $"WHERE {prefix}tenant_id=@TenantId AND {factoryFilter}";
  175. }, RegexOptions.IgnoreCase);
  176. }
  177. private static string ResolveSourcePrefix(string sql, int whereIndex)
  178. {
  179. var depth = 0;
  180. for (var i = whereIndex - 1; i >= 0; i--)
  181. {
  182. if (sql[i] == ')')
  183. {
  184. depth++;
  185. continue;
  186. }
  187. if (sql[i] == '(')
  188. {
  189. if (depth > 0) depth--;
  190. continue;
  191. }
  192. if (depth != 0 || i < 3) continue;
  193. if (!sql.AsSpan(i - 3, 4).Equals("FROM", StringComparison.OrdinalIgnoreCase)) continue;
  194. if (i > 3 && (char.IsLetterOrDigit(sql[i - 4]) || sql[i - 4] == '_')) continue;
  195. var match = FromAlias.Match(sql, i - 3);
  196. if (!match.Success || match.Index != i - 3) return string.Empty;
  197. var alias = match.Groups[1].Value.Trim('`');
  198. return alias is "LEFT" or "RIGHT" or "INNER" or "CROSS" or "JOIN" or "ON" or "AS" or "WHERE"
  199. ? string.Empty
  200. : alias + ".";
  201. }
  202. return string.Empty;
  203. }
  204. }
  205. public interface IModuleRebuildCapability
  206. {
  207. bool IsEnabled(string moduleCode);
  208. int MaxParallelScopes { get; }
  209. }
  210. public sealed class ModuleRebuildCapability : IModuleRebuildCapability, ISingleton
  211. {
  212. public bool IsEnabled(string moduleCode)
  213. {
  214. var mc = MdpRebuildScope.NormalizeModule(moduleCode);
  215. try
  216. {
  217. return Furion.App.GetConfig<bool?>($"AiDOP:MdpRebuild:Modules:{mc}:Enabled", true) ?? false;
  218. }
  219. catch
  220. {
  221. return false;
  222. }
  223. }
  224. public int MaxParallelScopes
  225. {
  226. get
  227. {
  228. try
  229. {
  230. return Math.Clamp(Furion.App.GetConfig<int?>("AiDOP:MdpRebuild:MaxParallelScopes", true) ?? 2, 1, 8);
  231. }
  232. catch
  233. {
  234. return 2;
  235. }
  236. }
  237. }
  238. }
  239. public sealed class AlwaysOnModuleRebuildCapability : IModuleRebuildCapability
  240. {
  241. public bool IsEnabled(string moduleCode) => true;
  242. public int MaxParallelScopes => 2;
  243. }