using InterfaceForward.Domain.Shared.Helpers; using Nest; using TrackCenter.Domain.Aggregates.WarningAggregate; using TrackCenter.Domain.Aggregates.WarningAggregate.Complete; using TrackCenter.Domain.Aggregates.WarningAggregate.Return; using TrackCenter.Domain.Aggregates.WarningAggregate.Transport; using TrackCenter.Domain.Consts; using TrackCenter.Domain.Dtos.Warning; using TrackCenter.Domain.Enums; using TrackCenter.Domain.Extensions; using TrackCenter.Repository.ES; namespace TrackCenter.Repository.Repositories; public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransientDependency { private const int SafelyLimit = GlobalConst.SafelyLimit; private readonly IElasticClient _client; public WarningDetailsEsRepository(ConnectionFactory connection) { _client = connection.GetContext(typeof(TransportWarningDetailsIndex)); } #region 运输时效 public async Task> GetPageTransportWarningDetailsAsync( GetPageTransportWarningDetailsInputVO input) { var musts = ListWrapperFactory .CreateListWrapper(input); // 精确匹配 musts.AddIf(x => !string.IsNullOrWhiteSpace(x.Customer), c => c.Match(f => f.Field(d => d.Customer).Query(input.Customer))); musts.AddIf(x => x.OrderNos?.Count > 0, c => c.Terms(f => f.Field(d => d.OrderNo.Suffix("keyword")).Terms(input.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.AppKey.IsNullOrWhiteSpace(), c => c.Term(f => f.AppKey, input.AppKey)); musts.AddIf(x => !string.IsNullOrWhiteSpace(x.WarningName), c => c.Match(f => f.Field(d => d.WarningName).Query(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 => !string.IsNullOrWhiteSpace(x.NotifyUserId), c => c.Match(f => f.Field(d => d.NotifyUsers).Query(input.NotifyUserId))); var r = await _client.SearchAsync(x => x .UseMonthlyIndex(GlobalConst.TransportWarningDetailsIndex) .Query(q => q.Bool(b => b.Must(musts))) .Sort(s => s.Ascending(f => f.Status).Descending(f => f.CreateTime)) .From((input.PageIndex - 1) * input.PageSize) .Size(input.PageSize) .TrackTotalHits(true) ); if (!r.IsValid) { if (r.IsIndexUnavailable()) return new PagedResult(0, []); throw new InvalidOperationException($"运输预警明细查询失败,原因:{r.DebugInformation}。"); } return new PagedResult(r.Total, r.ToIdList()); } public async Task> GetTransportWarningDetailsAsync( GetTransportWarningDetailsVO input) { var musts = ListWrapperFactory.CreateListWrapper(input); var isShouldQuery = false; if (input.HasDetailsIds) { musts.Add(c => c.Ids(i => i.Values(input.DetailsIds))); } else if (input.HasOrderIdStartTracksStatusIdDict) { foreach (var (orderId, trackStatusIds) in input.OrderIdStartTracksStatusIdsDict!) { musts.Add(c => c.Bool(b => b.Must ( mu => mu.Term(f => f.OrderId, orderId), mu => mu.Terms(t => t.Field(f => f.StartTrackStatusId).Terms(trackStatusIds)) ))); } isShouldQuery = true; } else if (input.HasOrderIdRuleItemIdsDict) { foreach (var (orderId, ruleItemIds) in input.OrderIdRuleItemIdsDict!) { musts.Add(c => c.Bool(b => b.Must ( mu => mu.Term(f => f.OrderId, orderId), mu => mu.Terms(t => t.Field(f => f.WarningSceneRuleItemId).Terms(ruleItemIds)) ))); } isShouldQuery = true; } else { musts.AddIf(x => x.HasRuleItemIds, c => c.Terms(t => t.Field(f => f.WarningSceneRuleItemId).Terms(input.RuleItemIds))); musts.AddIf(x => x.HasSceneIds, c => c.Terms(t => t.Field(f => f.WarningSceneId).Terms(input.SceneIds))); musts.AddIf(x => x.HasWarningIds, c => c.Terms(t => t.Field(f => f.WarningId).Terms(input.WarningIds))); musts.AddIf(x => x.HasOrderIds, c => c.Terms(t => t.Field(f => f.OrderId).Terms(input.OrderIds))); } ISearchResponse r; if (isShouldQuery) { r = await _client.SearchAsync(x => x .UseMonthlyIndex(GlobalConst.TransportWarningDetailsIndex) .Query(q => q.Bool(b => b .Must(mu => mu.Terms(t => t.Field(f => f.Status).Terms(input.Status))) .Should(musts) .MinimumShouldMatch(1))) .Size(SafelyLimit) ); } else { musts.Add(mu => mu.Terms(t => t.Field(f => f.Status).Terms(input.Status))); r = await _client.SearchAsync(x => x .UseMonthlyIndex(GlobalConst.TransportWarningDetailsIndex) .Query(q => q.Bool(b => b.Must(musts))) .Size(SafelyLimit) ); } if (!r.IsValid) { if (r.IsIndexUnavailable()) return []; throw new InvalidOperationException($"运输预警明细查询失败,原因:{r.DebugInformation}。"); } return r.ToIdList(); } public async Task UpdateTransportWarningDetailsAsync(TransportWarningDetailsIndex detail) { var r = await _client.UpdateAsync(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.AppKey, detail.Customer, detail.OrderId, detail.OrderNo, detail.OrderType, detail.StartTrackStatusId, detail.StartTrackContentId, detail.StartTrackTime, detail.StartTrackTimeZone, detail.EndTrackStatusId, detail.EndTrackContentId, detail.EndTrackContent, detail.EndTrackTime, detail.EndTrackTimeZone, // detail.LatestTrackContentId, // detail.LatestTrackContent, // detail.LatestTrackTime, // detail.LatestTrackTimeZone, // detail.LatestTrackPlace, detail.CumulativeTimeSpan, detail.Status, detail.CreateTime, })); if (!r.IsValid) { throw new InvalidOperationException($"运输预警明细更新失败{detail},原因:{r.DebugInformation}。"); } } public async Task BulkUpdateTransportWarningDetailsAsync(List 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 = params.WarningId; ctx._source.warningSceneId = params.WarningSceneId; ctx._source.warningSceneRuleItemId = params.WarningSceneRuleItemId; ctx._source.version = params.Version; ctx._source.warningName = params.WarningName; ctx._source.warningRuleDescription = params.WarningRuleDescription; ctx._source.transportTimeSpan = params.TransportTimeSpan; ctx._source.startTrackWarningTimeSpan = params.StartTrackWarningTimeSpan; //ctx._source.notifyModes = params.NotifyModes; ctx._source.notifyUsers = params.NotifyUsers; ctx._source.AppKey = params.AppKey; ctx._source.customer = params.Customer; ctx._source.customName = params.CustomName; ctx._source.orderId = params.OrderId; ctx._source.orderNo = params.OrderNo; ctx._source.orderType = params.OrderType; ctx._source.startTrackStatusId = params.StartTrackStatusId; ctx._source.startTrackContentId = params.StartTrackContentId; ctx._source.startTrackTime = params.StartTrackTime; ctx._source.startTrackTimeZone = params.StartTrackTimeZone; ctx._source.endTrackStatusId = params.EndTrackStatusId; ctx._source.endTrackContentId = params.EndTrackContentId; ctx._source.endTrackContent = params.EndTrackContent; ctx._source.endTrackTime = params.EndTrackTime; ctx._source.endTrackTimeZone = params.EndTrackTimeZone; //ctx._source.latestTrackContentId = params.LatestTrackContentId; //ctx._source.latestTrackContent = params.LatestTrackContent; //ctx._source.latestTrackTime = params.LatestTrackTime; //ctx._source.latestTrackTimeZone = params.LatestTrackTimeZone; //ctx._source.latestTrackPlace = params.LatestTrackPlace; ctx._source.cumulativeTimeSpan = params.CumulativeTimeSpan; ctx._source.status = params.Status; ctx._source.createTime = params.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("AppKey", detail.AppKey) .Add("Customer", detail.Customer) .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("EndTrackContent", detail.EndTrackContent) .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(BulkCloseTransportWarningDetailsVO input, int maxRetries = 3) { var musts = ListWrapperFactory .CreateListWrapper(input); // 关注状态,即将超时/已超时 musts.Add(c => c.Terms(t => t.Field(f => f.Status).Terms(input.Status))); musts.AddIf(x => x.HasDetailsIds, c => c.Ids(i => i.Values(input.DetailsIds))); musts.AddIf(x => x.HasRuleItemIds, c => c.Terms(t => t.Field(f => f.WarningSceneRuleItemId).Terms(input.RuleItemIds))); musts.AddIf(x => x.HasSceneIds, c => c.Terms(t => t.Field(f => f.WarningSceneId).Terms(input.SceneIds))); musts.AddIf(x => x.HasWarningIds, c => c.Terms(t => t.Field(f => f.WarningId).Terms(input.WarningIds))); musts.AddIf(x => x.HasOrderIds, c => c.Terms(t => t.Field(f => f.OrderId).Terms(input.OrderIds))); var r = await _client.UpdateByQueryAsync(u => u.UseMonthlyIndex(GlobalConst.TransportWarningDetailsIndex) .Query(q => q.Bool(b => b.Must(musts))) .Script(s => s .Source(@"ctx._source.status = params.status; if (ctx._source.startTrackTime != null) { // 替换日期字符串中的T String formattedDateString = ctx._source.startTrackTime.replace(""T"", "" ""); SimpleDateFormat sdf = new SimpleDateFormat(""yyyy-MM-dd HH:mm:ss""); // 更新 cumulativeTimeSpan 字段 ctx._source.cumulativeTimeSpan = new Date().getTime() - sdf.parse(formattedDateString).getTime() + 3600*8*1000; }" ) .Params(p => p.Add("status", (int) TransportWarningStatus.已关闭)) ) ); if (!r.IsValid && !r.IsIndexUnavailable()) { throw new InvalidOperationException( $"运输时效预警明细更新失败,input:{input.ToJson()},原因:{r.DebugInformation}。"); } } public async Task InsertTransportWarningDetailsAsync(TransportWarningDetailsIndex detail) { var indexName = $"{GlobalConst.TransportWarningDetailsIndex}-{DateTime.Now:yyyy.MM}"; 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 details, int maxRetries = 3) { // 按天写入es var indexName = $"{GlobalConst.TransportWarningDetailsIndex}-{DateTime.Now:yyyy.MM}"; 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> GetPageReturnWarningDetailsAsync( GetPageReturnWarningDetailsInputVO input) { var musts = ListWrapperFactory.CreateListWrapper(input); // 精确匹配 musts.AddIf(x => !x.AppKey.IsNullOrWhiteSpace(), c => c.Term(f => f.AppKey, input.AppKey)); musts.AddIf(x => x.OrderNos?.Count > 0, c => c.Terms(f => f.Field(d => d.OrderNo.Suffix("keyword")).Terms(input.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.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).GreaterThanOrEquals(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).GreaterThanOrEquals(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).GreaterThanOrEquals(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 => !string.IsNullOrWhiteSpace(x.NotifyUserId), c => c.Match(f => f.Field(d => d.NotifyUsers).Query(input.NotifyUserId))); musts.AddIf(x => !string.IsNullOrWhiteSpace(x.Customer), c => c.Match(f => f.Field(d => d.Customer).Query(input.Customer))); var r = await _client.SearchAsync(x => x .UseMonthlyIndex(GlobalConst.ReturnWarningDetailsIndex) .Query(q => q.Bool(b => b.Must(musts))) .Sort(s => s.Ascending(f => f.Status).Descending(f => f.CreateTime)) .From((input.PageIndex - 1) * input.PageSize) .Size(input.PageSize) .TrackTotalHits(true) ); if (!r.IsValid) { if (r.IsIndexUnavailable()) return new PagedResult(0, []); throw new InvalidOperationException($"回传预警明细查询失败,原因:{r.DebugInformation}。"); } return new PagedResult(r.Total, r.ToIdList()); } public async Task> GetReturnWarningDetailsAsync(GetReturnWarningDetailsVO input) { var isShouldQuery = false; var musts = ListWrapperFactory.CreateListWrapper(input); if (input.HasDetailsIds) { musts.Add(c => c.Ids(i => i.Values(input.DetailsIds))); } else if (input.HasTrackEsIds) { musts.Add(c => c.Terms(f => f.Field(d => d.TrackEsId.Suffix("keyword")).Terms(input.TrackEsIds))); } else if (input.HasOrderIdTrackContentIdsDict) { foreach (var (orderId, trackContentIds) in input.OrderIdTrackContentIdsDict!) { musts.Add(c => c.Bool(b => b.Must ( mu => mu.Term(f => f.OrderId, orderId), mu => mu.Terms(t => t.Field(f => f.TrackContentId).Terms(trackContentIds)) ))); } isShouldQuery = true; } else { musts.AddIf(x => x.HasRuleItemIds, c => c.Terms(t => t.Field(f => f.ReturnWarningRuleItemId).Terms(input.RuleItemIds))); musts.AddIf(x => x.HasWarningIds, c => c.Terms(t => t.Field(f => f.WarningId).Terms(input.WarningIds))); musts.AddIf(x => x.HasOrderIds, c => c.Terms(t => t.Field(f => f.OrderId).Terms(input.OrderIds))); musts.AddIf(x => x.HasPushContentConfigIds, c => c.Terms(f => f.Field(d => d.PushContentConfigId).Terms(input.PushContentConfigIds))); } ISearchResponse r; if (isShouldQuery) { r = await _client.SearchAsync(x => x .UseMonthlyIndex(GlobalConst.ReturnWarningDetailsIndex) .Query(q => q.Bool(b => b .Must(mu => mu.Terms(t => t.Field(f => f.Status).Terms(input.Status))) .Should(musts) .MinimumShouldMatch(1))) .Size(SafelyLimit) ); } else { musts.Add(c => c.Terms(t => t.Field(f => f.Status).Terms(input.Status))); r = await _client.SearchAsync(x => x .UseMonthlyIndex(GlobalConst.ReturnWarningDetailsIndex) .Query(q => q.Bool(b => b .Must(musts) )) .Size(SafelyLimit) ); } if (!r.IsValid) { if (r.IsIndexUnavailable()) return []; throw new InvalidOperationException($"回传预警明细查询失败,原因:{r.DebugInformation}。"); } return r.ToIdList(); } public async Task> GetReturnMessageRecordListByIdAsync(string esId) { var r = await _client.SearchAsync(x => x .UseMonthlyIndex(GlobalConst.ReturnMessageRecordIndex) .Query(q => q.Match(f => f.Field(d => d.WarningDetailsIds).Query(esId))) .Size(SafelyLimit) ); if (!r.IsValid) { if (r.IsIndexUnavailable()) return []; throw new InvalidOperationException($"回传预警明细查询失败,原因:{r.DebugInformation}。"); } return r.ToIdList(); } public async Task ReturnWarningDetailsRetryIncrementAsync(HashSet esIds) { var r = await _client.UpdateByQueryAsync(u => u.UseMonthlyIndex(GlobalConst.ReturnWarningDetailsIndex) .Query(q => q.Ids(i => i.Values(esIds))) .Script(s => s .Source("ctx._source.retryCount = ctx._source.retryCount+1;") ) ); if (!r.IsValid && !r.IsIndexUnavailable()) { throw new InvalidOperationException($"回传预警明细更新失败,原因:{r.DebugInformation}。"); } } public async Task UpdateReturnWarningDetailsAsync(ReturnWarningDetailsIndex detail) { detail.UpdateTime = DateTime.Now; var r = await _client.UpdateAsync(detail.Id, u => u.Index(detail.IndexName) .Doc(new { ReturnWarningRuleId = detail.ReturnWarningRuleItemId, detail.WarningId, detail.Version, detail.ReturnTimeSpan, //detail.NotifyModes, detail.NotifyUsers, detail.AppKey, detail.Customer, 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.PushContentConfigId, detail.PushTrackStatus, detail.PushTrackCode, detail.PushTrackContent, detail.PushTrackEnContent, detail.TrackReturnTime, detail.RetryCount, detail.Status, detail.ActualReturnTimeSpan, detail.SystemReturnTimeSpan, detail.TrackCreateTimeSpan, detail.CreateTime, detail.UpdateTime, })); if (!r.IsValid) { throw new InvalidOperationException($"回传预警明细更新失败{detail},原因:{r.DebugInformation}。"); } } public async Task BulkUpdateReturnWarningDetailsAsync(List 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.returnWarningRuleItemId = params.ReturnWarningRuleItemId; ctx._source.warningId = params.WarningId; ctx._source.version = params.Version; ctx._source.returnTimeSpan = params.ReturnTimeSpan; //ctx._source.notifyModes = params.NotifyModes; ctx._source.notifyUsers = params.NotifyUsers; ctx._source.AppKey = params.AppKey; ctx._source.customer = params.Customer; ctx._source.orderId = params.OrderId; ctx._source.orderNo = params.OrderNo; ctx._source.orderType = params.OrderType; ctx._source.trackStatusId = params.TrackStatusId; ctx._source.trackStatus = params.TrackStatus; ctx._source.compareFields = params.CompareFields; ctx._source.extra = params.Extra; ctx._source.trackContentId = params.TrackContentId; ctx._source.trackContent = params.TrackContent; ctx._source.trackCode = params.TrackCode; ctx._source.trackPlace = params.TrackPlace; ctx._source.trackTime = params.TrackTime; ctx._source.trackTimeZone = params.TrackTimeZone; ctx._source.trackCreateTime = params.TrackCreateTime; ctx._source.pushContentConfigId = params.PushContentConfigId; ctx._source.pushTrackStatus = params.PushTrackStatus; ctx._source.pushTrackCode = params.PushTrackCode; ctx._source.pushTrackContent = params.PushTrackContent; ctx._source.pushTrackEnContent = params.PushTrackEnContent; ctx._source.trackReturnTime = params.TrackReturnTime; ctx._source.retryCount = params.RetryCount; ctx._source.status = params.Status; ctx._source.actualReturnTimeSpan = params.ActualReturnTimeSpan; ctx._source.systemReturnTimeSpan = params.SystemReturnTimeSpan; ctx._source.trackCreateTimeSpan = params.TrackCreateTimeSpan; ctx._source.createTime = params.CreateTime; ctx._source.updateTime = params.UpdateTime; ").Params(p => p .Add("ReturnWarningRuleItemId", detail.ReturnWarningRuleItemId) .Add("WarningId", detail.WarningId) .Add("Version", detail.Version) .Add("ReturnTimeSpan", detail.ReturnTimeSpan) //.Add("NotifyModes", detail.NotifyModes) .Add("NotifyUsers", detail.NotifyUsers) .Add("AppKey", detail.AppKey) .Add("Customer", detail.Customer) .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("PushContentConfigId", detail.PushContentConfigId) .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", DateTime.Now) ) ) ); 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 BulkCloseReturnWarningDetailsAsync(BulkCloseReturnWarningDetailsVO input, int maxRetries = 3) { var musts = ListWrapperFactory.CreateListWrapper(input); // 关注状态,重试中/人工介入 musts.Add(c => c.Terms(t => t.Field(f => f.Status).Terms(input.Status))); musts.AddIf(x => x.HasDetailsIds, c => c.Ids(i => i.Values(input.DetailsIds))); musts.AddIf(x => x.HasRuleItemIds, c => c.Terms(t => t.Field(f => f.ReturnWarningRuleItemId).Terms(input.RuleItemIds))); musts.AddIf(x => x.HasWarningIds, c => c.Terms(t => t.Field(f => f.WarningId).Terms(input.WarningIds))); musts.AddIf(x => x.HasOrderIds, c => c.Terms(t => t.Field(f => f.OrderId).Terms(input.OrderIds))); var r = await _client.UpdateByQueryAsync(u => u.UseMonthlyIndex(GlobalConst.ReturnWarningDetailsIndex) .Query(q => q.Bool(b => b.Must(musts))) .Script(s => s .Source("ctx._source.status = params.status") .Params(p => p.Add("status", (int) ReturnWarningStatus.已关闭)) ) ); if (!r.IsValid && !r.IsIndexUnavailable()) { throw new InvalidOperationException( $"回传预警明细更新失败,input:{input.ToJson()},原因:{r.DebugInformation}。"); } } public async Task InsertReturnWarningDetailsAsync(ReturnWarningDetailsIndex detail) { var indexName = $"{GlobalConst.ReturnWarningDetailsIndex}-{DateTime.Now:yyyy.MM}"; 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 details, int maxRetries = 3) { // 按天写入es var indexName = $"{GlobalConst.ReturnWarningDetailsIndex}-{DateTime.Now:yyyy.MM}"; 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}"; 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 details, int maxRetries = 3) { // 按天写入es var indexName = $"{GlobalConst.ReturnMessageRecordIndex}-{DateTime.Now:yyyy.MM}"; 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> GetPageCompleteWarningDetailsAsync( GetPageCompleteWarningDetailsInputVO input) { var musts = ListWrapperFactory .CreateListWrapper(input); musts.AddIf(x => !x.AppKey.IsNullOrWhiteSpace(), c => c.Term(f => f.AppKey, input.AppKey)); musts.AddIf(x => x.OrderNos?.Count > 0, c => c.Terms(f => f.Field(d => d.OrderNo.Suffix("keyword")).Terms(input.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.WarningCreateTimeStart.HasValue, c => c.DateRange(r => r.Field(f => f.CreateTime).GreaterThanOrEquals(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).GreaterThanOrEquals(input.WarningHandleTimeStart))); musts.AddIf(x => x.WarningHandleTimeEnd.HasValue, c => c.DateRange(r => r.Field(f => f.ProcessedTime).LessThanOrEquals(input.WarningHandleTimeEnd))); // 模糊匹配userId musts.AddIf(x => !string.IsNullOrWhiteSpace(x.NotifyUserId), c => c.Match(f => f.Field(d => d.NotifyUsers).Query(input.NotifyUserId))); musts.AddIf(x => !string.IsNullOrWhiteSpace(x.Customer), c => c.Match(f => f.Field(d => d.Customer).Query(input.Customer))); // 匹配缺失轨迹状态 musts.AddIf(x => x.MissingTrackStatusId > 0, c => c.Script(s => s.Script(a => a .Source("doc['missingTrackStatusIds.keyword'].value.contains(params.value)") .Params(p => p.Add("value", input.MissingTrackStatusId.ToString()))))); var r = await _client.SearchAsync(x => x .UseMonthlyIndex(GlobalConst.CompleteWarningDetailsIndex) .Query(q => q.Bool(b => b.Must(musts))) .Sort(s => s.Ascending(f => f.Status).Descending(f => f.CreateTime)) .From((input.PageIndex - 1) * input.PageSize) .Size(input.PageSize) .TrackTotalHits(true) ); if (!r.IsValid) { if (r.IsIndexUnavailable()) return new PagedResult(0, []); throw new InvalidOperationException($"回传预警明细查询失败,原因:{r.DebugInformation}。"); } return new PagedResult(r.Total, r.ToIdList()); } public async Task> GetCompleteWarningDetailsAsync( GetCompleteWarningDetailsVO input) { var isShouldQuery = false; var musts = ListWrapperFactory.CreateListWrapper(input); if (input.HasDetailsIds) { musts.Add(c => c.Ids(i => i.Values(input.DetailsIds))); } else if (input.HasOrderIdCheckTracksStatusIdsDict) { foreach (var (orderId, checkTrackStatusIds) in input.OrderIdCheckTracksStatusIdsDict!) { musts.Add(c => c.Bool(b => b.Must ( mu => mu.Term(f => f.OrderId, orderId), mu => mu.Terms(t => t.Field(f => f.CheckPointTrackStatusId).Terms(checkTrackStatusIds)) ))); } isShouldQuery = true; } else if (input.HasOrderIdRuleItemIdsDict) { foreach (var (orderId, ruleItemIds) in input.OrderIdRuleItemIdsDict!) { musts.Add(c => c.Bool(b => b.Must ( mu => mu.Term(f => f.OrderId, orderId), mu => mu.Terms(t => t.Field(f => f.CompleteWarningRuleItemId).Terms(ruleItemIds)) ))); } isShouldQuery = true; } else { musts.AddIf(x => x.HasRuleItemIds, c => c.Terms(t => t.Field(f => f.CompleteWarningRuleItemId).Terms(input.RuleItemIds))); musts.AddIf(x => x.HasWarningIds, c => c.Terms(t => t.Field(f => f.WarningId).Terms(input.WarningIds))); musts.AddIf(x => x.HasOrderIds, c => c.Terms(t => t.Field(f => f.OrderId).Terms(input.OrderIds))); } ISearchResponse r; if (isShouldQuery) { r = await _client.SearchAsync(x => x .UseMonthlyIndex(GlobalConst.CompleteWarningDetailsIndex) .Query(q => q.Bool(b => b .Must(mu => mu.Terms(t => t.Field(f => f.Status).Terms(input.Status))) .Should(musts) .MinimumShouldMatch(1))) .Size(SafelyLimit) ); } else { musts.Add(c => c.Terms(t => t.Field(f => f.Status).Terms(input.Status))); r = await _client.SearchAsync(x => x .UseMonthlyIndex(GlobalConst.CompleteWarningDetailsIndex) .Query(q => q.Bool(b => b .Must(musts) )) .Size(SafelyLimit) ); } if (!r.IsValid) { if (r.IsIndexUnavailable()) return []; throw new InvalidOperationException($"完整预警明细查询失败,原因:{r.DebugInformation}。"); } return r.ToIdList(); } public async Task BulkCloseCompleteWarningDetailsAsync(BulkCloseCompleteWarningDetailsVO input, int maxRetries = 3) { var musts = ListWrapperFactory.CreateListWrapper(input); // 关注状态,待处理 musts.Add(c => c.Terms(t => t.Field(f => f.Status).Terms(input.Status))); musts.AddIf(x => x.HasDetailsIds, c => c.Ids(i => i.Values(input.DetailsIds))); musts.AddIf(x => x.HasRuleItemIds, c => c.Terms(t => t.Field(f => f.CompleteWarningRuleItemId).Terms(input.RuleItemIds))); musts.AddIf(x => x.HasWarningIds, c => c.Terms(t => t.Field(f => f.WarningId).Terms(input.WarningIds))); musts.AddIf(x => x.HasOrderIds, c => c.Terms(t => t.Field(f => f.OrderId).Terms(input.OrderIds))); var r = await _client.UpdateByQueryAsync(u => u.UseMonthlyIndex(GlobalConst.CompleteWarningDetailsIndex) .Query(q => q.Bool(b => b.Must(musts))) .Script(s => s .Source(@" ctx._source.status = params.status; ctx._source.processedTime = params.processedTime; ") .Params(p => p .Add("status", (int) input.SetStatus) .Add("processedTime", DateTime.Now) ) ) ); if (!r.IsValid && !r.IsIndexUnavailable()) { throw new InvalidOperationException( $"完整预警明细更新失败,input:{input.ToJson()},原因:{r.DebugInformation}。"); } } public async Task UpdateCompleteWarningDetailsAsync(CompleteWarningDetailsIndex detail) { var r = await _client.UpdateAsync(detail.Id, u => u.Index(detail.IndexName) .Doc(new { detail.WarningId, detail.CompleteWarningRuleItemId, detail.Version, //detail.NotifyModes, detail.NotifyUsers, detail.CheckPointTrackStatusId, detail.MissingTrackStatusIds, detail.AppKey, detail.Customer, detail.OrderId, detail.OrderNo, detail.OrderType, detail.Status, detail.CreateTime, detail.ProcessedTime, })); if (!r.IsValid) { throw new InvalidOperationException($"回传预警明细更新失败{detail},原因:{r.DebugInformation}。"); } } public async Task BulkUpdateCompleteWarningDetailsAsync(List 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 = params.WarningId; ctx._source.completeWarningRuleItemId = params.CompleteWarningRuleItemId; ctx._source.version = params.Version; //ctx._source.notifyModes = params.NotifyModes; ctx._source.notifyUsers = params.NotifyUsers; ctx._source.checkPointTrackStatusId = params.CheckPointTrackStatusId; ctx._source.missingTrackStatusIds = params.MissingTrackStatusIds; ctx._source.AppKey = params.AppKey; ctx._source.customer = params.Customer; ctx._source.orderId = params.OrderId; ctx._source.orderNo = params.OrderNo; ctx._source.orderType = params.OrderType; ctx._source.status = params.Status; ctx._source.createTime = params.CreateTime; ctx._source.processedTime = params.ProcessedTime; ") .Params(p => p .Add("WarningId", detail.WarningId) .Add("CompleteWarningRuleItemId", detail.CompleteWarningRuleItemId) .Add("Version", detail.Version) //.Add("NotifyModes", detail.NotifyModes) .Add("NotifyUsers", detail.NotifyUsers) .Add("CheckPointTrackStatusId", detail.CheckPointTrackStatusId) .Add("MissingTrackStatusIds", detail.MissingTrackStatusIds) .Add("AppKey", detail.AppKey) .Add("Customer", detail.Customer) .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}"; 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 details, int maxRetries = 3) { // 按天写入es var indexName = $"{GlobalConst.CompleteWarningDetailsIndex}-{DateTime.Now:yyyy.MM}"; 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}"; var indexName2 = $"{GlobalConst.ReturnWarningDetailsIndex}-{DateTime.Now:yyyy.MM}"; var indexName3 = $"{GlobalConst.CompleteWarningDetailsIndex}-{DateTime.Now:yyyy.MM}"; var indexName4 = $"{GlobalConst.ReturnMessageRecordIndex}-{DateTime.Now:yyyy.MM}"; var exists1 = await _client.Indices.ExistsAsync(indexName1); if (!exists1.Exists) { await _client.Indices.CreateAsync(indexName1); await _client.MapAsync(m => m.Index(indexName1).AutoMap()); } var exists2 = await _client.Indices.ExistsAsync(indexName2); if (!exists2.Exists) { await _client.Indices.CreateAsync(indexName2); await _client.MapAsync(m => m.Index(indexName2).AutoMap()); } var exists3 = await _client.Indices.ExistsAsync(indexName3); if (!exists3.Exists) { await _client.Indices.CreateAsync(indexName3); await _client.MapAsync(m => m.Index(indexName3).AutoMap()); } var exists4 = await _client.Indices.ExistsAsync(indexName4); if (!exists4.Exists) { await _client.Indices.CreateAsync(indexName4); await _client.MapAsync(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)) ); } }