order_site/apps/backend/scripts/sync-affiliate-dash-prices.ts

/**
 * 按上游自动发货商品价格，批量生成「自动发货采购成本」成本版本。
 *
 * 用法：
 *   npx tsx scripts/sync-affiliate-dash-prices.ts [价格 JSON 路径] [--cost-item-id=5] [--effective-from=YYYY-MM-DD] [--dry-run]
 *
 * 规则：
 * - 标准商品编码固定为 `aff:<sku>`，载体 affiliate_dash；
 * - 成本版本范围 = 标准商品，金额 = 上游 cost_amount（分）；
 * - 同一商品已有相同金额的生效版本时跳过；指定 --effective-from 时，
 *   若现版本生效时间更晚，则把首次基线版本的生效时间回填到该日期；
 * - 金额变化时先停用旧版本，再插入新版本（历史快照不变）。
 */
import fs from 'node:fs'
import path from 'node:path'
import process from 'node:process'

import { closeDb, query } from '../src/db/client.js'

type PriceRow = {
  sku: string
  name: string
  category: string
  costAmountFen: number
}

type CostItemRow = {
  id: number
  code: string
  name: string
  pricing_mode: string
}

function parseArgs(argv: string[]) {
  let filePath = ''
  let costItemId: number | null = null
  let effectiveFrom = ''
  let dryRun = false
  for (const arg of argv) {
    if (arg === '--dry-run') {
      dryRun = true
      continue
    }
    if (arg.startsWith('--cost-item-id=')) {
      const parsed = Number(arg.slice('--cost-item-id='.length))
      if (Number.isFinite(parsed) && parsed > 0) costItemId = parsed
      continue
    }
    if (arg.startsWith('--effective-from=')) {
      effectiveFrom = arg.slice('--effective-from='.length).trim()
      continue
    }
    if (!arg.startsWith('--') && !filePath) filePath = arg
  }
  return {
    filePath: path.resolve(filePath || 'scripts/data/affiliate-dash-prices.json'),
    costItemId,
    effectiveFrom,
    dryRun,
  }
}

async function resolveCostItem(costItemId: number | null): Promise<CostItemRow> {
  if (costItemId) {
    const result = await query<CostItemRow>('SELECT * FROM cost_items WHERE id = $1', [costItemId])
    const row = result.rows[0]
    if (!row) throw new Error(`成本项不存在: ${costItemId}`)
    if (row.pricing_mode !== 'per_item') {
      throw new Error(`成本项 ${row.code} 的计价方式不是按件，不能导入按件采购价`)
    }
    return row
  }

  const existing = await query<CostItemRow>(
    `SELECT * FROM cost_items
     WHERE cost_model = 'affiliate_purchase' AND status = 'active'
     ORDER BY id LIMIT 1`,
  )
  if (existing.rows[0]) return existing.rows[0]

  const now = new Date().toISOString()
  const inserted = await query<CostItemRow>(
    `INSERT INTO cost_items (
       code, name, cost_model, pricing_mode, unit_name, currency, status, note, created_at, updated_at
     ) VALUES ('auto_purchase', '自动发货采购成本', 'affiliate_purchase', 'per_item', '件', 'CNY', 'active', '上游自动发货价格同步', $1, $1)
     ON CONFLICT (code) DO NOTHING
     RETURNING *`,
    [now],
  )
  if (inserted.rows[0]) return inserted.rows[0]

  const fallback = await query<CostItemRow>(
    "SELECT * FROM cost_items WHERE code = 'auto_purchase' LIMIT 1",
  )
  if (!fallback.rows[0]) throw new Error('无法创建自动发货采购成本项')
  return fallback.rows[0]
}

async function main() {
  const { filePath, costItemId, effectiveFrom, dryRun } = parseArgs(process.argv.slice(2))
  const effectiveFromIso = normalizeEffectiveFrom(effectiveFrom)
  const rows = JSON.parse(fs.readFileSync(filePath, 'utf8')) as PriceRow[]

  const costItem = await resolveCostItem(costItemId)
  console.log(
    `[affiliate-price] 成本项 #${costItem.id} ${costItem.name}（${costItem.code}）；价格条目 ${rows.length} 条` +
      `${effectiveFromIso ? `；生效时间 ${effectiveFromIso}` : ''}${dryRun ? '（dry-run）' : ''}`,
  )

  let createdProductCount = 0
  let updatedProductCount = 0
  let createdVersionCount = 0
  let retiredVersionCount = 0
  let backdatedVersionCount = 0
  let skippedCount = 0

  for (const row of rows) {
    const now = new Date().toISOString()
    const canonicalSku = `aff:${row.sku}`
    const unitName = row.name.includes('个') ? '个' : '件'

    if (dryRun) {
      const existingProduct = await query<{ id: number }>(
        'SELECT id FROM product_catalog WHERE canonical_sku = $1',
        [canonicalSku],
      )
      const existingId = existingProduct.rows[0]?.id
      if (!existingId) {
        createdProductCount += 1
        createdVersionCount += 1
        continue
      }
      updatedProductCount += 1
      const current = await query<{ amount_fen: number }>(
        `SELECT amount_fen FROM cost_versions
         WHERE cost_item_id = $1 AND scope_type = 'product' AND canonical_product_id = $2 AND status = 'active'
         ORDER BY effective_from DESC, id DESC LIMIT 1`,
        [costItem.id, existingId],
      )
      if (current.rows[0] && Number(current.rows[0].amount_fen) === row.costAmountFen) {
        skippedCount += 1
      } else {
        createdVersionCount += 1
      }
      continue
    }

    const productResult = await query<{ id: number; inserted: boolean }>(
      `INSERT INTO product_catalog (
         canonical_sku, canonical_name, category, unit_name, status,
         attributes_json, created_at, updated_at
       ) VALUES ($1, $2, $3, $4, 'active', $5::jsonb, $6, $6)
       ON CONFLICT (canonical_sku) DO UPDATE
       SET canonical_name = EXCLUDED.canonical_name,
           category = EXCLUDED.category,
           unit_name = EXCLUDED.unit_name,
           attributes_json = product_catalog.attributes_json || EXCLUDED.attributes_json,
           updated_at = $6
       RETURNING id, (xmax = 0) AS inserted`,
      [
        canonicalSku,
        row.name,
        row.category,
        unitName,
        JSON.stringify({
          source: 'affiliate_dash_price',
          skuCode: row.sku,
          priceFen: row.costAmountFen,
        }),
        now,
      ],
    )
    const product = productResult.rows[0]
    if (!product) throw new Error(`商品写入失败: ${canonicalSku}`)
    if (product.inserted) createdProductCount += 1
    else updatedProductCount += 1

    const currentResult = await query<{ id: number; amount_fen: number; effective_from: string }>(
      `SELECT id, amount_fen, effective_from FROM cost_versions
       WHERE cost_item_id = $1 AND scope_type = 'product' AND canonical_product_id = $2 AND status = 'active'
       ORDER BY effective_from DESC, id DESC LIMIT 1`,
      [costItem.id, product.id],
    )
    const current = currentResult.rows[0]
    if (current && Number(current.amount_fen) === row.costAmountFen) {
      if (
        effectiveFromIso &&
        new Date(current.effective_from).getTime() > new Date(effectiveFromIso).getTime()
      ) {
        await query(
          `UPDATE cost_versions
           SET effective_from = $1, change_reason = '上游基线价格回填', updated_at = $2
           WHERE id = $3`,
          [effectiveFromIso, now, current.id],
        )
        backdatedVersionCount += 1
        continue
      }
      skippedCount += 1
      continue
    }

    if (current) {
      const retired = await query(
        `UPDATE cost_versions
         SET status = 'retired', change_reason = '上游价格更新', updated_at = $1
         WHERE cost_item_id = $2 AND canonical_product_id = $3 AND status = 'active'`,
        [now, costItem.id, product.id],
      )
      retiredVersionCount += retired.rowCount || 0
    }

    await query(
      `INSERT INTO cost_versions (
         cost_item_id, scope_type, provider, shop_id, canonical_product_id, work_product_rule_id,
         amount_fen, rate_bps, effective_from, effective_to, source, status, change_reason,
         created_by_admin_user_id, approved_by_admin_user_id, created_at, updated_at
       ) VALUES ($1, 'product', '', '', $2, NULL, $3, 0, $4, NULL, 'upstream_sync', 'active',
         '上游自动发货价格同步', NULL, NULL, $5, $5)`,
      [costItem.id, product.id, row.costAmountFen, effectiveFromIso || now, now],
    )
    createdVersionCount += 1
  }

  console.log(
    `[affiliate-price] 完成：新增商品 ${createdProductCount}、更新商品 ${updatedProductCount}、` +
      `新增版本 ${createdVersionCount}、停用旧版本 ${retiredVersionCount}、` +
      `回填生效时间 ${backdatedVersionCount}、金额未变跳过 ${skippedCount}`,
  )
}

function normalizeEffectiveFrom(value: string) {
  if (!value) return ''
  if (!/^\d{4}-\d{2}-\d{2}$/.test(value)) {
    throw new Error('--effective-from 必须是 YYYY-MM-DD')
  }
  const parsed = new Date(`${value}T00:00:00+08:00`)
  if (!Number.isFinite(parsed.getTime())) {
    throw new Error('--effective-from 日期无效')
  }
  return parsed.toISOString()
}

main()
  .then(async () => {
    await closeDb()
  })
  .catch(async (error) => {
    console.error(error)
    await closeDb()
    process.exitCode = 1
  })