ideship/admin
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master'

Browse Source
This commit is contained in:
xiaolipro 2026-07-22 14:19:56 +08:00
commit ed0bda4b7f
26 changed files with 665 additions and 712 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
db.sql
View File

@ -91,8 +91,6 @@ DROP TABLE IF EXISTS `t_order_weight`;
DROP TABLE IF EXISTS `t_order_declaration`;
DROP TABLE IF EXISTS `t_order_recipient`;
DROP TABLE IF EXISTS `t_order_sender`;
DROP TABLE IF EXISTS `t_order_task_fail`;
DROP TABLE IF EXISTS `t_order_task`;
DROP TABLE IF EXISTS `t_order_batch_job_item`;
DROP TABLE IF EXISTS `t_order_batch_job`;
DROP TABLE IF EXISTS `t_order_template`;
@ -224,42 +222,6 @@ CREATE TABLE `t_order_template` (
KEY `idx_ord_tpl_customer_type` (`CustomerId`, `Type`)
) COMMENT='下单模板表';
CREATE TABLE `t_order_task` (
`Id` INT NOT NULL AUTO_INCREMENT COMMENT '主键，自增，步长 1',
`TaskName` VARCHAR(128) NOT NULL COMMENT '任务名称',
`Quantity` INT NULL COMMENT '数量',
`SuccessCount` INT NULL COMMENT '成功数',
`FailCount` INT NULL COMMENT '失败数',
`Status` VARCHAR(16) NOT NULL COMMENT '任务状态：processing/completed/pending',
`TemplateId` INT NULL COMMENT '模板ID',
`CustomerId` INT NULL COMMENT '客户ID',
`IsDeleted` TINYINT NOT NULL DEFAULT 0 COMMENT '软删除：0-否，1-是',
`ConcurrencyStamp` CHAR(36) NOT NULL DEFAULT (UUID()) COMMENT '并发戳',
`CreateUserId` INT NOT NULL DEFAULT 0 COMMENT '创建人',
`CreateTime` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`UpdateUserId` INT NULL COMMENT '更新人',
`UpdateTime` DATETIME NULL COMMENT '更新时间',
PRIMARY KEY (`Id`),
KEY `idx_ord_task_customer_status_ctime` (`CustomerId`, `Status`, `CreateTime`),
KEY `idx_ord_task_template` (`TemplateId`)
) COMMENT='下单任务表';
CREATE TABLE `t_order_task_fail` (
`Id` INT NOT NULL AUTO_INCREMENT COMMENT '主键，自增，步长 1',
`TaskId` INT NOT NULL COMMENT '任务ID',
`RowNumber` INT NOT NULL COMMENT 'Excel行号',
`Reason` VARCHAR(512) NOT NULL COMMENT '失败原因',
`RawDataJson` JSON NULL COMMENT '原始行数据',
`IsDeleted` TINYINT NOT NULL DEFAULT 0 COMMENT '软删除：0-否，1-是',
`CreateUserId` INT NOT NULL DEFAULT 0 COMMENT '创建人',
`CreateTime` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`UpdateUserId` INT NULL COMMENT '更新人',
`UpdateTime` DATETIME NULL COMMENT '更新时间',
PRIMARY KEY (`Id`),
KEY `idx_ord_task_fail_task` (`TaskId`),
KEY `idx_ord_task_fail_row` (`TaskId`, `RowNumber`)
) COMMENT='下单任务失败明细表';
CREATE TABLE `t_order_batch_job` (
`Id` INT NOT NULL AUTO_INCREMENT COMMENT '主键，自增，步长 1',
`Action` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '批量操作类型：pay/pre-declare/intercept/parcel-create 等',
@ -288,6 +250,7 @@ CREATE TABLE `t_order_batch_job_item` (
`JobId` INT NOT NULL COMMENT '批量任务ID',
`OrderId` INT NOT NULL COMMENT '订单ID',
`ServiceNo` VARCHAR(64) NULL COMMENT '运单号',
`ClientNo` VARCHAR(64) NULL COMMENT '客户单号',
`Action` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '子项操作类型，通常与任务 Action 一致',
`Status` VARCHAR(32) NOT NULL DEFAULT 'pending' COMMENT '子项状态：pending/running/success/failed/cancelled',
`MessageId` VARCHAR(128) NULL COMMENT 'RabbitMQ 消息ID，用于幂等与去重',

9
design.md
View File

@ -65,13 +65,13 @@
- 我的订单：列表、详情、支付、预报、拦截、取消拦截、删除、打印、导出
- 我要查价：按目的国、运输方式、重量、货物类型返回价格方案
- 我要下单：模板下载、Excel 导入、任务跟踪
- 我要下单：模板下载、Excel 导入、批量任务跟踪
核心表：
- `t_order`
- `t_order_template`
- `t_order_task`
- `t_order_batch_job` / `t_order_batch_job_item`
- `t_order_sender`
- `t_order_recipient`
- `t_order_declaration`
@ -403,10 +403,9 @@ OMS 规格文档展示的是 kebab-case 路径示例，例如：
建议至少拆分为以下应用服务：
- `OrderService`
- `OrderService`（含批量任务 `OrderService.BatchJob`）
- `OrderPriceCheckService`
- `OrderTemplateService`
- `OrderTaskService`
- `SalesProductService`
- `SalesProductPriceService`
- `SalesProductAllocationService`
@ -432,7 +431,7 @@ OMS 规格文档展示的是 kebab-case 路径示例，例如：
以下场景必须显式事务：
- 创建订单时同时写主表、地址、申报、件重尺、费用
- 批量导入任务落库与任务结果回写
- 批量任务落库与子项结果回写（`t_order_batch_job` / `t_order_batch_job_item`）
- 保存价格项时同时覆盖因素与价格明细
- 发布与回滚时版本快照回写
- 充值审核通过并更新余额

1
scripts/order_batch_job_tables.sql
View File

@ -38,6 +38,7 @@ CREATE TABLE `t_order_batch_job_item` (
`JobId` INT NOT NULL COMMENT '批量任务ID',
`OrderId` INT NOT NULL COMMENT '订单ID',
`ServiceNo` VARCHAR(64) NULL COMMENT '运单号',
`ClientNo` VARCHAR(64) NULL COMMENT '客户单号',
`Action` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '子项操作类型，通常与任务 Action 一致',
`Status` VARCHAR(32) NOT NULL DEFAULT 'pending' COMMENT '子项状态：pending/running/success/failed/cancelled',
`MessageId` VARCHAR(128) NULL COMMENT 'RabbitMQ 消息ID，用于幂等与去重',

6
sql/migrate_drop_order_task_merge_batch.sql Normal file
View File

@ -0,0 +1,6 @@
-- 合并下单任务表到批量任务：删除 t_order_task / t_order_task_fail
-- 历史导入任务数据不再保留；Excel 导入统一走 t_order_batch_job（action=import-row）
-- 2026-07-22：触发重新发布，确保 API 不再访问已删表
DROP TABLE IF EXISTS `t_order_task_fail`;
DROP TABLE IF EXISTS `t_order_task`;

3
sql/migrate_order_batch_job_item_client_no.sql Normal file
View File

@ -0,0 +1,3 @@
-- 批量任务明细：增加客户单号字段（Excel 导入/小包创建反馈用）
ALTER TABLE `t_order_batch_job_item`
ADD COLUMN `ClientNo` VARCHAR(64) NULL COMMENT '客户单号' AFTER `ServiceNo`;

20
src/Oms.Application/Dtos/Orders/OrderBatchJobDtos.cs
View File

@ -29,6 +29,16 @@ public class OrderBatchJobOutput
public int FailCount { get; set; }
public int PendingCount { get; set; }
public int RunningCount { get; set; }
public int CancelledCount { get; set; }
public int? TemplateId { get; set; }
public string? TaskName { get; set; }
public DateTime? StartedAt { get; set; }
public DateTime? FinishedAt { get; set; }
@ -44,12 +54,22 @@ public class OrderBatchJobItemOutput
public string? ServiceNo { get; set; }
public string? ClientNo { get; set; }
public int? RowNumber { get; set; }
public string Action { get; set; } = string.Empty;
public string Status { get; set; } = string.Empty;
public string? FailReason { get; set; }
public string? ResultJson { get; set; }
public string? ResultMessage { get; set; }
public int RetryCount { get; set; }
public DateTime? StartedAt { get; set; }
public DateTime? FinishedAt { get; set; }

6
src/Oms.Application/Dtos/Orders/OrderBatchPayloadDtos.cs
View File

@ -9,6 +9,8 @@ public class OrderBatchPayloadItemInput
public string? ServiceNo { get; set; }
public string? ClientNo { get; set; }
public string PayloadJson { get; set; } = string.Empty;
public string? ItemKey { get; set; }
@ -32,8 +34,6 @@ public class OrderBatchPayloadJobCreateInput
/// </summary>
public class OrderBatchImportRowPayload
{
public int TaskId { get; set; }
public int RowNumber { get; set; }
public int CustomerId { get; set; }
@ -48,7 +48,7 @@ public class OrderBatchImportRowPayload
/// </summary>
public class OrderBatchImportJobPayload
{
public int TaskId { get; set; }
public string TaskName { get; set; } = string.Empty;
public int TemplateId { get; set; }

57
src/Oms.Application/Dtos/Orders/OrderTaskDetailOutput.cs
View File

@ -1,57 +0,0 @@
namespace Oms.Application.Dtos.Orders;
/// <summary>
/// 订单任务详情出参。
/// </summary>
public class OrderTaskDetailOutput
{
/// <summary>
/// 主键。
/// </summary>
public int Id { get; set; }
/// <summary>
/// 任务名称。
/// </summary>
public string TaskName { get; set; } = string.Empty;
/// <summary>
/// 数量。
/// </summary>
public int? Quantity { get; set; }
/// <summary>
/// 成功数量。
/// </summary>
public int? SuccessCount { get; set; }
/// <summary>
/// 失败数量。
/// </summary>
public int? FailCount { get; set; }
/// <summary>
/// 状态。
/// </summary>
public string Status { get; set; } = string.Empty;
/// <summary>
/// 模板主键。
/// </summary>
public int? TemplateId { get; set; }
/// <summary>
/// 创建时间。
/// </summary>
public DateTime CreateTime { get; set; }
/// <summary>
/// 更新时间。
/// </summary>
public DateTime? UpdateTime { get; set; }
/// <summary>
/// 失败明细集合。
/// </summary>
public List<OrderTaskFailOutput> FailList { get; set; } = [];
}

32
src/Oms.Application/Dtos/Orders/OrderTaskFailOutput.cs
View File

@ -1,32 +0,0 @@
namespace Oms.Application.Dtos.Orders;
/// <summary>
/// 订单任务失败项出参。
/// </summary>
public class OrderTaskFailOutput
{
/// <summary>
/// 主键。
/// </summary>
public int Id { get; set; }
/// <summary>
/// 任务主键。
/// </summary>
public int TaskId { get; set; }
/// <summary>
/// 行号。
/// </summary>
public int RowNumber { get; set; }
/// <summary>
/// 失败原因。
/// </summary>
public string Reason { get; set; } = string.Empty;
/// <summary>
/// 原始数据 JSON。
/// </summary>
public object? RawDataJson { get; set; }
}

27
src/Oms.Application/Dtos/Orders/OrderTaskListInput.cs
View File

@ -1,27 +0,0 @@
namespace Oms.Application.Dtos.Orders;
/// <summary>
/// 订单任务列表查询入参。
/// </summary>
public class OrderTaskListInput
{
/// <summary>
/// 页码，从 1 开始。
/// </summary>
public int Page { get; set; } = 1;
/// <summary>
/// 每页条数。
/// </summary>
public int PageSize { get; set; } = 20;
/// <summary>
/// 状态。
/// </summary>
public string? Status { get; set; }
/// <summary>
/// 关键词。
/// </summary>
public string? Keyword { get; set; }
}

42
src/Oms.Application/Dtos/Orders/OrderTaskOutput.cs
View File

@ -1,42 +0,0 @@
namespace Oms.Application.Dtos.Orders;
/// <summary>
/// 订单任务出参。
/// </summary>
public class OrderTaskOutput
{
/// <summary>
/// 主键。
/// </summary>
public int Id { get; set; }
/// <summary>
/// 任务名称。
/// </summary>
public string TaskName { get; set; } = string.Empty;
/// <summary>
/// 数量。
/// </summary>
public int? Quantity { get; set; }
/// <summary>
/// 状态。
/// </summary>
public string Status { get; set; } = string.Empty;
/// <summary>
/// 模板主键。
/// </summary>
public int? TemplateId { get; set; }
/// <summary>
/// 创建时间。
/// </summary>
public DateTime CreateTime { get; set; }
/// <summary>
/// 更新时间。
/// </summary>
public DateTime? UpdateTime { get; set; }
}

13
src/Oms.Application/Dtos/Orders/OrderUploadOutput.cs
View File

@ -1,22 +1,17 @@
namespace Oms.Application.Dtos.Orders;
/// <summary>
/// 订单结果出参。
/// 订单 Excel 导入出参（统一返回批量任务 ID）。
/// </summary>
public class OrderUploadOutput
{
/// <summary>
/// 任务主键。
/// 批量任务主键。
/// </summary>
public int TaskId { get; set; }
public int JobId { get; set; }
/// <summary>
/// 关联批量 Job（异步导入时返回）。
/// </summary>
public int? JobId { get; set; }
/// <summary>
/// 是否异步处理中。
/// 是否异步处理中（RabbitMQ 队列）。
/// </summary>
public bool Async { get; set; }
}

5
src/Oms.Application/Messaging/OrderBatchRabbitMqOptions.cs
View File

@ -11,4 +11,9 @@ public class OrderBatchRabbitMqOptions
/// 是否启用订单批量消息发布与处理。
/// </summary>
public bool Enabled { get; set; }
/// <summary>
/// 单个子项系统异常的最大重试次数（耗尽后标记失败并 Ack，避免子项长期停留在执行中）。
/// </summary>
public int MaxSystemRetries { get; set; } = 5;
}

181
src/Oms.Application/Services/Orders/Batch/OrderBatchJobCoordinator.cs
View File

@ -1,6 +1,7 @@
using Fake.Application;
using Fake.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using Newtonsoft.Json;
using Oms.Application.Dtos.Orders;
using Oms.Application.Dtos.Shared;
@ -23,6 +24,8 @@ public sealed class OrderBatchJobCoordinator(
IOrderRepository orderRepository,
ICountryRepository countryRepository,
IOrderBatchJobPublisher publisher,
OrderBatchJobProcessor batchJobProcessor,
IOptions<OrderBatchRabbitMqOptions> batchOptions,
ILogger<OrderBatchJobCoordinator> logger) : IScopedDependency
{
private const int DefaultMaxBatchSize = 200;
@ -113,6 +116,7 @@ public sealed class OrderBatchJobCoordinator(
JobId = job.Id,
OrderId = orderId,
ServiceNo = order?.ServiceNo,
ClientNo = order?.ClientNo,
Action = action,
Status = OrderBatchJobItemStatusCodes.Pending,
MessageId = $"{job.Id}:{orderId}:{action}",
@ -167,6 +171,83 @@ public sealed class OrderBatchJobCoordinator(
throw new BusinessException("订单批量队列未启用，请检查 RabbitMQ 配置");
}
var (job, items, itemsInput, created) = await InsertPayloadJobAsync(input, operatorUserId, cancellationToken);
if (!created)
{
return await MapJobOutputAsync(job, cancellationToken);
}
for (var index = 0; index < items.Count; index++)
{
await publisher.PublishAsync(
BuildPayloadMessage(job, items[index], itemsInput[index], input.CustomerId, operatorUserId),
cancellationToken);
}
logger.LogInformation(
"Created payload batch job {JobId} action={Action} total={Total}",
job.Id,
job.Action,
job.Total);
return await MapJobOutputAsync(job, cancellationToken);
}
/// <summary>
/// 同步创建并逐条处理 Payload 批量任务（MQ 未启用时的 Excel 导入等）。
/// </summary>
public async Task<OrderBatchJobOutput> CreatePayloadJobAndProcessSyncAsync(
OrderBatchPayloadJobCreateInput input,
int operatorUserId,
CancellationToken cancellationToken = default)
{
var (job, items, itemsInput, created) = await InsertPayloadJobAsync(input, operatorUserId, cancellationToken);
if (!created)
{
return await MapJobOutputAsync(job, cancellationToken);
}
var maxAttempts = Math.Max(1, batchOptions.Value.MaxSystemRetries);
for (var index = 0; index < items.Count; index++)
{
var item = items[index];
var message = BuildPayloadMessage(job, item, itemsInput[index], input.CustomerId, operatorUserId);
for (var attempt = 0; attempt < maxAttempts; attempt++)
{
try
{
await batchJobProcessor.ProcessAsync(message, cancellationToken);
break;
}
catch (InvalidOperationException) when (attempt < maxAttempts - 1)
{
var refreshed = await batchJobRepository.GetItemByIdAsync(item.Id, cancellationToken);
if (refreshed?.Status is OrderBatchJobItemStatusCodes.Failed or OrderBatchJobItemStatusCodes.Success)
{
break;
}
}
}
}
logger.LogInformation(
"Processed payload batch job synchronously {JobId} action={Action} total={Total}",
job.Id,
job.Action,
job.Total);
var refreshedJob = await batchJobRepository.GetByIdAsync(job.Id, cancellationToken)
?? throw new BusinessException("批量任务不存在");
return await MapJobOutputAsync(refreshedJob, cancellationToken);
}
private async Task<(OrderBatchJob Job, List<OrderBatchJobItem> Items, List<OrderBatchPayloadItemInput> ItemsInput, bool Created)> InsertPayloadJobAsync(
OrderBatchPayloadJobCreateInput input,
int operatorUserId,
CancellationToken cancellationToken)
{
var action = input.Action?.Trim() ?? string.Empty;
if (string.IsNullOrWhiteSpace(action))
{
@ -191,7 +272,8 @@ public sealed class OrderBatchJobCoordinator(
var existing = await batchJobRepository.GetByRequestIdAsync(input.RequestId.Trim(), cancellationToken);
if (existing != null)
{
return await MapJobOutputAsync(existing, cancellationToken);
var existingItems = await batchJobRepository.GetItemsByJobIdAsync(existing.Id, cancellationToken);
return (existing, existingItems, itemsInput, false);
}
}
@ -232,6 +314,7 @@ public sealed class OrderBatchJobCoordinator(
JobId = job.Id,
OrderId = source.OrderId,
ServiceNo = source.ServiceNo,
ClientNo = source.ClientNo,
Action = action,
Status = OrderBatchJobItemStatusCodes.Pending,
MessageId = itemKey,
@ -244,33 +327,43 @@ public sealed class OrderBatchJobCoordinator(
}
await batchJobRepository.InsertItemsAsync(items, cancellationToken);
return (job, items, itemsInput, true);
}
for (var index = 0; index < items.Count; index++)
private static OrderBatchJobMessage BuildPayloadMessage(
OrderBatchJob job,
OrderBatchJobItem item,
OrderBatchPayloadItemInput source,
int? customerId,
int operatorUserId) =>
new()
{
var item = items[index];
var source = itemsInput[index];
var message = new OrderBatchJobMessage
{
JobId = job.Id,
JobItemId = item.Id,
Action = action,
OrderId = item.OrderId,
CustomerId = input.CustomerId,
PayloadJson = source.PayloadJson,
OperatorUserId = operatorUserId,
MessageId = item.MessageId ?? $"{job.Id}:{index + 1}:{action}",
};
JobId = job.Id,
JobItemId = item.Id,
Action = job.Action,
OrderId = item.OrderId,
CustomerId = customerId,
PayloadJson = source.PayloadJson,
OperatorUserId = operatorUserId,
MessageId = item.MessageId ?? $"{job.Id}:{item.Id}:{job.Action}",
};
await publisher.PublishAsync(message, cancellationToken);
private static void ApplyImportMetadata(OrderBatchJobOutput output, OrderBatchJob job)
{
if (!string.Equals(job.Action, OrderBatchActionCodes.ImportRow, StringComparison.OrdinalIgnoreCase)
|| string.IsNullOrWhiteSpace(job.PayloadJson))
{
return;
}
logger.LogInformation(
"Created payload batch job {JobId} action={Action} total={Total}",
job.Id,
action,
job.Total);
var meta = JsonConvert.DeserializeObject<OrderBatchImportJobPayload>(job.PayloadJson);
if (meta == null)
{
return;
}
return await MapJobOutputAsync(job, cancellationToken);
output.TemplateId = meta.TemplateId > 0 ? meta.TemplateId : null;
output.TaskName = string.IsNullOrWhiteSpace(meta.TaskName) ? null : meta.TaskName.Trim();
}
public async Task PublishTrackUpsertAsync(SmallBagTrackUpsertRequest request, int orderId, int? customerId)
@ -309,18 +402,14 @@ public sealed class OrderBatchJobCoordinator(
public async Task<List<OrderBatchJobItemOutput>> GetJobItemsAsync(int jobId, CancellationToken cancellationToken = default)
{
var job = await batchJobRepository.GetByIdAsync(jobId, cancellationToken);
var jobAction = job?.Action ?? string.Empty;
var items = await batchJobRepository.GetItemsByJobIdAsync(jobId, cancellationToken);
return items.Select(x => new OrderBatchJobItemOutput
{
Id = x.Id,
OrderId = x.OrderId,
ServiceNo = x.ServiceNo,
Status = x.Status,
FailReason = x.FailReason,
ResultJson = x.ResultJson,
StartedAt = x.StartedAt,
FinishedAt = x.FinishedAt,
}).ToList();
return items
.Select(x => OrderBatchJobItemFeedback.MapToOutput(x, jobAction))
.OrderBy(x => x.RowNumber ?? int.MaxValue)
.ThenBy(x => x.Id)
.ToList();
}
public async Task<OrderBatchJobManageOutput> RetryFailedAsync(
@ -407,6 +496,14 @@ public sealed class OrderBatchJobCoordinator(
item.FinishedAt = now;
item.UpdateTime = now;
item.UpdateUserId = operatorUserId;
item.ResultJson = Newtonsoft.Json.JsonConvert.SerializeObject(new
{
success = false,
action = item.Action,
reason = item.FailReason,
clientNo = item.ClientNo,
orderId = item.OrderId > 0 ? item.OrderId : (int?)null,
});
await batchJobRepository.UpdateItemAsync(item, cancellationToken);
}
@ -440,6 +537,9 @@ public sealed class OrderBatchJobCoordinator(
var items = await batchJobRepository.GetItemsByJobIdAsync(job.Id, cancellationToken);
var successCount = items.Count(x => x.Status == OrderBatchJobItemStatusCodes.Success);
var failCount = items.Count(x => x.Status == OrderBatchJobItemStatusCodes.Failed);
var pendingCount = items.Count(x => x.Status == OrderBatchJobItemStatusCodes.Pending);
var runningCount = items.Count(x => x.Status == OrderBatchJobItemStatusCodes.Running);
var cancelledCount = items.Count(x => x.Status == OrderBatchJobItemStatusCodes.Cancelled);
if (job.SuccessCount != successCount || job.FailCount != failCount)
{
@ -452,14 +552,16 @@ public sealed class OrderBatchJobCoordinator(
or OrderBatchJobItemStatusCodes.Failed
or OrderBatchJobItemStatusCodes.Cancelled))
{
job.Status = failCount > 0 && successCount == 0
? OrderBatchJobStatusCodes.Failed
: OrderBatchJobStatusCodes.Completed;
job.Status = cancelledCount == items.Count
? OrderBatchJobStatusCodes.Cancelled
: failCount > 0 && successCount == 0
? OrderBatchJobStatusCodes.Failed
: OrderBatchJobStatusCodes.Completed;
job.FinishedAt ??= DateTime.Now;
await batchJobRepository.UpdateJobAsync(job, cancellationToken);
}
return new OrderBatchJobOutput
var output = new OrderBatchJobOutput
{
JobId = job.Id,
Action = job.Action,
@ -467,10 +569,15 @@ public sealed class OrderBatchJobCoordinator(
Total = job.Total,
SuccessCount = job.SuccessCount,
FailCount = job.FailCount,
PendingCount = pendingCount,
RunningCount = runningCount,
CancelledCount = cancelledCount,
StartedAt = job.StartedAt,
FinishedAt = job.FinishedAt,
CreateTime = job.CreateTime,
};
ApplyImportMetadata(output, job);
return output;
}
public async Task<OmsPageResult<OrderBatchJobOutput>> GetJobListAsync(

332
src/Oms.Application/Services/Orders/Batch/OrderBatchJobItemFeedback.cs Normal file
View File

@ -0,0 +1,332 @@
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
using Oms.Application.Dtos.Orders;
using Oms.Domain.OrdersAggregate.Consts;
using Oms.Domain.OrdersAggregate.Entities;
using Oms.Domain.OrdersAggregate.Repositories;
namespace Oms.Application.Services.Orders.Batch;
/// <summary>
/// 批量任务子项反馈：统一补全标识、写入结果快照、映射 API 输出。
/// </summary>
public static class OrderBatchJobItemFeedback
{
private static readonly string[] ClientNoRowKeys =
[
"clientNo", "ClientNo", "customerOrderNo", "CustomerOrderNo",
"客户单号", "客户订单号", "外部单号",
];
public static async Task EnrichIdentifiersAsync(
OrderBatchJobItem item,
string action,
IOrderRepository orderRepository,
CancellationToken cancellationToken = default)
{
EnrichFromPayload(item, action);
if (item.OrderId <= 0)
{
return;
}
if (!string.IsNullOrWhiteSpace(item.ClientNo) && !string.IsNullOrWhiteSpace(item.ServiceNo))
{
return;
}
var order = await orderRepository.FirstOrDefaultAsync(x => x.Id == item.OrderId, cancellationToken);
if (order == null)
{
return;
}
item.ClientNo ??= NullIfWhiteSpace(order.ClientNo);
item.ServiceNo ??= NullIfWhiteSpace(order.ServiceNo);
}
public static void EnrichFromPayload(OrderBatchJobItem item, string action)
{
if (string.IsNullOrWhiteSpace(item.PayloadJson))
{
return;
}
if (string.Equals(action, OrderBatchActionCodes.ImportRow, StringComparison.OrdinalIgnoreCase))
{
var payload = JsonConvert.DeserializeObject<OrderBatchImportRowPayload>(item.PayloadJson);
if (payload == null)
{
return;
}
if (string.IsNullOrWhiteSpace(item.ClientNo))
{
item.ClientNo = TryExtractClientNoFromRowJson(payload.RowDataJson);
}
return;
}
if (string.Equals(action, OrderBatchActionCodes.ParcelCreate, StringComparison.OrdinalIgnoreCase))
{
var input = JsonConvert.DeserializeObject<OrderParcelCreateInput>(item.PayloadJson);
if (input == null)
{
return;
}
item.ClientNo ??= NullIfWhiteSpace(input.ClientNo);
}
}
public static int? TryGetImportRowNumber(OrderBatchJobItem item, string action)
{
if (!string.Equals(action, OrderBatchActionCodes.ImportRow, StringComparison.OrdinalIgnoreCase)
|| string.IsNullOrWhiteSpace(item.PayloadJson))
{
return null;
}
var payload = JsonConvert.DeserializeObject<OrderBatchImportRowPayload>(item.PayloadJson);
return payload is { RowNumber: > 0 } ? payload.RowNumber : null;
}
public static void ApplySuccess(
OrderBatchJobItem item,
OrderBatchActionExecuteResult result,
string action)
{
item.Status = OrderBatchJobItemStatusCodes.Success;
item.FailReason = null;
if (result.CreatedOrderId is > 0)
{
item.OrderId = result.CreatedOrderId.Value;
}
if (!string.IsNullOrWhiteSpace(result.ClientNo))
{
item.ClientNo = result.ClientNo.Trim();
}
if (!string.IsNullOrWhiteSpace(result.ServiceNo))
{
item.ServiceNo = result.ServiceNo.Trim();
}
item.ResultJson = BuildSuccessResultJson(item, result, action);
}
public static void ApplyBusinessFailure(
OrderBatchJobItem item,
OrderBatchActionExecuteResult result,
string action)
{
item.Status = OrderBatchJobItemStatusCodes.Failed;
item.FailReason = string.IsNullOrWhiteSpace(result.FailReason) ? "操作失败" : result.FailReason.Trim();
if (result.CreatedOrderId is > 0)
{
item.OrderId = result.CreatedOrderId.Value;
}
item.ResultJson = BuildFailureResultJson(item, action, item.FailReason, retryCount: item.RetryCount);
}
public static void ApplySystemRetry(
OrderBatchJobItem item,
OrderBatchActionExecuteResult result,
string action,
int maxRetries)
{
item.RetryCount += 1;
var reason = string.IsNullOrWhiteSpace(result.FailReason) ? "系统异常" : result.FailReason.Trim();
item.FailReason = $"系统异常（第 {item.RetryCount}/{maxRetries} 次重试）：{reason}";
if (item.RetryCount >= maxRetries)
{
ApplySystemFailure(item, action, $"系统重试已达上限：{reason}");
return;
}
item.Status = OrderBatchJobItemStatusCodes.Pending;
item.ResultJson = BuildFailureResultJson(item, action, item.FailReason, retryCount: item.RetryCount, retrying: true);
}
public static void ApplySystemFailure(OrderBatchJobItem item, string action, string reason)
{
item.Status = OrderBatchJobItemStatusCodes.Failed;
item.FailReason = reason;
item.ResultJson = BuildFailureResultJson(item, action, reason, retryCount: item.RetryCount);
}
public static OrderBatchJobItemOutput MapToOutput(OrderBatchJobItem item, string jobAction)
{
EnrichFromPayload(item, jobAction);
return new OrderBatchJobItemOutput
{
Id = item.Id,
OrderId = item.OrderId,
ServiceNo = item.ServiceNo,
ClientNo = item.ClientNo,
RowNumber = TryGetImportRowNumber(item, jobAction),
Action = string.IsNullOrWhiteSpace(item.Action) ? jobAction : item.Action,
Status = item.Status,
FailReason = item.FailReason,
ResultJson = item.ResultJson,
ResultMessage = BuildResultMessage(item),
RetryCount = item.RetryCount,
StartedAt = item.StartedAt,
FinishedAt = item.FinishedAt,
};
}
public static string? BuildResultMessage(OrderBatchJobItem item)
{
if (item.Status == OrderBatchJobItemStatusCodes.Success)
{
return TryReadResultField(item.ResultJson, "message") ?? "操作成功";
}
if (item.Status == OrderBatchJobItemStatusCodes.Failed)
{
return item.FailReason;
}
if (item.Status == OrderBatchJobItemStatusCodes.Cancelled)
{
return item.FailReason ?? "已取消";
}
if (item.Status == OrderBatchJobItemStatusCodes.Running)
{
return "处理中";
}
return null;
}
private static string BuildSuccessResultJson(
OrderBatchJobItem item,
OrderBatchActionExecuteResult result,
string action)
{
if (!string.IsNullOrWhiteSpace(result.ResultJson))
{
return result.ResultJson;
}
return JsonConvert.SerializeObject(new
{
success = true,
action,
message = ResolveSuccessMessage(action),
clientNo = item.ClientNo ?? result.ClientNo,
serviceNo = item.ServiceNo ?? result.ServiceNo,
orderId = item.OrderId > 0 ? item.OrderId : result.CreatedOrderId,
trackNo = result.TrackNo,
labelUrl = result.LabelUrl,
rowNumber = TryGetImportRowNumber(item, action),
});
}
private static string BuildFailureResultJson(
OrderBatchJobItem item,
string action,
string reason,
int retryCount,
bool retrying = false)
{
return JsonConvert.SerializeObject(new
{
success = false,
action,
retrying,
retryCount,
reason,
clientNo = item.ClientNo,
serviceNo = item.ServiceNo,
orderId = item.OrderId > 0 ? item.OrderId : (int?)null,
rowNumber = TryGetImportRowNumber(item, action),
});
}
private static string ResolveSuccessMessage(string action) =>
action switch
{
OrderBatchActionCodes.ImportRow => "导入成功",
OrderBatchActionCodes.ParcelCreate => "创建成功",
OrderBatchActionCodes.Pay => "支付成功",
OrderBatchActionCodes.PreDeclare => "预报成功",
OrderBatchActionCodes.Intercept => "拦截成功",
OrderBatchActionCodes.CancelIntercept => "取消拦截成功",
OrderBatchActionCodes.Delete => "删除成功",
OrderBatchActionCodes.Print => "打印成功",
OrderBatchActionCodes.GenerateLabelPdf => "面单生成成功",
OrderBatchActionCodes.MarkProblematic => "问题件创建成功",
OrderBatchActionCodes.MarkClaimed => "理赔件创建成功",
OrderBatchActionCodes.HandleStatusUpdate => "处理状态更新成功",
OrderBatchActionCodes.DownstreamForwardCreateOne => "下游预报成功",
_ => "操作成功",
};
private static string? TryExtractClientNoFromRowJson(string? rowDataJson)
{
if (string.IsNullOrWhiteSpace(rowDataJson))
{
return null;
}
try
{
if (JsonConvert.DeserializeObject(rowDataJson) is not JObject obj)
{
return null;
}
foreach (var key in ClientNoRowKeys)
{
if (!obj.TryGetValue(key, StringComparison.OrdinalIgnoreCase, out var token))
{
continue;
}
var value = token?.ToString()?.Trim();
if (!string.IsNullOrWhiteSpace(value))
{
return value;
}
}
}
catch
{
// ignore malformed row snapshot
}
return null;
}
private static string? TryReadResultField(string? resultJson, string field)
{
if (string.IsNullOrWhiteSpace(resultJson))
{
return null;
}
try
{
var obj = JObject.Parse(resultJson);
return obj[field]?.ToString()?.Trim();
}
catch
{
return null;
}
}
private static string? NullIfWhiteSpace(string? value) =>
string.IsNullOrWhiteSpace(value) ? null : value.Trim();
}

150
src/Oms.Application/Services/Orders/Batch/OrderBatchJobProcessor.cs
View File

@ -1,6 +1,6 @@
using Fake.DependencyInjection;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;
using Microsoft.Extensions.Options;
using Oms.Application.Dtos.Orders;
using Oms.Application.Messaging;
using Oms.Domain.OrdersAggregate.Consts;
@ -11,18 +11,20 @@ namespace Oms.Application.Services.Orders.Batch;
/// <summary>
/// 消费 RabbitMQ 消息后更新任务状态并执行业务。
/// 正常返回由 Fake EventBus Ack；抛异常则框架重试后进 DLX。
/// 正常返回由 Fake EventBus Ack；系统失败可重试，耗尽后标记失败避免子项长期停留在执行中。
/// </summary>
public sealed class OrderBatchJobProcessor(
IOrderBatchJobRepository batchJobRepository,
IOrderTaskRepository orderTaskRepository,
IOrderRepository orderRepository,
IOrderBatchActionExecutor actionExecutor,
IOptions<OrderBatchRabbitMqOptions> batchOptions,
ILogger<OrderBatchJobProcessor> logger) : IScopedDependency
{
public async Task ProcessAsync(
OrderBatchJobMessage message,
CancellationToken cancellationToken = default)
{
var maxSystemRetries = Math.Max(1, batchOptions.Value.MaxSystemRetries);
OrderBatchJobItem? item = null;
if (message.JobItemId > 0)
{
@ -54,7 +56,6 @@ public sealed class OrderBatchJobProcessor(
if (message.JobItemId <= 0)
{
// fire-and-forget（如轨迹同步）：失败只记日志，不抛、不进 DLX
if (!result.Success)
{
logger.LogWarning(
@ -75,25 +76,12 @@ public sealed class OrderBatchJobProcessor(
item.FinishedAt = DateTime.Now;
item.UpdateTime = DateTime.Now;
var action = string.IsNullOrWhiteSpace(item.Action) ? message.Action : item.Action;
if (result.Success)
{
item.Status = OrderBatchJobItemStatusCodes.Success;
item.FailReason = null;
if (result.CreatedOrderId is > 0)
{
item.OrderId = result.CreatedOrderId.Value;
}
if (!string.IsNullOrWhiteSpace(result.ServiceNo))
{
item.ServiceNo = result.ServiceNo;
}
if (!string.IsNullOrWhiteSpace(result.ResultJson))
{
item.ResultJson = result.ResultJson;
}
OrderBatchJobItemFeedback.ApplySuccess(item, result, action);
await OrderBatchJobItemFeedback.EnrichIdentifiersAsync(item, action, orderRepository, cancellationToken);
await batchJobRepository.UpdateItemAsync(item, cancellationToken);
await RefreshJobSummaryAsync(item.JobId, cancellationToken);
return;
@ -101,20 +89,36 @@ public sealed class OrderBatchJobProcessor(
if (result.IsBusinessFailure)
{
item.Status = OrderBatchJobItemStatusCodes.Failed;
item.FailReason = result.FailReason;
await batchJobRepository.UpdateItemAsync(item, cancellationToken);
if (message.Action == OrderBatchActionCodes.ImportRow)
if (result.CreatedOrderId is > 0)
{
await RecordImportTaskFailAsync(message, result.FailReason, cancellationToken);
item.OrderId = result.CreatedOrderId.Value;
}
await OrderBatchJobItemFeedback.EnrichIdentifiersAsync(item, action, orderRepository, cancellationToken);
OrderBatchJobItemFeedback.ApplyBusinessFailure(item, result, action);
await batchJobRepository.UpdateItemAsync(item, cancellationToken);
await RefreshJobSummaryAsync(item.JobId, cancellationToken);
return;
}
// 系统失败：抛出让 EventBus 重试，耗尽后进 DLX
throw new InvalidOperationException(result.FailReason ?? "批量任务系统失败");
await OrderBatchJobItemFeedback.EnrichIdentifiersAsync(item, action, orderRepository, cancellationToken);
OrderBatchJobItemFeedback.ApplySystemRetry(item, result, action, maxSystemRetries);
await batchJobRepository.UpdateItemAsync(item, cancellationToken);
if (item.Status == OrderBatchJobItemStatusCodes.Failed)
{
logger.LogError(
"Batch job item {JobItemId} action={Action} failed after {RetryCount} system retries: {Reason}",
item.Id,
action,
item.RetryCount,
item.FailReason);
await RefreshJobSummaryAsync(item.JobId, cancellationToken);
return;
}
throw new InvalidOperationException(item.FailReason ?? result.FailReason ?? "批量任务系统失败");
}
private async Task RefreshJobSummaryAsync(int jobId, CancellationToken cancellationToken)
@ -134,9 +138,18 @@ public sealed class OrderBatchJobProcessor(
or OrderBatchJobItemStatusCodes.Failed
or OrderBatchJobItemStatusCodes.Cancelled))
{
job.Status = job.FailCount > 0 && job.SuccessCount == 0
? OrderBatchJobStatusCodes.Failed
: OrderBatchJobStatusCodes.Completed;
var cancelledCount = items.Count(x => x.Status == OrderBatchJobItemStatusCodes.Cancelled);
if (cancelledCount == items.Count)
{
job.Status = OrderBatchJobStatusCodes.Cancelled;
}
else
{
job.Status = job.FailCount > 0 && job.SuccessCount == 0
? OrderBatchJobStatusCodes.Failed
: OrderBatchJobStatusCodes.Completed;
}
job.FinishedAt = DateTime.Now;
}
else
@ -145,80 +158,5 @@ public sealed class OrderBatchJobProcessor(
}
await batchJobRepository.UpdateJobAsync(job, cancellationToken);
await SyncOrderTaskFromJobAsync(job, cancellationToken);
}
private async Task SyncOrderTaskFromJobAsync(OrderBatchJob job, CancellationToken cancellationToken)
{
if (job.Action != OrderBatchActionCodes.ImportRow || string.IsNullOrWhiteSpace(job.PayloadJson))
{
return;
}
var meta = JsonConvert.DeserializeObject<OrderBatchImportJobPayload>(job.PayloadJson);
if (meta == null || meta.TaskId <= 0)
{
return;
}
var task = await orderTaskRepository.FirstOrDefaultAsync(x => x.Id == meta.TaskId, cancellationToken);
if (task == null)
{
return;
}
task.SuccessCount = job.SuccessCount;
task.FailCount = job.FailCount;
task.UpdateTime = DateTime.Now;
if (job.Status is OrderBatchJobStatusCodes.Completed or OrderBatchJobStatusCodes.Failed)
{
task.Status = OrderTaskStatusCodes.Completed;
}
else if (job.Status == OrderBatchJobStatusCodes.Running)
{
task.Status = OrderTaskStatusCodes.Processing;
}
await orderTaskRepository.UpdateAsync(task, cancellationToken: cancellationToken);
}
private async Task RecordImportTaskFailAsync(
OrderBatchJobMessage message,
string? reason,
CancellationToken cancellationToken)
{
if (string.IsNullOrWhiteSpace(message.PayloadJson))
{
return;
}
var payload = JsonConvert.DeserializeObject<OrderBatchImportRowPayload>(message.PayloadJson);
if (payload == null || payload.TaskId <= 0)
{
return;
}
object? rawData;
try
{
rawData = JsonConvert.DeserializeObject(payload.RowDataJson) ?? payload.RowDataJson;
}
catch
{
rawData = payload.RowDataJson;
}
await orderTaskRepository.InsertFailListAsync([
new OrderTaskFail
{
TaskId = payload.TaskId,
RowNumber = payload.RowNumber,
Reason = reason ?? "导入失败",
RawDataJson = JsonConvert.SerializeObject(rawData),
CreateUserId = message.OperatorUserId > 0 ? message.OperatorUserId : 0,
CreateTime = DateTime.Now,
}
], cancellationToken);
}
}

219
src/Oms.Application/Services/Orders/OrderService.Helpers.cs
View File

@ -344,21 +344,23 @@ public partial class OrderService
throw new BusinessException("上传文件不能为空");
}
var batchInput = await BuildImportBatchJobInputAsync(input);
var operatorUserId = CurrentUser.GetId<int>();
if (BatchJobPublisher.IsEnabled)
{
return await ImportOrdersViaQueueAsync(input);
var job = await BatchJobCoordinator.CreatePayloadJobAsync(batchInput, operatorUserId);
return new OrderUploadOutput { JobId = job.JobId, Async = true };
}
return await ImportOrdersSyncAsync(input);
var syncJob = await BatchJobCoordinator.CreatePayloadJobAndProcessSyncAsync(batchInput, operatorUserId);
return new OrderUploadOutput { JobId = syncJob.JobId, Async = false };
}
private async Task<OrderUploadOutput> ImportOrdersViaQueueAsync(OrderUploadInput input)
private async Task<OrderBatchPayloadJobCreateInput> BuildImportBatchJobInputAsync(OrderUploadInput input)
{
var template = await OrderTemplateRepository.FirstOrDefaultAsync(x => x.Id == input.TemplateId);
if (template == null)
{
throw new BusinessException("模板不存在");
}
var template = await OrderTemplateRepository.FirstOrDefaultAsync(x => x.Id == input.TemplateId)
?? throw new BusinessException("模板不存在");
var effectiveCustomer = await ResolveEffectiveCustomerAsync();
EnsureTemplateAccessible(template, effectiveCustomer.Id);
@ -378,20 +380,12 @@ public partial class OrderService
throw new BusinessException($"单次最多导入 {ParcelBatchMaxCount} 行");
}
var task = await OrderTaskRepository.InsertAsync(new OrderTask
{
TaskName = $"批量下单_{DateTime.Now:yyyyMMddHHmmss}",
Quantity = parsedRows.Count,
SuccessCount = 0,
FailCount = 0,
Status = OrderTaskStatusCodes.Processing,
TemplateId = template.Id,
CustomerId = effectiveCustomer.Id
});
var taskName = $"批量下单_{DateTime.Now:yyyyMMddHHmmss}";
var requestId = $"import-{effectiveCustomer.Id}-{template.Id}-{DateTime.Now:yyyyMMddHHmmssfff}";
var jobPayload = JsonConvert.SerializeObject(new OrderBatchImportJobPayload
{
TaskId = task.Id,
TaskName = taskName,
TemplateId = template.Id,
CustomerId = effectiveCustomer.Id,
});
@ -403,10 +397,10 @@ public partial class OrderService
{
OrderId = 0,
ServiceNo = parcelInput.ServiceNo,
ItemKey = $"import:{task.Id}:{row.RowNumber}",
ClientNo = parcelInput.ClientNo,
ItemKey = $"import:{requestId}:{row.RowNumber}",
PayloadJson = JsonConvert.SerializeObject(new OrderBatchImportRowPayload
{
TaskId = task.Id,
RowNumber = row.RowNumber,
CustomerId = effectiveCustomer.Id,
TemplateId = template.Id,
@ -415,171 +409,13 @@ public partial class OrderService
};
}).ToList();
var job = await BatchJobCoordinator.CreatePayloadJobAsync(
new OrderBatchPayloadJobCreateInput
{
Action = OrderBatchActionCodes.ImportRow,
CustomerId = effectiveCustomer.Id,
JobPayloadJson = jobPayload,
RequestId = $"import-task-{task.Id}",
Items = payloadItems,
},
CurrentUser.GetId<int>());
return new OrderUploadOutput
return new OrderBatchPayloadJobCreateInput
{
TaskId = task.Id,
JobId = job.JobId,
Async = true,
};
}
private async Task<OrderUploadOutput> ImportOrdersSyncAsync(OrderUploadInput input)
{
var template = await OrderTemplateRepository.FirstOrDefaultAsync(x => x.Id == input.TemplateId);
if (template == null)
{
throw new BusinessException("模板不存在");
}
var effectiveCustomer = await ResolveEffectiveCustomerAsync();
EnsureTemplateAccessible(template, effectiveCustomer.Id);
OrderTemplateImportMapping.EnsureTemplateReadyForImport(template);
var mapping = OrderTemplateImportMapping.Resolve(template);
await using var stream = input.File!.OpenReadStream();
var parsedRows = await OrderTemplateExcelReader.ReadDataRowsAsync(stream, mapping);
var task = await OrderTaskRepository.InsertAsync(new OrderTask
{
TaskName = $"批量下单_{DateTime.Now:yyyyMMddHHmmss}",
Quantity = 0,
SuccessCount = 0,
FailCount = 0,
Status = OrderTaskStatusCodes.Processing,
TemplateId = template.Id,
CustomerId = effectiveCustomer.Id
});
var quantity = 0;
var successCount = 0;
var failCount = 0;
var failList = new List<OrderTaskFail>();
foreach (var row in parsedRows)
{
quantity++;
try
{
var parcelInput = OrderTemplateRowMapper.MapToParcelCreateInput(
row.SourceValues,
mapping,
effectiveCustomer.Id);
await CreateParcelCoreAsync(parcelInput);
successCount++;
}
catch (Exception ex)
{
failCount++;
failList.Add(BuildTaskFail(task.Id, row.RowNumber, ex.Message, row.SourceValues));
}
}
if (parsedRows.Count == 0)
{
quantity = 1;
failCount = 1;
failList.Add(BuildTaskFail(task.Id, 1, "Excel 中未读取到可用数据", new Dictionary<string, object>()));
}
if (failList.Count > 0)
{
await OrderTaskRepository.InsertFailListAsync(failList);
}
task.Quantity = quantity;
task.SuccessCount = successCount;
task.FailCount = failCount;
task.Status = OrderTaskStatusCodes.Completed;
await OrderTaskRepository.UpdateAsync(task);
return new OrderUploadOutput { TaskId = task.Id };
}
private async Task<OmsPageResult<OrderTaskOutput>> GetTaskOutputsAsync(OrderTaskListInput input)
{
var page = OmsDocHelper.NormalizePage(input.Page);
var pageSize = OmsDocHelper.NormalizePageSize(input.PageSize);
var keyword = input.Keyword?.Trim();
var customerId = await OmsSessionHelper.ResolveListCustomerIdAsync(
CustomerRepository,
CurrentUser,
null);
var hasCustomerFilter = customerId.HasValue;
var customerIdValue = customerId.GetValueOrDefault();
var predicate = (Expression<Func<OrderTask, bool>>)(x =>
(string.IsNullOrWhiteSpace(input.Status) || x.Status == input.Status) &&
(string.IsNullOrWhiteSpace(keyword) || x.TaskName.Contains(keyword)) &&
(!hasCustomerFilter || x.CustomerId == customerIdValue));
var tasks = await OrderTaskRepository.GetPagedListAsync(
predicate,
page,
pageSize,
new Dictionary<string, bool> { [nameof(OrderTask.UpdateTime)] = false, [nameof(OrderTask.Id)] = false });
var total = await OrderTaskRepository.CountAsync(predicate);
return new OmsPageResult<OrderTaskOutput>
{
Items = tasks.Select(x => new OrderTaskOutput
{
Id = x.Id,
TaskName = x.TaskName,
Quantity = x.Quantity,
Status = x.Status,
TemplateId = x.TemplateId,
CreateTime = x.CreateTime,
UpdateTime = OmsDocHelper.NullIfDefault(x.UpdateTime)
}).ToList(),
Total = total
};
}
private async Task<OrderTaskDetailOutput> GetTaskDetailCoreAsync(int taskId)
{
var task = await OrderTaskRepository.FirstOrDefaultAsync(x => x.Id == taskId);
if (task == null)
{
throw new BusinessException("任务不存在");
}
var customerId = await OmsSessionHelper.ResolveListCustomerIdAsync(
CustomerRepository,
CurrentUser,
null);
if (customerId.HasValue && task.CustomerId != customerId.Value)
{
throw new BusinessException("任务不存在");
}
var failList = await OrderTaskRepository.GetFailListAsync(task.Id);
return new OrderTaskDetailOutput
{
Id = task.Id,
TaskName = task.TaskName,
Quantity = task.Quantity,
SuccessCount = task.SuccessCount,
FailCount = task.FailCount,
Status = task.Status,
TemplateId = task.TemplateId,
CreateTime = task.CreateTime,
UpdateTime = OmsDocHelper.NullIfDefault(task.UpdateTime),
FailList = failList.Select(x => new OrderTaskFailOutput
{
Id = x.Id,
TaskId = x.TaskId,
RowNumber = x.RowNumber,
Reason = x.Reason,
RawDataJson = OmsDocHelper.DeserializeJsonObject(x.RawDataJson)
}).ToList()
Action = OrderBatchActionCodes.ImportRow,
CustomerId = effectiveCustomer.Id,
JobPayloadJson = jobPayload,
RequestId = requestId,
Items = payloadItems,
};
}
@ -782,19 +618,6 @@ public partial class OrderService
return forecast?.ChargeableWeight;
}
private static OrderTaskFail BuildTaskFail(int taskId, int rowNumber, string reason, object rawData)
{
return new OrderTaskFail
{
TaskId = taskId,
RowNumber = rowNumber,
Reason = reason,
RawDataJson = JsonConvert.SerializeObject(rawData),
CreateUserId = 0,
CreateTime = DateTime.Now
};
}
private static string? GetStringCell(IDictionary<string, object> row, string key) => row.TryGetValue(key, out var value) ? value?.ToString() : null;
private static int? GetintCell(IDictionary<string, object> row, string key) => int.TryParse(GetStringCell(row, key), out var value) ? value : null;
private static int? GetIntCell(IDictionary<string, object> row, string key) => int.TryParse(GetStringCell(row, key), out var value) ? value : null;

28
src/Oms.Application/Services/Orders/OrderService.cs
View File

@ -29,7 +29,6 @@ public partial class OrderService : ApplicationService
{
private IOrderRepository OrderRepository => LazyServiceProvider.GetRequiredService<IOrderRepository>();
private IOrderTemplateRepository OrderTemplateRepository => LazyServiceProvider.GetRequiredService<IOrderTemplateRepository>();
private IOrderTaskRepository OrderTaskRepository => LazyServiceProvider.GetRequiredService<IOrderTaskRepository>();
private IWarehouseAddressRepository WarehouseAddressRepository => LazyServiceProvider.GetRequiredService<IWarehouseAddressRepository>();
private ISalesProductRepository SalesProductRepository => LazyServiceProvider.GetRequiredService<ISalesProductRepository>();
private IDictRepository DictRepository => LazyServiceProvider.GetRequiredService<IDictRepository>();
@ -335,38 +334,16 @@ public partial class OrderService : ApplicationService
}
/// <summary>
/// 导入订单文件。
/// 导入订单文件，创建 import-row 批量任务并返回 jobId。
/// </summary>
/// <param name="input">入参：订单导入参数。</param>
/// <returns>订单导入结果。</returns>
/// <returns>批量任务 ID 及是否异步处理。</returns>
[HttpPost("/order/upload")]
public async Task<OrderUploadOutput> UploadAsync([FromForm] OrderUploadInput input)
{
return await ImportOrdersAsync(input);
}
/// <summary>
/// 查询订单任务列表。
/// </summary>
/// <param name="input">入参：订单任务列表查询条件。</param>
/// <returns>订单任务分页结果，包含列表数据和总记录数。</returns>
[HttpGet("/order/task/list")]
public async Task<OmsPageResult<OrderTaskOutput>> GetTaskListAsync([FromQuery] OrderTaskListInput input)
{
return await GetTaskOutputsAsync(input);
}
/// <summary>
/// 查询订单任务详情。
/// </summary>
/// <param name="taskId">入参：任务主键。</param>
/// <returns>订单任务详情。</returns>
[HttpGet("/order/task/detail")]
public async Task<OrderTaskDetailOutput> GetTaskDetailAsync([FromQuery] int taskId)
{
return await GetTaskDetailCoreAsync(taskId);
}
/// <summary>
/// 创建订单包裹。
/// </summary>
@ -412,6 +389,7 @@ public partial class OrderService : ApplicationService
{
OrderId = 0,
ServiceNo = request.ServiceNo,
ClientNo = request.ClientNo,
ItemKey = !string.IsNullOrWhiteSpace(request.RequestId)
? $"parcel:{request.RequestId}"
: $"parcel:{batchNo}:{index + 1}",

26
src/Oms.Domain/OrdersAggregate/Consts/OrderTaskStatusCodes.cs
View File

@ -1,26 +0,0 @@
namespace Oms.Domain.OrdersAggregate.Consts;
/// <summary>
/// 订单导入/批量任务状态（t_order_task.Status，字符串落库；与字典 order_task_status 对齐）。
/// </summary>
public static class OrderTaskStatusCodes
{
public const string Pending = "pending";
public const string Processing = "processing";
public const string Completed = "completed";
public static readonly HashSet<string> All = new(StringComparer.OrdinalIgnoreCase)
{
Pending,
Processing,
Completed
};
public static string GetDisplayName(string? code) => code?.Trim().ToLowerInvariant() switch
{
Pending => "待处理",
Processing => "处理中",
Completed => "已完成",
_ => code ?? string.Empty
};
}

3
src/Oms.Domain/OrdersAggregate/Entities/OrderBatchJobItem.cs
View File

@ -13,6 +13,9 @@ public class OrderBatchJobItem : OmsAuditedEntity
[SugarColumn(Length = 64, IsNullable = true)]
public string? ServiceNo { get; set; }
[SugarColumn(Length = 64, IsNullable = true)]
public string? ClientNo { get; set; }
public string Action { get; set; } = string.Empty;
public string Status { get; set; } = string.Empty;

22
src/Oms.Domain/OrdersAggregate/Entities/OrderTask.cs
View File

@ -1,22 +0,0 @@
using Oms.Domain.Shared;
using SqlSugar;
namespace Oms.Domain.OrdersAggregate.Entities;
[SugarTable("t_order_task")]
public class OrderTask : OmsAuditedAggregateRoot
{
public string TaskName { get; set; } = string.Empty;
public int? Quantity { get; set; }
public int? SuccessCount { get; set; }
public int? FailCount { get; set; }
public string Status { get; set; } = string.Empty;
public int? TemplateId { get; set; }
public int? CustomerId { get; set; }
}

17
src/Oms.Domain/OrdersAggregate/Entities/OrderTaskFail.cs
View File

@ -1,17 +0,0 @@
using Oms.Domain.Shared;
using SqlSugar;
namespace Oms.Domain.OrdersAggregate.Entities;
[SugarTable("t_order_task_fail")]
public class OrderTaskFail : OmsAuditedEntity
{
public int TaskId { get; set; }
public int RowNumber { get; set; }
public string Reason { get; set; } = string.Empty;
[SugarColumn(ColumnDataType = "json", IsNullable = true)]
public string? RawDataJson { get; set; }
}

15
src/Oms.Domain/OrdersAggregate/Repositories/IOrderTaskRepository.cs
View File

@ -1,15 +0,0 @@
using Fake.Domain.Repositories;
using Oms.Domain.OrdersAggregate.Entities;
namespace Oms.Domain.OrdersAggregate.Repositories;
public interface IOrderTaskRepository : IRepository<OrderTask>
{
Task<OrderTask?> GetByCustomerAndIdAsync(int customerId, int id, CancellationToken cancellationToken = default);
Task<bool> ExistsAsync(int id, CancellationToken cancellationToken = default);
Task<List<OrderTaskFail>> GetFailListAsync(int taskId, CancellationToken cancellationToken = default);
Task InsertFailListAsync(IEnumerable<OrderTaskFail> failList, CancellationToken cancellationToken = default);
}

40
src/Oms.SqlSugarCore/Orders/OrderTaskRepository.cs
View File

@ -1,40 +0,0 @@
using Fake.DependencyInjection;
using Oms.Domain.OrdersAggregate.Entities;
using Oms.Domain.OrdersAggregate.Repositories;
using Oms.SqlSugarCore.Repositories;
namespace Oms.SqlSugarCore.Orders;
public class OrderTaskRepository : OmsSqlSugarRepository<OrderTask>, IOrderTaskRepository, IScopedDependency
{
public Task<OrderTask?> GetByCustomerAndIdAsync(int customerId, int id, CancellationToken cancellationToken = default)
{
return FirstOrDefaultAsync(x => x.CustomerId == customerId && x.Id == id, cancellationToken);
}
public Task<bool> ExistsAsync(int id, CancellationToken cancellationToken = default)
{
return AnyAsync(x => x.Id == id, cancellationToken);
}
public async Task<List<OrderTaskFail>> GetFailListAsync(int taskId, CancellationToken cancellationToken = default)
{
var db = await GetDbContextAsync(cancellationToken);
return await db.Queryable<OrderTaskFail>()
.Where(x => x.TaskId == taskId)
.OrderBy(x => x.RowNumber)
.ToListAsync(cancellationToken);
}
public async Task InsertFailListAsync(IEnumerable<OrderTaskFail> failList, CancellationToken cancellationToken = default)
{
var items = failList.ToArray();
if (items.Length == 0)
{
return;
}
var db = await GetDbContextAsync(cancellationToken);
await db.Insertable(items).ExecuteCommandAsync(cancellationToken);
}
}

63
src/Oms.Tests/OrderBatchJobProcessorTests.cs
View File

@ -1,5 +1,6 @@
using FluentAssertions;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using Moq;
using Oms.Application.Messaging;
using Oms.Application.Services.Orders.Batch;
@ -7,12 +8,10 @@ using Oms.Domain.OrdersAggregate.Consts;
using Oms.Domain.OrdersAggregate.Entities;
using Oms.Domain.OrdersAggregate.Repositories;
namespace Oms.Tests;
public class OrderBatchJobProcessorTests
{
private readonly Mock<IOrderBatchJobRepository> _batchJobRepository = new(MockBehavior.Strict);
private readonly Mock<IOrderTaskRepository> _orderTaskRepository = new(MockBehavior.Loose);
private readonly Mock<IOrderRepository> _orderRepository = new(MockBehavior.Loose);
private readonly Mock<IOrderBatchActionExecutor> _actionExecutor = new(MockBehavior.Strict);
private readonly OrderBatchJobProcessor _processor;
@ -20,8 +19,9 @@ public class OrderBatchJobProcessorTests
{
_processor = new OrderBatchJobProcessor(
_batchJobRepository.Object,
_orderTaskRepository.Object,
_orderRepository.Object,
_actionExecutor.Object,
Options.Create(new OrderBatchRabbitMqOptions { MaxSystemRetries = 5 }),
NullLogger<OrderBatchJobProcessor>.Instance);
}
@ -40,10 +40,11 @@ public class OrderBatchJobProcessorTests
item.Status.Should().Be(OrderBatchJobItemStatusCodes.Failed);
item.FailReason.Should().Be("订单不存在");
item.ResultJson.Should().Contain("订单不存在");
}
[Fact]
public async Task ProcessAsync_SystemFailure_ShouldThrow()
public async Task ProcessAsync_SystemFailure_ShouldThrowForRetry()
{
var item = CreateItem(id: 11, status: OrderBatchJobItemStatusCodes.Pending);
var message = CreateMessage(item.Id);
@ -56,6 +57,33 @@ public class OrderBatchJobProcessorTests
await act.Should().ThrowAsync<InvalidOperationException>()
.WithMessage("*下游超时*");
item.RetryCount.Should().Be(1);
item.Status.Should().Be(OrderBatchJobItemStatusCodes.Pending);
}
[Fact]
public async Task ProcessAsync_SystemFailureExhausted_ShouldMarkFailedWithoutThrow()
{
var processor = new OrderBatchJobProcessor(
_batchJobRepository.Object,
_orderRepository.Object,
_actionExecutor.Object,
Options.Create(new OrderBatchRabbitMqOptions { MaxSystemRetries = 1 }),
NullLogger<OrderBatchJobProcessor>.Instance);
var item = CreateItem(id: 14, status: OrderBatchJobItemStatusCodes.Pending);
var message = CreateMessage(item.Id);
SetupItemLookup(item);
_actionExecutor
.Setup(x => x.ExecuteAsync(message, It.IsAny<CancellationToken>()))
.ReturnsAsync(OrderBatchActionExecuteResult.SystemFail("下游超时"));
SetupJobRefresh(item.JobId, [item]);
await processor.ProcessAsync(message);
item.Status.Should().Be(OrderBatchJobItemStatusCodes.Failed);
item.RetryCount.Should().Be(1);
item.FailReason.Should().Contain("系统重试已达上限");
}
[Fact]
@ -96,6 +124,31 @@ public class OrderBatchJobProcessorTests
Times.Never);
}
[Fact]
public async Task ProcessAsync_Success_ShouldPersistClientNo()
{
var item = CreateItem(id: 12, status: OrderBatchJobItemStatusCodes.Pending);
item.Action = OrderBatchActionCodes.ImportRow;
var message = CreateMessage(item.Id);
message.Action = OrderBatchActionCodes.ImportRow;
SetupItemLookup(item);
_actionExecutor
.Setup(x => x.ExecuteAsync(message, It.IsAny<CancellationToken>()))
.ReturnsAsync(OrderBatchActionExecuteResult.Ok(
createdOrderId: 9001,
serviceNo: "ID20260722001",
clientNo: "CUST-001"));
SetupJobRefresh(item.JobId, [item]);
await _processor.ProcessAsync(message);
item.Status.Should().Be(OrderBatchJobItemStatusCodes.Success);
item.OrderId.Should().Be(9001);
item.ClientNo.Should().Be("CUST-001");
item.ServiceNo.Should().Be("ID20260722001");
item.ResultJson.Should().Contain("CUST-001");
}
private void SetupItemLookup(OrderBatchJobItem item)
{
_batchJobRepository

21
系统结构说明.md
View File

@ -179,10 +179,14 @@ JWT 中业务用户整数 ID 的 Claim 名为 **`mapId`**（对应 `user.userid`
| 文件 | 主要 API | 作用 |
|------|----------|------|
| `OrderService.cs` | `/order/*`（主文件） | 订单列表/详情/支付/申报/拦截/导出/模板/上传/任务查询 |
| `OrderService.cs` | `/order/*`（主文件） | 订单列表/详情/支付/申报/拦截/导出/模板/上传 |
| `OrderService.BatchJob.cs` | `/order/batch/*` | 批量任务列表/详情/子项/重试/取消 |
| `OrderService.ParcelBilling.cs` | `/order/parcel/*` | 小包创建、批量创建、计重算价、费用快照写入 |
| `OrderService.ParcelQuote.cs` | `/order/price-check` | 价格试算：匹配产品与分区报价 |
| `OrderService.Helpers.cs` | — | 订单通用辅助（状态校验、映射等） |
| `OrderService.Helpers.cs` | — | 订单通用辅助（导入、状态校验、映射等） |
| `Batch/OrderBatchJobCoordinator.cs` | — | 批量任务创建、幂等、同步/异步入队 |
| `Batch/OrderBatchJobProcessor.cs` | — | 批量子项消费与结果回写 |
| `Batch/OrderBatchJobItemFeedback.cs` | — | 子项成功/失败/重试反馈封装 |
| `ParcelBillingCalculator.cs` | — | 静态计重算价引擎：材积重、分区、卖价费用 |
| `ParcelQuoteContext.cs` | — | 报价上下文（国家、邮编、尺寸重量等） |
| `ParcelQuoteResult.cs` | — | 单产品报价结果模型 |
@ -288,8 +292,8 @@ JWT 中业务用户整数 ID 的 Claim 名为 **`mapId`**（对应 `user.userid`
| `OrderParcelBatchCreateInput.cs` / `OrderParcelBatchCreateOutput.cs` / `OrderParcelBatchCreateItemOutput.cs` | 批量小包创建 |
| `OrderParcelItemInput.cs` / `OrderParcelReceiverInput.cs` / `OrderParcelSenderInput.cs` | 小包货品与地址 |
| `OrderTemplateListInput.cs` / `OrderTemplateOutput.cs` | 导入模板 |
| `OrderUploadInput.cs` / `OrderUploadOutput.cs` | Excel 批量上传 |
| `OrderTaskListInput.cs` / `OrderTaskOutput.cs` / `OrderTaskDetailOutput.cs` / `OrderTaskFailOutput.cs` | 异步导入任务 |
| `OrderUploadInput.cs` / `OrderUploadOutput.cs` | Excel 批量上传（返回 jobId、async） |
| `OrderBatchJobListInput.cs` / `OrderBatchJobDtos.cs` | 批量任务列表/详情/子项 DTO |
| `ProductBillingSnapshotDtos.cs` | 计费快照（产品、价格、分区等）内部 DTO |
#### Dtos/Recharge/
@ -360,15 +364,16 @@ JWT 中业务用户整数 ID 的 Claim 名为 **`mapId`**（对应 `user.userid`
| `Entities/OrderFee.cs` | 订单费用明细 |
| `Entities/OrderWeight.cs` | 重量尺寸（预报/复核等） |
| `Entities/OrderTemplate.cs` | 导入模板元数据 |
| `Entities/OrderTask.cs` | 批量导入任务 |
| `Entities/OrderTaskFail.cs` | 导入失败行 |
| `Entities/OrderBatchJob.cs` | 批量任务头（导入/支付/打印等） |
| `Entities/OrderBatchJobItem.cs` | 批量任务子项 |
| `Consts/OrderBatchJobStatusCodes.cs` | 批量任务状态常量 |
| `Consts/OrderStatusCodes.cs` | 订单状态常量 |
| `Consts/OrderFeeTypeCodes.cs` | 费用类型常量 |
| `Consts/OrderWeightTypeCodes.cs` | 重量类型常量 |
| `Consts/CargoAttributeCodes.cs` | 货物属性常量 |
| `Repositories/IOrderRepository.cs` | 订单仓储（含关联加载） |
| `Repositories/IOrderTemplateRepository.cs` | 模板仓储 |
| `Repositories/IOrderTaskRepository.cs` | 导入任务仓储 |
| `Repositories/IOrderBatchJobRepository.cs` | 批量任务仓储 |
### ChannelsAggregate/（渠道/产品/计价）
@ -449,7 +454,7 @@ JWT 中业务用户整数 ID 的 Claim 名为 **`mapId`**（对应 `user.userid`
| `Recharge/RechargeRecordRepository.cs` | `IRechargeRecordRepository` | 充值记录 |
| `Orders/OrderRepository.cs` | `IOrderRepository` | 订单及关联表 |
| `Orders/OrderTemplateRepository.cs` | `IOrderTemplateRepository` | 订单模板 |
| `Orders/OrderTaskRepository.cs` | `IOrderTaskRepository` | 导入任务 |
| `Orders/OrderBatchJobRepository.cs` | `IOrderBatchJobRepository` | 批量任务及子项 |
| `Channels/SalesProductRepository.cs` | `ISalesProductRepository` | 销售产品 |
| `Channels/ServiceChannelRepository.cs` | `IServiceChannelRepository` | 服务渠道 |
| `Channels/ServiceChannelGroupRepository.cs` | `IServiceChannelGroupRepository` | 渠道分组 |