update: 轨迹推送处理,根据code+比较字段作为标识,确保传递的轨迹唯一
This commit is contained in:
parent
7d7bc45c67
commit
e1e387e0d3
@ -352,37 +352,99 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
|
|||||||
if (config.IsAllPush)
|
if (config.IsAllPush)
|
||||||
{
|
{
|
||||||
var history = await _trackPushRecordEsRepository.GetLatestTrackPushRecordAsync(log.OrderId);
|
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)
|
if (history != default)
|
||||||
{
|
{
|
||||||
body = TextJsonHelper.Deserialize<TrackPushInput>(history.PushContent);
|
body = TextJsonHelper.Deserialize<TrackPushInput>(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)
|
if (config.IsAllPushWithLatest)
|
||||||
{
|
{
|
||||||
//同code同比较字段保证唯一
|
//历史轨迹中同一个code+比较字段联合作为key最新的一个
|
||||||
var old = body!.Data
|
var dict = new Dictionary<string, (string dateStr, object dict)>();
|
||||||
.Where(x =>
|
//todo-test:根据code和比较字段值分组,在每组里选出一个创建时间最新的
|
||||||
|
foreach (var x in body.Data)
|
||||||
{
|
{
|
||||||
//反序列化出的JsonElement和字典字符串类型做比较
|
var code = ((JsonElement) x["code"]!).GetString()!;
|
||||||
var a = ((JsonElement) x["code"]!).GetString();
|
var dateStr = ((JsonElement) x["date"]!).GetString()!;
|
||||||
var b = cur["code"]!.ToString(); //tips: 需要要将object类型的字符串转为string类型,否则使用的是引用比较,反序列化出的string一定和本地string引用不同
|
var key = string.Empty;
|
||||||
return a == b;
|
if (string.IsNullOrWhiteSpace(compareFields))
|
||||||
}).WhereIf(!compareFields.IsNullOrWhiteSpace(), x =>
|
key = code;
|
||||||
|
else
|
||||||
{
|
{
|
||||||
if (x["extra"] == null || log.Extra == null) return false;
|
var isMatch = ((JsonElement) x["extra"]!).TryGetProperty(compareFields, out var jsonElement);
|
||||||
return TextJsonHelper.EqByJsonElementField(x["extra"]!, TextJsonHelper.Deserialize<JsonElement>(log.Extra), compareFields!);
|
var extraValue = isMatch ? TextJsonHelper.GetString(jsonElement) : null;
|
||||||
})
|
key = code + extraValue;
|
||||||
.FirstOrDefault();
|
|
||||||
|
|
||||||
if (old != null)
|
|
||||||
{
|
|
||||||
body.Data.Remove(old);
|
|
||||||
}
|
|
||||||
//根据code和比较字段值分组,在每组里选出一个创建时间最新的
|
|
||||||
}
|
}
|
||||||
|
|
||||||
body!.Data.Add(cur);
|
//只保留最新的一条数据
|
||||||
|
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同比较字段保证唯一
|
||||||
|
// 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<JsonElement>(log.Extra), compareFields!);
|
||||||
|
// })
|
||||||
|
// .FirstOrDefault();
|
||||||
|
//
|
||||||
|
// if (old != null)
|
||||||
|
// {
|
||||||
|
// body.Data.Remove(old);
|
||||||
|
// }
|
||||||
|
|
||||||
|
//选出每个组最新的轨迹
|
||||||
|
body.Data = dict.Values.Select(x => x.dict).Cast<Dictionary<string, object?>>().ToList();
|
||||||
|
}
|
||||||
|
|
||||||
|
// //添加当前要推送的轨迹
|
||||||
|
// body.Data.Add(cur);
|
||||||
|
}
|
||||||
|
|
||||||
|
//todo-test:按推送映射顺序排序
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
log.PushContent = TextJsonHelper.Serialize(body)!;
|
log.PushContent = TextJsonHelper.Serialize(body)!;
|
||||||
|
|||||||
@ -1,3 +1,4 @@
|
|||||||
|
using System.Globalization;
|
||||||
using System.Text.Json;
|
using System.Text.Json;
|
||||||
|
|
||||||
namespace SJZY.Track.Application.Helpers;
|
namespace SJZY.Track.Application.Helpers;
|
||||||
@ -17,6 +18,54 @@ public static class TextJsonHelper
|
|||||||
return string.IsNullOrWhiteSpace(json) ? default : System.Text.Json.JsonSerializer.Deserialize<T>(json);
|
return string.IsNullOrWhiteSpace(json) ? default : System.Text.Json.JsonSerializer.Deserialize<T>(json);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public static Dictionary<string, object?>? DeserializeDictionary(string? dictString)
|
||||||
|
{
|
||||||
|
return string.IsNullOrWhiteSpace(dictString)
|
||||||
|
? null
|
||||||
|
: System.Text.Json.JsonSerializer.Deserialize<Dictionary<string, object?>>(dictString);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// 获取JsonElement字符串表达的值
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="el"></param>
|
||||||
|
/// <returns></returns>
|
||||||
|
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
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// 提取字典字符串中某个字段的值
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="dictString">字典json字符串</param>
|
||||||
|
/// <param name="fieldName"></param>
|
||||||
|
/// <returns></returns>
|
||||||
|
public static object? ExtractValue(string? dictString, string fieldName)
|
||||||
|
{
|
||||||
|
if (string.IsNullOrWhiteSpace(dictString))
|
||||||
|
{
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
|
||||||
|
var dict = System.Text.Json.JsonSerializer.Deserialize<Dictionary<string, object?>>(dictString);
|
||||||
|
if (dict == null || !dict.TryGetValue(fieldName, out var value))
|
||||||
|
{
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
|
||||||
|
return value;
|
||||||
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// 两个jsonElement进行比较
|
/// 两个jsonElement进行比较
|
||||||
/// </summary>
|
/// </summary>
|
||||||
|
|||||||
@ -24,4 +24,12 @@ public interface ITrackPushRecordEsRepository
|
|||||||
/// <param name="orderId"></param>
|
/// <param name="orderId"></param>
|
||||||
/// <returns></returns>
|
/// <returns></returns>
|
||||||
Task<TrackPushRecordIndex?> GetLatestTrackPushRecordAsync(int orderId);
|
Task<TrackPushRecordIndex?> GetLatestTrackPushRecordAsync(int orderId);
|
||||||
|
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// 获取历史中推送成功的记录
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="orderId"></param>
|
||||||
|
/// <returns></returns>
|
||||||
|
Task<List<TrackPushRecordIndex>> GetHistorySuccessTrackPushRecordAsync(int orderId);
|
||||||
}
|
}
|
||||||
|
|||||||
@ -71,4 +71,22 @@ public class TrackPushRecordEsRepository : ElasticSearchContextAbstract<TrackPus
|
|||||||
|
|
||||||
return data.Documents.FirstOrDefault();
|
return data.Documents.FirstOrDefault();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public async Task<List<TrackPushRecordIndex>> GetHistorySuccessTrackPushRecordAsync(int orderId)
|
||||||
|
{
|
||||||
|
var data = await Context.SearchAsync<TrackPushRecordIndex>(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();
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user