403 lines
12 KiB
C#
403 lines
12 KiB
C#
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 IGoodsService : IScopedDependency
|
|
{
|
|
Task<int> DeleteGoodsAsync(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<GoodsEntity> List, long Total)> SearchGoodsAsync(GoodsSearchRequest request);
|
|
Task<int> UpdateThumbUrlAsync(IEnumerable<GoodsEntity> goodsList);
|
|
}
|
|
|
|
internal class GoodsService : IGoodsService
|
|
{
|
|
private readonly GoodsRepository _repository;
|
|
private readonly IOpenSearchService _searchService;
|
|
private readonly IRawDataStorage _rawDataStorage;
|
|
private readonly IConfiguration _configuration;
|
|
|
|
public GoodsService(GoodsRepository repository, IOpenSearchService searchService, IRawDataStorage rawDataStorage, IConfiguration configuration)
|
|
{
|
|
_repository = repository;
|
|
_searchService = searchService;
|
|
_rawDataStorage = rawDataStorage;
|
|
_configuration = configuration;
|
|
}
|
|
|
|
public async Task<(List<GoodsEntity> List, long Total)> SearchGoodsAsync(GoodsSearchRequest request)
|
|
{
|
|
// 1. 从搜索引擎获取标题匹配的商品
|
|
var searchResult = await _searchService.SearchByTitleAsync(request);
|
|
|
|
var goodsList = searchResult.Documents.Select(a => new GoodsEntity
|
|
{
|
|
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();
|
|
|
|
return (goodsList, searchResult.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 documents = new List<GoodsSearchDocument>();
|
|
var indexedIds = new List<long>();
|
|
|
|
foreach (var item in items)
|
|
{
|
|
var goods = item?.Goods;
|
|
if (goods == null)
|
|
{
|
|
result.FailedCount++;
|
|
result.Errors.Add("商品数据为空");
|
|
continue;
|
|
}
|
|
|
|
var goodsId = goods.GoodsId;
|
|
if (goodsId == 0)
|
|
{
|
|
result.FailedCount++;
|
|
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)
|
|
{
|
|
documents.Add(new GoodsSearchDocument
|
|
{
|
|
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.SuccessCount++;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
result.FailedCount++;
|
|
result.Errors.Add($"GoodsId={goodsId}: {ex.Message}");
|
|
}
|
|
}
|
|
|
|
if (documents.Count > 0)
|
|
{
|
|
try
|
|
{
|
|
await _searchService.BulkIndexGoodsAsync(documents);
|
|
await MarkIndexedAsync(indexedIds);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
// 写入失败时保持“未同步”，由后台定时任务重试
|
|
result.FailedCount += documents.Count;
|
|
result.SuccessCount -= documents.Count;
|
|
result.Errors.Add("写入 OpenSearch 失败: " + ex.Message);
|
|
}
|
|
}
|
|
|
|
return result;
|
|
}
|
|
|
|
/// <summary>
|
|
/// 后台任务：批量处理未同步（IsSegmented=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 documents = pending
|
|
.Select(x => new GoodsSearchDocument
|
|
{
|
|
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 _searchService.BulkIndexGoodsAsync(documents);
|
|
|
|
await MarkIndexedAsync(pending.Select(x => x.GoodsId));
|
|
return pending.Count;
|
|
}
|
|
|
|
/// <summary>
|
|
/// 按平台 + 商品编号查找已存在记录，存在则更新，不存在则新增。
|
|
/// </summary>
|
|
private async Task<GoodsEntity> UpsertGoodsAsync(GoodsEntity 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>
|
|
/// 根据商品编号（GoodsId）列表读取商品。
|
|
/// </summary>
|
|
public async Task<List<GoodsEntity>> GetGoodsByGoodsIdsAsync(IEnumerable<long> goodsIds)
|
|
{
|
|
var idList = goodsIds?.ToList();
|
|
if (idList == null || idList.Count == 0)
|
|
return new List<GoodsEntity>();
|
|
|
|
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<GoodsEntity> 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 _searchService.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<GoodsEntity>()
|
|
.Set(x => x.Status, (byte)1)
|
|
.Where(x => idList.Contains(x.GoodsId))
|
|
.ExecuteAffrowsAsync();
|
|
}
|
|
else
|
|
{
|
|
// 硬删除：直接删除数据库记录
|
|
await _repository.Orm.Delete<GoodsEntity>()
|
|
.Where(x => idList.Contains(x.GoodsId))
|
|
.ExecuteAffrowsAsync();
|
|
}
|
|
|
|
// 无论软删除还是硬删除，都要删除 OpenSearch 中的对应文档
|
|
await _searchService.DeleteDocumentsAsync(idList);
|
|
return idList.Count;
|
|
}
|
|
|
|
/// <summary>
|
|
/// 批量更新商品缩略图。
|
|
/// </summary>
|
|
public async Task<int> UpdateThumbUrlAsync(IEnumerable<GoodsEntity> goodsList)
|
|
{
|
|
//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>
|
|
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<GoodsEntity>()
|
|
.Set(x => x.IsSegmented, (byte)1)
|
|
.Where(x => idList.Contains(x.GoodsId))
|
|
.ExecuteAffrowsAsync();
|
|
}
|
|
}
|
|
}
|