diff --git a/apps/backend/src/services/cost-center/product-normalize-service.ts b/apps/backend/src/services/cost-center/product-normalize-service.ts index 3800e6fa..6ea04101 100644 --- a/apps/backend/src/services/cost-center/product-normalize-service.ts +++ b/apps/backend/src/services/cost-center/product-normalize-service.ts @@ -319,3 +319,69 @@ async function loadMappingIndex(): Promise { })), ) } + +/** 指定订单项立即归一（渠道判定后调用，保证成本能马上算出来）。 */ +export async function normalizeProductsForItems(itemIds: number[]): Promise<{ + matched: number + unmatched: number +}> { + const ids = [ + ...new Set(itemIds.map(Number).filter((id) => Number.isInteger(id) && id > 0)), + ].slice(0, 2000) + if (ids.length === 0) return { matched: 0, unmatched: 0 } + const index = await loadMappingIndex() + const rows = await query( + ` + SELECT + oi.id, + o.provider, + o.shop_id, + oi.sku_code, + oi.sku_name, + oi.normalized_sku_name, + oi.normalized_sku_series, + oi.fulfillment_channel, + wpr.rule_key AS work_rule_key, + cfg.canonical_product_id AS manual_config_product_id + FROM order_items oi + JOIN orders o ON o.id = oi.order_id + LEFT JOIN work_orders wo ON wo.order_item_id = oi.id + LEFT JOIN work_product_rules wpr ON wpr.id = wo.product_rule_id + LEFT JOIN manual_fulfillment_configs cfg ON cfg.id = oi.fulfillment_config_id + WHERE oi.fulfillment_status = 'assigned' + AND oi.canonical_product_id IS NULL + AND oi.id = ANY($1::bigint[]) + `, + [ids], + ) + const matched: Array<{ id: number; productId: number }> = [] + let unmatched = 0 + for (const item of rows.rows) { + const productId = resolveProductId(item, index) + if (!productId) { + unmatched += 1 + await upsertAggregatedException({ + reason: 'product_mapping_missing', + scopeKey: `product_mapping_missing|${item.provider}|${item.shop_id}|${item.sku_code}|${item.fulfillment_channel || '-'}`, + orderItemId: item.id, + detail: { skuName: item.sku_name, channel: item.fulfillment_channel }, + }) + continue + } + matched.push({ id: item.id, productId }) + } + if (matched.length > 0) { + await query( + ` + UPDATE order_items oi + SET canonical_product_id = input.product_id, updated_at = NOW() + FROM unnest($1::bigint[], $2::bigint[]) AS input(id, product_id) + WHERE oi.id = input.id + AND oi.canonical_product_id IS NULL + `, + [matched.map((row) => row.id), matched.map((row) => row.productId)], + ) + await resolveExceptionsForItems(matched.map((row) => row.id)) + } + return { matched: matched.length, unmatched } +} diff --git a/apps/backend/src/services/finance/refresh-facts.ts b/apps/backend/src/services/finance/refresh-facts.ts new file mode 100644 index 00000000..89832af4 --- /dev/null +++ b/apps/backend/src/services/finance/refresh-facts.ts @@ -0,0 +1,71 @@ +import { query } from '../../db/client.js' +import { rebuildChannelSalesDailyFacts } from './sales-facts-channel-service.js' +import { syncFinanceLedgerFromFacts } from './finance-ledger-service.js' + +/** + * 按业务日回填销售事实与总账。 + * + * - 日期较多（>31）时按 min~max 整段重算，避免截断导致部分日期永远不更新； + * - 已关账期间会被跳过，返回 skippedClosed 供调用方提示。 + */ +export async function refreshFactsForDates( + dates: string[], +): Promise<{ days: number; skippedClosed: number }> { + const unique = [...new Set(dates.filter(Boolean))].sort() + if (unique.length === 0) return { days: 0, skippedClosed: 0 } + + if (unique.length > 31) { + const dateFrom = unique[0]! + const dateTo = unique[unique.length - 1]! + try { + const facts = await rebuildChannelSalesDailyFacts({ dateFrom, dateTo }) + await syncFinanceLedgerFromFacts({ dateFrom, dateTo }) + if (facts.skippedClosedPeriods.length > 0) { + console.warn('[finance] 部分日期已关账，事实未回填', { + dateFrom, + dateTo, + skipped: facts.skippedClosedPeriods, + }) + } + return { days: unique.length, skippedClosed: facts.skippedClosedPeriods.length } + } catch { + return { days: 0, skippedClosed: 0 } + } + } + + let done = 0 + let skippedClosed = 0 + for (const date of unique) { + try { + const facts = await rebuildChannelSalesDailyFacts({ dateFrom: date, dateTo: date }) + if (facts.skippedClosedPeriods.length > 0) { + skippedClosed += facts.skippedClosedPeriods.length + console.warn('[finance] 日期已关账，事实未回填', { date }) + continue + } + await syncFinanceLedgerFromFacts({ dateFrom: date, dateTo: date }) + done += 1 + } catch { + // 单日失败不阻断调用方，下一轮或人工重跑会再处理。 + } + } + return { days: done, skippedClosed } +} + +/** 批量取订单项所属业务日（Asia/Shanghai）。 */ +export async function orderItemBasisDates(itemIds: number[]): Promise { + const ids = [...new Set(itemIds.filter((id) => Number.isInteger(id) && id > 0))] + if (ids.length === 0) return [] + const rows = await query<{ basis_date: string }>( + ` + SELECT DISTINCT + (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date::text + AS basis_date + FROM order_items oi + JOIN orders o ON o.id = oi.order_id + WHERE oi.id = ANY($1::bigint[]) + `, + [ids], + ) + return rows.rows.map((row) => row.basis_date) +} diff --git a/apps/backend/src/services/fulfillment/channel-assign-service.ts b/apps/backend/src/services/fulfillment/channel-assign-service.ts index 7fc867e5..d6e90f3c 100644 --- a/apps/backend/src/services/fulfillment/channel-assign-service.ts +++ b/apps/backend/src/services/fulfillment/channel-assign-service.ts @@ -3,6 +3,9 @@ import { createHttpError } from '../../utils/http.js' import { nowIso } from '../../utils/time.js' import { upsertAggregatedException } from '../cost-center/cost-exception-service.js' import { readWatermarkItemId } from '../watermark-store.js' +import { syncCostSnapshotsForItems } from '../cost-center/cost-snapshot-service.js' +import { normalizeProductsForItems } from '../cost-center/product-normalize-service.js' +import { orderItemBasisDates, refreshFactsForDates } from '../finance/refresh-facts.js' import { FULFILLMENT_CHANNELS, MATCH_KEY_TYPES, @@ -145,6 +148,8 @@ export type AssignChannelsResult = { byChannel: Record /** 本轮被判定的订单所属业务日，供事实/总账按日回填。 */ affectedDates: string[] + /** 本轮新判定的订单项（供立即计算成本，最多 2000 条）。 */ + assignedItemIds: number[] } /** 批量判定：默认只扫 pending 且未锁定的订单项。 */ @@ -169,6 +174,7 @@ export async function assignPendingOrderItemChannels( nextCursor: cursor, byChannel: {}, affectedDates: [], + assignedItemIds: [], } const affectedDates = new Set() @@ -262,6 +268,7 @@ export async function assignPendingOrderItemChannels( ], ) result.assigned += updated.rowCount || 0 + result.assignedItemIds = assignments.map((row) => row.id).slice(0, 2000) } } @@ -305,7 +312,17 @@ export async function assignChannelToOrderItems(input: { [ids, input.channel, input.adminUserId, String(input.note || '').slice(0, 200)], ) - return { updated: result.rowCount || 0 } + const updated = result.rowCount || 0 + if (updated > 0) { + // 运营指定后先归一商品，再立即算成本并回填事实，避免旧单等冷通道。 + await normalizeProductsForItems(ids).catch(() => undefined) + const cost = await syncCostSnapshotsForItems(ids).catch(() => null) + const dates = [ + ...new Set([...(await orderItemBasisDates(ids)), ...(cost?.affectedDates || [])]), + ] + await refreshFactsForDates(dates).catch(() => undefined) + } + return { updated } } /** 运营解锁：允许自动判定重新接管。 */ diff --git a/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts b/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts index 853a2df1..70369011 100644 --- a/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts +++ b/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts @@ -3,10 +3,17 @@ 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 { refreshFactsForDates } from '../finance/refresh-facts.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 { + normalizeOrderItemProducts, + normalizeProductsForItems, +} from '../cost-center/product-normalize-service.js' +import { + syncChannelCostSnapshots, + syncCostSnapshotsForItems, +} from '../cost-center/cost-snapshot-service.js' import { upsertAggregatedException } from '../cost-center/cost-exception-service.js' import { formatShanghaiDate, shiftDate } from '../../utils/finance-date-range.js' import { @@ -47,7 +54,16 @@ export async function runChannelAssignJob(job: JsonObject = {}) { }) await touchWatermark('channel_assign', { lastItemId: result.nextCursor }) - const refreshed = await refreshFactsForDates(result.affectedDates) + + // 刚判定渠道的订单项立刻算成本，不等冷通道扫到。 + let costDates: string[] = [] + if (result.assignedItemIds.length > 0) { + // 先归一商品，再算成本：卡券/自动发货没有商品就无法解析采购价。 + await normalizeProductsForItems(result.assignedItemIds).catch(() => undefined) + const cost = await syncCostSnapshotsForItems(result.assignedItemIds).catch(() => null) + costDates = cost?.affectedDates || [] + } + const refreshed = await refreshFactsForDates([...result.affectedDates, ...costDates]) return { status: 'ok', message: @@ -60,57 +76,6 @@ export async function runChannelAssignJob(job: JsonObject = {}) { } } -/** - * 渠道/成本变更后，按业务日回填销售事实与总账。 - * - * 事实任务只回看最近几天；补配价格、迟到分类这类变更必须按受影响日期重算， - * 否则旧日期的毛利会一直停在缺失成本/未分类。 - */ -async function refreshFactsForDates( - dates: string[], -): Promise<{ days: number; skippedClosed: number }> { - const unique = [...new Set(dates.filter(Boolean))].sort() - if (unique.length === 0) return { days: 0, skippedClosed: 0 } - - // 日期过多时按整段重算，避免截断导致部分日期永远不更新。 - if (unique.length > 31) { - const dateFrom = unique[0]! - const dateTo = unique[unique.length - 1]! - try { - const facts = await rebuildChannelSalesDailyFacts({ dateFrom, dateTo }) - await syncFinanceLedgerFromFacts({ dateFrom, dateTo }) - if (facts.skippedClosedPeriods.length > 0) { - console.warn('[finance-pipeline] 部分日期已关账，事实未回填', { - dateFrom, - dateTo, - skipped: facts.skippedClosedPeriods, - }) - } - return { days: unique.length, skippedClosed: facts.skippedClosedPeriods.length } - } catch { - return { days: 0, skippedClosed: 0 } - } - } - - let done = 0 - let skippedClosed = 0 - for (const date of unique) { - try { - const facts = await rebuildChannelSalesDailyFacts({ dateFrom: date, dateTo: date }) - if (facts.skippedClosedPeriods.length > 0) { - skippedClosed += facts.skippedClosedPeriods.length - console.warn('[finance-pipeline] 日期已关账，事实未回填', { date }) - continue - } - await syncFinanceLedgerFromFacts({ dateFrom: date, dateTo: date }) - done += 1 - } catch { - // 单日失败不影响管道本身，下一轮或人工重跑会再处理。 - } - } - return { days: done, skippedClosed } -} - export async function runCarrierDispatchJob(job: JsonObject = {}) { const config = jobConfig(job) const limit = clampInt(config.limit, 200, 1, 2000)