using Nest; using SJZY.Expand.ABP.Core.PublicMethords; using SJZY.Expand.ElasticSearch.Services; using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; using SJZY.Track.Domain.Consts; using SJZY.Track.Domain.Dtos.TrackContents; using SJZY.Track.Domain.Enums; namespace SJZY.Track.Repository.Repositories; public class TrackDetailsEsRepository : ElasticSearchContextAbstract, ITrackDetailsEsRepository, ITransientDependency { const int SafelyLimit = GlobalConst.SafelyLimit; public TrackDetailsEsRepository(ConnectionFactory connectionFactory) : base(connectionFactory) { } public async Task> GetListAsync(int[] orderIds, string trackCode, bool forUpsert = false) { var r = !forUpsert ? await Context.SearchAsync(s => s .Index(GlobalConst.TrackDetailsIndex) .Source() .Size(SafelyLimit) .Query(q => q .Bool(b => b .Must( mu => mu.Terms(t => t.Field(f => f.OrderId).Terms(orderIds)), mu => mu.Term(t => t.Field(f => f.TrackCode.Suffix("keyword")).Value(trackCode)) ) ) ) ) : await Context.SearchAsync(s => s .Index(GlobalConst.TrackDetailsIndex) .Source(s2 => s2.Includes(fd => fd .Field(f => f.IndexName) .Field(f => f.OrderId) .Field(f => f.TrackTime) .Field(f => f.TrackPlace) )) .Size(SafelyLimit) .Query(q => q .Bool(b => b .Must( mu => mu.Terms(t => t.Field(f => f.OrderId).Terms(orderIds)), mu => mu.Term(t => t.Field(f => f.TrackCode.Suffix("keyword")).Value(trackCode)) ) ) ) ); if (!r.IsValid) { throw new InvalidOperationException( $"轨迹查询失败,orderIds:[{orderIds.JoinAsString(", ")}],trackCode:{trackCode},原因:{r.DebugInformation}。"); } return r.ToIdList(); } public async Task> GetLatestListAsync(int[] orderIds, List trackCodes) { var excludeTrackCodes = trackCodes.ToList(); excludeTrackCodes.Add(GlobalConst.EmptyString); var r = await Context.SearchAsync(x => x .Index(GlobalConst.TrackDetailsIndex) .Source(s => s.Includes(fd => fd .Field(f => f.TrackEnContent) .Field(f => f.TrackContent) .Field(f => f.TrackPlace) .Field(f => f.TrackTime) .Field(f => f.TrackTimeZone) .Field(f => f.OrderId) .Field(f => f.TrackCode) .Field(f => f.Id) .Field(f => f.IndexName) )) .Query(q => q .Bool(b => b .Must( mu => mu.Terms(t => t.Field(f => f.OrderId).Terms(orderIds)) ) .MustNot(mn => mn .Terms(t => t .Field(f => f.TrackCode.Suffix("keyword")) .Terms(excludeTrackCodes) ) ) ) ) .Size(SafelyLimit) .Aggregations(a => a .Terms("orders", t => t .Field(f => f.OrderId) .Aggregations(aa => aa .TopHits("latest_track", th => th .Size(1) .Sort(so => so .Field(f => f.ZeroTrackTime, SortOrder.Descending) ) .Source(s2 => s2.Includes(fd2 => fd2 .Field(f => f.TrackEnContent) .Field(f => f.TrackContent) .Field(f => f.TrackPlace) .Field(f => f.TrackTime) .Field(f => f.TrackTimeZone) .Field(f => f.OrderId) .Field(f => f.TrackCode) .Field(f => f.Id) .Field(f => f.IndexName) )) ) ) ) ) ); if (!r.IsValid) { throw new InvalidOperationException( $"最新轨迹查询失败,orderIds:[{orderIds.JoinAsString(", ")}],trackCodes:[{trackCodes.JoinAsString(", ")}],原因:{r.DebugInformation}。"); } return r.Documents.ToList(); } public async Task AnyAsync(int orderId, string originTrackDataHash) { var r = await Context.SearchAsync(x => x .Index(GlobalConst.TrackDetailsIndex) .Source(s => s.Includes(fd => fd.Field(f => f.TrackCode))) .Query(q => q.Term(f => f.OrderId, orderId) && q.Term(f => f.OriginTrackDataHash.Suffix("keyword"), originTrackDataHash) ) ); if (!r.IsValid) { throw new InvalidOperationException( $"轨迹查询失败,orderId:{orderId},hash:{originTrackDataHash},原因:{r.DebugInformation}。"); } return r.Hits.Count > 0; } public async Task UpdateAsync(TrackDetailsIndex track) { var r = await Context.UpdateAsync(track.Id, u => u .Index(track.IndexName) .Doc(new { track.TrackPlace, track.TrackTime, track.TrackTimeZone, track.UpdateUser, UpdateTime = DateTime.Now, track.TrackContent, track.TrackEnContent, track.IsVisibility, track.Remarks, track.Files, track.TrackPlaceCity, track.TrackPlaceCityId, track.TrackPlaceProvince, track.TrackPlaceProvinceId, track.TrackPlaceNation, track.TrackPlaceNationId, track.TrackPlacePostcode, track.DestinationTimeZone, track.EstimatedDeliveryDate }) ); if (!r.IsValid) { throw new InvalidOperationException($"轨迹更新失败,track:{track.ToJson()},原因:{r.DebugInformation}。"); } } //#4001批量更新时只会将所有id匹配的记录更新为同一内容,后续需要支持差异化更新 public async Task UpdateManyAsync(List indexes, int maxRetries = 3) { var retryCount = 0; var updateSuccessful = false; var track = indexes.First(); while (retryCount < maxRetries && !updateSuccessful) { var response = await Context.UpdateByQueryAsync(u => u .Index(GlobalConst.TrackDetailsIndex) .Query(q => q .Ids(t => t.Values(indexes.Select(x => x.Id))) ) .Script(s => s .Source(@" ctx._source.trackPlace = params.trackPlace; ctx._source.trackTime = params.trackTime; ctx._source.trackTimeZone = params.trackTimeZone; ctx._source.updateUser = params.updateUser; ctx._source.updateTime = params.updateTime; ctx._source.trackContent = params.trackContent; ctx._source.trackEnContent = params.trackEnContent; ctx._source.isVisibility = params.isVisibility; ctx._source.remarks = params.remarks; ctx._source.files = params.files; ctx._source.trackPlaceCity = params.trackPlaceCity; ctx._source.trackPlaceCityId = params.trackPlaceCityId; ctx._source.trackPlaceProvince = params.trackPlaceProvince; ctx._source.trackPlaceProvinceId = params.trackPlaceProvinceId; ctx._source.trackPlaceNation = params.trackPlaceNation; ctx._source.trackPlaceNationId = params.trackPlaceNationId; ctx._source.trackPlacePostcode = params.trackPlacePostcode; ctx._source.destinationTimeZone = params.destinationTimeZone; ctx._source.estimatedDeliveryDate = params.estimatedDeliveryDate; ") .Params(p => p .Add("trackPlace", track.TrackPlace) .Add("trackTime", track.TrackTime) .Add("trackTimeZone", track.TrackTimeZone) .Add("updateUser", track.UpdateUser) .Add("updateTime", DateTime.Now) //S06新增字段更新 .Add("trackContent", track.TrackContent) .Add("trackEnContent", track.TrackEnContent) .Add("isVisibility", track.IsVisibility) .Add("remarks", track.Remarks) .Add("files", track.Files) .Add("trackPlaceCity", track.TrackPlaceCity) .Add("trackPlaceCityId", track.TrackPlaceCityId) .Add("trackPlaceProvince", track.TrackPlaceProvince) .Add("trackPlaceProvinceId", track.TrackPlaceProvinceId) .Add("trackPlaceNation", track.TrackPlaceNation) .Add("trackPlaceNationId", track.TrackPlaceNationId) .Add("trackPlacePostcode", track.TrackPlacePostcode) .Add("destinationTimeZone", track.DestinationTimeZone) .Add("estimatedDeliveryDate", track.EstimatedDeliveryDate) ) ) ); if (response.IsValid) { updateSuccessful = true; } else if (response.ServerError?.Status == 409) // 409 表示版本冲突 { retryCount++; if (retryCount >= maxRetries) { throw new InvalidOperationException( $"轨迹更新冲突达到最大次数,track:{indexes.Count}条,原因:{response.DebugInformation}。"); } } else { throw new InvalidOperationException($"轨迹更新失败,track:{indexes.Count}条,原因:{response.DebugInformation}。"); } } } public async Task BulkUpdateAsync(List indexes, int maxRetries = 3) { if (indexes.Count == 0) return; //构建批量更新操作描述符 var desc = new BulkDescriptor(GlobalConst.TrackDetailsIndex); foreach (var track in indexes) { desc = desc.AddOperation(new BulkUpdateDescriptor() .Id(track.Id) .Index(track.IndexName) .Script(s => s.Source(@" ctx._source.trackPlace = params.trackPlace; ctx._source.trackTime = params.trackTime; ctx._source.trackTimeZone = params.trackTimeZone; ctx._source.updateUser = params.updateUser; ctx._source.updateTime = params.updateTime; ctx._source.trackContent = params.trackContent; ctx._source.trackEnContent = params.trackEnContent; ctx._source.isVisibility = params.isVisibility; ctx._source.remarks = params.remarks; ctx._source.files = params.files; ctx._source.trackPlaceCity = params.trackPlaceCity; ctx._source.trackPlaceCityId = params.trackPlaceCityId; ctx._source.trackPlaceProvince = params.trackPlaceProvince; ctx._source.trackPlaceProvinceId = params.trackPlaceProvinceId; ctx._source.trackPlaceNation = params.trackPlaceNation; ctx._source.trackPlaceNationId = params.trackPlaceNationId; ctx._source.trackPlacePostcode = params.trackPlacePostcode; ctx._source.destinationTimeZone = params.destinationTimeZone; ctx._source.estimatedDeliveryDate = params.estimatedDeliveryDate; ").Params(p => p .Add("trackPlace", track.TrackPlace) .Add("trackTime", track.TrackTime) .Add("trackTimeZone", track.TrackTimeZone) .Add("updateUser", track.UpdateUser) .Add("updateTime", DateTime.Now) //S06新增字段更新 .Add("trackContent", track.TrackContent) .Add("trackEnContent", track.TrackEnContent) .Add("isVisibility", track.IsVisibility) .Add("remarks", track.Remarks) .Add("files", track.Files) .Add("trackPlaceCity", track.TrackPlaceCity) .Add("trackPlaceCityId", track.TrackPlaceCityId) .Add("trackPlaceProvince", track.TrackPlaceProvince) .Add("trackPlaceProvinceId", track.TrackPlaceProvinceId) .Add("trackPlaceNation", track.TrackPlaceNation) .Add("trackPlaceNationId", track.TrackPlaceNationId) .Add("trackPlacePostcode", track.TrackPlacePostcode) .Add("destinationTimeZone", track.DestinationTimeZone) .Add("estimatedDeliveryDate", track.EstimatedDeliveryDate)) )); } var retryCount = 0; var updateSuccessful = false; while (retryCount < maxRetries && !updateSuccessful) { var response = await Context.BulkAsync(desc); if (response.IsValid) { updateSuccessful = true; } else if (response.ServerError?.Status == 409) // 409 表示版本冲突 { retryCount++; if (retryCount >= maxRetries) { throw new InvalidOperationException( $"轨迹更新冲突达到最大次数,track:{indexes.Count}条,原因:{response.DebugInformation}。"); } } else { throw new InvalidOperationException($"轨迹更新失败,track:{indexes.Count}条,原因:{response.DebugInformation}。"); } } } public async Task AddManyAsync(List tracks) { // 按天写入es var indexName = $"{GlobalConst.TrackDetailsIndex}-{DateTime.Now:yyyy.MM.dd}"; tracks.ForEach(x => x.IndexName = indexName); var r = await Context.IndexManyAsync(tracks, indexName); if (!r.IsValid) { throw new InvalidOperationException($"{tracks.Count}条轨迹新增失败,原因:{r.DebugInformation}。"); } } public async Task AddAsync(TrackDetailsIndex track) { // 按天写入es var indexName = $"{GlobalConst.TrackDetailsIndex}-{DateTime.Now:yyyy.MM.dd}"; track.IndexName = indexName; var r = await Context.IndexAsync(track, s => s.Index(indexName)); if (!r.IsValid) { throw new InvalidOperationException($"轨迹新增失败,track:{track.ToJson()},原因:{r.DebugInformation}。"); } } public async Task> GetListAsync(int subSystemId, int orderType, List orderNos) { if (orderNos.Count < 1) return new List(); var r = await Context.SearchAsync(x => x .Index(GlobalConst.TrackDetailsIndex) .Source(s => s.Includes(fd => fd .Field(f => f.OrderId) .Field(f => f.OrderNo) .Field(f => f.TrackStatus) .Field(f => f.TrackCode) .Field(f => f.TrackContent) .Field(f => f.OriginTrackContent) .Field(f => f.TrackPlace) .Field(f => f.TrackTime) .Field(f => f.ZeroTrackTime) .Field(f => f.TrackTimeZone) )) .Query(q => q.Bool(b => b .Must( mu => mu.Term(f => f.SubSystemId, subSystemId), mu => mu .Terms(t => t .Field(f => f.OrderNo.Suffix("keyword")) .Terms(orderNos) ), mu => mu .Term(f => f.OrderType, orderType) ) // tips:对外不暴露服务商轨迹 .MustNot(mn => mn .Term(t => t.TrackCode.Suffix("keyword"), GlobalConst.EmptyString) ) ) ) .Size(SafelyLimit) ); if (!r.IsValid) { throw new InvalidOperationException( $"轨迹查询失败,orderType:{orderType},orderNos:{orderNos},原因:{r.DebugInformation}。"); } return r.Documents.ToList(); } public async Task> GetListAsync(int orderId) { var r = await Context.SearchAsync(x => x .Index(GlobalConst.TrackDetailsIndex) .Query(q => q.Term(t => t.OrderId, orderId) ) .Size(SafelyLimit) .Sort(s => s.Field(f => f.ZeroTrackTime, SortOrder.Descending))); if (!r.IsValid) { throw new InvalidOperationException($"轨迹查询失败,orderId:{orderId},原因:{r.DebugInformation}。"); } return r.ToIdList(); } public async Task> GetListAsync(int[] orderIds, List? trackCodes = null, TrackSourceEnum? source = null) { var query = new QueryContainer(); if (orderIds.Length > 0) { query &= Query.Terms(t => t.Field(f => f.OrderId).Terms(orderIds)); } if (trackCodes != null && trackCodes.Count > 0) { query &= Query.Terms(t => t.Field(f => f.TrackCode.Suffix("keyword")).Terms(trackCodes)); } if (source.HasValue) { query &= Query.Term(t => t.Field(f => f.Source).Value(source.Value)); } var r = await Context.SearchAsync(x => x .Index(GlobalConst.TrackDetailsIndex) .Query(q => query) .Size(SafelyLimit) .Sort(s => s.Field(f => f.ZeroTrackTime, SortOrder.Descending))); if (!r.IsValid) { throw new InvalidOperationException($"轨迹查询失败,orderIds:[{string.Join(',', orderIds)}],原因:{r.DebugInformation}。"); } return r.ToIdList(); } public async Task> DeleteTrackAsync(int[] orderIds, List trackCodes) { // var deleteResponse = await Context.DeleteByQueryAsync(d => d // .Index($"{GlobalConst.TrackDetailsIndex}-*") // 使用通配符匹配所有相关索引 // .Query(q => // q.Terms(t => t.Field(f => f.OrderId).Terms(orderIds)) // && q.Terms(t => t.Field(f => f.TrackCode.Suffix("keyword")).Terms(trackCodes)) // ) // ); // // if (!deleteResponse.IsValid) // { // throw new InvalidOperationException( // $"删除轨迹失败,orderIds:[{string.Join(", ", orderIds)}],trackCodes:[{string.Join(", ", trackCodes)}],原因:{deleteResponse.DebugInformation}。"); // } var search = await Context.SearchAsync(x => x .Index(GlobalConst.TrackDetailsIndex) .Source(s => s.Includes(fd => fd .Field(f => f.IndexName) .Field(f => f.OrderId) .Field(f => f.TrackCode) )) .Size(SafelyLimit) .Query(q => q.Terms(t => t.Field(f => f.OrderId).Terms(orderIds)) && q.Terms(t => t.Field(f => f.TrackCode.Suffix("keyword")).Terms(trackCodes)) ) ); if (!search.IsValid) { throw new InvalidOperationException( $"轨迹查询失败,orderIds:[{orderIds.JoinAsString(", ")}],trackCodes:[{trackCodes.JoinAsString(", ")}],原因:{search.DebugInformation}。"); } //按订单id和code分组,只删除组元素数量为1的轨迹 var tracks = search.ToIdList().GroupBy(doc => new { doc.OrderId, doc.TrackCode }) .Where(g => g.Count() == 1) .Select(g => g.First()) // 只保留每组的第一个元素 .ToList(); foreach (var track in tracks) { await DeleteTrackByIdAsync(track.Id, track.IndexName); } return tracks.Select(x => x.OrderId).ToList(); } public async Task DeleteTrackByIdAsync(string id, string indexName) { var r = await Context.DeleteAsync(id, u => u .Index(indexName) ); if (!r.IsValid) { throw new InvalidOperationException($"轨迹删除失败,id:{indexName},原因:{r.DebugInformation}。"); } } public async Task> GetDataByIds(string[] ids) { var res = await Context.SearchAsync(m => m .Index(GlobalConst.TrackDetailsIndex) .Query( q => q.Ids(i => i.Values(ids)) )); return res.ToIdList(); } public async Task GetDetailAsync(string orderNo, string trackCode, int orderType, int subSystemId, DateTime trackTime) { var res = await Context.SearchAsync(s => s .Index("sjzy_track_details") .Query(q => q .Bool(b => b .Must(mu => mu .Terms(t => t .Field(f => f.OrderNo.Suffix("keyword")) .Terms(orderNo)), mu => mu .Terms(t => t .Field(f => f.TrackCode.Suffix("keyword")) .Terms(trackCode)), mu => mu .Terms(t => t .Field(f => f.OrderType) .Terms(orderType)), mu => mu .Terms(t => t .Field(f => f.SubSystemId) .Terms(subSystemId)), mu => mu .Terms(t => t .Field(f => f.TrackTime) .Terms(trackTime)) // 注意:这里你重复了OrderType的查询,你可能需要移除重复的查询 ) ) ) ); return res.Documents.FirstOrDefault(); } }