TracksParseMetricCore.cs 35 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798
  1. using System.Collections.Concurrent;
  2. using System.Data;
  3. using System.Threading;
  4. using Dapper;
  5. using dodohold.core;
  6. using YunhuiKit;
  7. namespace molilian.core
  8. {
  9. public partial class TracksCore
  10. {
  11. private const string ParseMetricTypeName = "转链";
  12. private const string ParseMetricRequest = "request";
  13. private const string ParseMetricCall = "call";
  14. private const string ParseMetricSuccess = "success";
  15. private const int ParseMetricFlushIntervalMs = 1000;
  16. private static readonly ConcurrentDictionary<ParseMetricCounterKey, ParseMetricCounter> PendingParseMetricCounters = new();
  17. private static readonly object ParseMetricFlushTimerLock = new();
  18. private static Timer? ParseMetricFlushTimer;
  19. private static int ParseMetricFlushRunning;
  20. private static int ParseMetricLifecycleHooked;
  21. public static Task<bool> RecordParseRequestAsync(UnionParseRequest request)
  22. {
  23. var dimension = ResolveParseMetricDimension(request.Channel, request.RiskStrategy, request.LaunchScene);
  24. return IncrementParseMetricAsync(ParseMetricRequest, dimension.Platform, dimension.RiskStrategy, dimension.LaunchScene, request.AccountId);
  25. }
  26. public static Task<bool> RecordParseAccountRequestAsync(string channel, string riskStrategy, int launchScene, int accountId)
  27. {
  28. if (accountId <= 0) return Task.FromResult(false);
  29. var dimension = ResolveParseMetricDimension(channel, riskStrategy, launchScene);
  30. return IncrementParseAccountMetricAsync(ParseMetricRequest, dimension.Platform, dimension.RiskStrategy, dimension.LaunchScene, accountId);
  31. }
  32. public static Task<bool> RecordParseResultAsync(string channel, string riskStrategy, int launchScene, int accountId, bool success)
  33. {
  34. var dimension = ResolveParseMetricDimension(channel, riskStrategy, launchScene);
  35. return IncrementParseResultMetricAsync(dimension.Platform, dimension.RiskStrategy, dimension.LaunchScene, accountId, success);
  36. }
  37. public static async Task<TrackParseMetricReportResult> GetParseMetricReportAsync(
  38. DateTime start,
  39. DateTime end,
  40. int page,
  41. int size,
  42. bool getTotal,
  43. int accountId = -1,
  44. bool accountBreakdownOnly = false,
  45. string platform = "",
  46. string riskStrategy = "",
  47. int? launchScene = null)
  48. {
  49. start = start.Date;
  50. end = end.Date;
  51. if (end < start) end = start;
  52. page = Math.Max(page, 1);
  53. size = Math.Max(size, 1);
  54. var list = new List<TrackParseMetricReportDTO>();
  55. for (var date = start; date <= end; date = date.AddDays(1))
  56. {
  57. string dateStr = date.ToString("yyyyMMdd");
  58. string indexKey = BuildParseMetricIndexKey("daily", dateStr);
  59. string[] members = await TkLogCore.GetTotalKeysAsync(indexKey) ?? [];
  60. foreach (var member in members)
  61. {
  62. if (!TryParseParseMetricIndexValue(member, out var dimension)) continue;
  63. if (!IsMatchedParseMetricDimension(dimension, accountId, accountBreakdownOnly, platform, riskStrategy, launchScene)) continue;
  64. var requestTask = TkLogCore.GetTotalAsync(BuildParseMetricCountKey(ParseMetricRequest, "daily", dateStr, dimension.Platform, dimension.RiskStrategy, dimension.LaunchScene, dimension.AccountId));
  65. var callTask = TkLogCore.GetTotalAsync(BuildParseMetricCountKey(ParseMetricCall, "daily", dateStr, dimension.Platform, dimension.RiskStrategy, dimension.LaunchScene, dimension.AccountId));
  66. var successTask = TkLogCore.GetTotalAsync(BuildParseMetricCountKey(ParseMetricSuccess, "daily", dateStr, dimension.Platform, dimension.RiskStrategy, dimension.LaunchScene, dimension.AccountId));
  67. await Task.WhenAll(requestTask, callTask, successTask);
  68. int requestCount = requestTask.Result;
  69. int callCount = callTask.Result;
  70. int successCount = successTask.Result;
  71. if (requestCount <= 0 && callCount <= 0 && successCount <= 0) continue;
  72. list.Add(new TrackParseMetricReportDTO
  73. {
  74. report_date = date,
  75. platform = dimension.Platform,
  76. typename = ParseMetricTypeName,
  77. risk_strategy = dimension.RiskStrategy,
  78. launch_scene = dimension.LaunchScene,
  79. account_id = dimension.AccountId,
  80. request_count = requestCount,
  81. call_count = callCount,
  82. success_count = successCount
  83. });
  84. }
  85. }
  86. list = list
  87. .OrderByDescending(item => item.report_date)
  88. .ThenBy(item => item.platform)
  89. .ThenBy(item => item.risk_strategy)
  90. .ThenBy(item => item.launch_scene)
  91. .ThenBy(item => item.account_id)
  92. .ToList();
  93. int count = list.Count;
  94. list = list
  95. .Skip((page - 1) * size)
  96. .Take(size)
  97. .ToList();
  98. return new TrackParseMetricReportResult
  99. {
  100. list = list,
  101. count = getTotal ? count : list.Count
  102. };
  103. }
  104. public static async Task<List<TrackParseMetricHourlyDTO>> GetParseMetricHourlyReportAsync(
  105. DateTime targetDate,
  106. string platform,
  107. string riskStrategy,
  108. int? launchScene = null,
  109. int accountId = 0)
  110. {
  111. targetDate = targetDate.Date;
  112. platform = NormalizeParseMetricPlatform(platform);
  113. riskStrategy = NormalizeReportScene(riskStrategy);
  114. accountId = Math.Max(accountId, 0);
  115. var list = new List<TrackParseMetricHourlyDTO>();
  116. string dateStr = targetDate.ToString("yyyyMMdd");
  117. for (int hour = 0; hour < 24; hour++)
  118. {
  119. string hourStr = $"{dateStr}{hour:00}";
  120. string indexKey = BuildParseMetricIndexKey("hour", hourStr);
  121. string[] members = await TkLogCore.GetTotalKeysAsync(indexKey) ?? [];
  122. long requestCount = 0;
  123. long callCount = 0;
  124. long successCount = 0;
  125. foreach (var member in members)
  126. {
  127. if (!TryParseParseMetricIndexValue(member, out var dimension)) continue;
  128. if (!IsMatchedParseMetricDimension(dimension, accountId, false, platform, riskStrategy, launchScene)) continue;
  129. var requestTask = TkLogCore.GetTotalAsync(BuildParseMetricCountKey(ParseMetricRequest, "hour", hourStr, dimension.Platform, dimension.RiskStrategy, dimension.LaunchScene, dimension.AccountId));
  130. var callTask = TkLogCore.GetTotalAsync(BuildParseMetricCountKey(ParseMetricCall, "hour", hourStr, dimension.Platform, dimension.RiskStrategy, dimension.LaunchScene, dimension.AccountId));
  131. var successTask = TkLogCore.GetTotalAsync(BuildParseMetricCountKey(ParseMetricSuccess, "hour", hourStr, dimension.Platform, dimension.RiskStrategy, dimension.LaunchScene, dimension.AccountId));
  132. await Task.WhenAll(requestTask, callTask, successTask);
  133. requestCount += requestTask.Result;
  134. callCount += callTask.Result;
  135. successCount += successTask.Result;
  136. }
  137. list.Add(new TrackParseMetricHourlyDTO
  138. {
  139. report_date = targetDate,
  140. platform = platform,
  141. typename = ParseMetricTypeName,
  142. risk_strategy = riskStrategy,
  143. launch_scene = launchScene ?? -1,
  144. account_id = accountId,
  145. hour = hour,
  146. request_count = SafeInt(requestCount),
  147. call_count = SafeInt(callCount),
  148. success_count = SafeInt(successCount)
  149. });
  150. }
  151. return list;
  152. }
  153. public static async Task<TrackParseMetricBackfillResult> BackfillParseMetricCountersAsync(
  154. DateTime targetDate,
  155. string platform = "",
  156. string riskStrategy = "",
  157. int? launchScene = null)
  158. {
  159. targetDate = targetDate.Date;
  160. platform = NormalizeParseMetricPlatform(platform);
  161. riskStrategy = NormalizeReportScene(riskStrategy);
  162. bool hasPlatformFilter = !string.IsNullOrWhiteSpace(platform);
  163. bool hasRiskStrategyFilter = !string.IsNullOrWhiteSpace(riskStrategy);
  164. bool hasLaunchSceneFilter = launchScene.HasValue;
  165. bool hasDimensionFilter = hasPlatformFilter || hasRiskStrategyFilter || hasLaunchSceneFilter;
  166. var result = new TrackParseMetricBackfillResult
  167. {
  168. report_date = targetDate.ToString("yyyy-MM-dd")
  169. };
  170. var allSources = new[]
  171. {
  172. new ParseMetricBackfillSource("tb", $"tk_parse_logs_{targetDate:yyyyMMdd}"),
  173. new ParseMetricBackfillSource("jd", $"jd_parse_logs_{targetDate:yyyyMMdd}"),
  174. new ParseMetricBackfillSource("pdd", $"pdd_parse_logs_{targetDate:yyyyMMdd}")
  175. };
  176. var sources = hasDimensionFilter
  177. ? allSources.Where(source => !hasPlatformFilter || string.Equals(source.Platform, platform, StringComparison.OrdinalIgnoreCase)).ToArray()
  178. : allSources;
  179. using var conn = DBContext.GetOpenConnection();
  180. DateTime start = targetDate;
  181. DateTime end = targetDate.AddDays(1);
  182. result.redis_cleared_keys = hasDimensionFilter
  183. ? await ClearParseMetricCountersForDimensionAsync(targetDate, platform, riskStrategy, launchScene)
  184. : await ClearParseMetricCountersForDateAsync(targetDate);
  185. foreach (var source in sources)
  186. {
  187. if (!TableExists(conn, source.TableName))
  188. {
  189. result.details.Add(new { platform = source.Platform, table = source.TableName, exists = false, rows = 0, dimensions = 0 });
  190. continue;
  191. }
  192. const string riskStrategySql = "COALESCE(NULLIF(TRIM(riskStrategy), ''), '')";
  193. const string rawLaunchSceneSql = "COALESCE(launchScene, -1)";
  194. string launchSceneSql = $"CASE WHEN LOWER({riskStrategySql}) IN ('os','tbpush','brwsimilar','icon') AND {rawLaunchSceneSql}=-1 THEN 0 ELSE {rawLaunchSceneSql} END";
  195. var dimensionWhereItems = new List<string>();
  196. if (hasRiskStrategyFilter) dimensionWhereItems.Add($"{riskStrategySql}=@riskStrategy");
  197. if (hasLaunchSceneFilter) dimensionWhereItems.Add($"{launchSceneSql}=@launchScene");
  198. string dimensionWhere = dimensionWhereItems.Count > 0
  199. ? "\n AND " + string.Join("\n AND ", dimensionWhereItems)
  200. : string.Empty;
  201. string sql = $@"
  202. SELECT
  203. {riskStrategySql} AS risk_strategy,
  204. {launchSceneSql} AS launch_scene,
  205. COALESCE(accountId, 0) AS account_id,
  206. DATE_FORMAT(create_time, '%Y%m%d%H') AS hour_key,
  207. CAST(COUNT(1) AS SIGNED) AS call_count,
  208. CAST(SUM(CASE WHEN success=1 THEN 1 ELSE 0 END) AS SIGNED) AS success_count
  209. FROM `{source.TableName}`
  210. WHERE create_time>=@start AND create_time<@end
  211. {dimensionWhere}
  212. GROUP BY
  213. {riskStrategySql},
  214. {launchSceneSql},
  215. COALESCE(accountId, 0),
  216. DATE_FORMAT(create_time, '%Y%m%d%H')";
  217. var args = new DynamicParameters();
  218. args.Add("start", start);
  219. args.Add("end", end);
  220. if (hasRiskStrategyFilter) args.Add("riskStrategy", riskStrategy);
  221. if (hasLaunchSceneFilter) args.Add("launchScene", launchScene!.Value);
  222. var rows = SqlMapper.Query<ParseMetricBackfillRow>(conn, sql, args).ToList();
  223. int mysqlRows = rows.Sum(row => SafeInt(row.call_count));
  224. int dimensions = ApplyParseMetricBackfillRows(source.Platform, targetDate, rows);
  225. result.mysql_rows += mysqlRows;
  226. result.redis_dimensions += dimensions;
  227. result.details.Add(new { platform = source.Platform, table = source.TableName, exists = true, rows = mysqlRows, dimensions });
  228. }
  229. await Task.CompletedTask;
  230. return result;
  231. }
  232. private static Task<bool> IncrementParseMetricAsync(string metric, string platform, string riskStrategy, int launchScene, int accountId = 0)
  233. {
  234. try
  235. {
  236. AddParseMetricCount(metric, DateTime.Now, platform, riskStrategy, launchScene, 0, 1);
  237. if (accountId > 0)
  238. {
  239. AddParseMetricCount(metric, DateTime.Now, platform, riskStrategy, launchScene, accountId, 1);
  240. }
  241. return Task.FromResult(true);
  242. }
  243. catch (Exception ex)
  244. {
  245. _ = new LoggerLibrary("TracksCore", "RecordParseMetric")
  246. .Info(ex.Message, ex.StackTrace)
  247. .SaveAsync();
  248. return Task.FromResult(false);
  249. }
  250. }
  251. private static Task<bool> IncrementParseAccountMetricAsync(string metric, string platform, string riskStrategy, int launchScene, int accountId)
  252. {
  253. try
  254. {
  255. accountId = Math.Max(accountId, 0);
  256. if (accountId <= 0) return Task.FromResult(false);
  257. AddParseMetricCount(metric, DateTime.Now, platform, riskStrategy, launchScene, accountId, 1);
  258. return Task.FromResult(true);
  259. }
  260. catch (Exception ex)
  261. {
  262. _ = new LoggerLibrary("TracksCore", "RecordParseAccountMetric")
  263. .Info(ex.Message, ex.StackTrace)
  264. .SaveAsync();
  265. return Task.FromResult(false);
  266. }
  267. }
  268. private static Task<bool> IncrementParseResultMetricAsync(string platform, string riskStrategy, int launchScene, int accountId, bool success)
  269. {
  270. try
  271. {
  272. accountId = Math.Max(accountId, 0);
  273. AddParseMetricCount(ParseMetricCall, DateTime.Now, platform, riskStrategy, launchScene, 0, 1);
  274. if (success) AddParseMetricCount(ParseMetricSuccess, DateTime.Now, platform, riskStrategy, launchScene, 0, 1);
  275. if (accountId > 0)
  276. {
  277. AddParseMetricCount(ParseMetricCall, DateTime.Now, platform, riskStrategy, launchScene, accountId, 1);
  278. if (success) AddParseMetricCount(ParseMetricSuccess, DateTime.Now, platform, riskStrategy, launchScene, accountId, 1);
  279. }
  280. return Task.FromResult(true);
  281. }
  282. catch (Exception ex)
  283. {
  284. _ = new LoggerLibrary("TracksCore", "RecordParseResultMetric")
  285. .Info(ex.Message, ex.StackTrace)
  286. .SaveAsync();
  287. return Task.FromResult(false);
  288. }
  289. }
  290. private static int ApplyParseMetricBackfillRows(string platform, DateTime targetDate, List<ParseMetricBackfillRow> rows)
  291. {
  292. string dateStr = targetDate.ToString("yyyyMMdd");
  293. int dimensions = 0;
  294. var dailyGroups = rows.GroupBy(row => new
  295. {
  296. RiskStrategy = NormalizeReportScene(row.risk_strategy),
  297. LaunchScene = NormalizeParseMetricLaunchScene(row.risk_strategy, row.launch_scene)
  298. });
  299. foreach (var group in dailyGroups)
  300. {
  301. int callCount = group.Sum(row => SafeInt(row.call_count));
  302. int successCount = group.Sum(row => SafeInt(row.success_count));
  303. SetParseMetricCount(ParseMetricCall, "daily", dateStr, platform, group.Key.RiskStrategy, group.Key.LaunchScene, 0, callCount);
  304. SetParseMetricCount(ParseMetricSuccess, "daily", dateStr, platform, group.Key.RiskStrategy, group.Key.LaunchScene, 0, successCount);
  305. dimensions++;
  306. }
  307. var dailyAccountGroups = rows
  308. .Where(row => row.account_id > 0)
  309. .GroupBy(row => new
  310. {
  311. RiskStrategy = NormalizeReportScene(row.risk_strategy),
  312. launch_scene = NormalizeParseMetricLaunchScene(row.risk_strategy, row.launch_scene),
  313. row.account_id
  314. });
  315. foreach (var group in dailyAccountGroups)
  316. {
  317. int callCount = group.Sum(row => SafeInt(row.call_count));
  318. int successCount = group.Sum(row => SafeInt(row.success_count));
  319. SetParseMetricCount(ParseMetricRequest, "daily", dateStr, platform, group.Key.RiskStrategy, group.Key.launch_scene, group.Key.account_id, callCount);
  320. SetParseMetricCount(ParseMetricCall, "daily", dateStr, platform, group.Key.RiskStrategy, group.Key.launch_scene, group.Key.account_id, callCount);
  321. SetParseMetricCount(ParseMetricSuccess, "daily", dateStr, platform, group.Key.RiskStrategy, group.Key.launch_scene, group.Key.account_id, successCount);
  322. dimensions++;
  323. }
  324. var hourlyGroups = rows.GroupBy(row => new
  325. {
  326. RiskStrategy = NormalizeReportScene(row.risk_strategy),
  327. launch_scene = NormalizeParseMetricLaunchScene(row.risk_strategy, row.launch_scene),
  328. row.hour_key
  329. });
  330. foreach (var group in hourlyGroups)
  331. {
  332. int callCount = group.Sum(row => SafeInt(row.call_count));
  333. int successCount = group.Sum(row => SafeInt(row.success_count));
  334. SetParseMetricCount(ParseMetricCall, "hour", group.Key.hour_key, platform, group.Key.RiskStrategy, group.Key.launch_scene, 0, callCount);
  335. SetParseMetricCount(ParseMetricSuccess, "hour", group.Key.hour_key, platform, group.Key.RiskStrategy, group.Key.launch_scene, 0, successCount);
  336. }
  337. var hourlyAccountGroups = rows
  338. .Where(row => row.account_id > 0)
  339. .GroupBy(row => new
  340. {
  341. RiskStrategy = NormalizeReportScene(row.risk_strategy),
  342. launch_scene = NormalizeParseMetricLaunchScene(row.risk_strategy, row.launch_scene),
  343. row.account_id,
  344. row.hour_key
  345. });
  346. foreach (var group in hourlyAccountGroups)
  347. {
  348. int callCount = group.Sum(row => SafeInt(row.call_count));
  349. int successCount = group.Sum(row => SafeInt(row.success_count));
  350. SetParseMetricCount(ParseMetricRequest, "hour", group.Key.hour_key, platform, group.Key.RiskStrategy, group.Key.launch_scene, group.Key.account_id, callCount);
  351. SetParseMetricCount(ParseMetricCall, "hour", group.Key.hour_key, platform, group.Key.RiskStrategy, group.Key.launch_scene, group.Key.account_id, callCount);
  352. SetParseMetricCount(ParseMetricSuccess, "hour", group.Key.hour_key, platform, group.Key.RiskStrategy, group.Key.launch_scene, group.Key.account_id, successCount);
  353. }
  354. return dimensions;
  355. }
  356. private static void AddParseMetricCount(string metric, DateTime metricTime, string platform, string riskStrategy, int launchScene, int accountId, int count)
  357. {
  358. if (count <= 0) return;
  359. QueueParseMetricCount(metric, "daily", metricTime.ToString("yyyyMMdd"), platform, riskStrategy, launchScene, accountId, count);
  360. QueueParseMetricCount(metric, "hour", metricTime.ToString("yyyyMMddHH"), platform, riskStrategy, launchScene, accountId, count);
  361. }
  362. private static void SetParseMetricCount(string metric, string bucketType, string bucketValue, string platform, string riskStrategy, int launchScene, int accountId, int count)
  363. {
  364. if (count < 0) return;
  365. SetOrIncrementParseMetricCount(metric, bucketType, bucketValue, platform, riskStrategy, launchScene, accountId, count, increment: false);
  366. }
  367. private static void SetOrIncrementParseMetricCount(string metric, string bucketType, string bucketValue, string platform, string riskStrategy, int launchScene, int accountId, long count, bool increment)
  368. {
  369. platform = Normalize(platform).ToLowerInvariant();
  370. riskStrategy = NormalizeReportScene(riskStrategy);
  371. launchScene = NormalizeParseMetricLaunchScene(riskStrategy, launchScene);
  372. accountId = Math.Max(accountId, 0);
  373. string countKey = BuildParseMetricCountKey(metric, bucketType, bucketValue, platform, riskStrategy, launchScene, accountId);
  374. string indexKey = BuildParseMetricIndexKey(bucketType, bucketValue);
  375. string indexValue = BuildParseMetricIndexValue(platform, riskStrategy, launchScene, accountId);
  376. int expireSeconds = bucketType == "hour" ? HourlyExpireSeconds : DailyExpireSeconds;
  377. if (increment)
  378. {
  379. IncrementParseMetricRedisKey(countKey, count);
  380. }
  381. else
  382. {
  383. RedisHelper.Set(countKey, count, expireSeconds);
  384. }
  385. RedisHelper.Expire(countKey, expireSeconds);
  386. RedisHelper.SAdd(indexKey, indexValue);
  387. RedisHelper.Expire(indexKey, expireSeconds);
  388. }
  389. public static int FlushParseMetricCounters()
  390. {
  391. if (Interlocked.Exchange(ref ParseMetricFlushRunning, 1) == 1) return 0;
  392. try
  393. {
  394. int flushed = 0;
  395. foreach (var item in PendingParseMetricCounters.ToArray())
  396. {
  397. long count = Interlocked.Exchange(ref item.Value.Count, 0);
  398. if (count <= 0) continue;
  399. try
  400. {
  401. SetOrIncrementParseMetricCount(
  402. item.Key.Metric,
  403. item.Key.BucketType,
  404. item.Key.BucketValue,
  405. item.Key.Platform,
  406. item.Key.RiskStrategy,
  407. item.Key.LaunchScene,
  408. item.Key.AccountId,
  409. count,
  410. increment: true);
  411. flushed++;
  412. }
  413. catch (Exception ex)
  414. {
  415. Interlocked.Add(ref item.Value.Count, count);
  416. _ = new LoggerLibrary("TracksCore", "FlushParseMetricCounters")
  417. .Info(ex.Message, ex.StackTrace)
  418. .SaveAsync();
  419. }
  420. }
  421. return flushed;
  422. }
  423. finally
  424. {
  425. Interlocked.Exchange(ref ParseMetricFlushRunning, 0);
  426. }
  427. }
  428. private static void QueueParseMetricCount(string metric, string bucketType, string bucketValue, string platform, string riskStrategy, int launchScene, int accountId, int count)
  429. {
  430. if (count <= 0) return;
  431. var key = new ParseMetricCounterKey(
  432. metric,
  433. bucketType,
  434. bucketValue,
  435. Normalize(platform).ToLowerInvariant(),
  436. NormalizeReportScene(riskStrategy),
  437. NormalizeParseMetricLaunchScene(riskStrategy, launchScene),
  438. Math.Max(accountId, 0));
  439. var counter = PendingParseMetricCounters.GetOrAdd(key, _ => new ParseMetricCounter());
  440. Interlocked.Add(ref counter.Count, count);
  441. EnsureParseMetricFlushTimer();
  442. }
  443. private static void EnsureParseMetricFlushTimer()
  444. {
  445. if (ParseMetricFlushTimer != null) return;
  446. lock (ParseMetricFlushTimerLock)
  447. {
  448. if (ParseMetricFlushTimer != null) return;
  449. ParseMetricFlushTimer = new Timer(
  450. _ => FlushParseMetricCounters(),
  451. null,
  452. ParseMetricFlushIntervalMs,
  453. ParseMetricFlushIntervalMs);
  454. if (Interlocked.Exchange(ref ParseMetricLifecycleHooked, 1) == 0)
  455. {
  456. AppDomain.CurrentDomain.ProcessExit += (_, _) => FlushParseMetricCounters();
  457. AppDomain.CurrentDomain.UnhandledException += (_, _) => FlushParseMetricCounters();
  458. }
  459. }
  460. }
  461. private static void IncrementParseMetricRedisKey(string key, long count)
  462. {
  463. while (count > 0)
  464. {
  465. int delta = count > int.MaxValue ? int.MaxValue : (int)count;
  466. RedisHelper.IncrBy(key, delta);
  467. count -= delta;
  468. }
  469. }
  470. private static async Task<int> ClearParseMetricCountersForDateAsync(DateTime targetDate)
  471. {
  472. string dateStr = targetDate.ToString("yyyyMMdd");
  473. int cleared = await ClearParseMetricBucketAsync("daily", dateStr);
  474. for (int hour = 0; hour < 24; hour++)
  475. {
  476. cleared += await ClearParseMetricBucketAsync("hour", $"{dateStr}{hour:00}");
  477. }
  478. return cleared;
  479. }
  480. private static async Task<int> ClearParseMetricCountersForDimensionAsync(
  481. DateTime targetDate,
  482. string platform,
  483. string riskStrategy,
  484. int? launchScene)
  485. {
  486. string dateStr = targetDate.ToString("yyyyMMdd");
  487. int cleared = await ClearParseMetricBucketAsync(
  488. "daily",
  489. dateStr,
  490. dimension => IsSameParseMetricDimension(dimension, platform, riskStrategy, launchScene));
  491. for (int hour = 0; hour < 24; hour++)
  492. {
  493. cleared += await ClearParseMetricBucketAsync(
  494. "hour",
  495. $"{dateStr}{hour:00}",
  496. dimension => IsSameParseMetricDimension(dimension, platform, riskStrategy, launchScene));
  497. }
  498. return cleared;
  499. }
  500. private static async Task<int> ClearParseMetricBucketAsync(
  501. string bucketType,
  502. string bucketValue,
  503. Func<ParseMetricDimension, bool>? predicate = null)
  504. {
  505. string indexKey = BuildParseMetricIndexKey(bucketType, bucketValue);
  506. string[] members = await TkLogCore.GetTotalKeysAsync(indexKey) ?? [];
  507. int cleared = 0;
  508. foreach (var member in members)
  509. {
  510. if (!TryParseParseMetricIndexValue(member, out var dimension)) continue;
  511. if (predicate != null && !predicate(dimension)) continue;
  512. RedisHelper.Del(BuildParseMetricCountKey(ParseMetricCall, bucketType, bucketValue, dimension.Platform, dimension.RiskStrategy, dimension.LaunchScene, dimension.AccountId));
  513. RedisHelper.Del(BuildParseMetricCountKey(ParseMetricSuccess, bucketType, bucketValue, dimension.Platform, dimension.RiskStrategy, dimension.LaunchScene, dimension.AccountId));
  514. cleared += 2;
  515. if (dimension.AccountId > 0)
  516. {
  517. RedisHelper.Del(BuildParseMetricCountKey(ParseMetricRequest, bucketType, bucketValue, dimension.Platform, dimension.RiskStrategy, dimension.LaunchScene, dimension.AccountId));
  518. cleared++;
  519. }
  520. }
  521. return cleared;
  522. }
  523. private static string BuildParseMetricCountKey(string metric, string bucketType, string bucketValue, string platform, string riskStrategy, int launchScene, int accountId = 0)
  524. {
  525. string key = $"{RedisPrefix}:parse_metric:{bucketType}:{metric}:{bucketValue}:{EncodeIndexPart(platform)}:{EncodeIndexPart(riskStrategy)}:{launchScene}";
  526. if (accountId > 0) key += $":account:{accountId}";
  527. return key;
  528. }
  529. private static string BuildParseMetricIndexKey(string bucketType, string bucketValue)
  530. {
  531. return $"{RedisPrefix}:parse_metric:{bucketType}:index:{bucketValue}";
  532. }
  533. private static string BuildParseMetricIndexValue(string platform, string riskStrategy, int launchScene, int accountId = 0)
  534. {
  535. string value = $"{EncodeIndexPart(platform)}|{EncodeIndexPart(riskStrategy)}|{launchScene}";
  536. if (accountId > 0) value += $"|{accountId}";
  537. return value;
  538. }
  539. private static bool TryParseParseMetricIndexValue(string value, out ParseMetricDimension dimension)
  540. {
  541. dimension = new ParseMetricDimension();
  542. var parts = (value ?? string.Empty).Split('|');
  543. if (parts.Length < 3 || parts.Length > 4) return false;
  544. if (!int.TryParse(parts[2], out int launchScene)) return false;
  545. int accountId = 0;
  546. if (parts.Length == 4 && (!int.TryParse(parts[3], out accountId) || accountId <= 0)) return false;
  547. dimension = new ParseMetricDimension
  548. {
  549. Platform = DecodeIndexPart(parts[0]),
  550. RiskStrategy = DecodeIndexPart(parts[1]),
  551. LaunchScene = launchScene,
  552. AccountId = accountId
  553. };
  554. return true;
  555. }
  556. private static bool IsMatchedParseMetricDimension(
  557. ParseMetricDimension dimension,
  558. int accountId,
  559. bool accountBreakdownOnly,
  560. string platform,
  561. string riskStrategy,
  562. int? launchScene)
  563. {
  564. if (accountBreakdownOnly)
  565. {
  566. if (dimension.AccountId <= 0) return false;
  567. }
  568. else if (accountId > 0)
  569. {
  570. if (dimension.AccountId != accountId) return false;
  571. }
  572. else if (dimension.AccountId > 0)
  573. {
  574. return false;
  575. }
  576. if (!string.IsNullOrWhiteSpace(platform) && !string.Equals(NormalizeParseMetricPlatform(platform), dimension.Platform, StringComparison.OrdinalIgnoreCase)) return false;
  577. if (!string.IsNullOrWhiteSpace(riskStrategy) && !string.Equals(NormalizeReportScene(riskStrategy), dimension.RiskStrategy, StringComparison.OrdinalIgnoreCase)) return false;
  578. if (launchScene.HasValue && NormalizeParseMetricLaunchScene(dimension.RiskStrategy, launchScene.Value) != NormalizeParseMetricLaunchScene(dimension.RiskStrategy, dimension.LaunchScene)) return false;
  579. return true;
  580. }
  581. private static bool IsSameParseMetricDimension(
  582. ParseMetricDimension dimension,
  583. string platform,
  584. string riskStrategy,
  585. int? launchScene)
  586. {
  587. if (!string.IsNullOrWhiteSpace(platform)
  588. && !string.Equals(NormalizeParseMetricPlatform(platform), dimension.Platform, StringComparison.OrdinalIgnoreCase))
  589. {
  590. return false;
  591. }
  592. if (!string.IsNullOrWhiteSpace(riskStrategy)
  593. && !string.Equals(NormalizeReportScene(riskStrategy), dimension.RiskStrategy, StringComparison.OrdinalIgnoreCase))
  594. {
  595. return false;
  596. }
  597. if (launchScene.HasValue
  598. && NormalizeParseMetricLaunchScene(dimension.RiskStrategy, launchScene.Value) != NormalizeParseMetricLaunchScene(dimension.RiskStrategy, dimension.LaunchScene))
  599. {
  600. return false;
  601. }
  602. return true;
  603. }
  604. private static ParseMetricDimension ResolveParseMetricDimension(string channel, string riskStrategy, int launchScene)
  605. {
  606. channel = Normalize(channel).ToLowerInvariant();
  607. riskStrategy = NormalizeReportScene(riskStrategy);
  608. switch (channel)
  609. {
  610. case "tbpush":
  611. case "brwsimilar":
  612. case "icon":
  613. riskStrategy = channel;
  614. launchScene = 0;
  615. channel = "tb";
  616. break;
  617. }
  618. switch (riskStrategy)
  619. {
  620. case "tbpush":
  621. case "brwsimilar":
  622. case "icon":
  623. if (launchScene == -1) launchScene = 0;
  624. break;
  625. }
  626. launchScene = NormalizeParseMetricLaunchScene(riskStrategy, launchScene);
  627. return new ParseMetricDimension
  628. {
  629. Platform = channel,
  630. RiskStrategy = riskStrategy,
  631. LaunchScene = launchScene,
  632. AccountId = 0
  633. };
  634. }
  635. private static int NormalizeParseMetricLaunchScene(string riskStrategy, int launchScene)
  636. {
  637. if (launchScene != -1) return launchScene;
  638. return NormalizeReportScene(riskStrategy).ToLowerInvariant() switch
  639. {
  640. "os" or "tbpush" or "brwsimilar" or "icon" => 0,
  641. _ => launchScene
  642. };
  643. }
  644. private static string NormalizeParseMetricPlatform(string platform)
  645. {
  646. string value = Normalize(platform).ToLowerInvariant();
  647. return value switch
  648. {
  649. "淘宝" or "taobao" or "tb" or "1" => "tb",
  650. "京东" or "jd" or "13" => "jd",
  651. "拼多多" or "pdd" or "9" => "pdd",
  652. _ => value
  653. };
  654. }
  655. private static bool TableExists(IDbConnection conn, string tableName)
  656. {
  657. const string sql = @"
  658. SELECT COUNT(1)
  659. FROM information_schema.TABLES
  660. WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=@tableName";
  661. return conn.ExecuteScalar<int>(sql, new { tableName }) > 0;
  662. }
  663. private static int SafeInt(long value)
  664. {
  665. if (value <= 0) return 0;
  666. return value > int.MaxValue ? int.MaxValue : (int)value;
  667. }
  668. private sealed record ParseMetricCounterKey(
  669. string Metric,
  670. string BucketType,
  671. string BucketValue,
  672. string Platform,
  673. string RiskStrategy,
  674. int LaunchScene,
  675. int AccountId);
  676. private sealed class ParseMetricCounter
  677. {
  678. public long Count = 0;
  679. }
  680. private sealed class ParseMetricDimension
  681. {
  682. public string Platform { get; set; } = string.Empty;
  683. public string RiskStrategy { get; set; } = string.Empty;
  684. public int LaunchScene { get; set; } = -1;
  685. public int AccountId { get; set; } = 0;
  686. }
  687. private sealed record ParseMetricBackfillSource(string Platform, string TableName);
  688. private sealed class ParseMetricBackfillRow
  689. {
  690. public string risk_strategy { get; set; } = string.Empty;
  691. public int launch_scene { get; set; } = -1;
  692. public int account_id { get; set; } = 0;
  693. public string hour_key { get; set; } = string.Empty;
  694. public long call_count { get; set; } = 0;
  695. public long success_count { get; set; } = 0;
  696. }
  697. }
  698. }