chfei/StoreWeb
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

更新框架，增加云库搜索模块

Browse Source
This commit is contained in:
chfei committed 2026-09-18 18:40:10 +08:00
1 parent 8526e5e0d2
commit f2caec6083
192 files changed
+3626 -804
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

Models/CloudProduct/Services/OpenSearchServer.cs
+498
View File
@@ -0,0 +1,498 @@
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;
namespace Store.Modules.CloudProduct.Services
{
public interface IOpenSearchServer: ISingletonDependency
{
Task AddOneRecordAsync(ProductSearchInfo goods);
Task BulkIndexGoodsAsync(IEnumerable<ProductSearchInfo> docs);
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);
}
internal class OpenSearchServer : IOpenSearchServer
{
private readonly string nodeUrl;
private IOpenSearchClient _client = null!;
private readonly string IndexName = "";
// 查询类操作（search/count/status/ping）统一超时
private static readonly TimeSpan QueryTimeout = TimeSpan.FromSeconds(5);
// 批量写入/建索引/删除索引等管理类操作超时（数据量大时需要更长，可挪到配置文件）
private static readonly TimeSpan AdminTimeout = TimeSpan.FromSeconds(30);
public OpenSearchServer(IConfiguration configuration)
{
var config = configuration.GetSection("Search");
nodeUrl = config.GetValue<string>("Url") ?? "";
IndexName = config.GetValue<string>("IndexName") ?? "";
_client = CreateClient();
}
private IOpenSearchClient CreateClient()
{
var pool = new SingleNodeConnectionPool(new Uri(nodeUrl));
var settings = new ConnectionSettings(pool)
.DefaultIndex(IndexName)
.DefaultMappingFor<ProductSearchInfo>(m => m.IndexName(IndexName))
.RequestTimeout(QueryTimeout)
.MaxRetryTimeout(QueryTimeout)
.PrettyJson()
.DisableDirectStreaming(); // 便于调试打印请求/响应报文
return new OpenSearchClient(settings);
}
/// <summary>
/// ============ 1. 创建索引 ============
/// </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())
{
Console.WriteLine($"索引 [{IndexName}] 已存在且映射正确，无需重复创建。");
return true;
}
Console.WriteLine($"索引 [{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)
{
Console.WriteLine("IK 分词插件不可用，已回退到 standard 分析器，中文分词将失效，请检查集群是否安装 analysis-ik 插件。");
createResponse = await TryCreateIndexAsync(useIk: false);
}
if (createResponse.IsValid)
{
Console.WriteLine($"索引 [{IndexName}] 创建成功。");
return true;
}
else
{
throw new AppException("索引创建失败: " + createResponse.DebugInformation);
}
}
/// <summary>
/// 校验索引映射是否健康：title 字段必须使用 IK 分析器
/// </summary>
private async Task<bool> IsIndexMappingHealthyAsync()
{
try
{
var getResponse = await _client.Indices.GetAsync(IndexName, d => d
.RequestConfiguration(r => r.RequestTimeout(QueryTimeout)));
if (!getResponse.IsValid || !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;
}
return textProperty.Analyzer == "ik_max_word" && textProperty.SearchAnalyzer == "ik_smart";
}
catch
{
return false;
}
}
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<ProductSearchInfo>(m => m
.Properties(p => p
.Number(n => n.Name(x => x.Id).Type(NumberType.Long))
.Text(t => t
.Name(x => x.Title)
.Analyzer(indexAnalyzer)
.SearchAnalyzer(searchAnalyzer)
)
.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))
)
)
.RequestConfiguration(r => r.RequestTimeout(AdminTimeout))
);
}
/// <summary>
/// ============ 2. 删除索引 ============
/// </summary>
public async Task DeleteIndexAsync()
{
var response = await _client.Indices.DeleteAsync(IndexName, d => d
.RequestConfiguration(r => r.RequestTimeout(AdminTimeout)));
if (!response.IsValid)
{
throw new AppException("删除失败: " + response.DebugInformation);
}
}
/// <summary>
/// ============ 3.1 批量写入商品搜索文档（存在则覆盖） ============
/// </summary>
public async Task BulkIndexGoodsAsync(IEnumerable<ProductSearchInfo> docs)
{
if (docs == null) throw new ArgumentNullException(nameof(docs));
var list = docs.ToList();
if (list.Count == 0) return;
// 确保索引已按 IK 分词映射创建，否则中文分词无效
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("写入 OpenSearch 失败: " + bulkResponse.DebugInformation);
}
// 刷新索引，保证数据立即可搜索（仅用于演示，生产环境不建议频繁 Refresh）
await _client.Indices.RefreshAsync(IndexName, d => d
.RequestConfiguration(r => r.RequestTimeout(AdminTimeout)));
}
/// <summary>
/// ============ 3.2 根据数据库商品重建搜索索引 ============
/// </summary>
public async Task<long> RebuildIndexAsync(IEnumerable<ProductEntity> goodsList)
{
if (goodsList == null) throw new ArgumentNullException(nameof(goodsList));
var docs = goodsList
.Where(x => x != null && x.GoodsId > 0)
.Select(x => new ProductSearchInfo
{
Id = x.GoodsId,
Title = x.Title ?? string.Empty,
CatId1 = x.CatId1,
CatId2 = x.CatId2,
CatId3 = x.CatId3,
CatId4 = x.CatId4,
MinPrice = x.MinPrice,
SoldQuantity = x.SoldQuantity
})
.ToList();
if (docs.Count == 0) return 0;
await BulkIndexGoodsAsync(docs);
return docs.Count;
}
/// <summary>
/// ============ 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
.RequestConfiguration(r => r.RequestTimeout(QueryTimeout)));
if (!exists.Exists)
{
return;
}
var response = await _client.BulkAsync(b =>
{
foreach (var id in list)
{
b.Delete<ProductSearchInfo>(d => d.Index(IndexName).Id(id));
}
b.RequestConfiguration(r => r.RequestTimeout(AdminTimeout));
return b;
});
if (!response.IsValid)
{
throw new AppException("删除 OpenSearch 文档失败: " + response.DebugInformation);
}
await _client.Indices.RefreshAsync(IndexName, d => d
.RequestConfiguration(r => r.RequestTimeout(AdminTimeout)));
}
/// <summary>
/// ============ 4. 手动新增一条 ============
/// </summary>
public async Task AddOneRecordAsync(ProductSearchInfo goods)
{
if (goods == null) throw new ArgumentNullException(nameof(goods));
var response = await _client.IndexAsync(goods, i => i
.Index(IndexName)
.Id(goods.Id)
.RequestConfiguration(r => r.RequestTimeout(QueryTimeout))
);
if (response.IsValid)
{
await _client.Indices.RefreshAsync(IndexName, d => d
.RequestConfiguration(r => r.RequestTimeout(QueryTimeout)));
}
else
{
throw new AppException("新增失败: " + response.DebugInformation);
}
}
/// <summary>
/// ============ 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="sortBy">排序方式</param>
/// <param name="pageIndex">页码，从 1 开始</param>
/// <param name="pageSize">每页条数</param>
public async Task<(long Total, List<ProductSearchInfo> list)> SearchByTitleAsync(ProductSearchRequest request,
GoodsSearchSortBy sortBy = GoodsSearchSortBy.Relevance)
{
string 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? 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;
try
{
searchResponse = await _client.SearchAsync<ProductSearchInfo>(s => s
.Index(IndexName)
.Query(q => q
.Bool(b =>
{
if (string.IsNullOrWhiteSpace(keyword) == false)
{
b.Must(m => m.Match(mm => mm.Field(f => f.Title).Query(keyword)));
}
var filters = new List<Func<QueryContainerDescriptor<ProductSearchInfo>, 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)));
if (catId2.HasValue) filters.Add(f => f.Term(t => t.Field(x => x.CatId2).Value(catId2.Value)));
if (catId3.HasValue) filters.Add(f => f.Term(t => t.Field(x => x.CatId3).Value(catId3.Value)));
if (catId4.HasValue) filters.Add(f => f.Term(t => t.Field(x => x.CatId4).Value(catId4.Value)));
if (minPrice.HasValue || maxPrice.HasValue)
{
filters.Add(f => f.Range(r =>
{
var range = r.Field(x => x.MinPrice);
if (minPrice.HasValue) range = range.GreaterThanOrEquals(minPrice.Value);
if (maxPrice.HasValue) range = range.LessThanOrEquals(maxPrice.Value);
return range;
}));
}
if (minsold.HasValue || maxsold.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);
return range;
}));
}
if (filters.Count > 0) b.Filter(filters.ToArray());
return b;
})
)
.Sort(sort => sortBy switch
{
GoodsSearchSortBy.PriceAsc => sort.Ascending(f => f.MinPrice),
GoodsSearchSortBy.PriceDesc => sort.Descending(f => f.MinPrice),
GoodsSearchSortBy.SoldDesc => sort.Descending(f => f.SoldQuantity),
_ => sort.Descending(SortSpecialField.Score)
})
.From(from)
.Size(pageSize)
.TrackTotalHits(true) // 命中数超过 10000 时默认不精确统计，需要显式开启
.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);
}
//var ids = searchResponse.Documents.Select(d => d.Id).ToList();
return (searchResponse.Total, searchResponse.Documents.ToList());
}
/// <summary>
/// ============ 6. 统计文档数 ============
/// </summary>
public async Task<long> CountDocumentsAsync()
{
var countResponse = await _client.CountAsync<ProductSearchInfo>(c => c
.Index(IndexName)
.RequestConfiguration(r => r.RequestTimeout(QueryTimeout))
);
if (countResponse.IsValid)
{
Console.WriteLine($"索引 [{IndexName}] 当前文档总数: {countResponse.Count}");
return countResponse.Count;
}
else
{
Console.WriteLine("统计失败: " + countResponse.DebugInformation);
throw new AppException(countResponse.DebugInformation);
}
}
/// <summary>
/// ============ 7. 查询 OpenSearch 状态 ============
/// </summary>
public async Task<OpenSearchStatusDto> GetStatusAsync()
{
var status = new OpenSearchStatusDto
{
NodeUrl = nodeUrl,
IndexName = IndexName,
};
try
{
var ping = await _client.PingAsync(p => p
.RequestConfiguration(r => r.RequestTimeout(QueryTimeout)));
status.Connected = ping.IsValid;
if (!ping.IsValid)
{
status.Message = ping.DebugInformation;
return status;
}
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)
.RequestConfiguration(r => r.RequestTimeout(QueryTimeout)));
status.DocumentCount = count.IsValid ? count.Count : -1;
if (!count.IsValid)
status.Message = count.DebugInformation;
}
}
catch (Exception ex)
{
status.Connected = false;
status.Message = ex.Message;
}
return status;
}
/// <summary>
/// 清空索引：删除后按统一映射重建，重建期间该索引不可查（由后台定时任务负责回填数据，接受短暂空窗）
/// </summary>
public async Task<bool> ClearIndexAsync()
{
var existsResponse = await _client.Indices.ExistsAsync(IndexName, d => d
.RequestConfiguration(r => r.RequestTimeout(QueryTimeout)));
if (!existsResponse.Exists)
{
Console.WriteLine($"索引 {IndexName} 不存在");
return false;
}
var deleteResponse = await _client.Indices.DeleteAsync(IndexName, d => d
.RequestConfiguration(r => r.RequestTimeout(AdminTimeout)));
if (!deleteResponse.IsValid)
{
Console.WriteLine($"删除索引 {IndexName} 失败: {deleteResponse.DebugInformation}");
return false;
}
Console.WriteLine($"索引 {IndexName} 已删除");
return await CreateIndexAsync();
}
}
}
Models/CloudProduct/Services/ProductRepository.cs
+13
View File
@@ -0,0 +1,13 @@
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)
{
}
}
}
Models/CloudProduct/Services/ProductService.cs
+409
View File
@@ -0,0 +1,409 @@
using MAX.Core.DependencyInjection;
using Microsoft.Extensions.Configuration;
using Store.Modules.CloudProduct.Dtos;
using Store.Modules.CloudProduct.Entities;
namespace Store.Modules.CloudProduct.Services
{
public interface IProductService : IScopedDependency
{
Task<int> DeleteGoodsAsync(IEnumerable<long> ids);
Task<List<ProductEntity>> GetGoodsByIdsAsync(IEnumerable<long> ids);
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);
}
internal class ProductService : IProductService
{
private readonly ProductRepository _repository;
private readonly OpenSearchServer _search;
private readonly RawDataStorage _rawDataStorage;
private readonly IConfiguration _configuration;
public ProductService(ProductRepository repository, OpenSearchServer search, RawDataStorage rawDataStorage, IConfiguration configuration)
{
_repository = repository;
_search = search;
_rawDataStorage = rawDataStorage;
_configuration = configuration;
}
public async Task<(List<ProductEntity> List, long Total)> SearchGoods(ProductSearchRequest request)
{
// 1. 从搜索引擎获取标题匹配的商品ID列表
List<ProductEntity> goodsList = new();
long total = 0;
// 从搜索引擎查询标题匹配的ID
(long Total, List<ProductSearchInfo> list) searchRet = (await _search.SearchByTitleAsync(request));
goodsList = searchRet.list.Select(a => new ProductEntity
{
Platform = a.Platform,
GoodsId = a.Id,
Title = a.Title,
CatId1 = a.CatId1,
CatId2 = a.CatId2,
CatId3 = a.CatId3,
CatId4 = a.CatId4,
MinPrice = a.MinPrice,
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);
//// 2. 构建数据库查询（只查正常状态，排除软删除）
//var query = _repository.Select.Where(x => x.Status == 0);
//// 如果有标题搜索，先用ID过滤
//if (goodsIds.Any())
//{
// query = query.Where(x => goodsIds.Contains(x.GoodsId));
//}
//// 3. 价格区间查询
//if (request.MinPrice.HasValue)
//{
// query = query.Where(x => x.MinPrice >= request.MinPrice.Value);
//}
//if (request.MaxPrice.HasValue)
//{
// query = query.Where(x => x.MaxPrice <= request.MaxPrice.Value);
//}
//// 4. 类目查询
//if (request.CatId1.HasValue)
//{
// query = query.Where(x => x.CatId1 == request.CatId1);
//}
//if (request.CatId2.HasValue)
//{
// query = query.Where(x => x.CatId2 == request.CatId2);
//}
//if (request.CatId3.HasValue)
//{
// query = query.Where(x => x.CatId3 == request.CatId3);
//}
//if (request.CatId4.HasValue)
//{
// query = query.Where(x => x.CatId4 == request.CatId4);
//}
//// 5. 销量区间查询
//if (request.MinSoldQuantity.HasValue)
//{
// query = query.Where(x => x.SoldQuantity >= request.MinSoldQuantity.Value);
//}
//if (request.MaxSoldQuantity.HasValue)
//{
// query = query.Where(x => x.SoldQuantity <= request.MaxSoldQuantity.Value);
//}
//// 6. 平台过滤
//if (request.Platform.HasValue)
//{
// query = query.Where(x => x.Platform == request.Platform);
//}
//// 8. 获取总数
//total = await query.CountAsync();
//// 9. 排序
//if (!string.IsNullOrEmpty(request.SortBy))
//{
// var isAsc = request.SortOrder?.ToLower() == "asc";
// query = query.OrderByPropertyName(request.SortBy, isAsc);
//}
//else
//{
// query = query.OrderByDescending(x => x.CreatedTime);
//}
//// 10. 分页查询
//var list = await query
// //.Skip((request.PageIndex - 1) * request.PageSize)
// //.Limit(request.PageSize)
// .ToListAsync();
////内存排序
//var orderMap = goodsIds.Select((id, idx) => new { id, idx })
// .ToDictionary(x => x.id, x => x.idx);
//list = list.OrderBy(x => orderMap.TryGetValue(x.GoodsId, out var idx) ? idx : int.MaxValue)
// .ToList();
//return (list, total);
}
/// <summary>
/// 批量导入商品：先落库（按平台 + 商品编号去重）+ 保存原始数据，
/// 然后根据配置决定“立即分词”还是“标记未同步交给后台任务分词”。
/// </summary>
public async Task<GoodsImportResult> ImportGoodsAsync(IEnumerable<GoodsImportItem> items)
{
var result = new GoodsImportResult();
if (items == null)
return result;
var indexImmediately = IsImmediateIndexing();
var docs = new List<ProductSearchInfo>();
var indexedIds = new List<long>();
foreach (var item in items)
{
var goods = item?.ProductEntity;
if (goods == null)
{
result.Failed++;
result.Errors.Add("商品数据为空");
continue;
}
var goodsId = goods.GoodsId;
if (goodsId == 0)
{
result.Failed++;
result.Errors.Add("商品编号（GoodsId）不能为空");
continue;
}
try
{
// 导入时统一先标记“未同步”；立即分词成功后再置为“已同步”，
// 后台模式下保持“未同步”，由 Hangfire 定时任务分词。
goods.IsSegmented = 0;
var saved = await UpsertGoodsAsync(goods);
var rawData = item?.RawData ?? string.Empty;
if (!string.IsNullOrWhiteSpace(rawData))
{
await _rawDataStorage.SaveAsync(saved.GoodsId, rawData);
}
if (indexImmediately)
{
docs.Add(new ProductSearchInfo
{
Platform = saved.Platform,
Id = saved.GoodsId,
Title = saved.Title ?? string.Empty,
ThumbUrl = saved.ThumbUrl,
CatId1 = saved.CatId1,
CatId2 = saved.CatId2,
CatId3 = saved.CatId3,
CatId4 = saved.CatId4,
MinPrice = saved.MinPrice,
SoldQuantity = saved.SoldQuantity,
});
indexedIds.Add(saved.GoodsId);
}
result.Success++;
}
catch (Exception ex)
{
result.Failed++;
result.Errors.Add($"GoodsId={goodsId}: {ex.Message}");
}
}
if (docs.Count > 0)
{
try
{
await _search.BulkIndexGoodsAsync(docs);
await MarkIndexedAsync(indexedIds);
}
catch (Exception ex)
{
// 写入失败时保持“未同步”，由后台定时任务重试
result.Failed += docs.Count;
result.Success -= docs.Count;
result.Errors.Add("写入 OpenSearch 失败: " + ex.Message);
}
}
return result;
}
/// <summary>
/// 后台任务：批量处理未同步（IndexStatus=0）的商品，写入 OpenSearch 后标记为已同步。
/// </summary>
public async Task<int> IndexPendingGoodsAsync(int batchSize)
{
if (batchSize <= 0)
batchSize = 200;
var pending = await _repository.Select
.Where(x => x.IsSegmented == 0 && x.Status == 0)
.OrderBy(x => x.GoodsId)
.Limit(batchSize)
.ToListAsync();
if (pending.Count == 0)
return 0;
var docs = pending
.Select(x => new ProductSearchInfo
{
Platform = x.Platform,
Id = x.GoodsId,
Title = x.Title ?? string.Empty,
ThumbUrl = x.ThumbUrl,
CatId1 = x.CatId1,
CatId2 = x.CatId2,
CatId3 = x.CatId3,
CatId4 = x.CatId4,
MinPrice = x.MinPrice,
SoldQuantity = x.SoldQuantity,
})
.ToList();
await _search.BulkIndexGoodsAsync(docs);
await MarkIndexedAsync(pending.Select(x => x.GoodsId));
return pending.Count;
}
/// <summary>
/// 按平台 + 商品编号查找已存在记录，存在则更新，不存在则新增。
/// </summary>
private async Task<ProductEntity> UpsertGoodsAsync(ProductEntity goods)
{
var existing = await _repository.Select
.Where(x => x.Platform == goods.Platform && x.GoodsId == goods.GoodsId)
.FirstAsync();
if (existing != null)
{
goods.Id = existing.Id;
goods.GoodsId = existing.GoodsId;
goods.CreatedTime = existing.CreatedTime;
goods.UpdatedTime = DateTime.Now;
await _repository.UpdateAsync(goods);
return goods;
}
goods.CreatedTime = DateTime.Now;
goods.UpdatedTime = DateTime.Now;
return await _repository.InsertAsync(goods);
}
/// <summary>
/// 根据数据库主键 ID 列表读取商品。
/// </summary>
public async Task<List<ProductEntity>> GetGoodsByIdsAsync(IEnumerable<long> ids)
{
var idList = ids?.ToList();
if (idList == null || idList.Count == 0)
return new List<ProductEntity>();
return await _repository.Select
.Where(x => idList.Contains(x.GoodsId))
.ToListAsync();
}
/// <summary>
/// 重新从数据库读取指定商品，覆盖 OpenSearch 中已有的分词并重新分词。
/// ids 为空时重建全部商品。重建成功的记录标记为已同步。
/// </summary>
public async Task<long> RebuildSearchIndexAsync(IEnumerable<long>? ids = null)
{
List<ProductEntity> goodsList;
var idList = ids?.ToList();
if (idList == null || idList.Count == 0)
{
goodsList = await _repository.Select.ToListAsync();
}
else
{
goodsList = await _repository.Select
.Where(x => idList.Contains(x.GoodsId))
.ToListAsync();
}
var count = await _search.RebuildIndexAsync(goodsList);
await MarkIndexedAsync(goodsList.Where(x => x.GoodsId > 0).Select(x => x.GoodsId));
return count;
}
/// <summary>
/// 删除商品：按配置决定软删除（Status=1）还是硬删除（物理删除），
/// 两种方式都会删除 OpenSearch 中对应的文档。
/// </summary>
public async Task<int> DeleteGoodsAsync(IEnumerable<long> ids)
{
var idList = ids?.Distinct().ToList();
if (idList == null || idList.Count == 0)
return 0;
if (IsSoftDelete())
{
// 软删除：仅用 Status 标记（0正常 1已删除）
await _repository.Orm.Update<ProductEntity>()
.Set(x => x.Status, (byte)1)
.Where(x => idList.Contains(x.GoodsId))
.ExecuteAffrowsAsync();
}
else
{
// 硬删除：直接删除数据库记录
await _repository.Orm.Delete<ProductEntity>()
.Where(x => idList.Contains(x.GoodsId))
.ExecuteAffrowsAsync();
}
// 无论软删除还是硬删除，都要删除 OpenSearch 中的对应文档
await _search.DeleteDocumentsAsync(idList);
return idList.Count;
}
public async Task<int> UpdateImage(IEnumerable<ProductEntity> goods)
{
//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)
.UpdateColumns(a => new { a.ThumbUrl })
.ExecuteAffrowsAsync();
}
/// <summary>
/// 是否导入时立即分词（Indexing:Mode = Immediate）。
/// </summary>
private bool IsImmediateIndexing()
{
var mode = _configuration.GetValue<string>("Indexing:Mode");
return string.Equals(mode, "Immediate", StringComparison.OrdinalIgnoreCase);
}
/// <summary>
/// 是否软删除（Delete:SoftDelete，默认 true）。
/// </summary>
private bool IsSoftDelete()
{
return _configuration.GetValue<bool?>("Delete:SoftDelete") ?? true;
}
private async Task MarkIndexedAsync(IEnumerable<long> ids)
{
var idList = ids?.ToList();
if (idList == null || idList.Count == 0)
return;
await _repository.Orm.Update<ProductEntity>()
.Set(x => x.IsSegmented, (byte)1)
.Where(x => idList.Contains(x.GoodsId))
.ExecuteAffrowsAsync();
}
}
}
Models/CloudProduct/Services/RawDataStorage.cs
+77
View File
@@ -0,0 +1,77 @@
using MAX.Core.DependencyInjection;
using Microsoft.Extensions.Configuration;
using System.IO.Compression;
using System.Text;
namespace Store.Modules.CloudProduct.Services
{
/// <summary>
/// 商品原始数据文件存储：按 goodsId % 100 分目录，文件名即商品ID，GZip 压缩保存。
/// </summary>
public class RawDataStorage : IScopedDependency
{
private readonly string _rootPath;
public RawDataStorage(IConfiguration configuration)
{
var configured = configuration.GetValue<string>("RawData:Path");
if (string.IsNullOrWhiteSpace(configured))
configured = "RawData";
_rootPath = Path.GetFullPath(configured);
}
/// <summary>
/// 保存原始数据，返回保存后的完整路径。
/// </summary>
public async Task<string> SaveAsync(long goodsId, string rawData)
{
var filePath = GetFilePath(goodsId);
var dir = Path.GetDirectoryName(filePath)!;
Directory.CreateDirectory(dir);
await using var fileStream = new FileStream(filePath, FileMode.Create, FileAccess.Write, FileShare.None);
await using var gzip = new GZipStream(fileStream, CompressionLevel.Fastest);
await using var writer = new StreamWriter(gzip, new UTF8Encoding(false));
await writer.WriteAsync(rawData);
return filePath;
}
/// <summary>
/// 读取并解压原始数据；文件不存在时返回 null。
/// </summary>
public async Task<string?> ReadAsync(long goodsId)
{
var filePath = GetFilePath(goodsId);
if (!File.Exists(filePath))
return null;
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();
}
/// <summary>
/// 判断指定商品的原始数据文件是否存在。
/// </summary>
public bool Exists(long goodsId)
{
return File.Exists(GetFilePath(goodsId));
}
private string GetFilePath(long goodsId)
{
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();
}
}
}