443 lines
16 KiB
C#
443 lines
16 KiB
C#
using MAX.Plugins;
|
|
using Microsoft.Extensions.DependencyInjection;
|
|
using MAX.Core.DependencyInjection;
|
|
using MAX.Framework.Commons;
|
|
using MAX.Framework.Dtos;
|
|
using Microsoft.Extensions.Configuration;
|
|
using Microsoft.Extensions.Logging;
|
|
using OpenSearch.Client;
|
|
using OpenSearch.Net;
|
|
using Store.Modules.CloudProduct.Dtos;
|
|
using Store.Modules.CloudProduct.Enums;
|
|
|
|
namespace Store.Modules.CloudProduct.Services
|
|
{
|
|
/// <summary>
|
|
/// 用户收藏商品索引服务：负责收藏索引（Search:FavoritesIndexName）的
|
|
/// 创建、写入、删除、重建与检索，索引文档由 user_goods_favorites 关联 goods 补全而来。
|
|
/// </summary>
|
|
public interface IUserFavoritesIndexService : ISingletonDependency
|
|
{
|
|
/// <summary>检查索引是否存在</summary>
|
|
Task<bool> IndexExistsAsync();
|
|
|
|
/// <summary>检查索引映射（分析器）是否符合当前服务要求</summary>
|
|
Task<bool> IsIndexMappingHealthyAsync();
|
|
|
|
/// <summary>创建索引（含 IK 分词映射；已存在且映射正常时跳过）</summary>
|
|
Task<bool> CreateIndexAsync();
|
|
|
|
/// <summary>删除并重建索引（映射统一，重建期间该索引不可查）</summary>
|
|
Task<bool> ClearIndexAsync();
|
|
|
|
/// <summary>批量写入收藏文档（存在则覆盖）</summary>
|
|
Task IndexDocumentsAsync(IEnumerable<UserFavoritesSearchDocument> documents);
|
|
|
|
/// <summary>更新单条收藏的安全字段并等待对检索可见。</summary>
|
|
Task UpdateSafetyAsync(string documentId, bool isSafe);
|
|
|
|
/// <summary>按文档Id批量删除收藏文档</summary>
|
|
Task DeleteDocumentsAsync(IEnumerable<string> documentIds);
|
|
|
|
/// <summary>统计索引文档数</summary>
|
|
Task<long> CountDocumentsAsync();
|
|
|
|
/// <summary>按用户检索收藏商品（标题/销量/价格/类目/分组）</summary>
|
|
Task<(long Total, List<UserFavoritesSearchDocument> Documents)> SearchAsync(UserFavoritesSearchRequest request, UserFavoritesSortBy sortBy);
|
|
}
|
|
|
|
internal class UserFavoritesIndexService : IUserFavoritesIndexService
|
|
{
|
|
private readonly string _nodeUrl;
|
|
private readonly string _indexName;
|
|
private readonly IOpenSearchClient _client = null!;
|
|
private readonly ILogger<UserFavoritesIndexService> _logger;
|
|
|
|
// 查询类操作（search/count/exists）统一超时
|
|
private static readonly TimeSpan QueryTimeout = TimeSpan.FromSeconds(5);
|
|
// 批量写入/建索引/删除索引等管理类操作超时（数据量大时需要更长）
|
|
private static readonly TimeSpan AdminTimeout = TimeSpan.FromSeconds(30);
|
|
|
|
public UserFavoritesIndexService([FromKeyedServices("CloudProduct")] PluginContext context, ILogger<UserFavoritesIndexService> logger)
|
|
{
|
|
var config = context.Configuration.GetSection("Search");
|
|
_nodeUrl = config.GetValue<string>("Url") ?? "";
|
|
_indexName = config.GetValue<string>("FavoritesIndexName") ?? "";
|
|
_logger = logger;
|
|
_client = CreateClient();
|
|
}
|
|
|
|
private IOpenSearchClient CreateClient()
|
|
{
|
|
var pool = new SingleNodeConnectionPool(new Uri(_nodeUrl));
|
|
var settings = new ConnectionSettings(pool)
|
|
.DefaultIndex(_indexName)
|
|
.DefaultMappingFor<UserFavoritesSearchDocument>(m => m.IndexName(_indexName))
|
|
.RequestTimeout(QueryTimeout)
|
|
.MaxRetryTimeout(QueryTimeout)
|
|
.PrettyJson()
|
|
.DisableDirectStreaming();
|
|
|
|
return new OpenSearchClient(settings);
|
|
}
|
|
|
|
/// <summary>
|
|
/// ============ 1. 索引是否存在 ============
|
|
/// </summary>
|
|
public async Task<bool> IndexExistsAsync()
|
|
{
|
|
var response = await _client.Indices.ExistsAsync(_indexName, d => d
|
|
.RequestConfiguration(r => r.RequestTimeout(QueryTimeout)));
|
|
return response.Exists;
|
|
}
|
|
|
|
/// <summary>
|
|
/// ============ 2. 索引映射是否健康 ============
|
|
/// </summary>
|
|
public async Task<bool> IsIndexMappingHealthyAsync()
|
|
{
|
|
var getResponse = await _client.Indices.GetAsync(_indexName, d => d
|
|
.RequestConfiguration(r => r.RequestTimeout(QueryTimeout)));
|
|
if (!getResponse.IsValid)
|
|
{
|
|
throw new AppException("读取收藏索引映射失败: " + getResponse.DebugInformation);
|
|
}
|
|
if (!getResponse.Indices.TryGetValue(_indexName, out var indexState))
|
|
{
|
|
return false;
|
|
}
|
|
|
|
var mapping = indexState.Mappings;
|
|
if (mapping?.Properties == null
|
|
|| !mapping.Properties.TryGetValue(new PropertyName("title"), out var property)
|
|
|| property is not ITextProperty textProperty)
|
|
{
|
|
return false;
|
|
}
|
|
|
|
var indexAnalyzer = textProperty.Analyzer;
|
|
var searchAnalyzer = textProperty.SearchAnalyzer;
|
|
var usesIk = indexAnalyzer == "ik_max_word" && searchAnalyzer == "ik_smart";
|
|
var usesStandard = indexAnalyzer == "standard" && searchAnalyzer == "standard";
|
|
return usesIk || usesStandard;
|
|
}
|
|
|
|
/// <summary>
|
|
/// ============ 3. 创建索引 ============
|
|
/// </summary>
|
|
public async Task<bool> CreateIndexAsync()
|
|
{
|
|
var existsResponse = await _client.Indices.ExistsAsync(_indexName, d => d
|
|
.RequestConfiguration(r => r.RequestTimeout(AdminTimeout)));
|
|
if (existsResponse.Exists)
|
|
{
|
|
if (await IsIndexMappingHealthyAsync())
|
|
{
|
|
return true;
|
|
}
|
|
|
|
_logger.LogWarning("收藏索引 [{Index}] 的映射不完整，正在重建...", _indexName);
|
|
var deleteResponse = await _client.Indices.DeleteAsync(_indexName, d => d
|
|
.RequestConfiguration(r => r.RequestTimeout(AdminTimeout)));
|
|
if (!deleteResponse.IsValid)
|
|
{
|
|
throw new AppException("删除损坏的收藏索引失败: " + deleteResponse.DebugInformation);
|
|
}
|
|
}
|
|
|
|
// 优先尝试使用 IK 中文分词（需集群安装 analysis-ik 插件）
|
|
var createResponse = await TryCreateIndexAsync(useIk: true);
|
|
|
|
if (!createResponse.IsValid)
|
|
{
|
|
_logger.LogWarning("IK 分词插件不可用，收藏索引已回退到 standard 分析器，中文分词将失效。");
|
|
createResponse = await TryCreateIndexAsync(useIk: false);
|
|
}
|
|
|
|
if (!createResponse.IsValid)
|
|
{
|
|
throw new AppException("收藏索引创建失败: " + createResponse.DebugInformation);
|
|
}
|
|
|
|
_logger.LogInformation("收藏索引 [{Index}] 创建成功。", _indexName);
|
|
return true;
|
|
}
|
|
|
|
private async Task<CreateIndexResponse> TryCreateIndexAsync(bool useIk)
|
|
{
|
|
var indexAnalyzer = useIk ? "ik_max_word" : "standard";
|
|
var searchAnalyzer = useIk ? "ik_smart" : "standard";
|
|
|
|
return await _client.Indices.CreateAsync(_indexName, c => c
|
|
.Settings(s => s
|
|
.NumberOfShards(1)
|
|
.NumberOfReplicas(0)
|
|
)
|
|
.Map<UserFavoritesSearchDocument>(m => m
|
|
.Properties(p => p
|
|
.Boolean(b => b.Name(x => x.IsSafe))
|
|
.Keyword(k => k.Name(x => x.Id))
|
|
.Number(n => n.Name(x => x.FavId).Type(NumberType.Long))
|
|
.Keyword(k => k.Name(x => x.UserName))
|
|
.Number(n => n.Name(x => x.GroupId).Type(NumberType.Integer))
|
|
.Keyword(k => k.Name(x => x.GroupName))
|
|
.Number(n => n.Name(x => x.GoodsId).Type(NumberType.Long))
|
|
.Number(n => n.Name(x => x.Platform).Type(NumberType.Byte))
|
|
.Text(t => t
|
|
.Name(x => x.Title)
|
|
.Analyzer(indexAnalyzer)
|
|
.SearchAnalyzer(searchAnalyzer)
|
|
)
|
|
.Text(t => t.Name(x => x.ThumbUrl))
|
|
.Number(n => n.Name(x => x.CatId1).Type(NumberType.Integer))
|
|
.Number(n => n.Name(x => x.CatId2).Type(NumberType.Integer))
|
|
.Number(n => n.Name(x => x.CatId3).Type(NumberType.Integer))
|
|
.Number(n => n.Name(x => x.CatId4).Type(NumberType.Integer))
|
|
.Number(n => n.Name(x => x.MinPrice).Type(NumberType.Long))
|
|
.Number(n => n.Name(x => x.SoldQuantity).Type(NumberType.Integer))
|
|
.Date(d => d.Name(x => x.CreatedAt))
|
|
)
|
|
)
|
|
.RequestConfiguration(r => r.RequestTimeout(AdminTimeout))
|
|
);
|
|
}
|
|
|
|
/// <summary>
|
|
/// ============ 4. 删除并重建索引 ============
|
|
/// </summary>
|
|
public async Task<bool> ClearIndexAsync()
|
|
{
|
|
var existsResponse = await _client.Indices.ExistsAsync(_indexName, d => d
|
|
.RequestConfiguration(r => r.RequestTimeout(QueryTimeout)));
|
|
if (!existsResponse.Exists)
|
|
{
|
|
return await CreateIndexAsync();
|
|
}
|
|
|
|
var deleteResponse = await _client.Indices.DeleteAsync(_indexName, d => d
|
|
.RequestConfiguration(r => r.RequestTimeout(AdminTimeout)));
|
|
if (!deleteResponse.IsValid)
|
|
{
|
|
throw new AppException($"删除收藏索引 {_indexName} 失败: " + deleteResponse.DebugInformation);
|
|
}
|
|
|
|
_logger.LogInformation("收藏索引 [{Index}] 已删除，准备重建。", _indexName);
|
|
return await CreateIndexAsync();
|
|
}
|
|
|
|
public async Task UpdateSafetyAsync(string documentId, bool isSafe)
|
|
{
|
|
var response = await _client.UpdateAsync<UserFavoritesSearchDocument, object>(documentId, u => u
|
|
.Index(_indexName).Doc(new { is_safe = isSafe }).Refresh(Refresh.WaitFor));
|
|
if (!response.IsValid)
|
|
throw new AppException("更新收藏安全状态索引失败: " + response.DebugInformation);
|
|
}
|
|
|
|
private async Task RefreshIndexAsync()
|
|
{
|
|
var response = await _client.Indices.RefreshAsync(_indexName, d => d
|
|
.RequestConfiguration(r => r.RequestTimeout(AdminTimeout)));
|
|
if (!response.IsValid)
|
|
{
|
|
throw new AppException("刷新收藏索引失败: " + response.DebugInformation);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// ============ 5. 批量写入收藏文档（存在则覆盖） ============
|
|
/// </summary>
|
|
public async Task IndexDocumentsAsync(IEnumerable<UserFavoritesSearchDocument> documents)
|
|
{
|
|
if (documents == null) throw new ArgumentNullException(nameof(documents));
|
|
|
|
var list = documents.Where(x => x != null && string.IsNullOrWhiteSpace(x.Id) == false).ToList();
|
|
if (list.Count == 0) return;
|
|
|
|
await CreateIndexAsync();
|
|
|
|
var bulkResponse = await _client.BulkAsync(b => b
|
|
.Index(_indexName)
|
|
.IndexMany(list, (descriptor, doc) => descriptor.Id(doc.Id))
|
|
.RequestConfiguration(r => r.RequestTimeout(AdminTimeout))
|
|
);
|
|
|
|
if (!bulkResponse.IsValid)
|
|
{
|
|
throw new AppException("批量写入收藏索引失败: " + bulkResponse.DebugInformation);
|
|
}
|
|
|
|
await RefreshIndexAsync();
|
|
}
|
|
|
|
/// <summary>
|
|
/// ============ 6. 按文档Id批量删除收藏文档 ============
|
|
/// </summary>
|
|
public async Task DeleteDocumentsAsync(IEnumerable<string> documentIds)
|
|
{
|
|
var list = documentIds?.Where(x => string.IsNullOrWhiteSpace(x) == false).Distinct().ToList() ?? new List<string>();
|
|
if (list.Count == 0) return;
|
|
|
|
if (await IndexExistsAsync() == false)
|
|
{
|
|
return;
|
|
}
|
|
|
|
var response = await _client.BulkAsync(b =>
|
|
{
|
|
foreach (var id in list)
|
|
{
|
|
b.Delete<UserFavoritesSearchDocument>(d => d.Index(_indexName).Id(id));
|
|
}
|
|
b.RequestConfiguration(r => r.RequestTimeout(AdminTimeout));
|
|
return b;
|
|
});
|
|
|
|
if (!response.IsValid)
|
|
{
|
|
throw new AppException("删除收藏索引文档失败: " + response.DebugInformation);
|
|
}
|
|
|
|
await RefreshIndexAsync();
|
|
}
|
|
|
|
/// <summary>
|
|
/// ============ 7. 统计文档数 ============
|
|
/// </summary>
|
|
public async Task<long> CountDocumentsAsync()
|
|
{
|
|
var countResponse = await _client.CountAsync<UserFavoritesSearchDocument>(c => c
|
|
.Index(_indexName)
|
|
.RequestConfiguration(r => r.RequestTimeout(QueryTimeout))
|
|
);
|
|
if (!countResponse.IsValid)
|
|
{
|
|
throw new AppException("统计收藏索引失败: " + countResponse.DebugInformation);
|
|
}
|
|
|
|
return countResponse.Count;
|
|
}
|
|
|
|
/// <summary>
|
|
/// ============ 8. 检索用户收藏商品（标题/价格/销量/类目/分组 + 排序 + 分页） ============
|
|
/// </summary>
|
|
public async Task<(long Total, List<UserFavoritesSearchDocument> Documents)> SearchAsync(UserFavoritesSearchRequest request, UserFavoritesSortBy sortBy)
|
|
{
|
|
if (request == null) throw new ArgumentNullException(nameof(request));
|
|
if (string.IsNullOrWhiteSpace(request.UserName))
|
|
{
|
|
throw new AppException("用户名不能为空", ResultCode.ValidationError);
|
|
}
|
|
|
|
if (request.SafeStatus is < 0 or > 3)
|
|
throw new AppException("状态必须为 0（全部）、1（安全）、2（违规）或 3（未检测）", ResultCode.ValidationError);
|
|
|
|
var keyword = request.Title;
|
|
int? groupId = request.GroupId;
|
|
var groupName = request.GroupName;
|
|
int pageIndex = request.PageIndex;
|
|
int pageSize = request.PageSize;
|
|
|
|
if (pageIndex < 1) pageIndex = 1;
|
|
if (pageSize < 1) pageSize = 20;
|
|
if (pageSize > 2000) pageSize = 2000; // 避免深分页/单页过大拖垮性能
|
|
|
|
var from = (pageIndex - 1) * pageSize;
|
|
|
|
ISearchResponse<UserFavoritesSearchDocument> searchResponse;
|
|
try
|
|
{
|
|
searchResponse = await _client.SearchAsync<UserFavoritesSearchDocument>(s => s
|
|
.Index(_indexName)
|
|
.Query(q => q
|
|
.Bool(b =>
|
|
{
|
|
var filters = new List<Func<QueryContainerDescriptor<UserFavoritesSearchDocument>, QueryContainer>>();
|
|
|
|
// 只检索当前用户自己的收藏 —— 放进 filters，避免被后面的 Filter 覆盖
|
|
filters.Add(f => f.Term(t => t.Field(x => x.UserName).Value(request.UserName)));
|
|
|
|
// is_safe 为 null 或缺失时没有索引值，两者均表示暂无检测结论。
|
|
if (request.SafeStatus is 1 or 2)
|
|
filters.Add(f => f.Term(t => t.Field(x => x.IsSafe).Value(request.SafeStatus == 1)));
|
|
else if (request.SafeStatus == 3)
|
|
filters.Add(f => f.Bool(b => b.MustNot(q => q.Exists(e => e.Field(x => x.IsSafe)))));
|
|
|
|
if (groupId.HasValue)
|
|
{
|
|
filters.Add(f => f.Term(t => t.Field(x => x.GroupId).Value(groupId.Value)));
|
|
}
|
|
else if (string.IsNullOrWhiteSpace(groupName) == false)
|
|
{
|
|
filters.Add(f => f.Term(t => t.Field(x => x.GroupName).Value(groupName)));
|
|
}
|
|
|
|
if (request.Platform.HasValue && request.Platform.Value > 0)
|
|
{
|
|
filters.Add(f => f.Term(t => t.Field(x => x.Platform).Value(request.Platform.Value)));
|
|
}
|
|
|
|
if (request.CatId1.HasValue) filters.Add(f => f.Term(t => t.Field(x => x.CatId1).Value(request.CatId1.Value)));
|
|
if (request.CatId2.HasValue) filters.Add(f => f.Term(t => t.Field(x => x.CatId2).Value(request.CatId2.Value)));
|
|
if (request.CatId3.HasValue) filters.Add(f => f.Term(t => t.Field(x => x.CatId3).Value(request.CatId3.Value)));
|
|
if (request.CatId4.HasValue) filters.Add(f => f.Term(t => t.Field(x => x.CatId4).Value(request.CatId4.Value)));
|
|
|
|
if (request.MinPrice.HasValue || request.MaxPrice.HasValue)
|
|
{
|
|
filters.Add(f => f.Range(r =>
|
|
{
|
|
var range = r.Field(x => x.MinPrice);
|
|
if (request.MinPrice.HasValue) range = range.GreaterThanOrEquals(request.MinPrice.Value);
|
|
if (request.MaxPrice.HasValue) range = range.LessThanOrEquals(request.MaxPrice.Value);
|
|
return range;
|
|
}));
|
|
}
|
|
|
|
if (request.MinSoldQuantity.HasValue || request.MaxSoldQuantity.HasValue)
|
|
{
|
|
filters.Add(f => f.Range(r =>
|
|
{
|
|
var range = r.Field(x => x.SoldQuantity);
|
|
if (request.MinSoldQuantity.HasValue) range = range.GreaterThanOrEquals(request.MinSoldQuantity.Value);
|
|
if (request.MaxSoldQuantity.HasValue) range = range.LessThanOrEquals(request.MaxSoldQuantity.Value);
|
|
return range;
|
|
}));
|
|
}
|
|
|
|
// 统一一次性应用所有 filter，避免覆盖
|
|
b.Filter(filters.ToArray());
|
|
|
|
if (string.IsNullOrWhiteSpace(keyword) == false)
|
|
{
|
|
b.Must(m => m.Match(mm => mm.Field(f => f.Title).Query(keyword)));
|
|
}
|
|
|
|
return b;
|
|
})
|
|
)
|
|
.Sort(sort => sortBy switch
|
|
{
|
|
UserFavoritesSortBy.CreatedTimeAsc => sort.Ascending(f => f.CreatedAt),
|
|
UserFavoritesSortBy.PriceAsc => sort.Ascending(f => f.MinPrice),
|
|
UserFavoritesSortBy.PriceDesc => sort.Descending(f => f.MinPrice),
|
|
UserFavoritesSortBy.SoldDesc => sort.Descending(f => f.SoldQuantity),
|
|
_ => sort.Descending(f => f.CreatedAt)
|
|
})
|
|
.From(from)
|
|
.Size(pageSize)
|
|
.TrackTotalHits(true)
|
|
.RequestConfiguration(r => r.RequestTimeout(QueryTimeout))
|
|
);}
|
|
catch (OpenSearchClientException ex) when (ex.Response?.Success == false)
|
|
{
|
|
throw new AppException($"收藏检索超时或失败（{QueryTimeout.TotalSeconds}秒）: " + ex.Message);
|
|
}
|
|
|
|
if (!searchResponse.IsValid)
|
|
{
|
|
throw new AppException("收藏检索失败: " + searchResponse.DebugInformation);
|
|
}
|
|
|
|
return (searchResponse.Total, searchResponse.Documents.ToList());
|
|
}
|
|
}
|
|
} |