diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs index 5aa98422..78c97f41 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs @@ -352,37 +352,99 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe if (config.IsAllPush) { var history = await _trackPushRecordEsRepository.GetLatestTrackPushRecordAsync(log.OrderId); + //找到历史中推送成功的轨迹 + var trackPushRecords = await _trackPushRecordEsRepository.GetHistorySuccessTrackPushRecordAsync(log.OrderId); + //构建标识符，code+比较字段的值(如果提供的比较字段,否则使用轨迹时间) + var pushIdentitySuccess = trackPushRecords.Select(x => + { + //寻找映射配置 + var code = _pushConfigs.First(y => y.TrackCode == x.TrackCode).TrackContentCodeValue; + if (string.IsNullOrWhiteSpace(compareFields)) + return code; + + //尝试从扩展中提取目标值 + var extraValue = TextJsonHelper.ExtractValue(x.Extra, compareFields); + return code + extraValue; + }).ToHashSet(); + if (history != default) { body = TextJsonHelper.Deserialize(history.PushContent); + //todo-test:在历史选出推送成功的历史轨迹 + body!.Data = body.Data.Where(x => + { + var code = ((JsonElement) x["code"]!).GetString(); + if (string.IsNullOrWhiteSpace(compareFields)) + return pushIdentitySuccess.Contains(code); + + var isMatch = ((JsonElement) x["extra"]!).TryGetProperty(compareFields, out var jsonElement); + var extraValue = isMatch ? TextJsonHelper.GetString(jsonElement) : null; + return pushIdentitySuccess.Contains(code + extraValue); + }).ToList(); + + //添加当前要推送的轨迹 + body.Data.Add(cur); - //在历史选出推送成功的历史轨迹 if (config.IsAllPushWithLatest) { - //同code同比较字段保证唯一 - var old = body!.Data - .Where(x => - { - //反序列化出的JsonElement和字典字符串类型做比较 - var a = ((JsonElement) x["code"]!).GetString(); - var b = cur["code"]!.ToString(); //tips: 需要要将object类型的字符串转为string类型，否则使用的是引用比较，反序列化出的string一定和本地string引用不同 - return a == b; - }).WhereIf(!compareFields.IsNullOrWhiteSpace(), x => - { - if (x["extra"] == null || log.Extra == null) return false; - return TextJsonHelper.EqByJsonElementField(x["extra"]!, TextJsonHelper.Deserialize(log.Extra), compareFields!); - }) - .FirstOrDefault(); - - if (old != null) + //历史轨迹中同一个code+比较字段联合作为key最新的一个 + var dict = new Dictionary(); + //todo-test:根据code和比较字段值分组，在每组里选出一个创建时间最新的 + foreach (var x in body.Data) { - body.Data.Remove(old); + var code = ((JsonElement) x["code"]!).GetString()!; + var dateStr = ((JsonElement) x["date"]!).GetString()!; + var key = string.Empty; + if (string.IsNullOrWhiteSpace(compareFields)) + key = code; + else + { + var isMatch = ((JsonElement) x["extra"]!).TryGetProperty(compareFields, out var jsonElement); + var extraValue = isMatch ? TextJsonHelper.GetString(jsonElement) : null; + key = code + extraValue; + } + + //只保留最新的一条数据 + if (dict.TryGetValue(key, out var tuple)) + { + //如果当前值创建时间大于历史值，则替换历史值 + if (DateTime.Parse(dateStr) > DateTime.Parse(tuple.dateStr)) + dict[key] = (dateStr, x); + } + else + dict[key] = (dateStr, x); } - //根据code和比较字段值分组，在每组里选出一个创建时间最新的 + + // //同code同比较字段保证唯一 + // var old = body!.Data + // .Where(x => + // { + // //反序列化出的JsonElement和字典字符串类型做比较 + // var a = ((JsonElement) x["code"]!).GetString(); + // var b = cur["code"]!.ToString(); //tips: 需要要将object类型的字符串转为string类型，否则使用的是引用比较，反序列化出的string一定和本地string引用不同 + // return a == b; + // }).WhereIf(!compareFields.IsNullOrWhiteSpace(), x => + // { + // if (x["extra"] == null || log.Extra == null) return false; + // return TextJsonHelper.EqByJsonElementField(x["extra"]!, TextJsonHelper.Deserialize(log.Extra), compareFields!); + // }) + // .FirstOrDefault(); + // + // if (old != null) + // { + // body.Data.Remove(old); + // } + + //选出每个组最新的轨迹 + body.Data = dict.Values.Select(x => x.dict).Cast>().ToList(); } - body!.Data.Add(cur); + // //添加当前要推送的轨迹 + // body.Data.Add(cur); } + + //todo-test:按推送映射顺序排序 + } log.PushContent = TextJsonHelper.Serialize(body)!; diff --git a/src/SJZY.Track.Application/Helpers/TextJsonHelper.cs b/src/SJZY.Track.Application/Helpers/TextJsonHelper.cs index 387a9032..e63b0360 100644 --- a/src/SJZY.Track.Application/Helpers/TextJsonHelper.cs +++ b/src/SJZY.Track.Application/Helpers/TextJsonHelper.cs @@ -1,3 +1,4 @@ +using System.Globalization; using System.Text.Json; namespace SJZY.Track.Application.Helpers; @@ -17,6 +18,54 @@ public static class TextJsonHelper return string.IsNullOrWhiteSpace(json) ? default : System.Text.Json.JsonSerializer.Deserialize(json); } + public static Dictionary? DeserializeDictionary(string? dictString) + { + return string.IsNullOrWhiteSpace(dictString) + ? null + : System.Text.Json.JsonSerializer.Deserialize>(dictString); + } + + /// + /// 获取JsonElement字符串表达的值 + /// + /// + /// + public static string? GetString(object? el) + { + if (el is not JsonElement jsonElement) + return default; + return jsonElement.ValueKind switch + { + JsonValueKind.Undefined or JsonValueKind.Null => default, + JsonValueKind.True or JsonValueKind.False => jsonElement.GetBoolean().ToString(), + JsonValueKind.Object or JsonValueKind.Array or JsonValueKind.Number => jsonElement.GetRawText(), + JsonValueKind.String => jsonElement.GetString(), + _ => default + }; + } + + /// + /// 提取字典字符串中某个字段的值 + /// + /// 字典json字符串 + /// + /// + public static object? ExtractValue(string? dictString, string fieldName) + { + if (string.IsNullOrWhiteSpace(dictString)) + { + return null; + } + + var dict = System.Text.Json.JsonSerializer.Deserialize>(dictString); + if (dict == null || !dict.TryGetValue(fieldName, out var value)) + { + return null; + } + + return value; + } + /// /// 两个jsonElement进行比较 /// diff --git a/src/SJZY.Track.Domain/Aggregates/TrackPushAggregate/ITrackPushRecordEsRepository.cs b/src/SJZY.Track.Domain/Aggregates/TrackPushAggregate/ITrackPushRecordEsRepository.cs index ee5840f5..261599f3 100644 --- a/src/SJZY.Track.Domain/Aggregates/TrackPushAggregate/ITrackPushRecordEsRepository.cs +++ b/src/SJZY.Track.Domain/Aggregates/TrackPushAggregate/ITrackPushRecordEsRepository.cs @@ -24,4 +24,12 @@ public interface ITrackPushRecordEsRepository /// /// Task GetLatestTrackPushRecordAsync(int orderId); + + + /// + /// 获取历史中推送成功的记录 + /// + /// + /// + Task> GetHistorySuccessTrackPushRecordAsync(int orderId); } diff --git a/src/SJZY.Track.Repository/Repositories/TrackPushRecordEsRepository.cs b/src/SJZY.Track.Repository/Repositories/TrackPushRecordEsRepository.cs index 5e448eaa..b606bc29 100644 --- a/src/SJZY.Track.Repository/Repositories/TrackPushRecordEsRepository.cs +++ b/src/SJZY.Track.Repository/Repositories/TrackPushRecordEsRepository.cs @@ -71,4 +71,22 @@ public class TrackPushRecordEsRepository : ElasticSearchContextAbstract> GetHistorySuccessTrackPushRecordAsync(int orderId) + { + var data = await Context.SearchAsync(x => x + .Index(GlobalConst.TrackPushRecordIndex) + .Source(s => + s.Excludes(fs => + fs.Field(f => f.PushContent))) + .Query(q => q.Bool(b => + b.Must( + mu => mu.Term(t => t.IsSuccess, true), + mu => mu.Term(t => t.OrderId, orderId) + )) + ) + .Size(1000) + .Sort(s => s.Field(f => f.CreateTime, SortOrder.Descending))); + return data.ToIdList(); + } }