khhao 自动同步

2026-05-02 20:38:52 +08:00
parent 17b96dfef6
commit dc32eef08b
12 changed files with 502 additions and 10 deletions
@@ -8,6 +8,7 @@ import claimsRouter from './routes/claims.js'
 import webhooksRouter from './routes/webhooks.js'
 import { ensureAdminUsersBootstrapped } from './services/admin/admin-auth-service.js'
 import { ensureFulfillmentCatalogBootstrapped } from './services/bootstrap/fulfillment-bootstrap-service.js'
+import { startKhhaoAutoSyncLoop, stopKhhaoAutoSyncLoop } from './services/platforms/khhao/auto-sync-service.js'
 import { closeLocalOcrWorker, warmupLocalOcrWorker } from './services/session/ocr.js'
 import { closeAllTencentBrowserSessions, warmupTencentBrowser } from './services/session/session.js'
 import { buildSuccessPayload } from './utils/http.js'
@@ -160,6 +161,7 @@ async function bootstrapCoreServices() {
        attempt: startupState.core.attemptCount,
      })

+      startKhhaoAutoSyncLoop()
      void bootstrapBrowser()
      void bootstrapOcr()
      break
@@ -368,6 +370,12 @@ async function shutdown(signal) {
    logError('[shutdown]', 'failed to close local OCR worker', error)
  }

+  try {
+    stopKhhaoAutoSyncLoop()
+  } catch (error) {
+    logError('[shutdown]', 'failed to stop khhao auto sync loop', error)
+  }
+
  if (!server.listening) {
    return
  }
@@ -13,6 +13,7 @@ import { mapKhhaoOrderPreviewList } from '../platforms/khhao/order-mapper-servic
 import { queryKhhaoOrderList } from '../platforms/khhao/order-query-service.js'
 import { syncKhhaoOrders } from '../platforms/khhao/order-sync-service.js'
 import { getKhhaoSourceConfig, getKhhaoSourcesFilePath, saveKhhaoSourceConfig } from '../platforms/khhao/source-config-service.js'
+import { ensureKhhaoSyncBaseline, getKhhaoSyncState, getKhhaoSyncStateFilePath } from '../platforms/khhao/sync-state-service.js'
 import { loginKhhaoSession } from '../platforms/khhao/session-service.js'
 import { normalizeAgisoMessageTemplate } from '../platforms/agiso/xianyu/message-service.js'
 import {
@@ -186,15 +187,33 @@ function applyOptionalStringField(target, key, source) {

 export function getAdminKhhaoSourceConfig() {
  const config = getKhhaoSourceConfig()
+  const syncState = getKhhaoSyncState()

  return {
    filePath: getKhhaoSourcesFilePath(),
+    stateFilePath: getKhhaoSyncStateFilePath(),
    source: {
      enabled: config.enabled !== false,
      baseUrl: String(config.baseUrl || '').trim(),
      username: String(config.username || '').trim(),
      password: String(config.password || '').trim(),
      maxCaptchaAttempts: Number(config.maxCaptchaAttempts || 3) || 3,
+      autoSync: {
+        enabled: config.autoSync?.enabled === true,
+        intervalMinutes: Number(config.autoSync?.intervalMinutes || 3) || 3,
+        pages: Number(config.autoSync?.pages || 2) || 2,
+        pageSize: Number(config.autoSync?.pageSize || 20) || 20,
+      },
+    },
+    syncState: {
+      syncFromCreatedAt: String(syncState.syncFromCreatedAt || '').trim(),
+      lastRunStartedAt: String(syncState.lastRunStartedAt || '').trim(),
+      lastRunFinishedAt: String(syncState.lastRunFinishedAt || '').trim(),
+      lastRunStatus: String(syncState.lastRunStatus || '').trim(),
+      lastErrorMessage: String(syncState.lastErrorMessage || '').trim(),
+      fetchedCount: Number(syncState.fetchedCount || 0),
+      syncedCount: Number(syncState.syncedCount || 0),
+      ignoredCount: Number(syncState.ignoredCount || 0),
    },
  }
 }
@@ -207,16 +226,40 @@ export function updateAdminKhhaoSourceConfig(payload = /** @type {AdminKhhaoSour
    username: String(payload.username || '').trim(),
    password: String(payload.password || '').trim(),
    maxCaptchaAttempts: Number(payload.maxCaptchaAttempts || 3) || 3,
+    autoSync: {
+      enabled: payload.autoSync?.enabled === true,
+      intervalMinutes: Number(payload.autoSync?.intervalMinutes || 3) || 3,
+      pages: Number(payload.autoSync?.pages || 2) || 2,
+      pageSize: Number(payload.autoSync?.pageSize || 20) || 20,
+    },
  })
+  const syncState = saved.autoSync?.enabled ? ensureKhhaoSyncBaseline() : getKhhaoSyncState()

  return {
    filePath: getKhhaoSourcesFilePath(),
+    stateFilePath: getKhhaoSyncStateFilePath(),
    source: {
      enabled: saved.enabled !== false,
      baseUrl: String(saved.baseUrl || '').trim(),
      username: String(saved.username || '').trim(),
      password: String(saved.password || '').trim(),
      maxCaptchaAttempts: Number(saved.maxCaptchaAttempts || 3) || 3,
+      autoSync: {
+        enabled: saved.autoSync?.enabled === true,
+        intervalMinutes: Number(saved.autoSync?.intervalMinutes || 3) || 3,
+        pages: Number(saved.autoSync?.pages || 2) || 2,
+        pageSize: Number(saved.autoSync?.pageSize || 20) || 20,
+      },
+    },
+    syncState: {
+      syncFromCreatedAt: String(syncState.syncFromCreatedAt || '').trim(),
+      lastRunStartedAt: String(syncState.lastRunStartedAt || '').trim(),
+      lastRunFinishedAt: String(syncState.lastRunFinishedAt || '').trim(),
+      lastRunStatus: String(syncState.lastRunStatus || '').trim(),
+      lastErrorMessage: String(syncState.lastErrorMessage || '').trim(),
+      fetchedCount: Number(syncState.fetchedCount || 0),
+      syncedCount: Number(syncState.syncedCount || 0),
+      ignoredCount: Number(syncState.ignoredCount || 0),
    },
  }
 }
@@ -0,0 +1,139 @@
+// @ts-check
+
+import { getKhhaoSourceConfig } from './source-config-service.js'
+import { ensureKhhaoSyncBaseline, getKhhaoSyncState, saveKhhaoSyncState } from './sync-state-service.js'
+import { syncKhhaoOrders } from './order-sync-service.js'
+import { logError, logInfo, logWarn } from '../../../utils/logger.js'
+import { nowIso } from '../../../utils/time.js'
+
+let timer = null
+let running = false
+
+export function startKhhaoAutoSyncLoop() {
+  scheduleNextRun(5_000)
+}
+
+export function stopKhhaoAutoSyncLoop() {
+  if (timer) {
+    clearTimeout(timer)
+    timer = null
+  }
+}
+
+async function runKhhaoAutoSyncCycle() {
+  const source = getKhhaoSourceConfig()
+
+  if (!source.enabled || !source.autoSync?.enabled) {
+    scheduleNextRun(resolveIntervalMs(source.autoSync?.intervalMinutes))
+    return
+  }
+
+  if (running) {
+    logWarn('[khhao/auto-sync]', '检测到上一次自动同步仍在运行，本轮跳过')
+    scheduleNextRun(resolveIntervalMs(source.autoSync?.intervalMinutes))
+    return
+  }
+
+  if (!String(source.username || '').trim() || !String(source.password || '').trim()) {
+    logWarn('[khhao/auto-sync]', 'khhao 自动同步已启用，但缺少账号或密码')
+    scheduleNextRun(resolveIntervalMs(source.autoSync?.intervalMinutes))
+    return
+  }
+
+  running = true
+  const baselineState = ensureKhhaoSyncBaseline()
+  const startedAt = nowIso()
+
+  saveKhhaoSyncState({
+    ...baselineState,
+    lastRunStartedAt: startedAt,
+    lastRunStatus: 'running',
+    lastErrorMessage: '',
+  })
+
+  try {
+    const result = await syncKhhaoOrders({
+      baseUrl: source.baseUrl,
+      username: source.username,
+      password: source.password,
+      maxCaptchaAttempts: source.maxCaptchaAttempts,
+      pages: source.autoSync.pages,
+      limit: source.autoSync.pageSize,
+      createdAfter: baselineState.syncFromCreatedAt,
+    })
+
+    const latestCreatedAt = resolveLatestOrderCreatedAt(result.results)
+
+    saveKhhaoSyncState({
+      ...getKhhaoSyncState(),
+      syncFromCreatedAt: baselineState.syncFromCreatedAt,
+      lastRunStartedAt: startedAt,
+      lastRunFinishedAt: nowIso(),
+      lastRunStatus: 'success',
+      lastErrorMessage: '',
+      fetchedCount: result.fetchedCount,
+      syncedCount: result.syncedCount,
+      ignoredCount: result.ignoredCount,
+      lastOrderCreatedAt: latestCreatedAt,
+    })
+
+    logInfo('[khhao/auto-sync]', 'khhao 自动同步完成', {
+      fetchedCount: result.fetchedCount,
+      syncedCount: result.syncedCount,
+      ignoredCount: result.ignoredCount,
+      syncFromCreatedAt: baselineState.syncFromCreatedAt,
+    })
+  } catch (error) {
+    const message = error instanceof Error ? error.message : String(error || 'khhao 自动同步失败')
+    saveKhhaoSyncState({
+      ...getKhhaoSyncState(),
+      syncFromCreatedAt: baselineState.syncFromCreatedAt,
+      lastRunStartedAt: startedAt,
+      lastRunFinishedAt: nowIso(),
+      lastRunStatus: 'failed',
+      lastErrorMessage: message,
+    })
+    logError('[khhao/auto-sync]', 'khhao 自动同步失败', error)
+  } finally {
+    running = false
+    scheduleNextRun(resolveIntervalMs(source.autoSync?.intervalMinutes))
+  }
+}
+
+function scheduleNextRun(delayMs) {
+  stopKhhaoAutoSyncLoop()
+  timer = setTimeout(() => {
+    void runKhhaoAutoSyncCycle()
+  }, delayMs)
+}
+
+function resolveIntervalMs(value) {
+  const minutes = Number(value)
+  const normalized = Number.isInteger(minutes) && minutes > 0 ? minutes : 3
+  return normalized * 60 * 1000
+}
+
+function resolveLatestOrderCreatedAt(results = []) {
+  let latestText = ''
+  let latestTime = 0
+
+  for (const item of Array.isArray(results) ? results : []) {
+    const text = String(item?.rawOrderCreatedAt || item?.orderCreatedAt || '').trim()
+    const timestamp = parseKhhaoOrderCreatedAt(text)
+    if (Number.isFinite(timestamp) && timestamp > latestTime) {
+      latestTime = timestamp
+      latestText = new Date(timestamp).toISOString()
+    }
+  }
+
+  return latestText
+}
+
+function parseKhhaoOrderCreatedAt(value) {
+  const text = String(value || '').trim()
+  if (!text) {
+    return NaN
+  }
+
+  return Date.parse(`${text.replace(' ', 'T')}+08:00`)
+}
@@ -12,7 +12,9 @@ import { loginKhhaoSession } from './session-service.js'
 *   password?: string
 *   page?: number | string
 *   limit?: number | string
+ *   pages?: number | string
 *   maxCaptchaAttempts?: number | string
+ *   createdAfter?: string
 * }} [payload]
 */
 export async function syncKhhaoOrders(payload = {}) {
@@ -26,25 +28,58 @@ export async function syncKhhaoOrders(payload = {}) {

  const queryResult = await queryKhhaoOrderList({
    baseUrl: payload.baseUrl,
-    page: payload.page,
-    limit: payload.limit,
+    page: 1,
+    limit: resolvePageSize(payload.limit),
    session,
  })

+  const createdAfter = String(payload.createdAfter || '').trim()
+  const pages = Math.max(1, Math.min(Number(payload.pages || payload.page || 1) || 1, 5))
  const results = []
+  const allItems = [...queryResult.items]

-  for (const item of queryResult.items) {
+  for (let page = 2; page <= pages; page += 1) {
+    const nextPage = await queryKhhaoOrderList({
+      baseUrl: payload.baseUrl,
+      page,
+      limit: resolvePageSize(payload.limit),
+      session,
+    })
+    allItems.push(...nextPage.items)
+  }
+
+  for (const item of allItems) {
    const preview = mapKhhaoOrderPreview(item)
+
+    if (createdAfter && !isKhhaoOrderAfter(preview.orderCreatedAt, createdAfter)) {
+      results.push({
+        platformOrderId: preview.platformOrderId,
+        provider: 'khhao',
+        platform: preview.platform || 'unknown',
+        shopId: preview.shopId,
+        skuCode: preview.skuCode,
+        itemTitle: preview.itemTitle,
+        rawOrderCreatedAt: preview.orderCreatedAt,
+        ignored: true,
+        ignoreReason: 'before_sync_baseline',
+        orderId: null,
+        orderItemCount: 0,
+        taskCount: 0,
+      })
+      continue
+    }
+
    const sourceEvent = mapKhhaoOrderToSourceEvent(item)
    const upsertResult = await upsertOrderFromSource(sourceEvent, { sourceLabel: 'khhao-sync' })

-    results.push({
+      results.push({
      platformOrderId: preview.platformOrderId,
      provider: sourceEvent.provider,
      platform: sourceEvent.platform,
      shopId: sourceEvent.shopId,
      skuCode: preview.skuCode,
      itemTitle: preview.itemTitle,
+      rawOrderCreatedAt: preview.orderCreatedAt,
      ignored: Boolean(upsertResult.ignored),
      ignoreReason: String(upsertResult.ignoreReason || '').trim(),
      orderId: Number(upsertResult.order?.id || 0) || null,
@@ -55,12 +90,42 @@ export async function syncKhhaoOrders(payload = {}) {

  return {
    baseUrl: session.baseUrl,
-    page: queryResult.page,
-    limit: queryResult.limit,
+    page: 1,
+    limit: resolvePageSize(payload.limit),
    total: queryResult.total,
-    fetchedCount: queryResult.items.length,
+    fetchedCount: allItems.length,
    syncedCount: results.filter((item) => !item.ignored && item.orderId).length,
    ignoredCount: results.filter((item) => item.ignored).length,
    results,
  }
 }
+
+function resolvePageSize(value) {
+  const parsed = Number(value)
+  if (!Number.isInteger(parsed) || parsed <= 0) {
+    return 20
+  }
+
+  return Math.min(parsed, 50)
+}
+
+function isKhhaoOrderAfter(orderCreatedAt, baselineIso) {
+  const orderTime = parseKhhaoOrderCreatedAt(orderCreatedAt)
+  const baselineTime = Date.parse(String(baselineIso || '').trim())
+
+  if (!Number.isFinite(orderTime) || !Number.isFinite(baselineTime)) {
+    return true
+  }
+
+  return orderTime > baselineTime
+}
+
+function parseKhhaoOrderCreatedAt(value) {
+  const text = String(value || '').trim()
+  if (!text) {
+    return NaN
+  }
+
+  const isoText = text.replace(' ', 'T')
+  return Date.parse(`${isoText}+08:00`)
+}
@@ -44,6 +44,7 @@ function normalizeKhhaoSourceConfig(rawValue) {
    username: String(source.username || '').trim(),
    password: String(source.password || '').trim(),
    maxCaptchaAttempts: normalizePositiveInteger(source.maxCaptchaAttempts, 3),
+    autoSync: normalizeAutoSyncConfig(source.autoSync),
  }
 }

@@ -54,6 +55,18 @@ function createDefaultKhhaoSourceConfig() {
    username: '',
    password: '',
    maxCaptchaAttempts: 3,
+    autoSync: normalizeAutoSyncConfig({}),
+  }
+}
+
+function normalizeAutoSyncConfig(rawValue) {
+  const source = isPlainObject(rawValue) ? rawValue : {}
+
+  return {
+    enabled: typeof source.enabled === 'boolean' ? source.enabled : false,
+    intervalMinutes: normalizePositiveInteger(source.intervalMinutes, 3),
+    pages: normalizePositiveInteger(source.pages, 2),
+    pageSize: normalizePositiveInteger(source.pageSize, 20),
  }
 }

@@ -0,0 +1,92 @@
+// @ts-check
+
+import fs from 'node:fs'
+import path from 'node:path'
+
+import { PROJECT_ROOT } from '../../../config/runtime.js'
+import { nowIso } from '../../../utils/time.js'
+
+const KHHAO_SYNC_STATE_FILE_PATH = path.join(PROJECT_ROOT, 'data', 'khhao-sync-state.json')
+
+export function getKhhaoSyncStateFilePath() {
+  return KHHAO_SYNC_STATE_FILE_PATH
+}
+
+export function getKhhaoSyncState() {
+  return loadKhhaoSyncStateFromFile()
+}
+
+export function saveKhhaoSyncState(rawValue) {
+  const normalized = normalizeKhhaoSyncState(rawValue)
+  fs.mkdirSync(path.dirname(KHHAO_SYNC_STATE_FILE_PATH), { recursive: true })
+  fs.writeFileSync(KHHAO_SYNC_STATE_FILE_PATH, `${JSON.stringify(normalized, null, 2)}\n`, 'utf8')
+  return normalized
+}
+
+export function ensureKhhaoSyncBaseline() {
+  const current = getKhhaoSyncState()
+  if (String(current.syncFromCreatedAt || '').trim()) {
+    return current
+  }
+
+  return saveKhhaoSyncState({
+    ...current,
+    syncFromCreatedAt: nowIso(),
+  })
+}
+
+function loadKhhaoSyncStateFromFile() {
+  if (!fs.existsSync(KHHAO_SYNC_STATE_FILE_PATH)) {
+    return createDefaultKhhaoSyncState()
+  }
+
+  try {
+    return normalizeKhhaoSyncState(JSON.parse(fs.readFileSync(KHHAO_SYNC_STATE_FILE_PATH, 'utf8')))
+  } catch {
+    return createDefaultKhhaoSyncState()
+  }
+}
+
+function createDefaultKhhaoSyncState() {
+  return {
+    syncFromCreatedAt: '',
+    lastRunStartedAt: '',
+    lastRunFinishedAt: '',
+    lastRunStatus: 'idle',
+    lastErrorMessage: '',
+    fetchedCount: 0,
+    syncedCount: 0,
+    ignoredCount: 0,
+    lastOrderCreatedAt: '',
+  }
+}
+
+function normalizeKhhaoSyncState(rawValue) {
+  const source = isPlainObject(rawValue) ? rawValue : {}
+
+  return {
+    syncFromCreatedAt: String(source.syncFromCreatedAt || '').trim(),
+    lastRunStartedAt: String(source.lastRunStartedAt || '').trim(),
+    lastRunFinishedAt: String(source.lastRunFinishedAt || '').trim(),
+    lastRunStatus: normalizeStatus(source.lastRunStatus),
+    lastErrorMessage: String(source.lastErrorMessage || '').trim(),
+    fetchedCount: normalizeNonNegativeInteger(source.fetchedCount),
+    syncedCount: normalizeNonNegativeInteger(source.syncedCount),
+    ignoredCount: normalizeNonNegativeInteger(source.ignoredCount),
+    lastOrderCreatedAt: String(source.lastOrderCreatedAt || '').trim(),
+  }
+}
+
+function normalizeStatus(value) {
+  const normalized = String(value || '').trim()
+  return normalized || 'idle'
+}
+
+function normalizeNonNegativeInteger(value) {
+  const parsed = Number(value)
+  return Number.isInteger(parsed) && parsed >= 0 ? parsed : 0
+}
+
+function isPlainObject(value) {
+  return Object.prototype.toString.call(value) === '[object Object]'
+}
@@ -74,6 +74,12 @@ export {}
 *   username?: string
 *   password?: string
 *   maxCaptchaAttempts?: number | string
+ *   autoSync?: {
+ *     enabled?: boolean
+ *     intervalMinutes?: number | string
+ *     pages?: number | string
+ *     pageSize?: number | string
+ *   }
 * }} AdminKhhaoSourceConfigInput
 */