gx
This commit is contained in:
1 parent
f2caec6083
commit
75d0e85218
16 files changed
+316
-320
No files matched your search
+51
-58
@@ -2,44 +2,42 @@ using MAX.Core.DependencyInjection;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using Store.Modules.CloudProduct.Dtos;
|
||||
using Store.Modules.CloudProduct.Entities;
|
||||
using Store.Modules.CloudProduct.Repositorys;
|
||||
|
||||
namespace Store.Modules.CloudProduct.Services
|
||||
{
|
||||
public interface IProductService : IScopedDependency
|
||||
public interface IGoodsService : IScopedDependency
|
||||
{
|
||||
Task<int> DeleteGoodsAsync(IEnumerable<long> ids);
|
||||
Task<List<ProductEntity>> GetGoodsByIdsAsync(IEnumerable<long> ids);
|
||||
Task<List<GoodsEntity>> GetGoodsByGoodsIdsAsync(IEnumerable<long> goodsIds);
|
||||
Task<GoodsImportResult> ImportGoodsAsync(IEnumerable<GoodsImportItem> items);
|
||||
Task<int> IndexPendingGoodsAsync(int batchSize);
|
||||
Task<long> RebuildSearchIndexAsync(IEnumerable<long>? ids = null);
|
||||
Task<(List<ProductEntity> List, long Total)> SearchGoods(ProductSearchRequest request);
|
||||
Task<int> UpdateImage(IEnumerable<ProductEntity> goods);
|
||||
Task<(List<GoodsEntity> List, long Total)> SearchGoodsAsync(GoodsSearchRequest request);
|
||||
Task<int> UpdateThumbUrlAsync(IEnumerable<GoodsEntity> goodsList);
|
||||
}
|
||||
|
||||
internal class ProductService : IProductService
|
||||
internal class GoodsService : IGoodsService
|
||||
{
|
||||
private readonly ProductRepository _repository;
|
||||
private readonly OpenSearchServer _search;
|
||||
private readonly RawDataStorage _rawDataStorage;
|
||||
private readonly GoodsRepository _repository;
|
||||
private readonly IOpenSearchService _searchService;
|
||||
private readonly IRawDataStorage _rawDataStorage;
|
||||
private readonly IConfiguration _configuration;
|
||||
|
||||
public ProductService(ProductRepository repository, OpenSearchServer search, RawDataStorage rawDataStorage, IConfiguration configuration)
|
||||
public GoodsService(GoodsRepository repository, IOpenSearchService searchService, IRawDataStorage rawDataStorage, IConfiguration configuration)
|
||||
{
|
||||
_repository = repository;
|
||||
_search = search;
|
||||
_searchService = searchService;
|
||||
_rawDataStorage = rawDataStorage;
|
||||
_configuration = configuration;
|
||||
}
|
||||
|
||||
public async Task<(List<ProductEntity> List, long Total)> SearchGoods(ProductSearchRequest request)
|
||||
public async Task<(List<GoodsEntity> List, long Total)> SearchGoodsAsync(GoodsSearchRequest request)
|
||||
{
|
||||
// 1. 从搜索引擎获取标题匹配的商品ID列表
|
||||
List<ProductEntity> goodsList = new();
|
||||
long total = 0;
|
||||
// 1. 从搜索引擎获取标题匹配的商品
|
||||
var searchResult = await _searchService.SearchByTitleAsync(request);
|
||||
|
||||
// 从搜索引擎查询标题匹配的ID
|
||||
(long Total, List<ProductSearchInfo> list) searchRet = (await _search.SearchByTitleAsync(request));
|
||||
goodsList = searchRet.list.Select(a => new ProductEntity
|
||||
var goodsList = searchResult.Documents.Select(a => new GoodsEntity
|
||||
{
|
||||
Platform = a.Platform,
|
||||
GoodsId = a.Id,
|
||||
@@ -52,17 +50,9 @@ namespace Store.Modules.CloudProduct.Services
|
||||
MaxPrice = 0,
|
||||
SoldQuantity = a.SoldQuantity,
|
||||
ThumbUrl = a.ThumbUrl,
|
||||
|
||||
}).ToList();
|
||||
//goodsIds = searchRet.list.Select(d => d.Id).ToList();
|
||||
total = searchRet.Total;
|
||||
//// 如果没有匹配的结果,直接返回空
|
||||
//if (!goodsIds.Any())
|
||||
//{
|
||||
// return (new List<ProductEntity>(), 0);
|
||||
//}
|
||||
|
||||
return (goodsList, total);
|
||||
return (goodsList, searchResult.Total);
|
||||
|
||||
//// 2. 构建数据库查询(只查正常状态,排除软删除)
|
||||
//var query = _repository.Select.Where(x => x.Status == 0);
|
||||
@@ -155,15 +145,15 @@ namespace Store.Modules.CloudProduct.Services
|
||||
return result;
|
||||
|
||||
var indexImmediately = IsImmediateIndexing();
|
||||
var docs = new List<ProductSearchInfo>();
|
||||
var documents = new List<GoodsSearchDocument>();
|
||||
var indexedIds = new List<long>();
|
||||
|
||||
foreach (var item in items)
|
||||
{
|
||||
var goods = item?.ProductEntity;
|
||||
var goods = item?.Goods;
|
||||
if (goods == null)
|
||||
{
|
||||
result.Failed++;
|
||||
result.FailedCount++;
|
||||
result.Errors.Add("商品数据为空");
|
||||
continue;
|
||||
}
|
||||
@@ -171,7 +161,7 @@ namespace Store.Modules.CloudProduct.Services
|
||||
var goodsId = goods.GoodsId;
|
||||
if (goodsId == 0)
|
||||
{
|
||||
result.Failed++;
|
||||
result.FailedCount++;
|
||||
result.Errors.Add("商品编号(GoodsId)不能为空");
|
||||
continue;
|
||||
}
|
||||
@@ -191,7 +181,7 @@ namespace Store.Modules.CloudProduct.Services
|
||||
|
||||
if (indexImmediately)
|
||||
{
|
||||
docs.Add(new ProductSearchInfo
|
||||
documents.Add(new GoodsSearchDocument
|
||||
{
|
||||
Platform = saved.Platform,
|
||||
Id = saved.GoodsId,
|
||||
@@ -207,27 +197,27 @@ namespace Store.Modules.CloudProduct.Services
|
||||
indexedIds.Add(saved.GoodsId);
|
||||
}
|
||||
|
||||
result.Success++;
|
||||
result.SuccessCount++;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
result.Failed++;
|
||||
result.FailedCount++;
|
||||
result.Errors.Add($"GoodsId={goodsId}: {ex.Message}");
|
||||
}
|
||||
}
|
||||
|
||||
if (docs.Count > 0)
|
||||
if (documents.Count > 0)
|
||||
{
|
||||
try
|
||||
{
|
||||
await _search.BulkIndexGoodsAsync(docs);
|
||||
await _searchService.BulkIndexGoodsAsync(documents);
|
||||
await MarkIndexedAsync(indexedIds);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
// 写入失败时保持“未同步”,由后台定时任务重试
|
||||
result.Failed += docs.Count;
|
||||
result.Success -= docs.Count;
|
||||
result.FailedCount += documents.Count;
|
||||
result.SuccessCount -= documents.Count;
|
||||
result.Errors.Add("写入 OpenSearch 失败: " + ex.Message);
|
||||
}
|
||||
}
|
||||
@@ -236,7 +226,7 @@ namespace Store.Modules.CloudProduct.Services
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 后台任务:批量处理未同步(IndexStatus=0)的商品,写入 OpenSearch 后标记为已同步。
|
||||
/// 后台任务:批量处理未同步(IsSegmented=0)的商品,写入 OpenSearch 后标记为已同步。
|
||||
/// </summary>
|
||||
public async Task<int> IndexPendingGoodsAsync(int batchSize)
|
||||
{
|
||||
@@ -252,8 +242,8 @@ namespace Store.Modules.CloudProduct.Services
|
||||
if (pending.Count == 0)
|
||||
return 0;
|
||||
|
||||
var docs = pending
|
||||
.Select(x => new ProductSearchInfo
|
||||
var documents = pending
|
||||
.Select(x => new GoodsSearchDocument
|
||||
{
|
||||
Platform = x.Platform,
|
||||
Id = x.GoodsId,
|
||||
@@ -268,7 +258,7 @@ namespace Store.Modules.CloudProduct.Services
|
||||
})
|
||||
.ToList();
|
||||
|
||||
await _search.BulkIndexGoodsAsync(docs);
|
||||
await _searchService.BulkIndexGoodsAsync(documents);
|
||||
|
||||
await MarkIndexedAsync(pending.Select(x => x.GoodsId));
|
||||
return pending.Count;
|
||||
@@ -277,7 +267,7 @@ namespace Store.Modules.CloudProduct.Services
|
||||
/// <summary>
|
||||
/// 按平台 + 商品编号查找已存在记录,存在则更新,不存在则新增。
|
||||
/// </summary>
|
||||
private async Task<ProductEntity> UpsertGoodsAsync(ProductEntity goods)
|
||||
private async Task<GoodsEntity> UpsertGoodsAsync(GoodsEntity goods)
|
||||
{
|
||||
var existing = await _repository.Select
|
||||
.Where(x => x.Platform == goods.Platform && x.GoodsId == goods.GoodsId)
|
||||
@@ -299,13 +289,13 @@ namespace Store.Modules.CloudProduct.Services
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 根据数据库主键 ID 列表读取商品。
|
||||
/// 根据商品编号(GoodsId)列表读取商品。
|
||||
/// </summary>
|
||||
public async Task<List<ProductEntity>> GetGoodsByIdsAsync(IEnumerable<long> ids)
|
||||
public async Task<List<GoodsEntity>> GetGoodsByGoodsIdsAsync(IEnumerable<long> goodsIds)
|
||||
{
|
||||
var idList = ids?.ToList();
|
||||
var idList = goodsIds?.ToList();
|
||||
if (idList == null || idList.Count == 0)
|
||||
return new List<ProductEntity>();
|
||||
return new List<GoodsEntity>();
|
||||
|
||||
return await _repository.Select
|
||||
.Where(x => idList.Contains(x.GoodsId))
|
||||
@@ -318,7 +308,7 @@ namespace Store.Modules.CloudProduct.Services
|
||||
/// </summary>
|
||||
public async Task<long> RebuildSearchIndexAsync(IEnumerable<long>? ids = null)
|
||||
{
|
||||
List<ProductEntity> goodsList;
|
||||
List<GoodsEntity> goodsList;
|
||||
|
||||
var idList = ids?.ToList();
|
||||
if (idList == null || idList.Count == 0)
|
||||
@@ -332,7 +322,7 @@ namespace Store.Modules.CloudProduct.Services
|
||||
.ToListAsync();
|
||||
}
|
||||
|
||||
var count = await _search.RebuildIndexAsync(goodsList);
|
||||
var count = await _searchService.RebuildIndexAsync(goodsList);
|
||||
await MarkIndexedAsync(goodsList.Where(x => x.GoodsId > 0).Select(x => x.GoodsId));
|
||||
return count;
|
||||
}
|
||||
@@ -350,7 +340,7 @@ namespace Store.Modules.CloudProduct.Services
|
||||
if (IsSoftDelete())
|
||||
{
|
||||
// 软删除:仅用 Status 标记(0正常 1已删除)
|
||||
await _repository.Orm.Update<ProductEntity>()
|
||||
await _repository.Orm.Update<GoodsEntity>()
|
||||
.Set(x => x.Status, (byte)1)
|
||||
.Where(x => idList.Contains(x.GoodsId))
|
||||
.ExecuteAffrowsAsync();
|
||||
@@ -358,25 +348,28 @@ namespace Store.Modules.CloudProduct.Services
|
||||
else
|
||||
{
|
||||
// 硬删除:直接删除数据库记录
|
||||
await _repository.Orm.Delete<ProductEntity>()
|
||||
await _repository.Orm.Delete<GoodsEntity>()
|
||||
.Where(x => idList.Contains(x.GoodsId))
|
||||
.ExecuteAffrowsAsync();
|
||||
}
|
||||
|
||||
// 无论软删除还是硬删除,都要删除 OpenSearch 中的对应文档
|
||||
await _search.DeleteDocumentsAsync(idList);
|
||||
await _searchService.DeleteDocumentsAsync(idList);
|
||||
return idList.Count;
|
||||
}
|
||||
|
||||
public async Task<int> UpdateImage(IEnumerable<ProductEntity> goods)
|
||||
/// <summary>
|
||||
/// 批量更新商品缩略图。
|
||||
/// </summary>
|
||||
public async Task<int> UpdateThumbUrlAsync(IEnumerable<GoodsEntity> goodsList)
|
||||
{
|
||||
//string sql = _repository.Orm.Update<ProductEntity>().SetSource(goods, a => a.GoodsId).UpdateColumns(a => new { a.ThumbUrl }).ToSql();
|
||||
return await _repository.Orm.Update<ProductEntity>()
|
||||
.SetSource(goods, a => a.GoodsId)
|
||||
//string sql = _repository.Orm.Update<GoodsEntity>().SetSource(goodsList, a => a.GoodsId).UpdateColumns(a => new { a.ThumbUrl }).ToSql();
|
||||
return await _repository.Orm.Update<GoodsEntity>()
|
||||
.SetSource(goodsList, a => a.GoodsId)
|
||||
.UpdateColumns(a => new { a.ThumbUrl })
|
||||
.ExecuteAffrowsAsync();
|
||||
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 是否导入时立即分词(Indexing:Mode = Immediate)。
|
||||
/// </summary>
|
||||
@@ -400,10 +393,10 @@ namespace Store.Modules.CloudProduct.Services
|
||||
if (idList == null || idList.Count == 0)
|
||||
return;
|
||||
|
||||
await _repository.Orm.Update<ProductEntity>()
|
||||
await _repository.Orm.Update<GoodsEntity>()
|
||||
.Set(x => x.IsSegmented, (byte)1)
|
||||
.Where(x => idList.Contains(x.GoodsId))
|
||||
.ExecuteAffrowsAsync();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
+76
-92
@@ -1,53 +1,53 @@
|
||||
using MAX.Core.DependencyInjection;
|
||||
using MAX.Framework.Commons;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using MySqlX.XDevAPI;
|
||||
using OpenSearch.Client;
|
||||
using OpenSearch.Net;
|
||||
using Store.Modules.CloudProduct.Dtos;
|
||||
using Store.Modules.CloudProduct.Entities;
|
||||
using Store.Modules.CloudProduct.Enums;
|
||||
|
||||
namespace Store.Modules.CloudProduct.Services
|
||||
{
|
||||
public interface IOpenSearchServer: ISingletonDependency
|
||||
public interface IOpenSearchService : ISingletonDependency
|
||||
{
|
||||
Task AddOneRecordAsync(ProductSearchInfo goods);
|
||||
Task BulkIndexGoodsAsync(IEnumerable<ProductSearchInfo> docs);
|
||||
Task AddDocumentAsync(GoodsSearchDocument document);
|
||||
Task BulkIndexGoodsAsync(IEnumerable<GoodsSearchDocument> documents);
|
||||
Task<bool> ClearIndexAsync();
|
||||
Task<long> CountDocumentsAsync();
|
||||
Task<bool> CreateIndexAsync();
|
||||
Task DeleteDocumentsAsync(IEnumerable<long> ids);
|
||||
Task DeleteIndexAsync();
|
||||
Task<OpenSearchStatusDto> GetStatusAsync();
|
||||
Task<long> RebuildIndexAsync(IEnumerable<ProductEntity> goodsList);
|
||||
Task<(long Total, List<ProductSearchInfo> list)> SearchByTitleAsync(ProductSearchRequest request, GoodsSearchSortBy sortBy = GoodsSearchSortBy.Relevance);
|
||||
Task<long> RebuildIndexAsync(IEnumerable<GoodsEntity> goodsList);
|
||||
Task<(long Total, List<GoodsSearchDocument> Documents)> SearchByTitleAsync(GoodsSearchRequest request, GoodsSearchSortBy sortBy = GoodsSearchSortBy.Relevance);
|
||||
}
|
||||
|
||||
internal class OpenSearchServer : IOpenSearchServer
|
||||
internal class OpenSearchService : IOpenSearchService
|
||||
{
|
||||
private readonly string nodeUrl;
|
||||
private IOpenSearchClient _client = null!;
|
||||
private readonly string IndexName = "";
|
||||
private readonly string _nodeUrl;
|
||||
private readonly string _indexName;
|
||||
private readonly IOpenSearchClient _client = null!;
|
||||
|
||||
// 查询类操作(search/count/status/ping)统一超时
|
||||
private static readonly TimeSpan QueryTimeout = TimeSpan.FromSeconds(5);
|
||||
// 批量写入/建索引/删除索引等管理类操作超时(数据量大时需要更长,可挪到配置文件)
|
||||
private static readonly TimeSpan AdminTimeout = TimeSpan.FromSeconds(30);
|
||||
|
||||
public OpenSearchServer(IConfiguration configuration)
|
||||
public OpenSearchService(IConfiguration configuration)
|
||||
{
|
||||
var config = configuration.GetSection("Search");
|
||||
nodeUrl = config.GetValue<string>("Url") ?? "";
|
||||
IndexName = config.GetValue<string>("IndexName") ?? "";
|
||||
_nodeUrl = config.GetValue<string>("Url") ?? "";
|
||||
_indexName = config.GetValue<string>("IndexName") ?? "";
|
||||
_client = CreateClient();
|
||||
}
|
||||
|
||||
private IOpenSearchClient CreateClient()
|
||||
{
|
||||
var pool = new SingleNodeConnectionPool(new Uri(nodeUrl));
|
||||
var pool = new SingleNodeConnectionPool(new Uri(_nodeUrl));
|
||||
var settings = new ConnectionSettings(pool)
|
||||
.DefaultIndex(IndexName)
|
||||
.DefaultMappingFor<ProductSearchInfo>(m => m.IndexName(IndexName))
|
||||
.DefaultIndex(_indexName)
|
||||
.DefaultMappingFor<GoodsSearchDocument>(m => m.IndexName(_indexName))
|
||||
.RequestTimeout(QueryTimeout)
|
||||
.MaxRetryTimeout(QueryTimeout)
|
||||
.PrettyJson()
|
||||
@@ -61,18 +61,18 @@ namespace Store.Modules.CloudProduct.Services
|
||||
/// </summary>
|
||||
public async Task<bool> CreateIndexAsync()
|
||||
{
|
||||
var existsResponse = await _client.Indices.ExistsAsync(IndexName, d => d
|
||||
var existsResponse = await _client.Indices.ExistsAsync(_indexName, d => d
|
||||
.RequestConfiguration(r => r.RequestTimeout(AdminTimeout)));
|
||||
if (existsResponse.Exists)
|
||||
{
|
||||
if (await IsIndexMappingHealthyAsync())
|
||||
{
|
||||
Console.WriteLine($"索引 [{IndexName}] 已存在且映射正确,无需重复创建。");
|
||||
Console.WriteLine($"索引 [{_indexName}] 已存在且映射正确,无需重复创建。");
|
||||
return true;
|
||||
}
|
||||
|
||||
Console.WriteLine($"索引 [{IndexName}] 的映射不完整(可能由清空索引等操作造成),正在重建...");
|
||||
var deleteResponse = await _client.Indices.DeleteAsync(IndexName, d => d
|
||||
Console.WriteLine($"索引 [{_indexName}] 的映射不完整(可能由清空索引等操作造成),正在重建...");
|
||||
var deleteResponse = await _client.Indices.DeleteAsync(_indexName, d => d
|
||||
.RequestConfiguration(r => r.RequestTimeout(AdminTimeout)));
|
||||
if (!deleteResponse.IsValid)
|
||||
{
|
||||
@@ -91,7 +91,7 @@ namespace Store.Modules.CloudProduct.Services
|
||||
|
||||
if (createResponse.IsValid)
|
||||
{
|
||||
Console.WriteLine($"索引 [{IndexName}] 创建成功。");
|
||||
Console.WriteLine($"索引 [{_indexName}] 创建成功。");
|
||||
return true;
|
||||
}
|
||||
else
|
||||
@@ -107,9 +107,9 @@ namespace Store.Modules.CloudProduct.Services
|
||||
{
|
||||
try
|
||||
{
|
||||
var getResponse = await _client.Indices.GetAsync(IndexName, d => d
|
||||
var getResponse = await _client.Indices.GetAsync(_indexName, d => d
|
||||
.RequestConfiguration(r => r.RequestTimeout(QueryTimeout)));
|
||||
if (!getResponse.IsValid || !getResponse.Indices.TryGetValue(IndexName, out var indexState))
|
||||
if (!getResponse.IsValid || !getResponse.Indices.TryGetValue(_indexName, out var indexState))
|
||||
{
|
||||
return false;
|
||||
}
|
||||
@@ -135,12 +135,12 @@ namespace Store.Modules.CloudProduct.Services
|
||||
var indexAnalyzer = useIk ? "ik_max_word" : "standard";
|
||||
var searchAnalyzer = useIk ? "ik_smart" : "standard";
|
||||
|
||||
return await _client.Indices.CreateAsync(IndexName, c => c
|
||||
return await _client.Indices.CreateAsync(_indexName, c => c
|
||||
.Settings(s => s
|
||||
.NumberOfShards(1)
|
||||
.NumberOfReplicas(0)
|
||||
)
|
||||
.Map<ProductSearchInfo>(m => m
|
||||
.Map<GoodsSearchDocument>(m => m
|
||||
.Properties(p => p
|
||||
.Number(n => n.Name(x => x.Id).Type(NumberType.Long))
|
||||
.Text(t => t
|
||||
@@ -165,7 +165,7 @@ namespace Store.Modules.CloudProduct.Services
|
||||
/// </summary>
|
||||
public async Task DeleteIndexAsync()
|
||||
{
|
||||
var response = await _client.Indices.DeleteAsync(IndexName, d => d
|
||||
var response = await _client.Indices.DeleteAsync(_indexName, d => d
|
||||
.RequestConfiguration(r => r.RequestTimeout(AdminTimeout)));
|
||||
if (!response.IsValid)
|
||||
{
|
||||
@@ -176,18 +176,18 @@ namespace Store.Modules.CloudProduct.Services
|
||||
/// <summary>
|
||||
/// ============ 3.1 批量写入商品搜索文档(存在则覆盖) ============
|
||||
/// </summary>
|
||||
public async Task BulkIndexGoodsAsync(IEnumerable<ProductSearchInfo> docs)
|
||||
public async Task BulkIndexGoodsAsync(IEnumerable<GoodsSearchDocument> documents)
|
||||
{
|
||||
if (docs == null) throw new ArgumentNullException(nameof(docs));
|
||||
if (documents == null) throw new ArgumentNullException(nameof(documents));
|
||||
|
||||
var list = docs.ToList();
|
||||
var list = documents.ToList();
|
||||
if (list.Count == 0) return;
|
||||
|
||||
// 确保索引已按 IK 分词映射创建,否则中文分词无效
|
||||
await CreateIndexAsync();
|
||||
|
||||
var bulkResponse = await _client.BulkAsync(b => b
|
||||
.Index(IndexName)
|
||||
.Index(_indexName)
|
||||
.IndexMany(list, (descriptor, doc) => descriptor.Id(doc.Id))
|
||||
.RequestConfiguration(r => r.RequestTimeout(AdminTimeout))
|
||||
);
|
||||
@@ -198,20 +198,20 @@ namespace Store.Modules.CloudProduct.Services
|
||||
}
|
||||
|
||||
// 刷新索引,保证数据立即可搜索(仅用于演示,生产环境不建议频繁 Refresh)
|
||||
await _client.Indices.RefreshAsync(IndexName, d => d
|
||||
await _client.Indices.RefreshAsync(_indexName, d => d
|
||||
.RequestConfiguration(r => r.RequestTimeout(AdminTimeout)));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// ============ 3.2 根据数据库商品重建搜索索引 ============
|
||||
/// </summary>
|
||||
public async Task<long> RebuildIndexAsync(IEnumerable<ProductEntity> goodsList)
|
||||
public async Task<long> RebuildIndexAsync(IEnumerable<GoodsEntity> goodsList)
|
||||
{
|
||||
if (goodsList == null) throw new ArgumentNullException(nameof(goodsList));
|
||||
|
||||
var docs = goodsList
|
||||
var documents = goodsList
|
||||
.Where(x => x != null && x.GoodsId > 0)
|
||||
.Select(x => new ProductSearchInfo
|
||||
.Select(x => new GoodsSearchDocument
|
||||
{
|
||||
Id = x.GoodsId,
|
||||
Title = x.Title ?? string.Empty,
|
||||
@@ -224,21 +224,21 @@ namespace Store.Modules.CloudProduct.Services
|
||||
})
|
||||
.ToList();
|
||||
|
||||
if (docs.Count == 0) return 0;
|
||||
if (documents.Count == 0) return 0;
|
||||
|
||||
await BulkIndexGoodsAsync(docs);
|
||||
return docs.Count;
|
||||
await BulkIndexGoodsAsync(documents);
|
||||
return documents.Count;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// ============ 3.3 按主键批量删除 OpenSearch 文档(软/硬删除共用) ============
|
||||
/// ============ 3.3 按商品编号批量删除 OpenSearch 文档(软/硬删除共用) ============
|
||||
/// </summary>
|
||||
public async Task DeleteDocumentsAsync(IEnumerable<long> ids)
|
||||
{
|
||||
var list = ids?.Distinct().ToList() ?? new List<long>();
|
||||
if (list.Count == 0) return;
|
||||
|
||||
var exists = await _client.Indices.ExistsAsync(IndexName, d => d
|
||||
var exists = await _client.Indices.ExistsAsync(_indexName, d => d
|
||||
.RequestConfiguration(r => r.RequestTimeout(QueryTimeout)));
|
||||
if (!exists.Exists)
|
||||
{
|
||||
@@ -249,7 +249,7 @@ namespace Store.Modules.CloudProduct.Services
|
||||
{
|
||||
foreach (var id in list)
|
||||
{
|
||||
b.Delete<ProductSearchInfo>(d => d.Index(IndexName).Id(id));
|
||||
b.Delete<GoodsSearchDocument>(d => d.Index(_indexName).Id(id));
|
||||
}
|
||||
b.RequestConfiguration(r => r.RequestTimeout(AdminTimeout));
|
||||
return b;
|
||||
@@ -260,25 +260,26 @@ namespace Store.Modules.CloudProduct.Services
|
||||
throw new AppException("删除 OpenSearch 文档失败: " + response.DebugInformation);
|
||||
}
|
||||
|
||||
await _client.Indices.RefreshAsync(IndexName, d => d
|
||||
await _client.Indices.RefreshAsync(_indexName, d => d
|
||||
.RequestConfiguration(r => r.RequestTimeout(AdminTimeout)));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// ============ 4. 手动新增一条 ============
|
||||
/// ============ 4. 手动新增一条搜索文档 ============
|
||||
/// </summary>
|
||||
public async Task AddOneRecordAsync(ProductSearchInfo goods)
|
||||
public async Task AddDocumentAsync(GoodsSearchDocument document)
|
||||
{
|
||||
if (goods == null) throw new ArgumentNullException(nameof(goods));
|
||||
var response = await _client.IndexAsync(goods, i => i
|
||||
.Index(IndexName)
|
||||
.Id(goods.Id)
|
||||
if (document == null) throw new ArgumentNullException(nameof(document));
|
||||
|
||||
var response = await _client.IndexAsync(document, i => i
|
||||
.Index(_indexName)
|
||||
.Id(document.Id)
|
||||
.RequestConfiguration(r => r.RequestTimeout(QueryTimeout))
|
||||
);
|
||||
|
||||
if (response.IsValid)
|
||||
{
|
||||
await _client.Indices.RefreshAsync(IndexName, d => d
|
||||
await _client.Indices.RefreshAsync(_indexName, d => d
|
||||
.RequestConfiguration(r => r.RequestTimeout(QueryTimeout)));
|
||||
}
|
||||
else
|
||||
@@ -288,51 +289,37 @@ namespace Store.Modules.CloudProduct.Services
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// ============ 5. 分词搜索(分页 + 类目/价格过滤 + 排序) ============
|
||||
/// ============ 5. 分词搜索(分页 + 类目/价格/销量/平台过滤 + 排序) ============
|
||||
/// </summary>
|
||||
/// <param name="keyword">搜索关键词</param>
|
||||
/// <param name="catId1">一级类目,不传则不限制</param>
|
||||
/// <param name="catId2">二级类目,不传则不限制</param>
|
||||
/// <param name="catId3">三级类目,不传则不限制</param>
|
||||
/// <param name="catId4">四级类目,不传则不限制</param>
|
||||
/// <param name="minPrice">最低价(含),不传则不限制</param>
|
||||
/// <param name="maxPrice">最高价(含),不传则不限制</param>
|
||||
/// <param name="request">搜索条件(关键词、类目、价格、销量、平台、分页)</param>
|
||||
/// <param name="sortBy">排序方式</param>
|
||||
/// <param name="pageIndex">页码,从 1 开始</param>
|
||||
/// <param name="pageSize">每页条数</param>
|
||||
public async Task<(long Total, List<ProductSearchInfo> list)> SearchByTitleAsync(ProductSearchRequest request,
|
||||
public async Task<(long Total, List<GoodsSearchDocument> Documents)> SearchByTitleAsync(GoodsSearchRequest request,
|
||||
GoodsSearchSortBy sortBy = GoodsSearchSortBy.Relevance)
|
||||
{
|
||||
string keyword = request.Title;
|
||||
var keyword = request.Title;
|
||||
int? catId1 = request.CatId1;
|
||||
int? catId2 = request.CatId2;
|
||||
int? catId3 = request.CatId3;
|
||||
int? catId4 = request.CatId4;
|
||||
long? minPrice = request.MinPrice;
|
||||
long? maxPrice = request.MaxPrice;
|
||||
int? minsold = request.MinSoldQuantity;
|
||||
int? maxsold = request.MaxSoldQuantity;
|
||||
int? minSoldQuantity = request.MinSoldQuantity;
|
||||
int? maxSoldQuantity = request.MaxSoldQuantity;
|
||||
int? platform = request.Platform;
|
||||
int pageIndex = request.PageIndex;
|
||||
int pageSize = request.PageSize;
|
||||
|
||||
|
||||
//if (string.IsNullOrWhiteSpace(keyword))
|
||||
//{
|
||||
// throw new AppException("关键词不能为空。");
|
||||
//}
|
||||
|
||||
if (pageIndex < 1) pageIndex = 1;
|
||||
if (pageSize < 1) pageSize = 20;
|
||||
if (pageSize > 2000) pageSize = 2000; // 避免深分页/单页过大拖垮性能,按需调整上限
|
||||
|
||||
var from = (pageIndex - 1) * pageSize;
|
||||
|
||||
ISearchResponse<ProductSearchInfo> searchResponse;
|
||||
ISearchResponse<GoodsSearchDocument> searchResponse;
|
||||
try
|
||||
{
|
||||
searchResponse = await _client.SearchAsync<ProductSearchInfo>(s => s
|
||||
.Index(IndexName)
|
||||
searchResponse = await _client.SearchAsync<GoodsSearchDocument>(s => s
|
||||
.Index(_indexName)
|
||||
.Query(q => q
|
||||
.Bool(b =>
|
||||
{
|
||||
@@ -341,7 +328,7 @@ namespace Store.Modules.CloudProduct.Services
|
||||
b.Must(m => m.Match(mm => mm.Field(f => f.Title).Query(keyword)));
|
||||
}
|
||||
|
||||
var filters = new List<Func<QueryContainerDescriptor<ProductSearchInfo>, QueryContainer>>();
|
||||
var filters = new List<Func<QueryContainerDescriptor<GoodsSearchDocument>, QueryContainer>>();
|
||||
if (platform.HasValue && platform!.Value > 0) filters.Add(f => f.Term(t => t.Field(x => x.Platform).Value(platform.Value)));
|
||||
|
||||
if (catId1.HasValue) filters.Add(f => f.Term(t => t.Field(x => x.CatId1).Value(catId1.Value)));
|
||||
@@ -360,19 +347,17 @@ namespace Store.Modules.CloudProduct.Services
|
||||
}));
|
||||
}
|
||||
|
||||
if (minsold.HasValue || maxsold.HasValue)
|
||||
if (minSoldQuantity.HasValue || maxSoldQuantity.HasValue)
|
||||
{
|
||||
filters.Add(f => f.Range(r =>
|
||||
{
|
||||
var range = r.Field(x => x.SoldQuantity);
|
||||
if (minsold.HasValue) range = range.GreaterThanOrEquals(minsold.Value);
|
||||
if (maxsold.HasValue) range = range.LessThanOrEquals(maxsold.Value);
|
||||
if (minSoldQuantity.HasValue) range = range.GreaterThanOrEquals(minSoldQuantity.Value);
|
||||
if (maxSoldQuantity.HasValue) range = range.LessThanOrEquals(maxSoldQuantity.Value);
|
||||
return range;
|
||||
}));
|
||||
}
|
||||
|
||||
|
||||
|
||||
if (filters.Count > 0) b.Filter(filters.ToArray());
|
||||
return b;
|
||||
})
|
||||
@@ -400,7 +385,6 @@ namespace Store.Modules.CloudProduct.Services
|
||||
throw new AppException("搜索失败: " + searchResponse.DebugInformation);
|
||||
}
|
||||
|
||||
//var ids = searchResponse.Documents.Select(d => d.Id).ToList();
|
||||
return (searchResponse.Total, searchResponse.Documents.ToList());
|
||||
}
|
||||
|
||||
@@ -409,13 +393,13 @@ namespace Store.Modules.CloudProduct.Services
|
||||
/// </summary>
|
||||
public async Task<long> CountDocumentsAsync()
|
||||
{
|
||||
var countResponse = await _client.CountAsync<ProductSearchInfo>(c => c
|
||||
.Index(IndexName)
|
||||
var countResponse = await _client.CountAsync<GoodsSearchDocument>(c => c
|
||||
.Index(_indexName)
|
||||
.RequestConfiguration(r => r.RequestTimeout(QueryTimeout))
|
||||
);
|
||||
if (countResponse.IsValid)
|
||||
{
|
||||
Console.WriteLine($"索引 [{IndexName}] 当前文档总数: {countResponse.Count}");
|
||||
Console.WriteLine($"索引 [{_indexName}] 当前文档总数: {countResponse.Count}");
|
||||
return countResponse.Count;
|
||||
}
|
||||
else
|
||||
@@ -432,8 +416,8 @@ namespace Store.Modules.CloudProduct.Services
|
||||
{
|
||||
var status = new OpenSearchStatusDto
|
||||
{
|
||||
NodeUrl = nodeUrl,
|
||||
IndexName = IndexName,
|
||||
NodeUrl = _nodeUrl,
|
||||
IndexName = _indexName,
|
||||
};
|
||||
|
||||
try
|
||||
@@ -447,14 +431,14 @@ namespace Store.Modules.CloudProduct.Services
|
||||
return status;
|
||||
}
|
||||
|
||||
var exists = await _client.Indices.ExistsAsync(IndexName, d => d
|
||||
var exists = await _client.Indices.ExistsAsync(_indexName, d => d
|
||||
.RequestConfiguration(r => r.RequestTimeout(QueryTimeout)));
|
||||
status.IndexExists = exists.Exists;
|
||||
|
||||
if (exists.Exists)
|
||||
{
|
||||
var count = await _client.CountAsync<ProductSearchInfo>(c => c
|
||||
.Index(IndexName)
|
||||
var count = await _client.CountAsync<GoodsSearchDocument>(c => c
|
||||
.Index(_indexName)
|
||||
.RequestConfiguration(r => r.RequestTimeout(QueryTimeout)));
|
||||
status.DocumentCount = count.IsValid ? count.Count : -1;
|
||||
if (!count.IsValid)
|
||||
@@ -475,24 +459,24 @@ namespace Store.Modules.CloudProduct.Services
|
||||
/// </summary>
|
||||
public async Task<bool> ClearIndexAsync()
|
||||
{
|
||||
var existsResponse = await _client.Indices.ExistsAsync(IndexName, d => d
|
||||
var existsResponse = await _client.Indices.ExistsAsync(_indexName, d => d
|
||||
.RequestConfiguration(r => r.RequestTimeout(QueryTimeout)));
|
||||
if (!existsResponse.Exists)
|
||||
{
|
||||
Console.WriteLine($"索引 {IndexName} 不存在");
|
||||
Console.WriteLine($"索引 {_indexName} 不存在");
|
||||
return false;
|
||||
}
|
||||
|
||||
var deleteResponse = await _client.Indices.DeleteAsync(IndexName, d => d
|
||||
var deleteResponse = await _client.Indices.DeleteAsync(_indexName, d => d
|
||||
.RequestConfiguration(r => r.RequestTimeout(AdminTimeout)));
|
||||
if (!deleteResponse.IsValid)
|
||||
{
|
||||
Console.WriteLine($"删除索引 {IndexName} 失败: {deleteResponse.DebugInformation}");
|
||||
Console.WriteLine($"删除索引 {_indexName} 失败: {deleteResponse.DebugInformation}");
|
||||
return false;
|
||||
}
|
||||
Console.WriteLine($"索引 {IndexName} 已删除");
|
||||
Console.WriteLine($"索引 {_indexName} 已删除");
|
||||
|
||||
return await CreateIndexAsync();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,13 +0,0 @@
|
||||
using FreeSql;
|
||||
using MAX.Core.DependencyInjection;
|
||||
using Store.Modules.CloudProduct.Entities;
|
||||
|
||||
namespace Store.Modules.CloudProduct.Services
|
||||
{
|
||||
public class ProductRepository : BaseRepository<ProductEntity>, IScopedDependency
|
||||
{
|
||||
public ProductRepository(IFreeSql fsql) : base(fsql)
|
||||
{
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,4 +1,4 @@
|
||||
using MAX.Core.DependencyInjection;
|
||||
using MAX.Core.DependencyInjection;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using System.IO.Compression;
|
||||
using System.Text;
|
||||
@@ -8,7 +8,25 @@ namespace Store.Modules.CloudProduct.Services
|
||||
/// <summary>
|
||||
/// 商品原始数据文件存储:按 goodsId % 100 分目录,文件名即商品ID,GZip 压缩保存。
|
||||
/// </summary>
|
||||
public class RawDataStorage : IScopedDependency
|
||||
public interface IRawDataStorage : IScopedDependency
|
||||
{
|
||||
/// <summary>
|
||||
/// 保存原始数据,返回保存后的完整路径。
|
||||
/// </summary>
|
||||
Task<string> SaveAsync(long goodsId, string rawData);
|
||||
|
||||
/// <summary>
|
||||
/// 读取并解压原始数据;文件不存在时返回 null。
|
||||
/// </summary>
|
||||
Task<string?> ReadAsync(long goodsId);
|
||||
|
||||
/// <summary>
|
||||
/// 判断指定商品的原始数据文件是否存在。
|
||||
/// </summary>
|
||||
bool Exists(long goodsId);
|
||||
}
|
||||
|
||||
internal class RawDataStorage : IRawDataStorage
|
||||
{
|
||||
private readonly string _rootPath;
|
||||
|
||||
@@ -21,9 +39,7 @@ namespace Store.Modules.CloudProduct.Services
|
||||
_rootPath = Path.GetFullPath(configured);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 保存原始数据,返回保存后的完整路径。
|
||||
/// </summary>
|
||||
/// <inheritdoc />
|
||||
public async Task<string> SaveAsync(long goodsId, string rawData)
|
||||
{
|
||||
var filePath = GetFilePath(goodsId);
|
||||
@@ -38,9 +54,7 @@ namespace Store.Modules.CloudProduct.Services
|
||||
return filePath;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 读取并解压原始数据;文件不存在时返回 null。
|
||||
/// </summary>
|
||||
/// <inheritdoc />
|
||||
public async Task<string?> ReadAsync(long goodsId)
|
||||
{
|
||||
var filePath = GetFilePath(goodsId);
|
||||
@@ -53,9 +67,7 @@ namespace Store.Modules.CloudProduct.Services
|
||||
return await reader.ReadToEndAsync();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 判断指定商品的原始数据文件是否存在。
|
||||
/// </summary>
|
||||
/// <inheritdoc />
|
||||
public bool Exists(long goodsId)
|
||||
{
|
||||
return File.Exists(GetFilePath(goodsId));
|
||||
@@ -66,12 +78,5 @@ namespace Store.Modules.CloudProduct.Services
|
||||
var subDir = (goodsId % 100).ToString();
|
||||
return Path.Combine(_rootPath, subDir, goodsId.ToString());
|
||||
}
|
||||
public async Task<string?> ReadAsync(string filePath)
|
||||
{
|
||||
await using var fileStream = new FileStream(filePath, FileMode.Open, FileAccess.Read, FileShare.Read);
|
||||
await using var gzip = new GZipStream(fileStream, CompressionMode.Decompress);
|
||||
using var reader = new StreamReader(gzip, new UTF8Encoding(false));
|
||||
return await reader.ReadToEndAsync();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user