diff --git a/apps/backend/src/routes/admin/fulfillment-channel.ts b/apps/backend/src/routes/admin/fulfillment-channel.ts index 2a7bf804..e6aa9fc1 100644 --- a/apps/backend/src/routes/admin/fulfillment-channel.ts +++ b/apps/backend/src/routes/admin/fulfillment-channel.ts @@ -87,6 +87,7 @@ router.post( channel, adminUserId, note, + createRule: body.createRule !== false, }) } diff --git a/apps/backend/src/services/cost-center/cost-exception-service.ts b/apps/backend/src/services/cost-center/cost-exception-service.ts index e4044631..49f69d0e 100644 --- a/apps/backend/src/services/cost-center/cost-exception-service.ts +++ b/apps/backend/src/services/cost-center/cost-exception-service.ts @@ -18,6 +18,8 @@ export const COST_EXCEPTION_REASONS = [ 'product_mapping_missing', 'cost_missing', 'manual_config_missing', + 'manual_cost_missing', + 'carrier_missing', ] as const export type CostExceptionReason = (typeof COST_EXCEPTION_REASONS)[number] diff --git a/apps/backend/src/services/cost-center/manual-fulfillment-config-service.ts b/apps/backend/src/services/cost-center/manual-fulfillment-config-service.ts index f1d9dd5b..af3e25b7 100644 --- a/apps/backend/src/services/cost-center/manual-fulfillment-config-service.ts +++ b/apps/backend/src/services/cost-center/manual-fulfillment-config-service.ts @@ -127,7 +127,7 @@ export async function scanManualSkus(options: ScanOptions) { oi.manual_sku_key AS sku_key FROM order_items oi JOIN orders o ON o.id = oi.order_id - WHERE oi.fulfillment_mode = 'manual' + WHERE oi.fulfillment_status = 'assigned' AND oi.fulfillment_channel = 'manual' AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date BETWEEN $1::date AND $2::date ), @@ -240,7 +240,7 @@ export async function listManualSkus(input: Record = {}) { conditions.push(`EXISTS ( SELECT 1 FROM order_items oi JOIN orders o ON o.id = oi.order_id - WHERE oi.fulfillment_mode = 'manual' + WHERE oi.fulfillment_status = 'assigned' AND oi.fulfillment_channel = 'manual' AND o.provider = $${params.length} AND oi.manual_sku_key = r.sku_key )`) @@ -378,6 +378,9 @@ export async function saveManualFulfillmentConfig(payload: JsonObject) { const config = requiredRow(result.rows[0], '保存人工履约配置失败') await syncRegistryConfigId(config.sku_key, config.id) + // 同步渠道规则:人工配置就是「这个内部 SKU 走人工渠道」的唯一声明, + // 少了这一步,新配的 SKU 会一直停在待分类。 + await syncManualChannelRule(config) // 配置保存后立刻回填该 SKU 最近 30 天的订单项,避免必须再点一次「重新匹配」。 // 历史更早的订单项不自动回填(历史人工成本不补录);需要时再显式点「重新匹配」。 const today = formatShanghaiDate(new Date()) @@ -403,6 +406,7 @@ export async function deleteManualFulfillmentConfig(skuKey: string) { `DELETE FROM manual_fulfillment_configs WHERE sku_key = $1 RETURNING id`, [key], ) + await query(`DELETE FROM sku_channel_rules WHERE source = 'manual' AND match_key = $1`, [key]) await syncRegistryConfigId(key, null) return { skuKey: key, deleted: deleted.rowCount || 0 } } @@ -431,7 +435,7 @@ export async function rematchManualFulfillments(input: Record = JOIN orders o ON o.id = oi.order_id JOIN manual_fulfillment_configs cfg ON cfg.sku_key = oi.manual_sku_key - WHERE oi.fulfillment_mode = 'manual' + WHERE oi.fulfillment_status = 'assigned' AND oi.fulfillment_channel = 'manual' AND cfg.enabled = TRUE AND ($1 = '' OR cfg.sku_key = $1) AND ($2 = FALSE OR oi.fulfillment_config_id IS NULL) @@ -699,3 +703,36 @@ function requiredRow(row: T | undefined, message: string): T { } return row } + +/** + * 人工配置 → 渠道规则(内部 SKU 精确匹配)。 + * + * 幂等:按唯一键 (provider, shop, match_key_type, match_key, channel) upsert。 + */ +async function syncManualChannelRule(config: { + id: number + sku_key: string + enabled: boolean + note: string +}): Promise { + const skuKey = String(config.sku_key || '').trim() + if (!skuKey) return + + await query( + ` + INSERT INTO sku_channel_rules ( + provider, shop_id, match_key_type, match_key, channel, priority, + source, source_ref_id, enabled, note, created_at, updated_at + ) + VALUES ('', '', 'internal_sku', $1, 'manual', 70, 'manual', $2, $3, $4, NOW(), NOW()) + ON CONFLICT (provider, shop_id, match_key_type, match_key, channel) + DO UPDATE SET + enabled = EXCLUDED.enabled, + source = 'manual', + source_ref_id = EXCLUDED.source_ref_id, + note = EXCLUDED.note, + updated_at = NOW() + `, + [skuKey, config.id, config.enabled !== false, String(config.note || '').slice(0, 200)], + ) +} diff --git a/apps/backend/src/services/finance/manual-fulfillment-service.ts b/apps/backend/src/services/finance/manual-fulfillment-service.ts index 6655b125..75b33d69 100644 --- a/apps/backend/src/services/finance/manual-fulfillment-service.ts +++ b/apps/backend/src/services/finance/manual-fulfillment-service.ts @@ -96,7 +96,11 @@ export async function listManualFulfillmentPending( const keyword = String(input.keyword || '').trim() const status = String(input.status || '').trim() - const conditions = ["oi.fulfillment_mode = 'manual'", 'cfg.id IS NOT NULL'] + const conditions = [ + "oi.fulfillment_status = 'assigned'", + "oi.fulfillment_channel = 'manual'", + 'cfg.id IS NOT NULL', + ] const params: unknown[] = [] if (status === 'pending') { @@ -219,7 +223,7 @@ export async function confirmManualFulfillment(payload: JsonObject, session: Adm COALESCE(oi.revenue_amount_fen, 0)::bigint AS revenue_fen, COALESCE(o.paid_at, o.created_at) AS basis_at, oi.canonical_product_id, - oi.fulfillment_mode, + COALESCE(oi.fulfillment_channel, CASE WHEN oi.fulfillment_status = 'archived' THEN 'archived' ELSE 'pending' END) AS fulfillment_mode, oi.fulfillment_config_id AS config_id, COALESCE(cfg.platform_key, '') AS platform_key, COALESCE(cfg.credit_quantity, 0)::bigint AS credit_quantity, @@ -450,6 +454,21 @@ export async function confirmManualFulfillment(payload: JsonObject, session: Adm }) } + // 人工确认即渠道成立:写新三列,避免订单一直停在 pending。 + await client.query( + ` + UPDATE order_items + SET fulfillment_channel = 'manual', + fulfillment_status = 'assigned', + fulfillment_source = 'manual_confirm', + fulfillment_decided_at = $2, + updated_at = $2 + WHERE id = $1 + AND fulfillment_status <> 'archived' + `, + [orderItemId, now], + ) + // 人工缺成本异常立即收口,不等下一轮定时任务。 await client.query( `UPDATE cost_exceptions @@ -461,6 +480,20 @@ export async function confirmManualFulfillment(payload: JsonObject, session: Adm [orderItemId, now], ) + // v2 聚合异常:删除本条明细,没人再引用的异常头自动关闭。 + await client.query(`DELETE FROM cost_exception_items WHERE order_item_id = $1`, [orderItemId]) + await client.query( + ` + UPDATE cost_exceptions ce + SET status = 'resolved', resolved_at = $1, item_count = 0, last_seen_at = $1 + WHERE ce.status <> 'resolved' + AND NOT EXISTS ( + SELECT 1 FROM cost_exception_items i WHERE i.exception_id = ce.id + ) + `, + [now], + ) + return { orderItemId, orderId: Number(row.order_id_resolved), @@ -582,7 +615,7 @@ async function matchManualFulfillmentConfigWithClient( ) ) WHERE oi.id = $1 - AND oi.fulfillment_mode = 'manual' + AND oi.fulfillment_status = 'assigned' AND oi.fulfillment_channel = 'manual' AND oi.fulfillment_config_id IS NULL ORDER BY (cfg.canonical_product_id IS NOT NULL @@ -820,7 +853,7 @@ export async function getManualFulfillmentCountsByOrderIds( LEFT JOIN manual_fulfillment_configs cfg ON cfg.id = oi.fulfillment_config_id LEFT JOIN fulfilled ON fulfilled.order_item_id = oi.id WHERE oi.order_id = ANY($1::bigint[]) - AND oi.fulfillment_mode = 'manual' + AND oi.fulfillment_status = 'assigned' AND oi.fulfillment_channel = 'manual' GROUP BY oi.order_id `, [ids], diff --git a/apps/backend/src/services/fulfillment/channel-assign-service.ts b/apps/backend/src/services/fulfillment/channel-assign-service.ts index 50b43051..16d844c5 100644 --- a/apps/backend/src/services/fulfillment/channel-assign-service.ts +++ b/apps/backend/src/services/fulfillment/channel-assign-service.ts @@ -371,7 +371,8 @@ export async function assignChannelBySku(input: { channel: string adminUserId: number | null note?: string -}): Promise<{ updated: number }> { + createRule?: boolean +}): Promise<{ updated: number; ruleCreated: boolean }> { if (!isFulfillmentChannel(input.channel)) { throw createHttpError('渠道无效', { statusCode: 400, errorCode: 'invalid_channel' }) } @@ -395,7 +396,26 @@ export async function assignChannelBySku(input: { `, [input.provider, input.shopId, input.skuCode, input.channel, input.adminUserId], ) - return { updated: result.rowCount || 0 } + + // 默认同时落一条渠道规则:否则同 SKU 的下一单还是会进待分类。 + let ruleCreated = false + if (input.createRule !== false) { + const rule = await upsertChannelRule({ + provider: input.provider, + shopId: input.shopId, + matchKeyType: 'sku_code', + matchKey: input.skuCode, + channel: input.channel, + priority: 10, + source: 'operator', + enabled: true, + note: `运营在待分类队列指定:${String(input.note || '').slice(0, 120)}`, + adminUserId: input.adminUserId, + }) + ruleCreated = rule.id > 0 + } + + return { updated: result.rowCount || 0, ruleCreated } } export async function listChannelRules( diff --git a/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts b/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts index 638d25e1..a03d4631 100644 --- a/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts +++ b/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts @@ -110,12 +110,74 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) { }) } + // 已配置人工 SKU、但还没确认履约的单:进入「人工成本未记录」队列。 + const pendingManualCost = await query<{ id: number; sku_key: string }>( + ` + SELECT oi.id, oi.manual_sku_key AS sku_key + FROM order_items oi + WHERE oi.fulfillment_status = 'assigned' + AND oi.fulfillment_channel = 'manual' + AND oi.fulfillment_config_id IS NOT NULL + AND NOT EXISTS ( + SELECT 1 FROM cost_snapshots cs + WHERE cs.subject_type = 'manual_case' + AND cs.order_item_id = oi.id + AND cs.superseded_at IS NULL + ) + ORDER BY oi.id + LIMIT $1 + `, + [limit], + ) + + for (const row of pendingManualCost.rows) { + await upsertAggregatedException({ + reason: 'manual_cost_missing', + scopeKey: `manual_cost_missing|-|-|${row.sku_key}|manual`, + orderItemId: row.id, + detail: { skuKey: row.sku_key }, + }) + } + + // 自动发货 / 接单平台被运营指定后却没有载体:不能静默卡住,进问题队列。 + const missingCarrier = await query<{ id: number; channel: string; sku_code: string }>( + ` + SELECT oi.id, oi.fulfillment_channel AS channel, oi.sku_code + FROM order_items oi + WHERE oi.fulfillment_status = 'assigned' + AND oi.fulfillment_channel IN ('affiliate_dash', 'work_order') + AND NOT EXISTS ( + SELECT 1 FROM fulfillment_tasks ft + WHERE ft.order_item_id = oi.id + AND ft.executor_key IN ('affiliate_dash', 'third_party_voucher') + ) + AND NOT EXISTS ( + SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id + ) + ORDER BY oi.id + LIMIT $1 + `, + [limit], + ) + + for (const row of missingCarrier.rows) { + await upsertAggregatedException({ + reason: 'carrier_missing', + scopeKey: `carrier_missing|${row.channel}|${row.sku_code}`, + orderItemId: row.id, + detail: { channel: row.channel, skuCode: row.sku_code }, + }) + } + await touchWatermark('carrier_dispatch') return { status: 'ok', - message: `人工载体挂载 ${attached},缺配置 ${missingManual.rows.length}`, + message: + `人工载体挂载 ${attached},缺配置 ${missingManual.rows.length},` + + `缺履约载体 ${missingCarrier.rows.length}`, attached, missingManual: missingManual.rows.length, + missingCarrier: missingCarrier.rows.length, } } @@ -153,7 +215,7 @@ export async function runSalesFactsJob(job: JsonObject = {}) { const config = jobConfig(job) const days = clampInt(config.days, 3, 1, 60) const dateTo = formatShanghaiDate(new Date()) - const dateFrom = shiftDate(dateTo, -(days - 1)) + const dateFrom = await resolveCatchUpStart('sales_facts', days, dateTo) const result = await rebuildChannelSalesDailyFacts({ dateFrom, dateTo }) await touchWatermark('sales_facts', { lastStatDate: result.dateTo }) @@ -168,7 +230,7 @@ export async function runLedgerPostJob(job: JsonObject = {}) { const config = jobConfig(job) const days = clampInt(config.days, 3, 1, 60) const dateTo = formatShanghaiDate(new Date()) - const dateFrom = shiftDate(dateTo, -(days - 1)) + const dateFrom = await resolveCatchUpStart('ledger_post', days, dateTo) const result = await syncFinanceLedgerFromFacts({ dateFrom, dateTo }) await touchWatermark('ledger_post', { lastStatDate: dateTo }) @@ -215,3 +277,25 @@ export async function runSingleFinancePipeline( throw error } } + +/** + * 计算本轮起始日期:默认回看 N 天;若水位更早(例如停机数天), + * 从水位那天补起,最多补 90 天,避免一次扫描过长。 + */ +async function resolveCatchUpStart( + pipeline: 'sales_facts' | 'ledger_post', + days: number, + dateTo: string, +): Promise { + const windowStart = shiftDate(dateTo, -(days - 1)) + const result = await query<{ last_stat_date: string | null }>( + `SELECT last_stat_date::text FROM finance_watermarks WHERE pipeline = $1`, + [pipeline], + ) + const lastDate = result.rows[0]?.last_stat_date || '' + if (!lastDate) return windowStart + + const earliest = shiftDate(dateTo, -89) + const candidate = lastDate < windowStart ? lastDate : windowStart + return candidate < earliest ? earliest : candidate +}