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 49f69d0e..a21b6dda 100644 --- a/apps/backend/src/services/cost-center/cost-exception-service.ts +++ b/apps/backend/src/services/cost-center/cost-exception-service.ts @@ -175,6 +175,37 @@ export async function listExceptionQueue(input: { * * 批量版本:一次处理一批订单项,避免逐条 UPDATE。 */ +export async function resolveExceptionsForItemsByReason( + orderItemIds: number[], + reason: string, +): Promise { + const ids = orderItemIds.filter((id) => Number.isInteger(id) && id > 0) + if (ids.length === 0) return 0 + const now = nowIso() + const result = await query<{ id: number }>( + ` + WITH removed AS ( + DELETE FROM cost_exception_items i + USING cost_exceptions ce + WHERE i.exception_id = ce.id + AND i.order_item_id = ANY($1::bigint[]) + AND ce.reason = $3 + RETURNING i.exception_id + ) + UPDATE cost_exceptions ce + SET status = 'resolved', resolved_at = $2, item_count = 0, last_seen_at = $2 + WHERE ce.id IN (SELECT DISTINCT exception_id FROM removed) + AND ce.status <> 'resolved' + AND NOT EXISTS ( + SELECT 1 FROM cost_exception_items i WHERE i.exception_id = ce.id + ) + RETURNING ce.id + `, + [ids, now, reason], + ) + return result.rowCount || 0 +} + export async function resolveExceptionsForItems(orderItemIds: number[]): Promise { const ids = orderItemIds.filter((id) => Number.isInteger(id) && id > 0) if (ids.length === 0) return 0 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 af3e25b7..989775fc 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 @@ -1,4 +1,5 @@ import { query, withTransaction } from '../../db/client.js' +import { resolveExceptionsForItemsByReason } from './cost-exception-service.js' import type { JsonObject } from '../../types/json.js' import { createHttpError } from '../../utils/http.js' import { formatShanghaiDate, shiftDate } from '../../utils/finance-date-range.js' @@ -424,7 +425,7 @@ export async function rematchManualFulfillments(input: Record = const dateTo = normalizeOptionalDate(input.dateTo ?? input.date_to, '结束日期') const now = nowIso() - const result = await query<{ updated_count: number }>( + const result = await query<{ id: number }>( ` WITH target AS ( SELECT @@ -454,11 +455,17 @@ export async function rematchManualFulfillments(input: Record = updated_at = $5 FROM target WHERE oi.id = target.order_item_id - RETURNING 1 + RETURNING oi.id `, [skuKey, onlyUnmatched, dateFrom, dateTo, now], ) + // 配置补上以后,「人工 SKU 未配置」异常要立即收口,不能等人工去点。 + const updatedIds = result.rows.map((row) => Number(row.id)) + if (updatedIds.length > 0) { + await resolveExceptionsForItemsByReason(updatedIds, 'manual_config_missing') + } + // 配置生效后,把 SKU 对应的成本版本挂上(若已配置商品)。 await attachCostVersionsForManualConfigs() diff --git a/apps/backend/src/services/fulfillment/channel-assign-service.test.ts b/apps/backend/src/services/fulfillment/channel-assign-service.test.ts index 9d135354..111b2a00 100644 --- a/apps/backend/src/services/fulfillment/channel-assign-service.test.ts +++ b/apps/backend/src/services/fulfillment/channel-assign-service.test.ts @@ -18,6 +18,7 @@ function buildItem(patch: Partial = {}): ChannelCandidateI normalized_sku_name: '套装-测试', has_work_order: false, has_affiliate_task: false, + basis_date: '2026-10-31', ...patch, } } diff --git a/apps/backend/src/services/fulfillment/channel-assign-service.ts b/apps/backend/src/services/fulfillment/channel-assign-service.ts index c79bae0c..97700fb1 100644 --- a/apps/backend/src/services/fulfillment/channel-assign-service.ts +++ b/apps/backend/src/services/fulfillment/channel-assign-service.ts @@ -50,6 +50,7 @@ export type ChannelCandidateItem = { normalized_sku_name: string | null has_work_order: boolean has_affiliate_task: boolean + basis_date: string } export type ChannelDecision = { @@ -142,6 +143,8 @@ export type AssignChannelsResult = { /** 下一轮起点;到达表尾时回绕到 0,保证漏判的订单还能被重新评估。 */ nextCursor: number byChannel: Record + /** 本轮被判定的订单所属业务日,供事实/总账按日回填。 */ + affectedDates: string[] } /** 批量判定:默认只扫 pending 且未锁定的订单项。 */ @@ -165,7 +168,9 @@ export async function assignPendingOrderItemChannels( lastId: cursor, nextCursor: cursor, byChannel: {}, + affectedDates: [], } + const affectedDates = new Set() for (let batch = 0; batch < maxBatches; batch += 1) { const rows = await query( @@ -178,6 +183,8 @@ export async function assignPendingOrderItemChannels( oi.sku_name, oi.manual_sku_key, oi.normalized_sku_name, + (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date::text + AS basis_date, EXISTS ( SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id ) AS has_work_order, @@ -213,6 +220,7 @@ export async function assignPendingOrderItemChannels( for (const item of rows.rows) { const decision = resolveChannelDecision(item, rules) + if (decision.channel) affectedDates.add(item.basis_date) if (decision.conflict) { result.conflicted += 1 await upsertAggregatedException({ @@ -258,6 +266,7 @@ export async function assignPendingOrderItemChannels( } result.nextCursor = reachedEnd ? 0 : cursor + result.affectedDates = [...affectedDates].sort() return result } @@ -543,3 +552,45 @@ async function loadEnabledRules(): Promise { export const CHANNEL_RULE_KEY_TYPES = MATCH_KEY_TYPES export const CHANNEL_RULE_SOURCES = RULE_SOURCES export const CHANNEL_VALUES = FULFILLMENT_CHANNELS + +/** + * 为卡券商品(tpv: 前缀)批量补渠道规则。 + * + * 卡券没有履约载体证据,只能靠规则判定;商品目录建好后一次补齐, + * 避免运营按 SKU 逐条点「指定渠道」。 + */ +export async function ensureVoucherChannelRules(): Promise<{ + created: number + productCount: number +}> { + const created = await query<{ id: number }>( + ` + INSERT INTO sku_channel_rules ( + provider, shop_id, match_key_type, match_key, channel, priority, + source, enabled, note, created_at, updated_at + ) + SELECT + '', '', 'sku_code', substring(p.canonical_sku from 5), 'third_party_voucher', 70, + 'voucher', TRUE, '卡券商品自动生成', NOW(), NOW() + FROM product_catalog p + WHERE p.status = 'active' + AND p.canonical_sku LIKE 'tpv:%' + AND btrim(substring(p.canonical_sku from 5)) <> '' + AND NOT EXISTS ( + SELECT 1 FROM sku_channel_rules r + WHERE r.provider = '' + AND r.shop_id = '' + AND r.match_key_type = 'sku_code' + AND r.match_key = substring(p.canonical_sku from 5) + AND r.channel = 'third_party_voucher' + ) + ON CONFLICT DO NOTHING + RETURNING id + `, + ) + const total = await query<{ count: number }>( + `SELECT COUNT(*)::int AS count FROM product_catalog + WHERE status = 'active' AND canonical_sku LIKE 'tpv:%'`, + ) + return { created: created.rowCount || 0, productCount: Number(total.rows[0]?.count || 0) } +} diff --git a/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts b/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts index 568681ca..a332ef82 100644 --- a/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts +++ b/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts @@ -1,11 +1,13 @@ import type { JsonObject } from '../../types/json.js' import { query } from '../../db/client.js' import { assignPendingOrderItemChannels } from '../fulfillment/channel-assign-service.js' +import { resolveExceptionsForItemsByReason } from '../cost-center/cost-exception-service.js' +import { readWatermarkItemId } from '../watermark-store.js' +import { rebuildChannelSalesDailyFacts } from '../finance/sales-facts-channel-service.js' +import { syncFinanceLedgerFromFacts } from '../finance/finance-ledger-service.js' import { normalizeOrderItemProducts } from '../cost-center/product-normalize-service.js' import { syncChannelCostSnapshots } from '../cost-center/cost-snapshot-service.js' import { upsertAggregatedException } from '../cost-center/cost-exception-service.js' -import { rebuildChannelSalesDailyFacts } from '../finance/sales-facts-channel-service.js' -import { syncFinanceLedgerFromFacts } from '../finance/finance-ledger-service.js' import { formatShanghaiDate, shiftDate } from '../../utils/finance-date-range.js' import { recordPipelineError, @@ -45,18 +47,46 @@ export async function runChannelAssignJob(job: JsonObject = {}) { }) await touchWatermark('channel_assign', { lastItemId: result.nextCursor }) + const refreshed = await refreshFactsForDates(result.affectedDates) return { status: 'ok', message: `渠道判定:扫描 ${result.scanned},已定 ${result.assigned},` + - `冲突 ${result.conflicted},仍待定 ${result.stillPending}`, + `冲突 ${result.conflicted},仍待定 ${result.stillPending},回填 ${refreshed} 天`, ...result, + refreshedDays: refreshed, } } +/** + * 渠道/成本变更后,按业务日回填销售事实与总账。 + * + * 事实任务只回看最近几天;补配价格、迟到分类这类变更必须按受影响日期重算, + * 否则旧日期的毛利会一直停在缺失成本/未分类。 + */ +async function refreshFactsForDates(dates: string[]): Promise { + const unique = [...new Set(dates.filter(Boolean))].sort().slice(0, 31) + let done = 0 + for (const date of unique) { + try { + await rebuildChannelSalesDailyFacts({ dateFrom: date, dateTo: date }) + await syncFinanceLedgerFromFacts({ dateFrom: date, dateTo: date }) + done += 1 + } catch { + // 单日失败不影响管道本身,下一轮或人工重跑会再处理。 + } + } + return done +} + export async function runCarrierDispatchJob(job: JsonObject = {}) { const config = jobConfig(job) const limit = clampInt(config.limit, 200, 1, 2000) + const cursor = + config.afterId != null + ? Math.max(0, Math.trunc(Number(config.afterId))) + : await readWatermarkItemId('carrier_dispatch') + let maxId = 0 const manual = await query<{ id: number; sku_key: string; config_id: number }>( ` @@ -66,10 +96,11 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) { WHERE oi.fulfillment_status = 'assigned' AND oi.fulfillment_channel = 'manual' AND oi.fulfillment_config_id IS NULL + AND oi.id > $2 ORDER BY oi.id LIMIT $1 `, - [limit], + [limit, cursor], ) let attached = 0 @@ -105,10 +136,11 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) { AND ce.reason = 'manual_config_missing' AND ce.status <> 'resolved' ) + AND oi.id > $2 ORDER BY oi.id LIMIT $1 `, - [limit], + [limit, cursor], ) for (const row of missingManual.rows) { @@ -142,10 +174,11 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) { AND ce.reason = 'manual_cost_missing' AND ce.status <> 'resolved' ) + AND oi.id > $2 ORDER BY oi.id LIMIT $1 `, - [limit], + [limit, cursor], ) for (const row of pendingManualCost.rows) { @@ -180,10 +213,11 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) { AND ce.reason = 'carrier_missing' AND ce.status <> 'resolved' ) + AND oi.id > $2 ORDER BY oi.id LIMIT $1 `, - [limit], + [limit, cursor], ) for (const row of missingCarrier.rows) { @@ -195,18 +229,122 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) { }) } - await touchWatermark('carrier_dispatch') + for (const row of [ + ...manual.rows, + ...missingManual.rows, + ...pendingManualCost.rows, + ...missingCarrier.rows, + ]) { + maxId = Math.max(maxId, Number(row.id)) + } + + const resolved = await resolveCarrierDispatchExceptions() + const nextCursor = maxId === 0 ? 0 : maxId + await touchWatermark('carrier_dispatch', { lastItemId: nextCursor }) return { status: 'ok', message: `人工载体挂载 ${attached},缺配置 ${missingManual.rows.length},` + - `缺履约载体 ${missingCarrier.rows.length}`, + `缺履约载体 ${missingCarrier.rows.length},异常收口 ${resolved}`, attached, missingManual: missingManual.rows.length, missingCarrier: missingCarrier.rows.length, + resolved, } } +/** + * 载体/配置补上后,把对应异常自动关掉(每轮最多处理 500 条)。 + */ +async function resolveCarrierDispatchExceptions(): Promise { + // 游标会跳过已扫过的订单项;配置是后来补的,这里兜底把配置挂上。 + await query( + ` + UPDATE order_items oi + SET fulfillment_config_id = cfg.id, updated_at = NOW() + FROM manual_fulfillment_configs cfg + WHERE cfg.sku_key = oi.manual_sku_key + AND cfg.enabled + AND oi.fulfillment_status = 'assigned' + AND oi.fulfillment_channel = 'manual' + AND oi.fulfillment_config_id IS NULL + AND EXISTS ( + SELECT 1 + FROM cost_exception_items i + JOIN cost_exceptions ce ON ce.id = i.exception_id + WHERE i.order_item_id = oi.id + AND ce.reason = 'manual_config_missing' + AND ce.status <> 'resolved' + ) + `, + ) + + const [carrier, config, manualCost] = await Promise.all([ + query<{ order_item_id: number }>( + ` + SELECT DISTINCT i.order_item_id + FROM cost_exception_items i + JOIN cost_exceptions ce ON ce.id = i.exception_id + JOIN order_items oi ON oi.id = i.order_item_id + WHERE ce.reason = 'carrier_missing' + AND ce.status <> 'resolved' + AND ( + EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id) + OR EXISTS ( + SELECT 1 FROM fulfillment_tasks ft + WHERE ft.order_item_id = oi.id + AND ft.executor_key IN ('affiliate_dash', 'third_party_voucher') + ) + ) + LIMIT 500 + `, + ), + query<{ order_item_id: number }>( + ` + SELECT DISTINCT i.order_item_id + FROM cost_exception_items i + JOIN cost_exceptions ce ON ce.id = i.exception_id + JOIN order_items oi ON oi.id = i.order_item_id + WHERE ce.reason = 'manual_config_missing' + AND ce.status <> 'resolved' + AND oi.fulfillment_config_id IS NOT NULL + LIMIT 500 + `, + ), + query<{ order_item_id: number }>( + ` + SELECT DISTINCT i.order_item_id + FROM cost_exception_items i + JOIN cost_exceptions ce ON ce.id = i.exception_id + WHERE ce.reason = 'manual_cost_missing' + AND ce.status <> 'resolved' + AND EXISTS ( + SELECT 1 FROM cost_snapshots cs + WHERE cs.subject_type = 'manual_case' + AND cs.order_item_id = i.order_item_id + AND cs.superseded_at IS NULL + ) + LIMIT 500 + `, + ), + ]) + + let resolved = 0 + resolved += await resolveExceptionsForItemsByReason( + carrier.rows.map((row) => Number(row.order_item_id)), + 'carrier_missing', + ) + resolved += await resolveExceptionsForItemsByReason( + config.rows.map((row) => Number(row.order_item_id)), + 'manual_config_missing', + ) + resolved += await resolveExceptionsForItemsByReason( + manualCost.rows.map((row) => Number(row.order_item_id)), + 'manual_cost_missing', + ) + return resolved +} + export async function runProductNormalizeJob(job: JsonObject = {}) { const config = jobConfig(job) const result = await normalizeOrderItemProducts({ @@ -230,10 +368,14 @@ export async function runCostSnapshotJob(job: JsonObject = {}) { }) await touchWatermark('cost_snapshot', { lastItemId: result.nextCursor }) + const refreshed = await refreshFactsForDates(result.affectedDates) return { status: 'ok', - message: `成本快照:扫描 ${result.scanned},生成 ${result.created},缺失 ${result.missing}`, + message: + `成本快照:扫描 ${result.scanned},生成 ${result.created},` + + `替换 ${result.superseded},缺失 ${result.missing},回填 ${refreshed} 天`, ...result, + refreshedDays: refreshed, } }