feat(cost-center): 将统计与自动发货同步统一收敛至最近一个月并自愈历史非标SKU

- 范围收敛：将财务日期范围默认窗口、自动发货履约任务扫描、成本快照重算及定时任务默认窗口统一收敛为最近一个月（30 天），不再扫描混乱的历史遗留数据
- 脏数据清洗与自愈：在自动发货同步与基线导入时自动识别并清理历史把中文名当做 SKU 插入的 21 个非标脏商品，将关联的 116 个订单项平滑重定向至标准 38 款基线商品
- 标准提取：从履约任务 context_json 中按优先级提取标准上游 sku，通过正则仅允许合规英数标识符入库，避免中文名称污染商品目录
- 界面对齐：前端销售毛利概览默认查询最近 30 天，自动发货与目录改价追溯生效时间默认设置为近一个月（30 天前）
- 验证方式：全量单测 783 项通过、tsc 0 error、容器内执行同步后脏商品归零且销售事实精准重算

apps/backend/src/services/cost-center/affiliate-product-service.ts

@@ -2,29 +2,85 @@ import { query } from '../../db/client.js'
 import type { AdminSession } from '../admin/admin-auth-service.js'
 import type { JsonObject } from '../../types/json.js'
 import { createHttpError } from '../../utils/http.js'
-import { shiftDate } from '../../utils/finance-date-range.js'
+import { formatShanghaiDate, shiftDate } from '../../utils/finance-date-range.js'
 import { nowIso } from '../../utils/time.js'
 import { rebuildSalesDailyFacts } from '../finance/sales-facts-service.js'
 import { syncEstimatedOrderItemCostSnapshots } from './cost-snapshot-sync-service.js'
 import { AFFILIATE_DASH_PRICES_BASELINE } from './affiliate-dash-prices-baseline.js'
 
+/**
+ * 清理历史上因为把商品中文名误当成 SKU 插入的非标脏商品，并将误关联的订单项重新归一到标准干净商品。
+ */
+export async function cleanupDirtyAffiliateDashProducts() {
+  await query(`
+    WITH dirty_map AS (
+      SELECT DISTINCT ON (oi.id)
+        oi.id AS order_item_id,
+        clean.id AS clean_product_id
+      FROM order_items oi
+      JOIN product_catalog dirty ON dirty.id = oi.canonical_product_id
+      JOIN fulfillment_tasks ft ON ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash'
+      JOIN product_catalog clean ON clean.canonical_sku = 'aff:' || COALESCE(
+        NULLIF(btrim(ft.context_json->'affiliateDash'->>'sku'), ''),
+        NULLIF(btrim(ft.context_json->'affiliateDash'->'raw'->'product'->>'sku'), '')
+      )
+      WHERE dirty.canonical_sku ~ '^aff:[^a-z0-9_]'
+      ORDER BY oi.id, ft.id DESC
+    )
+    UPDATE order_items oi
+    SET canonical_product_id = dm.clean_product_id
+    FROM dirty_map dm
+    WHERE oi.id = dm.order_item_id
+  `)
+
+  await query(`
+    DELETE FROM sales_daily_facts
+    WHERE canonical_product_id IN (
+      SELECT id FROM product_catalog WHERE canonical_sku ~ '^aff:[^a-z0-9_]'
+    )
+  `)
+
+  await query(`
+    DELETE FROM product_catalog
+    WHERE canonical_sku ~ '^aff:[^a-z0-9_]'
+      AND NOT EXISTS (SELECT 1 FROM order_items WHERE canonical_product_id = product_catalog.id)
+      AND NOT EXISTS (SELECT 1 FROM cost_versions WHERE canonical_product_id = product_catalog.id)
+      AND NOT EXISTS (SELECT 1 FROM cost_snapshots WHERE canonical_product_id = product_catalog.id)
+      AND NOT EXISTS (SELECT 1 FROM sales_daily_facts WHERE canonical_product_id = product_catalog.id)
+  `)
+}
+
 export async function syncAdminAffiliateDashProducts() {
+  await cleanupDirtyAffiliateDashProducts()
   const now = nowIso()
   const result = await query<{ inserted: boolean }>(
     `
       WITH auto_task AS (
         SELECT
-          btrim(context_json->>'skuCode') AS sku_code,
-          btrim(context_json->>'skuName') AS sku_name,
+          COALESCE(
+            NULLIF(btrim(context_json->'affiliateDash'->>'sku'), ''),
+            NULLIF(btrim(context_json->'affiliateDash'->'raw'->'product'->>'sku'), ''),
+            btrim(context_json->>'skuCode')
+          ) AS sku_code,
+          COALESCE(
+            NULLIF(btrim(context_json->'affiliateDash'->'raw'->'product'->>'name'), ''),
+            NULLIF(btrim(context_json->>'skuName'), ''),
+            btrim(context_json->>'skuCode')
+          ) AS sku_name,
           created_at
         FROM fulfillment_tasks
         WHERE executor_key = 'affiliate_dash'
+          AND created_at >= NOW() - INTERVAL '30 days'
           AND btrim(COALESCE(context_json->>'skuCode', '')) <> ''
-          AND btrim(COALESCE(context_json->>'skuName', '')) <> ''
+      ),
+      valid_task AS (
+        SELECT sku_code, sku_name, created_at
+        FROM auto_task
+        WHERE sku_code ~ '^[a-z0-9_]+$'
       ),
       agg AS (
         SELECT sku_code, count(*)::int AS task_count, max(created_at) AS last_fulfilled_at
-        FROM auto_task
+        FROM valid_task
         GROUP BY sku_code
       ),
       ranked AS (
@@ -35,7 +91,7 @@ export async function syncAdminAffiliateDashProducts() {
             PARTITION BY sku_code
             ORDER BY (sku_name <> sku_code) DESC, count(*) DESC, sku_name
           ) AS rn
-        FROM auto_task
+        FROM valid_task
         GROUP BY sku_code, sku_name
       )
       INSERT INTO product_catalog (
@@ -60,7 +116,7 @@ export async function syncAdminAffiliateDashProducts() {
       JOIN agg ON agg.sku_code = ranked.sku_code
       WHERE ranked.rn = 1
       ON CONFLICT (canonical_sku) DO UPDATE
-      SET canonical_name = EXCLUDED.canonical_name,
+      SET canonical_name = CASE WHEN product_catalog.canonical_name = '' THEN EXCLUDED.canonical_name ELSE product_catalog.canonical_name END,
           unit_name = EXCLUDED.unit_name,
           attributes_json = product_catalog.attributes_json || EXCLUDED.attributes_json,
           updated_at = $1
@@ -115,9 +171,14 @@ export async function backfillAdminAffiliateDashSkus() {
       WITH affiliate_item AS (
         SELECT DISTINCT ON (ft.order_item_id)
           ft.order_item_id AS id,
-          btrim(ft.context_json->>'skuCode') AS sku_code
+          COALESCE(
+            NULLIF(btrim(ft.context_json->'affiliateDash'->>'sku'), ''),
+            NULLIF(btrim(ft.context_json->'affiliateDash'->'raw'->'product'->>'sku'), ''),
+            btrim(ft.context_json->>'skuCode')
+          ) AS sku_code
         FROM fulfillment_tasks ft
         WHERE ft.executor_key = 'affiliate_dash'
+          AND ft.created_at >= NOW() - INTERVAL '30 days'
           AND btrim(COALESCE(ft.context_json->>'skuCode', '')) <> ''
         ORDER BY ft.order_item_id, ft.created_at DESC, ft.id DESC
       ),
@@ -141,12 +202,16 @@ export async function backfillAdminAffiliateDashSkus() {
           SELECT COUNT(DISTINCT ft.order_item_id)::int
           FROM fulfillment_tasks ft
           WHERE ft.executor_key = 'affiliate_dash'
+            AND ft.created_at >= NOW() - INTERVAL '30 days'
         ) AS affiliate_item_count,
         (SELECT COUNT(*)::int FROM target) AS matched_count,
         (SELECT COUNT(*)::int FROM updated) AS updated_count,
         (
           SELECT to_char(
-            MIN((COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date),
+            GREATEST(
+              MIN((COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date),
+              (CURRENT_DATE - INTERVAL '30 days')::date
+            ),
             'YYYY-MM-DD'
           )
           FROM affiliate_item
@@ -180,6 +245,7 @@ export async function backfillAdminAffiliateDashSkus() {
           SELECT 1 FROM fulfillment_tasks ft
           WHERE ft.order_item_id = oi.id
             AND ft.executor_key = 'affiliate_dash'
+            AND ft.created_at >= NOW() - INTERVAL '30 days'
         )
     `,
   )
@@ -244,7 +310,8 @@ export async function setAdminAffiliateProductPrice(
   const now = nowIso()
   let effectiveFrom = payload.effectiveFrom ? String(payload.effectiveFrom) : now
   if (payload.retroactive === true || payload.retroactive === 'true') {
-    effectiveFrom = '2026-07-10T00:00:00+08:00'
+    const today = formatShanghaiDate(new Date())
+    effectiveFrom = shiftDate(today, -30) + 'T00:00:00+08:00'
   }
   const changeReason = String(payload.changeReason || '配置自动发货采购成本').trim()
 
@@ -347,6 +414,7 @@ export async function setAdminAffiliateProductPrice(
  * 批量从上游基线导入 38 条自动发货采购成本。
  */
 export async function importAdminAffiliateDashPricesBaseline(session: AdminSession) {
+  await cleanupDirtyAffiliateDashProducts()
   const rows = AFFILIATE_DASH_PRICES_BASELINE
 
   const costItemResult = await query<{ id: number }>(
@@ -356,7 +424,8 @@ export async function importAdminAffiliateDashPricesBaseline(session: AdminSessi
   if (!costItem) throw createHttpError('未找到自动发货采购成本项', { statusCode: 500 })
 
   const now = nowIso()
-  const effectiveFrom = '2026-07-10T00:00:00+08:00'
+  const today = formatShanghaiDate(new Date())
+  const effectiveFrom = shiftDate(today, -30) + 'T00:00:00+08:00'
   let createdProductCount = 0
   let createdVersionCount = 0
   let updatedVersionCount = 0
@@ -460,13 +529,18 @@ export async function importAdminAffiliateDashPricesBaseline(session: AdminSessi
 
 /**
  * 一键重算自动发货预估成本快照并重算销售事实。
+ * 默认统计最新的一个月（30 天）数据，避免扫描混乱的历史遗留数据。
  */
-export async function syncAdminAffiliateDashCostSnapshots() {
+export async function syncAdminAffiliateDashCostSnapshots(options: { days?: number } = {}) {
+  const days = options.days && options.days > 0 ? options.days : 30
+  const today = formatShanghaiDate(new Date())
+  const fromDate = shiftDate(today, -(days - 1))
+
   const snapshotResult = await syncEstimatedOrderItemCostSnapshots({
-    dateFrom: '2026-07-10',
-    dateTo: '2026-10-26',
+    dateFrom: fromDate,
+    dateTo: today,
   })
-  const factsResult = await rebuildFactsForRange('2026-07-10', '2026-10-26')
+  const factsResult = await rebuildFactsForRange(fromDate, today)
 
   return {
     snapshotResult,

apps/backend/src/services/cost-center/cost-center-constants.ts

@@ -119,11 +119,16 @@ export const PRODUCT_CATEGORY_JOINS = `
 export const AFFILIATE_TASK_AGG_JOIN = `
   LEFT JOIN (
     SELECT
-      btrim(context_json->>'skuCode') AS sku_code,
+      COALESCE(
+        NULLIF(btrim(context_json->'affiliateDash'->>'sku'), ''),
+        NULLIF(btrim(context_json->'affiliateDash'->'raw'->'product'->>'sku'), ''),
+        btrim(context_json->>'skuCode')
+      ) AS sku_code,
       COUNT(*)::int AS task_count,
       MAX(created_at) AS last_fulfilled_at
     FROM fulfillment_tasks
     WHERE executor_key = 'affiliate_dash'
+      AND created_at >= NOW() - INTERVAL '30 days'
       AND btrim(COALESCE(context_json->>'skuCode', '')) <> ''
     GROUP BY 1
   ) aff ON p.canonical_sku LIKE 'aff:%'

apps/backend/src/services/cost-center/product-catalog-service.ts

@@ -10,6 +10,7 @@ import { query } from '../../db/client.js'
 import type { AdminSession } from '../admin/admin-auth-service.js'
 import type { JsonObject } from '../../types/json.js'
 import { createHttpError } from '../../utils/http.js'
+import { formatShanghaiDate, shiftDate } from '../../utils/finance-date-range.js'
 import { nowIso } from '../../utils/time.js'
 import { normalizeMatchText, normalizeSkuSeriesEquivalence } from '../product-match/normalize.js'
 import { resolveSkuNameQuantity } from '../product-match/pricing.js'
@@ -1314,7 +1315,8 @@ export async function setAdminProductPrice(
   const now = nowIso()
   let effectiveFrom = payload.effectiveFrom ? String(payload.effectiveFrom) : now
   if (payload.retroactive === true || payload.retroactive === 'true') {
-    effectiveFrom = '2026-07-10T00:00:00+08:00'
+    const today = formatShanghaiDate(new Date())
+    effectiveFrom = shiftDate(today, -30) + 'T00:00:00+08:00'
   }
   const changeReason = String(payload.changeReason || '在商品与成本中心统一调整价格').trim()
 

apps/backend/src/services/scheduler/cost-finance-sync-job.ts

@@ -13,7 +13,7 @@ import { reconcilePendingFulfillmentModes } from '../fulfillment/fulfillment-mod
  */
 export async function runCostFinanceSyncJob(job: JsonObject = {}) {
   const config = job.config && typeof job.config === 'object' ? (job.config as JsonObject) : {}
-  const days = clampInteger(config.days, 7, 1, 90)
+  const days = clampInteger(config.days, 30, 1, 90)
 
   const fulfillment = await reconcilePendingFulfillmentModes({
     batchSize: 2000,

apps/backend/src/utils/finance-date-range.ts

@@ -16,7 +16,7 @@ type ResolveFinanceDateRangeOptions = {
  */
 export function resolveFinanceDateRange(
   input: Record<string, unknown> = {},
-  { defaultDays = 7, maxDays = 400 }: ResolveFinanceDateRangeOptions = {},
+  { defaultDays = 30, maxDays = 400 }: ResolveFinanceDateRangeOptions = {},
 ): FinanceDateRange {
   const today = formatShanghaiDate(new Date())
   const rawFrom = normalizeDateInput(input.dateFrom ?? input.from)