using Nest; using SJZY.Expand.ElasticSearch.Services; using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; using SJZY.Track.Domain.Consts; using SJZY.Track.Domain.Dtos.TrackDetails; namespace SJZY.Track.Repository.Repositories; internal class TrackDetailsRepository : ElasticSearchContextAbstract, ITrackDetailsRepository, ITransientDependency { public TrackDetailsRepository(ConnectionFactory connectionFactory) : base(connectionFactory) { } public async Task> GetListAsync(int[] orderIds, string trackCode) { var r = await Context.SearchAsync(s => s .Source() .Size(1000) .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 Exception($"轨迹查询,原因:{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 .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) )) .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(orderIds.Length) .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) ) ) ) ) ) ); if (!r.IsValid) { throw new Exception($"轨迹查询,原因:{r.DebugInformation}。"); } return r.Documents.ToList(); } public async Task AnyAsync(int orderId, DateTime trackTime) { var r = await Context.SearchAsync(x => x .Source(s => s.Includes(fd => fd.Field(f => f.TrackCode))) .Query(q => q.Term(f => f.OrderId, orderId) && q.Term(f => f.TrackTime, trackTime) ) ); if (!r.IsValid) { throw new Exception($"轨迹查询,原因:{r.DebugInformation}。"); } return r.Hits.Count > 0; } public async Task AnyAsync(int orderId, string originTrackData) { var r = await Context.SearchAsync(x => x .Source(s => s.Includes(fd => fd.Field(f => f.TrackCode))) .Query(q => q.Term(f => f.OrderId, orderId) && q.Term(f => f.OriginTrackData.Suffix("keyword"), originTrackData) ) ); if (!r.IsValid) { throw new Exception($"轨迹查询,原因:{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(track) //.Refresh(Refresh.True) // 可选:立即刷新索引使更改立即可见 ); if (!r.IsValid) { throw new Exception($"轨迹更新失败,原因:{r.DebugInformation}。"); } } public async Task AddManyAsync(List tracks) { if (!tracks.Any()) return; // 按天写入es var indexName = $"{GlobalConst.TrackIndex}-{DateTime.Now:yyyy.MM.dd}"; tracks.ForEach(x => x.IndexName = indexName); var r = await Context.IndexManyAsync(tracks, indexName); if (!r.IsValid) { throw new Exception($"{tracks.Count}条轨迹新增失败,原因:{r.DebugInformation}。"); } } public async Task AddAsync(TrackDetailsIndex track) { // 按天写入es var indexName = $"{GlobalConst.TrackIndex}-{DateTime.Now:yyyy.MM.dd}"; track.IndexName = indexName; var r = await Context.IndexAsync(track, s => s.Index(indexName)); if (!r.IsValid) { throw new Exception($"轨迹新增失败,原因:{r.DebugInformation}。"); } } public async Task> GetListAsync(int orderType, List orderNos) { if (orderNos.Count < 1) return new List(); var r = await Context.SearchAsync(x => x .Source(s => s.Includes(fd => fd .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.TrackTimeZone) )) .Query(q => q.Terms(t => t .Field(f => f.OrderNo.Suffix("keyword")) .Terms(orderNos)) && q.Term(f => f.OrderType, orderType) ) .Size(1000) ); if (!r.IsValid) { throw new Exception($"轨迹查询,原因:{r.DebugInformation}。"); } return r.Documents.ToList(); } public async Task> GetListAsync(int orderId) { var r = await Context.SearchAsync(x => x .Query(q => q.Term(t => t.OrderId, orderId) ) .Size(1000) .Sort(s => s.Field(f => f.ZeroTrackTime, SortOrder.Descending))); if (!r.IsValid) { throw new Exception($"轨迹查询,原因:{r.DebugInformation}。"); } return r.Documents.ToList(); } public async Task DeleteTrackAsync(int[] orderIds, List trackCodes) { var r = await Context.UpdateByQueryAsync(x => x .Script(s => s .Source("ctx.op = 'delete'") ) .Query( q => q.Terms(t => t.Field(f => f.OrderId).Terms(orderIds)) && q.Term(t => t.IsCanDelete, true) && q.Terms(t => t.Field(f => f.TrackCode.Suffix("keyword")).Terms(trackCodes))) ); if (!r.IsValid) throw new Exception($"轨迹数据删除失败,原因:{r.DebugInformation}"); } public async Task<(List list, long total)> GetListAsync(TrackDetailsQueryInput input) { if (input.OrderNos.Count < 1) return default; var distinctOrderNumbers = input.OrderNos.Distinct(); var finalSize = input.PageSize > 500 ? 500 : input.PageSize; var data = await Context.SearchAsync(x => x .Query(q => q.Terms(t => t .Field(f => f.OrderNo) .Terms(distinctOrderNumbers)) ) .Sort(s => s.Field(f => f.CreateTime, SortOrder.Descending)) .From((input.PageIndex - 1) * input.PageSize) .Size(finalSize) .TrackTotalHits() // todo: 聚合获取匹配查询条件的文档总数,而不受"index.max_result_window"的限制 ); return (data.Documents.ToList(), data.Total); } }