708 lines
34 KiB
C#
708 lines
34 KiB
C#
using Nest;
|
||
using SJZY.Expand.ElasticSearch.Services;
|
||
using SJZY.Track.Domain.Aggregates.WarningAggregate;
|
||
using SJZY.Track.Domain.Aggregates.WarningAggregate.Complete;
|
||
using SJZY.Track.Domain.Aggregates.WarningAggregate.Return;
|
||
using SJZY.Track.Domain.Aggregates.WarningAggregate.Transport;
|
||
using SJZY.Track.Domain.Consts;
|
||
using SJZY.Track.Domain.Dtos.Warning;
|
||
using SJZY.Track.Domain.Enums;
|
||
using SJZY.Track.Domain.Extensions;
|
||
|
||
namespace SJZY.Track.Repository.Repositories;
|
||
|
||
public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransientDependency
|
||
{
|
||
private readonly IElasticClient _client;
|
||
|
||
public WarningDetailsEsRepository(ConnectionFactory connection)
|
||
{
|
||
_client = connection.GetContext(typeof(TransportWarningDetailsIndex));
|
||
}
|
||
|
||
#region 运输时效
|
||
|
||
public async Task<UnifyPageResultDto<TransportWarningDetailsIndex>> GetPageTransportWarningDetailsAsync(GetPageTransportWarningDetailsInputVO input)
|
||
{
|
||
var musts = ListWrapperFactory.CreateListWrapper<TransportWarningDetailsIndex, GetPageTransportWarningDetailsInputVO>(input);
|
||
|
||
var orderNos = input.OrderNos?.Split(',');
|
||
|
||
// 精确匹配
|
||
musts.AddIf(x => x.CustomId > 0, c => c.Term(f => f.CustomId, input.CustomId));
|
||
musts.AddIf(_ => orderNos?.Length > 0, c => c.Terms(f => f.Field(d => d.OrderNo.Suffix("keyword")).Terms(orderNos)));
|
||
musts.AddIf(x => x.OrderType > 0, c => c.Term(f => f.OrderType, input.OrderType));
|
||
musts.AddIf(x => x.Status > 0, c => c.Term(f => f.Status, input.Status));
|
||
musts.AddIf(x => x.CustomId > 0, c => c.Term(f => f.CustomId, input.CustomId));
|
||
musts.AddIf(x => !string.IsNullOrWhiteSpace(x.WarningName), c => c.Term(f => f.WarningName, input.WarningName));
|
||
musts.AddIf(x => x.CreateTimeStart.HasValue, c => c.DateRange(r => r.Field(f => f.CreateTime).GreaterThanOrEquals(input.CreateTimeStart)));
|
||
musts.AddIf(x => x.CreateTimeEnd.HasValue, c => c.DateRange(r => r.Field(f => f.CreateTime).LessThanOrEquals(input.CreateTimeEnd)));
|
||
|
||
// 模糊匹配userId
|
||
musts.AddIf(x => x.NotifyUserId.HasValue, c => c.Match(f => f.Field(d => d.NotifyUsers).Query(input.NotifyUserId.ToString())));
|
||
|
||
var resp = await _client.SearchAsync<TransportWarningDetailsIndex>(x => x
|
||
.Index(GlobalConst.TransportWarningDetailsIndex)
|
||
.Query(q => q.Bool(b => b.Must(musts)))
|
||
.From((input.PageIndex - 1) * input.PageSize)
|
||
.Size(input.PageSize));
|
||
|
||
return new UnifyPageResultDto<TransportWarningDetailsIndex>(resp.ToIdList(), input.PageIndex, input.PageSize, (int) resp.Total);
|
||
}
|
||
|
||
public async Task<List<TransportWarningDetailsIndex>> GetTransportWarningDetailsAsync(HashSet<int> ruleItemIds, HashSet<int> status)
|
||
{
|
||
var r = await _client.SearchAsync<TransportWarningDetailsIndex>(x => x
|
||
.Index(GlobalConst.TransportWarningDetailsIndex)
|
||
.Query(q => q
|
||
.Bool(b => b
|
||
.Must(
|
||
mu => mu.Terms(t => t.Field(f => f.WarningSceneRuleItemId).Terms(ruleItemIds)),
|
||
mu => mu.Terms(t => t.Field(f => f.Status).Terms(status)))
|
||
)));
|
||
|
||
return r.Documents.ToList();
|
||
}
|
||
|
||
public async Task UpdateTransportWarningDetailsAsync(TransportWarningDetailsIndex detail)
|
||
{
|
||
var r = await _client.UpdateAsync<TransportWarningDetailsIndex, object>(detail.Id,
|
||
u => u.Index(detail.IndexName)
|
||
.Doc(new
|
||
{
|
||
detail.WarningId,
|
||
detail.WarningSceneId,
|
||
detail.WarningSceneRuleItemId,
|
||
detail.Version,
|
||
detail.WarningName,
|
||
detail.WarningRuleDescription,
|
||
detail.TransportTimeSpan,
|
||
detail.StartTrackWarningTimeSpan,
|
||
//detail.NotifyModes,
|
||
detail.NotifyUsers,
|
||
detail.SubSystemId,
|
||
detail.CustomId,
|
||
detail.CustomName,
|
||
detail.OrderId,
|
||
detail.OrderNo,
|
||
detail.OrderType,
|
||
detail.StartTrackStatusId,
|
||
detail.StartTrackContentId,
|
||
detail.StartTrackTime,
|
||
detail.StartTrackTimeZone,
|
||
detail.EndTrackStatusId,
|
||
detail.EndTrackContentId,
|
||
detail.EndTrackTime,
|
||
detail.EndTrackTimeZone,
|
||
// detail.LatestTrackContentId,
|
||
// detail.LatestTrackContent,
|
||
// detail.LatestTrackTime,
|
||
// detail.LatestTrackTimeZone,
|
||
// detail.LatestTrackPlace,
|
||
detail.CumulativeTimeSpan,
|
||
detail.Status,
|
||
detail.CreateTime,
|
||
}));
|
||
}
|
||
|
||
public async Task BulkUpdateTransportWarningDetailsAsync(List<TransportWarningDetailsIndex> details, int maxRetries = 3)
|
||
{
|
||
if (details.Count == 0) return;
|
||
|
||
//构建批量更新操作描述符
|
||
var desc = new BulkDescriptor().UpdateMany(details,
|
||
(u, detail) => u.Id(detail.Id)
|
||
.Index(detail.IndexName)
|
||
.Script(s => s.Source(@"
|
||
ctx._source.WarningId = param.WarningId;
|
||
ctx._source.WarningSceneId = param.WarningSceneId;
|
||
ctx._source.WarningSceneRuleItemId = param.WarningSceneRuleItemId;
|
||
ctx._source.Version = param.Version;
|
||
ctx._source.WarningName = param.WarningName;
|
||
ctx._source.WarningRuleDescription = param.WarningRuleDescription;
|
||
ctx._source.TransportTimeSpan = param.TransportTimeSpan;
|
||
ctx._source.StartTrackWarningTimeSpa = param.StartTrackWarningTimeSpa;n
|
||
//ctx._source.NotifyModes = param.NotifyModes;
|
||
ctx._source.NotifyUsers = param.NotifyUsers;
|
||
ctx._source.SubSystemId = param.SubSystemId;
|
||
ctx._source.CustomId = param.CustomId;
|
||
ctx._source.CustomName = param.CustomName;
|
||
ctx._source.OrderId = param.OrderId;
|
||
ctx._source.OrderNo = param.OrderNo;
|
||
ctx._source.OrderType = param.OrderType;
|
||
ctx._source.StartTrackStatusId = param.StartTrackStatusId;
|
||
ctx._source.StartTrackContentId = param.StartTrackContentId;
|
||
ctx._source.StartTrackTime = param.StartTrackTime;
|
||
ctx._source.StartTrackTimeZone = param.StartTrackTimeZone;
|
||
ctx._source.EndTrackStatusId = param.EndTrackStatusId;
|
||
ctx._source.EndTrackContentId = param.EndTrackContentId;
|
||
ctx._source.EndTrackTime = param.EndTrackTime;
|
||
ctx._source.EndTrackTimeZone = param.EndTrackTimeZone;
|
||
//ctx._source.LatestTrackContentId = param.LatestTrackContentId;
|
||
//ctx._source.LatestTrackContent = param.LatestTrackContent;
|
||
//ctx._source.LatestTrackTime = param.LatestTrackTime;
|
||
//ctx._source.LatestTrackTimeZone = param.LatestTrackTimeZone;
|
||
//ctx._source.LatestTrackPlace = param.LatestTrackPlace;
|
||
ctx._source.CumulativeTimeSpan = param.CumulativeTimeSpan;
|
||
ctx._source.Status = param.Status;
|
||
ctx._source.CreateTime = param.CreateTime;
|
||
").Params(p => p
|
||
.Add("WarningId", detail.WarningId)
|
||
.Add("WarningSceneId", detail.WarningSceneId)
|
||
.Add("WarningSceneRuleItemId", detail.WarningSceneRuleItemId)
|
||
.Add("Version", detail.Version)
|
||
.Add("WarningName", detail.WarningName)
|
||
.Add("WarningRuleDescription", detail.WarningRuleDescription)
|
||
.Add("TransportTimeSpan", detail.TransportTimeSpan)
|
||
.Add("StartTrackWarningTimeSpan", detail.StartTrackWarningTimeSpan)
|
||
//.Add("NotifyModes", detail.NotifyModes)
|
||
.Add("NotifyUsers", detail.NotifyUsers)
|
||
.Add("SubSystemId", detail.SubSystemId)
|
||
.Add("CustomId", detail.CustomId)
|
||
.Add("CustomName", detail.CustomName)
|
||
.Add("OrderId", detail.OrderId)
|
||
.Add("OrderNo", detail.OrderNo)
|
||
.Add("OrderType", detail.OrderType)
|
||
.Add("StartTrackStatusId", detail.StartTrackStatusId)
|
||
.Add("StartTrackContentId", detail.StartTrackContentId)
|
||
.Add("StartTrackTime", detail.StartTrackTime)
|
||
.Add("StartTrackTimeZone", detail.StartTrackTimeZone)
|
||
.Add("EndTrackStatusId", detail.EndTrackStatusId)
|
||
.Add("EndTrackContentId", detail.EndTrackContentId)
|
||
.Add("EndTrackTime", detail.EndTrackTime)
|
||
.Add("EndTrackTimeZone", detail.EndTrackTimeZone)
|
||
// .Add("LatestTrackContentId", detail.LatestTrackContentId)
|
||
// .Add("LatestTrackContent", detail.LatestTrackContent)
|
||
// .Add("LatestTrackTime", detail.LatestTrackTime)
|
||
// .Add("LatestTrackTimeZone", detail.LatestTrackTimeZone)
|
||
// .Add("LatestTrackPlace", detail.LatestTrackPlace)
|
||
// .Add("CumulativeTimeSpan", detail.CumulativeTimeSpan)
|
||
.Add("Status", detail.Status)
|
||
.Add("CreateTime", detail.CreateTime)
|
||
)));
|
||
|
||
await PollyHelper.RetryPollyExecAsync(() => _client.BulkAsync(desc), response =>
|
||
{
|
||
if (response.IsValid) // 认为成功
|
||
return true;
|
||
|
||
if (response.ServerError.Status == 409) //版本冲突认为失败，则重试
|
||
return false;
|
||
|
||
throw new InvalidOperationException($"ES更新失败：{details.Count}条，原因：{response.DebugInformation}。");
|
||
});
|
||
}
|
||
|
||
public async Task BulkCloseTransportWarningDetailsAsync(HashSet<string> detailIds, int maxRetries = 3)
|
||
{
|
||
var status = new HashSet<int>(2)
|
||
{
|
||
TransportWarningStatus.即将超时.GetHashCode(),
|
||
TransportWarningStatus.已超时.GetHashCode()
|
||
};
|
||
|
||
var r = await _client.UpdateByQueryAsync<TransportWarningDetailsIndex>(u =>
|
||
u.Index(GlobalConst.TransportWarningDetailsIndex)
|
||
.Query(q => q
|
||
.Bool(b => b.Must(
|
||
mu => mu.Terms(t => t.Field(f => f.Status).Terms(status)),
|
||
mu => mu.Terms(t => t.Field(f => f.Id).Terms(detailIds))
|
||
)
|
||
)
|
||
)
|
||
.Script(s => s
|
||
.Source("ctx._source.status = params.status")
|
||
.Params(p => p.Add("status", (int) TransportWarningStatus.已关闭))
|
||
)
|
||
);
|
||
|
||
if (!r.IsValid)
|
||
{
|
||
throw new InvalidOperationException(
|
||
$"运输时效预警明细更新失败，detailsIds：{detailIds}，原因：{r.DebugInformation}。");
|
||
}
|
||
}
|
||
|
||
public async Task BulkCloseTransportWarningDetailsAsync(HashSet<int> ruleItemIds, int maxRetries = 3)
|
||
{
|
||
var status = new HashSet<int>(2)
|
||
{
|
||
TransportWarningStatus.即将超时.GetHashCode(),
|
||
TransportWarningStatus.已超时.GetHashCode()
|
||
};
|
||
|
||
var r = await _client.UpdateByQueryAsync<TransportWarningDetailsIndex>(u =>
|
||
u.Index(GlobalConst.TransportWarningDetailsIndex)
|
||
.Query(q => q
|
||
.Bool(b => b.Must(
|
||
mu => mu.Terms(t => t.Field(f => f.Status).Terms(status)),
|
||
mu => mu.Terms(t => t.Field(f => f.WarningSceneRuleItemId).Terms(ruleItemIds))
|
||
)
|
||
)
|
||
)
|
||
.Script(s => s
|
||
.Source("ctx._source.status = params.status")
|
||
.Params(p => p.Add("status", (int) TransportWarningStatus.已关闭))
|
||
)
|
||
);
|
||
|
||
if (!r.IsValid)
|
||
{
|
||
throw new InvalidOperationException(
|
||
$"运输时效预警明细更新失败，ruleItemIds：{ruleItemIds}，原因：{r.DebugInformation}。");
|
||
}
|
||
}
|
||
|
||
public async Task InsertTransportWarningDetailsAsync(TransportWarningDetailsIndex detail)
|
||
{
|
||
var indexName = $"{GlobalConst.TransportWarningDetailsIndex}-{DateTime.Now:yyyy.MM.dd}";
|
||
detail.IndexName = indexName;
|
||
var r = await _client.IndexAsync(detail, s => s.Index(indexName));
|
||
|
||
if (!r.IsValid)
|
||
{
|
||
throw new InvalidOperationException($"运输预警明细新增失败，{detail}，原因：{r.DebugInformation}。");
|
||
}
|
||
}
|
||
|
||
public async Task BulkInsertTransportWarningDetailsAsync(List<TransportWarningDetailsIndex> details, int maxRetries = 3)
|
||
{
|
||
// 按天写入es
|
||
var indexName = $"{GlobalConst.TransportWarningDetailsIndex}-{DateTime.Now:yyyy.MM.dd}";
|
||
details.ForEach(x => x.IndexName = indexName);
|
||
|
||
var r = await _client.IndexManyAsync(details, indexName);
|
||
|
||
if (!r.IsValid)
|
||
{
|
||
throw new InvalidOperationException($"运输预警明细新增失败，原因：{r.DebugInformation}。");
|
||
}
|
||
}
|
||
|
||
#endregion
|
||
|
||
#region 回传时效
|
||
|
||
public async Task<UnifyPageResultDto<ReturnWarningDetailsIndex>> GetPageReturnWarningDetailsAsync(GetPageReturnWarningDetailsInputVO input)
|
||
{
|
||
var musts = ListWrapperFactory.CreateListWrapper<ReturnWarningDetailsIndex, GetPageReturnWarningDetailsInputVO>(input);
|
||
|
||
var orderNos = input.OrderNos?.Split(',');
|
||
|
||
// 精确匹配
|
||
musts.AddIf(x => x.SubSystemId > 0, c => c.Term(f => f.SubSystemId, input.SubSystemId));
|
||
musts.AddIf(_ => orderNos?.Length > 0, c => c.Terms(f => f.Field(d => d.OrderNo.Suffix("keyword")).Terms(orderNos)));
|
||
musts.AddIf(x => x.OrderType > 0, c => c.Term(f => f.OrderType, input.OrderType));
|
||
musts.AddIf(x => x.Status > 0, c => c.Term(f => f.Status, input.Status));
|
||
musts.AddIf(x => x.CustomId > 0, c => c.Term(f => f.CustomId, input.CustomId));
|
||
musts.AddIf(x => x.TrackStatusId > 0, c => c.Term(f => f.TrackStatusId, input.TrackStatusId));
|
||
musts.AddIf(x => x.TrackContentId > 0, c => c.Term(f => f.TrackContentId, input.TrackContentId));
|
||
musts.AddIf(x => x.IsKpi.HasValue, c => c.Term(f => f.IsExistsKpi, input.IsKpi!.Value));
|
||
musts.AddIf(x => x.TrackTimeStart.HasValue, c => c.DateRange(r => r.Field(f => f.TrackTime).GreaterThanOrEquals(input.TrackTimeStart)));
|
||
musts.AddIf(x => x.TrackTimeEnd.HasValue, c => c.DateRange(r => r.Field(f => f.TrackTime).LessThanOrEquals(input.TrackTimeEnd)));
|
||
musts.AddIf(x => x.TrackCreateTimeStart.HasValue, c => c.DateRange(r => r.Field(f => f.TrackCreateTime).LessThanOrEquals(input.TrackCreateTimeStart)));
|
||
musts.AddIf(x => x.TrackCreateTimeEnd.HasValue, c => c.DateRange(r => r.Field(f => f.TrackCreateTime).LessThanOrEquals(input.TrackCreateTimeEnd)));
|
||
musts.AddIf(x => x.TrackReturnTimeStart.HasValue, c => c.DateRange(r => r.Field(f => f.TrackReturnTime).LessThanOrEquals(input.TrackReturnTimeStart)));
|
||
musts.AddIf(x => x.TrackReturnTimeEnd.HasValue, c => c.DateRange(r => r.Field(f => f.TrackReturnTime).LessThanOrEquals(input.TrackReturnTimeEnd)));
|
||
musts.AddIf(x => x.WarningCreateTimeStart.HasValue, c => c.DateRange(r => r.Field(f => f.CreateTime).LessThanOrEquals(input.WarningCreateTimeStart)));
|
||
musts.AddIf(x => x.WarningCreateTimeEnd.HasValue, c => c.DateRange(r => r.Field(f => f.CreateTime).LessThanOrEquals(input.WarningCreateTimeEnd)));
|
||
|
||
// 模糊匹配userId（json字符串包含userId）
|
||
musts.AddIf(x => x.NotifyUserId.HasValue, c => c.Match(f => f.Field(d => d.NotifyUsers).Query(input.NotifyUserId.ToString())));
|
||
|
||
var resp = await _client.SearchAsync<ReturnWarningDetailsIndex>(x => x
|
||
.Index(GlobalConst.ReturnWarningDetailsIndex)
|
||
.Query(q => q.Bool(b => b.Must(musts)))
|
||
.From((input.PageIndex - 1) * input.PageSize)
|
||
.Size(input.PageSize));
|
||
|
||
return new UnifyPageResultDto<ReturnWarningDetailsIndex>(resp.ToIdList(), input.PageIndex, input.PageSize, (int) resp.Total);
|
||
}
|
||
|
||
public async Task<List<ReturnMessageRecordIndex>> GetReturnMessageRecordListByIdAsync(string esId)
|
||
{
|
||
var resp = await _client.SearchAsync<ReturnMessageRecordIndex>(x => x
|
||
.Index(GlobalConst.ReturnMessageRecordIndex)
|
||
.Query(q => q.Term(t => t.WarningDetailsId.Suffix("keyword"), esId))
|
||
.Size(1000)
|
||
);
|
||
|
||
return resp.ToIdList();
|
||
}
|
||
|
||
public async Task UpdateReturnWarningDetailsAsync(ReturnWarningDetailsIndex detail)
|
||
{
|
||
var r = await _client.UpdateAsync<ReturnWarningDetailsIndex, object>(detail.Id,
|
||
u => u.Index(detail.IndexName)
|
||
.Doc(new
|
||
{
|
||
detail.ReturnWarningRuleId,
|
||
detail.WarningId,
|
||
detail.Version,
|
||
detail.ReturnTimeSpan,
|
||
//detail.NotifyModes,
|
||
detail.NotifyUsers,
|
||
detail.SubSystemId,
|
||
detail.CustomId,
|
||
detail.OrderId,
|
||
detail.OrderNo,
|
||
detail.OrderType,
|
||
detail.TrackStatusId,
|
||
detail.TrackStatus,
|
||
detail.CompareFields,
|
||
detail.Extra,
|
||
detail.TrackContentId,
|
||
detail.TrackContent,
|
||
detail.TrackCode,
|
||
detail.TrackPlace,
|
||
detail.TrackTime,
|
||
detail.TrackTimeZone,
|
||
detail.TrackCreateTime,
|
||
detail.PushTrackStatus,
|
||
detail.PushTrackCode,
|
||
detail.PushTrackContent,
|
||
detail.PushTrackEnContent,
|
||
detail.TrackReturnTime,
|
||
detail.RetryCount,
|
||
detail.Status,
|
||
detail.ActualReturnTimeSpan,
|
||
detail.SystemReturnTimeSpan,
|
||
detail.TrackCreateTimeSpan,
|
||
detail.CreateTime,
|
||
detail.UpdateTime,
|
||
}));
|
||
}
|
||
|
||
public async Task BulkUpdateReturnWarningDetailsAsync(List<ReturnWarningDetailsIndex> details, int maxRetries = 3)
|
||
{
|
||
if (details.Count == 0) return;
|
||
|
||
//构建批量更新操作描述符
|
||
var desc = new BulkDescriptor().UpdateMany(details,
|
||
(u, detail) => u.Id(detail.Id)
|
||
.Index(detail.IndexName)
|
||
.Script(s => s.Source(@"
|
||
ctx._source.ReturnWarningRuleId = param.ReturnWarningRuleId;
|
||
ctx._source.WarningId = param.WarningId;
|
||
ctx._source.Version = param.Version;
|
||
ctx._source.ReturnTimeSpan = param.ReturnTimeSpan;
|
||
//ctx._source.NotifyModes = param.NotifyModes;
|
||
ctx._source.NotifyUsers = param.NotifyUsers;
|
||
ctx._source.SubSystemId = param.SubSystemId;
|
||
ctx._source.CustomId = param.CustomId;
|
||
ctx._source.OrderId = param.OrderId;
|
||
ctx._source.OrderNo = param.OrderNo;
|
||
ctx._source.OrderType = param.OrderType;
|
||
ctx._source.TrackStatusId = param.TrackStatusId;
|
||
ctx._source.TrackStatus = param.TrackStatus;
|
||
ctx._source.CompareFields = param.CompareFields;
|
||
ctx._source.Extra = param.Extra;
|
||
ctx._source.TrackContentId = param.TrackContentId;
|
||
ctx._source.TrackContent = param.TrackContent;
|
||
ctx._source.TrackCode = param.TrackCode;
|
||
ctx._source.TrackPlace = param.TrackPlace;
|
||
ctx._source.TrackTime = param.TrackTime;
|
||
ctx._source.TrackTimeZone = param.TrackTimeZone;
|
||
ctx._source.TrackCreateTime = param.TrackCreateTime;
|
||
ctx._source.PushTrackStatus = param.PushTrackStatus;
|
||
ctx._source.PushTrackCode = param.PushTrackCode;
|
||
ctx._source.PushTrackContent = param.PushTrackContent;
|
||
ctx._source.PushTrackEnContent = param.PushTrackEnContent;
|
||
ctx._source.TrackReturnTime = param.TrackReturnTime;
|
||
ctx._source.RetryCount = param.RetryCount;
|
||
ctx._source.Status = param.Status;
|
||
ctx._source.ActualReturnTimeSpan = param.ActualReturnTimeSpan;
|
||
ctx._source.SystemReturnTimeSpan = param.SystemReturnTimeSpan;
|
||
ctx._source.TrackCreateTimeSpan = param.TrackCreateTimeSpan;
|
||
ctx._source.CreateTime = param.CreateTime;
|
||
ctx._source.UpdateTime = param.UpdateTime;
|
||
").Params(p => p
|
||
.Add("ReturnWarningRuleId", detail.ReturnWarningRuleId)
|
||
.Add("WarningId", detail.WarningId)
|
||
.Add("Version", detail.Version)
|
||
.Add("ReturnTimeSpan", detail.ReturnTimeSpan)
|
||
//.Add("NotifyModes", detail.NotifyModes)
|
||
.Add("NotifyUsers", detail.NotifyUsers)
|
||
.Add("SubSystemId", detail.SubSystemId)
|
||
.Add("CustomId", detail.CustomId)
|
||
.Add("OrderId", detail.OrderId)
|
||
.Add("OrderNo", detail.OrderNo)
|
||
.Add("OrderType", detail.OrderType)
|
||
.Add("TrackStatusId", detail.TrackStatusId)
|
||
.Add("TrackStatus", detail.TrackStatus)
|
||
.Add("CompareFields", detail.CompareFields)
|
||
.Add("Extra", detail.Extra)
|
||
.Add("TrackContentId", detail.TrackContentId)
|
||
.Add("TrackContent", detail.TrackContent)
|
||
.Add("TrackCode", detail.TrackCode)
|
||
.Add("TrackPlace", detail.TrackPlace)
|
||
.Add("TrackTime", detail.TrackTime)
|
||
.Add("TrackTimeZone", detail.TrackTimeZone)
|
||
.Add("TrackCreateTime", detail.TrackCreateTime)
|
||
.Add("PushTrackStatus", detail.PushTrackStatus)
|
||
.Add("PushTrackCode", detail.PushTrackCode)
|
||
.Add("PushTrackContent", detail.PushTrackContent)
|
||
.Add("PushTrackEnContent", detail.PushTrackEnContent)
|
||
.Add("TrackReturnTime", detail.TrackReturnTime)
|
||
.Add("RetryCount", detail.RetryCount)
|
||
.Add("Status", detail.Status)
|
||
.Add("ActualReturnTimeSpan", detail.ActualReturnTimeSpan)
|
||
.Add("SystemReturnTimeSpan", detail.SystemReturnTimeSpan)
|
||
.Add("TrackCreateTimeSpan", detail.TrackCreateTimeSpan)
|
||
.Add("CreateTime", detail.CreateTime)
|
||
.Add("UpdateTime", detail.UpdateTime)
|
||
)
|
||
)
|
||
);
|
||
|
||
await PollyHelper.RetryPollyExecAsync(() => _client.BulkAsync(desc), response =>
|
||
{
|
||
if (response.IsValid) // 认为成功
|
||
return true;
|
||
|
||
if (response.ServerError.Status == 409) //版本冲突认为失败，则重试
|
||
return false;
|
||
|
||
throw new InvalidOperationException($"ES更新失败：{details.Count}条，原因：{response.DebugInformation}。");
|
||
});
|
||
}
|
||
|
||
public async Task InsertReturnWarningDetailsAsync(ReturnWarningDetailsIndex detail)
|
||
{
|
||
var indexName = $"{GlobalConst.ReturnWarningDetailsIndex}-{DateTime.Now:yyyy.MM.dd}";
|
||
detail.IndexName = indexName;
|
||
var r = await _client.IndexAsync(detail, s => s.Index(indexName));
|
||
|
||
if (!r.IsValid)
|
||
{
|
||
throw new InvalidOperationException($"回传预警明细新增失败，{detail}，原因：{r.DebugInformation}。");
|
||
}
|
||
}
|
||
|
||
public async Task BulkInsertReturnWarningDetailsAsync(List<ReturnWarningDetailsIndex> details, int maxRetries = 3)
|
||
{
|
||
// 按天写入es
|
||
var indexName = $"{GlobalConst.ReturnWarningDetailsIndex}-{DateTime.Now:yyyy.MM.dd}";
|
||
details.ForEach(x => x.IndexName = indexName);
|
||
|
||
var r = await _client.IndexManyAsync(details, indexName);
|
||
|
||
if (!r.IsValid)
|
||
{
|
||
throw new InvalidOperationException($"回传预警明细新增失败，原因：{r.DebugInformation}。");
|
||
}
|
||
}
|
||
|
||
public async Task InsertReturnMessageRecordAsync(ReturnMessageRecordIndex detail)
|
||
{
|
||
var indexName = $"{GlobalConst.ReturnMessageRecordIndex}-{DateTime.Now:yyyy.MM.dd}";
|
||
detail.IndexName = indexName;
|
||
var r = await _client.IndexAsync(detail, s => s.Index(indexName));
|
||
|
||
if (!r.IsValid)
|
||
{
|
||
throw new InvalidOperationException($"回传日志新增失败，{detail}，原因：{r.DebugInformation}。");
|
||
}
|
||
}
|
||
|
||
public async Task BulkInsertReturnMessageRecordAsync(List<ReturnMessageRecordIndex> details, int maxRetries = 3)
|
||
{
|
||
// 按天写入es
|
||
var indexName = $"{GlobalConst.ReturnMessageRecordIndex}-{DateTime.Now:yyyy.MM.dd}";
|
||
details.ForEach(x => x.IndexName = indexName);
|
||
|
||
var r = await _client.IndexManyAsync(details, indexName);
|
||
|
||
if (!r.IsValid)
|
||
{
|
||
throw new InvalidOperationException($"回传日志新增失败，原因：{r.DebugInformation}。");
|
||
}
|
||
}
|
||
|
||
#endregion
|
||
|
||
#region 完整
|
||
|
||
public async Task<UnifyPageResultDto<CompleteWarningDetailsIndex>> GetPageCompleteWarningDetailsAsync(GetPageCompleteWarningDetailsInputVO input)
|
||
{
|
||
var musts = ListWrapperFactory.CreateListWrapper<CompleteWarningDetailsIndex, GetPageCompleteWarningDetailsInputVO>(input);
|
||
|
||
var orderNos = input.OrderNos?.Split(',');
|
||
|
||
musts.AddIf(x => x.SubSystemId > 0, c => c.Term(f => f.SubSystemId, input.SubSystemId));
|
||
musts.AddIf(_ => orderNos?.Length > 0, c => c.Terms(f => f.Field(d => d.OrderNo.Suffix("keyword")).Terms(orderNos)));
|
||
musts.AddIf(x => x.OrderType > 0, c => c.Term(f => f.OrderType, input.OrderType));
|
||
musts.AddIf(x => x.Status > 0, c => c.Term(f => f.Status, input.Status));
|
||
musts.AddIf(x => x.CustomId > 0, c => c.Term(f => f.CustomId, input.CustomId));
|
||
musts.AddIf(x => x.MissingTrackStatusId > 0, c => c.Match(f => f.Field(d => d.MissingTrackStatusIds).Query(input.MissingTrackStatusId.ToString())));
|
||
musts.AddIf(x => x.WarningCreateTimeStart.HasValue, c => c.DateRange(r => r.Field(f => f.CreateTime).LessThanOrEquals(input.WarningCreateTimeStart)));
|
||
musts.AddIf(x => x.WarningCreateTimeEnd.HasValue, c => c.DateRange(r => r.Field(f => f.CreateTime).LessThanOrEquals(input.WarningCreateTimeEnd)));
|
||
musts.AddIf(x => x.WarningHandleTimeStart.HasValue, c => c.DateRange(r => r.Field(f => f.ProcessedTime).LessThanOrEquals(input.WarningHandleTimeStart)));
|
||
musts.AddIf(x => x.WarningHandleTimeEnd.HasValue, c => c.DateRange(r => r.Field(f => f.ProcessedTime).LessThanOrEquals(input.WarningHandleTimeEnd)));
|
||
|
||
// 模糊匹配userId
|
||
musts.AddIf(x => x.NotifyUserId.HasValue, c => c.Match(f => f.Field(d => d.NotifyUsers).Query(input.NotifyUserId.ToString())));
|
||
|
||
var resp = await _client.SearchAsync<CompleteWarningDetailsIndex>(x => x
|
||
.Index(GlobalConst.CompleteWarningDetailsIndex)
|
||
.Query(q => q.Bool(b => b.Must(musts)))
|
||
.From((input.PageIndex - 1) * input.PageSize)
|
||
.Size(input.PageSize));
|
||
|
||
return new UnifyPageResultDto<CompleteWarningDetailsIndex>(resp.ToIdList(), input.PageIndex, input.PageSize, (int) resp.Total);
|
||
}
|
||
|
||
public async Task UpdateCompleteWarningDetailsAsync(CompleteWarningDetailsIndex detail)
|
||
{
|
||
var r = await _client.UpdateAsync<CompleteWarningDetailsIndex, object>(detail.Id,
|
||
u => u.Index(detail.IndexName)
|
||
.Doc(new
|
||
{
|
||
detail.WarningId,
|
||
detail.CompleteWarningRuleId,
|
||
detail.Version,
|
||
//detail.NotifyModes,
|
||
detail.NotifyUsers,
|
||
detail.CheckPointTrackStatusId,
|
||
detail.MissingTrackStatusIds,
|
||
detail.SubSystemId,
|
||
detail.CustomId,
|
||
detail.OrderId,
|
||
detail.OrderNo,
|
||
detail.OrderType,
|
||
detail.Status,
|
||
detail.CreateTime,
|
||
detail.ProcessedTime,
|
||
}));
|
||
}
|
||
|
||
public async Task BulkUpdateCompleteWarningDetailsAsync(List<CompleteWarningDetailsIndex> details, int maxRetries = 3)
|
||
{
|
||
if (details.Count == 0) return;
|
||
|
||
//构建批量更新操作描述符
|
||
var desc = new BulkDescriptor().UpdateMany(details,
|
||
(u, detail) => u.Id(detail.Id)
|
||
.Index(detail.IndexName)
|
||
.Script(s => s.Source(@"
|
||
ctx._source.WarningId = param.WarningId;
|
||
ctx._source.CompleteWarningRuleId = param.CompleteWarningRuleId;
|
||
ctx._source.Version = param.Version;
|
||
//ctx._source.NotifyModes = param.NotifyModes;
|
||
ctx._source.NotifyUsers = param.NotifyUsers;
|
||
ctx._source.CheckPointTrackStatusId = param.CheckPointTrackStatusId;
|
||
ctx._source.MissingTrackStatusIds = param.MissingTrackStatusIds;
|
||
ctx._source.SubSystemId = param.SubSystemId;
|
||
ctx._source.CustomId = param.CustomId;
|
||
ctx._source.OrderId = param.OrderId;
|
||
ctx._source.OrderNo = param.OrderNo;
|
||
ctx._source.OrderType = param.OrderType;
|
||
ctx._source.Status = param.Status;
|
||
ctx._source.CreateTime = param.CreateTime;
|
||
ctx._source.ProcessedTime = param.ProcessedTime;
|
||
")
|
||
.Params(p => p
|
||
.Add("WarningId", detail.WarningId)
|
||
.Add("CompleteWarningRuleId", detail.CompleteWarningRuleId)
|
||
.Add("Version", detail.Version)
|
||
//.Add("NotifyModes", detail.NotifyModes)
|
||
.Add("NotifyUsers", detail.NotifyUsers)
|
||
.Add("CheckPointTrackStatusId", detail.CheckPointTrackStatusId)
|
||
.Add("MissingTrackStatusIds", detail.MissingTrackStatusIds)
|
||
.Add("SubSystemId", detail.SubSystemId)
|
||
.Add("CustomId", detail.CustomId)
|
||
.Add("OrderId", detail.OrderId)
|
||
.Add("OrderNo", detail.OrderNo)
|
||
.Add("OrderType", detail.OrderType)
|
||
.Add("Status", detail.Status)
|
||
.Add("CreateTime", detail.CreateTime)
|
||
.Add("ProcessedTime", detail.ProcessedTime)
|
||
)
|
||
)
|
||
);
|
||
|
||
await PollyHelper.RetryPollyExecAsync(() => _client.BulkAsync(desc), response =>
|
||
{
|
||
if (response.IsValid) // 认为成功
|
||
return true;
|
||
|
||
if (response.ServerError.Status == 409) //版本冲突认为失败，则重试
|
||
return false;
|
||
|
||
throw new InvalidOperationException($"ES更新失败：{details.Count}条，原因：{response.DebugInformation}。");
|
||
});
|
||
}
|
||
|
||
public async Task InsertCompleteWarningDetailsAsync(CompleteWarningDetailsIndex detail)
|
||
{
|
||
var indexName = $"{GlobalConst.CompleteWarningDetailsIndex}-{DateTime.Now:yyyy.MM.dd}";
|
||
detail.IndexName = indexName;
|
||
var r = await _client.IndexAsync(detail, s => s.Index(indexName));
|
||
|
||
if (!r.IsValid)
|
||
{
|
||
throw new InvalidOperationException($"完整预警明细新增失败，{detail}，原因：{r.DebugInformation}。");
|
||
}
|
||
}
|
||
|
||
public async Task BulkInsertCompleteWarningDetailsAsync(List<CompleteWarningDetailsIndex> details, int maxRetries = 3)
|
||
{
|
||
// 按天写入es
|
||
var indexName = $"{GlobalConst.CompleteWarningDetailsIndex}-{DateTime.Now:yyyy.MM.dd}";
|
||
details.ForEach(x => x.IndexName = indexName);
|
||
|
||
var r = await _client.IndexManyAsync(details, indexName);
|
||
|
||
if (!r.IsValid)
|
||
{
|
||
throw new InvalidOperationException($"完整预警明细新增失败，原因：{r.DebugInformation}。");
|
||
}
|
||
}
|
||
|
||
#endregion
|
||
|
||
public async Task CreateAliasAsync()
|
||
{
|
||
var indexName1 = $"{GlobalConst.TransportWarningDetailsIndex}-{DateTime.Now:yyyy.MM.dd}";
|
||
var indexName2 = $"{GlobalConst.ReturnWarningDetailsIndex}-{DateTime.Now:yyyy.MM.dd}";
|
||
var indexName3 = $"{GlobalConst.CompleteWarningDetailsIndex}-{DateTime.Now:yyyy.MM.dd}";
|
||
var indexName4 = $"{GlobalConst.ReturnMessageRecordIndex}-{DateTime.Now:yyyy.MM.dd}";
|
||
|
||
var exists1 = await _client.Indices.ExistsAsync(indexName1);
|
||
if (!exists1.Exists)
|
||
{
|
||
await _client.Indices.CreateAsync(indexName1);
|
||
await _client.MapAsync<TransportWarningDetailsIndex>(m => m.Index(indexName1).AutoMap());
|
||
}
|
||
|
||
var exists2 = await _client.Indices.ExistsAsync(indexName2);
|
||
if (!exists2.Exists)
|
||
{
|
||
await _client.Indices.CreateAsync(indexName2);
|
||
await _client.MapAsync<ReturnWarningDetailsIndex>(m => m.Index(indexName2).AutoMap());
|
||
}
|
||
|
||
var exists3 = await _client.Indices.ExistsAsync(indexName3);
|
||
if (!exists3.Exists)
|
||
{
|
||
await _client.Indices.CreateAsync(indexName3);
|
||
await _client.MapAsync<CompleteWarningDetailsIndex>(m => m.Index(indexName3).AutoMap());
|
||
}
|
||
|
||
var exists4 = await _client.Indices.ExistsAsync(indexName4);
|
||
if (!exists4.Exists)
|
||
{
|
||
await _client.Indices.CreateAsync(indexName4);
|
||
await _client.MapAsync<ReturnMessageRecordIndex>(m => m.Index(indexName4).AutoMap());
|
||
}
|
||
|
||
var r = await _client.Indices.BulkAliasAsync(b =>
|
||
b.Add(a => a.Alias(GlobalConst.TransportWarningDetailsIndex).Index(indexName1))
|
||
.Add(a => a.Alias(GlobalConst.ReturnWarningDetailsIndex).Index(indexName2))
|
||
.Add(a => a.Alias(GlobalConst.CompleteWarningDetailsIndex).Index(indexName3))
|
||
.Add(a => a.Alias(GlobalConst.ReturnMessageRecordIndex).Index(indexName4))
|
||
);
|
||
|
||
}
|
||
}
|