feat: reconcile workflow product ids daily

2026-09-17 16:14:14 +08:00 · 2026-09-17 16:14:14 +08:00 · d0cd7ab979
commit d0cd7ab979
parent 35a301a45e
12 changed files with 263 additions and 62 deletions
--- a/PRODUCT.md
+++ b/PRODUCT.md
@ -30,7 +30,7 @@ web
 - “进行中项目”项目明细按业务导出表展示审批标题、审批编号、需求人、需求部门、平台、品牌、类目、SPU、需求类型、标准时效、当前节点与负责人；需求人取流程发起人，需求部门按 HR 人员架构匹配。
 - 美术设计流程的明细开始时间取“品牌策划主管分配任务”节点创建时间，结束时间取“品牌策划主管审核”节点完成时间；内容视频流程的明细开始时间取已确认的“内容主管分配任务”节点创建时间，结束时间按规则分支取“编导审核”或“内容主管审核”节点完成时间。内容流程中多个主管分配节点、编导制作节点及普通领导者任务均不等同于最终审核；未知 activityId 不得仅凭负责人身份、AGREE 结果或完成状态推断为审核节点。
 - 美术设计申请流程的 `ding_WorkflowTask.activityId` 按已确认流程定义映射为精确中文节点名并写入本地快照；未知节点继续使用同步字段推断或明确回退，不改写上游只读数据。
- 钉钉流程快照每小时从只读报表库的 `ding_Workflow`、`ding_WorkflowTask`、`ding_WorkflowScheme` 联合增量同步，只接收已批准的美术设计申请和内容视频流程；三表复合游标写入现有同步状态并继续复用同一批生产快照数据。
+- 钉钉流程快照每小时从只读报表库的 `ding_Workflow`、`ding_WorkflowTask`、`ding_WorkflowScheme` 联合增量同步，只接收已批准的美术设计申请和内容视频流程；三表复合游标写入现有同步状态，并每 24 小时全量校准流程、品牌和商品 ID，修正上游未推进游标的历史表单改动。
 - “本月上新”只读取表单字段“需求类型”为“新品”的六品牌美术设计流程，不以流程标题关键词判断，也不计“优化”等其他需求类型；分子为本月整条流程已完结数量，分母为本月已完结数量与当前进行中数量之和；其中进行中流程直接复用“进行中项目”的美术设计流程状态、时间与当前品牌中心处理人条件。
 - “按时交付”只统计对应主管/规则审核节点在本月完成、审批状态为已完结且审批结论未拒绝的美术设计与内容视频流程；待计算流程不进入分母。详情页可在全部流程、美术设计流程和内容视频流程之间筛选，公式、数量与明细必须共用当前筛选口径。
 - “首轮通过”读取作品过稿数据，分子为 V1 版本审核通过的作品数，分母为当前已通过的作品数；详情页列出部门范围内所有含 V1 的作品，搜索只过滤明细，不改变顶部完整口径。
--- a/components/department-dashboard/PerformanceConversionDashboard.test.tsx
+++ b/components/department-dashboard/PerformanceConversionDashboard.test.tsx
@ -16,7 +16,11 @@ const emptyList = {
  page: 1,
  pageSize: 20,
  hasMore: false,
-  lastWorkflowSyncedAt: '2026-09-17T00:00:00.000Z',
+  workflowSync: {
+    status: 'ready' as const,
+    lastCompletedAt: '2099-09-17T01:00:00.000Z',
+    lastFullReconciledAt: '2099-09-17T00:00:00.000Z',
+  },
 }

 const workflowItem = {
@ -65,6 +69,33 @@ describe('PerformanceConversionDashboard filters', () => {
    expect(requestUrl).toContain('completedTo=2026-07-31')
    expect(window.location.search).toContain('completedFrom=2026-07-01')
    expect(window.location.search).toContain('completedTo=2026-07-31')
+    expect(screen.getByText('流程与商品 ID 每日增量同步')).toBeInTheDocument()
+    expect(screen.getByText('最近成功 2099/09/17 09:00')).toBeInTheDocument()
+  })
+
+  it.each([
+    ['syncing', '正在同步流程与商品 ID'],
+    ['error', '流程与商品 ID 同步异常'],
+    ['pending', '等待首次流程与商品 ID 同步'],
+  ] as const)('shows the %s workflow synchronization state', async (status, label) => {
+    global.fetch = jest.fn().mockResolvedValue({
+      ok: true,
+      json: async () => ({
+        success: true,
+        data: {
+          ...emptyList,
+          workflowSync: {
+            status,
+            lastCompletedAt: status === 'pending' ? null : '2099-09-17T01:00:00.000Z',
+            lastFullReconciledAt: status === 'pending' ? null : '2099-09-17T00:00:00.000Z',
+          },
+        },
+      }),
+    } as Response)
+
+    render(<PerformanceConversionDashboard />)
+
+    expect(await screen.findByText(label)).toBeInTheDocument()
  })

  it('opens each product ID on a separate detail page and preserves the list context', async () => {
--- a/components/department-dashboard/PerformanceConversionDashboard.tsx
+++ b/components/department-dashboard/PerformanceConversionDashboard.tsx
@ -4,15 +4,19 @@ import Link from 'next/link'
 import { FormEvent, useCallback, useEffect, useRef, useState } from 'react'
 import {
  Activity,
+  AlertTriangle,
  CalendarDays,
+  CheckCircle2,
  ChevronLeft,
  ChevronRight,
+  Clock3,
  RefreshCw,
  Search,
 } from 'lucide-react'
 import DepartmentShell from './DepartmentShell'
 import {
  BUSINESS_TREND_BRANDS,
+  type PerformanceConversionWorkflowSync,
  type PerformanceConversionWorkflowItem,
  type PerformanceConversionWorkflowList,
  isValidDateKey,
@ -43,6 +47,41 @@ function formatDate(value: string | null, includeTime = false) {
  }).format(date)
 }

+function WorkflowSyncStatus({ sync }: { sync: PerformanceConversionWorkflowSync }) {
+  const latestSuccess = sync.lastCompletedAt || sync.lastFullReconciledAt
+  const latestSuccessLabel = latestSuccess ? formatDate(latestSuccess, true) : null
+  const latestReconciliation = sync.lastFullReconciledAt || sync.lastCompletedAt
+  const isDelayed = sync.status === 'ready'
+    && latestReconciliation !== null
+    && Date.now() - new Date(latestReconciliation).getTime() > 30 * 60 * 60 * 1000
+
+  const view = sync.status === 'syncing'
+    ? { label: '正在同步流程与商品 ID', detail: latestSuccessLabel ? `上次成功 ${latestSuccessLabel}` : '首次同步进行中', tone: 'text-[#0050cb]', icon: RefreshCw, spin: true }
+    : sync.status === 'error'
+      ? { label: '流程与商品 ID 同步异常', detail: latestSuccessLabel ? `暂显示 ${latestSuccessLabel} 的数据` : '暂无可用同步记录', tone: 'text-amber-700', icon: AlertTriangle, spin: false }
+      : sync.status === 'pending'
+        ? { label: '等待首次流程与商品 ID 同步', detail: '后台任务尚未完成', tone: 'text-slate-500', icon: Clock3, spin: false }
+        : isDelayed
+          ? { label: '流程与商品 ID 每日校准延迟', detail: `最近增量成功 ${latestSuccessLabel}`, tone: 'text-amber-700', icon: AlertTriangle, spin: false }
+          : { label: '流程与商品 ID 每日增量同步', detail: latestSuccessLabel ? `最近成功 ${latestSuccessLabel}` : '等待首次成功记录', tone: 'text-emerald-700', icon: CheckCircle2, spin: false }
+  const Icon = view.icon
+
+  return (
+    <div
+      role="status"
+      aria-live="polite"
+      className={`min-w-0 text-left text-xs leading-5 lg:text-right ${view.tone}`}
+      title={latestSuccess ? `最近同步成功：${latestSuccessLabel}（北京时间）` : undefined}
+    >
+      <p className="inline-flex items-center gap-1.5 font-bold">
+        <Icon className={`h-3.5 w-3.5 shrink-0 ${view.spin ? 'animate-spin' : ''}`} aria-hidden="true" />
+        <span>{view.label}</span>
+      </p>
+      <p className="text-slate-500">{view.detail}</p>
+    </div>
+  )
+}
+
 function WorkflowTable({ items, detailHref }: {
  items: PerformanceConversionWorkflowItem[]
  detailHref: (item: PerformanceConversionWorkflowItem) => string
@ -178,7 +217,7 @@ export default function PerformanceConversionDashboard() {
          <aside aria-label="品牌导航" className="border-b border-slate-200 bg-slate-50 lg:border-b-0 lg:border-r"><div className="border-b border-slate-200 px-4 py-4"><p className="text-xs font-bold text-slate-500">美术申请流程</p><p className="mt-1 text-sm font-bold text-slate-900">六品牌</p></div><nav className="flex gap-1 overflow-x-auto p-2 lg:block lg:space-y-1 lg:overflow-visible" aria-label="选择品牌">{BUSINESS_TREND_BRANDS.map((brand) => { const count = listState.data?.brands.find((item) => item.brand === brand)?.count; return <button key={brand} type="button" aria-current={selectedBrand === brand ? 'page' : undefined} onClick={() => { const nextFilters = { ...appliedFilters, query: '' }; setSelectedBrand(brand); setQuery(''); setCompletedFrom(nextFilters.completedFrom); setCompletedTo(nextFilters.completedTo); setAppliedFilters(nextFilters); setFilterError(null); setPage(1); void loadList(brand, 1, nextFilters) }} className={`flex min-h-11 shrink-0 items-center gap-2 rounded-lg px-3 text-left text-sm font-bold focus-visible:outline-none focus-visible:ring-2 focus-visible:ring-[#1677ff] lg:w-full ${selectedBrand === brand ? 'bg-[#1677ff] text-white' : 'text-slate-600 hover:bg-slate-200/70 hover:text-slate-950'}`}><span className="min-w-0 flex-1 truncate">{brand}</span>{count !== undefined && <span className={`tabular-nums text-xs ${selectedBrand === brand ? 'text-white/80' : 'text-slate-400'}`}>{count}</span>}</button> })}</nav></aside>

          <section className="min-w-0 bg-white" aria-labelledby="performance-brand-title" aria-busy={listState.loading}>
-            <header className="flex flex-col gap-4 border-b border-slate-200 px-5 py-5 sm:px-7 lg:flex-row lg:items-end lg:justify-between"><div><p className="text-xs font-bold text-[#9a4d08]">当前品牌</p><h2 id="performance-brand-title" className="mt-1 text-xl font-bold tracking-normal text-slate-950">{selectedBrand}</h2><p className="mt-1 text-xs text-slate-500">{appliedFilters.completedFrom || appliedFilters.completedTo ? '区间内 ' : ''}{selectedBrandCount} 个有效流程商品 ID{listState.data?.invalidProductIdCount ? ` · ${listState.data.invalidProductIdCount} 条流程未填写有效 ID` : ''}</p></div><div className="text-xs leading-5 text-slate-500">流程同步 {formatDate(listState.data?.lastWorkflowSyncedAt || null, true)}</div></header>
+            <header className="flex flex-col gap-4 border-b border-slate-200 px-5 py-5 sm:px-7 lg:flex-row lg:items-end lg:justify-between"><div><p className="text-xs font-bold text-[#9a4d08]">当前品牌</p><h2 id="performance-brand-title" className="mt-1 text-xl font-bold tracking-normal text-slate-950">{selectedBrand}</h2><p className="mt-1 text-xs text-slate-500">{appliedFilters.completedFrom || appliedFilters.completedTo ? '区间内 ' : ''}{selectedBrandCount} 个有效流程商品 ID{listState.data?.invalidProductIdCount ? ` · ${listState.data.invalidProductIdCount} 条流程未填写有效 ID` : ''}</p></div>{listState.data && <WorkflowSyncStatus sync={listState.data.workflowSync} />}</header>
            <form onSubmit={submitSearch} className="grid gap-3 border-b border-slate-200 bg-slate-50 px-5 py-4 sm:px-7 md:grid-cols-[minmax(0,1fr)_auto] md:items-end xl:grid-cols-[minmax(280px,0.9fr)_minmax(220px,1fr)_auto]">
              <fieldset className="min-w-0 md:col-span-2 xl:col-span-1">
                <legend className="mb-1 text-xs font-bold text-slate-600">完结日期</legend>
--- a/components/department-dashboard/PerformanceConversionDetail.test.tsx
+++ b/components/department-dashboard/PerformanceConversionDetail.test.tsx
@ -104,6 +104,7 @@ describe('PerformanceConversionDetail', () => {
      expect.objectContaining({ cache: 'no-store' }),
    )
    expect(screen.getByRole('link', { name: '返回业绩转化' })).toHaveAttribute('href', returnHref)
+    expect(screen.getByText(/经营数据更新至/)).toBeInTheDocument()
    expect(screen.getAllByText('12.50%').length).toBeGreaterThan(0)
    expect(screen.getAllByText('¥1,299.50').length).toBeGreaterThan(0)
    expect(screen.getByRole('img', { name: /支付金额从/ })).toBeInTheDocument()
--- a/components/department-dashboard/PerformanceConversionDetail.tsx
+++ b/components/department-dashboard/PerformanceConversionDetail.tsx
@ -161,7 +161,7 @@ function DetailContent({ data }: { data: PerformanceConversionTimeseries }) {
          <dl className="grid shrink-0 gap-x-5 gap-y-2 text-left text-xs leading-5 text-slate-500 sm:grid-cols-3 lg:block lg:text-right">
            <div><dt className="sr-only">审批编号</dt><dd>审批编号 {data.workflow.approvalNumber || '—'}</dd></div>
            <div><dt className="sr-only">流程完结</dt><dd>流程完结 {formatDate(data.workflow.completedAt, true)}</dd></div>
-            <div><dt className="sr-only">数据更新</dt><dd>数据更新至 {formatDate(data.coverage.sourceLatestDate)}</dd></div>
+            <div><dt className="sr-only">经营数据更新</dt><dd>经营数据更新至 {formatDate(data.coverage.sourceLatestDate)}</dd></div>
          </dl>
        </div>
        {data.product.brandMatchesWorkflow === false && (
--- a/components/department-dashboard/performance-conversion.ts
+++ b/components/department-dashboard/performance-conversion.ts
@ -1,8 +1,13 @@
 import { BUSINESS_TREND_BRANDS, type BusinessTrendBrandName } from './business-trends'
-
 export { BUSINESS_TREND_BRANDS }
 export type PerformanceConversionBrand = BusinessTrendBrandName

+export type PerformanceConversionWorkflowSync = {
+  status: 'pending' | 'syncing' | 'ready' | 'error'
+  lastCompletedAt: string | null
+  lastFullReconciledAt: string | null
+}
+
 export const PERFORMANCE_CONVERSION_METRICS = [
  { key: 'impressions', label: '展现量', group: '流量', format: 'integer' },
  { key: 'visitors', label: '商品访客数', group: '流量', format: 'integer' },
@ -38,7 +43,7 @@ export type PerformanceConversionWorkflowList = {
  page: number
  pageSize: number
  hasMore: boolean
-  lastWorkflowSyncedAt: string | null
+  workflowSync: PerformanceConversionWorkflowSync
 }

 export type PerformanceConversionDailyMetric = {
--- a/instrumentation.ts
+++ b/instrumentation.ts
@ -120,12 +120,12 @@ export async function register() {
      console.error('[Instrumentation] 调度组织架构每日同步任务失败:', error)
    }

-    // 钉钉审批流程快照：启动后建立基线，之后每小时增量同步。
+    // 钉钉审批流程快照：每小时增量同步，并每 24 小时全量校准流程与商品 ID。
    // 看板只读本地 PostgreSQL 快照，避免用户请求直接扫描上游无索引大表。
    try {
      const { ensureHourlyDingWorkflowSyncStarted } = await import('./lib/ding-workflow-sync')
      ensureHourlyDingWorkflowSyncStarted()
-      console.log('[Instrumentation] 钉钉流程每小时增量同步任务已调度')
+      console.log('[Instrumentation] 钉钉流程增量同步与每日全量校准任务已调度')
    } catch (error) {
      console.error('[Instrumentation] 调度钉钉流程每小时同步任务失败:', error)
    }
--- a/lib/tests/ding-workflow-sync.test.ts
+++ b/lib/tests/ding-workflow-sync.test.ts
@ -28,7 +28,9 @@ import {
  parseDingWorkflowSourceCursor,
  resolveDingWorkflowInitiator,
  resolveDingWorkflowDetailMilestones,
+  selectLatestDingWorkflowSchemeValue,
  serializeDingWorkflowSourceCursor,
+  shouldRunDingWorkflowDailyReconciliation,
  sourceWorkflowTimestampToIso,
  storedWorkflowTimestampToIso,
 } from '@/lib/ding-workflow-sync'
@ -182,29 +184,32 @@ describe('ding workflow snapshot sync semantics', () => {

  it('upgrades the legacy workflow-only cursor to a three-table cursor', () => {
    expect(parseDingWorkflowSourceCursor('2026-08-26 19:42:36')).toEqual({
-      version: 3,
+      version: 4,
      workflowUpdatedAt: '2026-08-26 19:42:36',
      taskId: null,
      schemeId: null,
+      lastFullReconciledAt: null,
    })
  })

-  it('round-trips the Workflow, Task and Scheme incremental cursors', () => {
+  it('round-trips the Workflow, Task and Scheme cursors with the daily reconciliation watermark', () => {
    const cursor = {
-      version: 3 as const,
+      version: 4 as const,
      workflowUpdatedAt: '2026-08-26 19:42:36',
      taskId: 2248339,
      schemeId: 10634556,
+      lastFullReconciledAt: '2026-09-16T18:00:00.000Z',
    }
    expect(parseDingWorkflowSourceCursor(serializeDingWorkflowSourceCursor(cursor))).toEqual(cursor)
  })

  it('preserves empty Task and Scheme cursors so the next run performs alignment', () => {
    const cursor = {
-      version: 3 as const,
+      version: 4 as const,
      workflowUpdatedAt: '2026-08-26 19:42:36',
      taskId: null,
      schemeId: null,
+      lastFullReconciledAt: null,
    }
    expect(parseDingWorkflowSourceCursor(serializeDingWorkflowSourceCursor(cursor))).toEqual(cursor)
  })
@ -216,27 +221,46 @@ describe('ding workflow snapshot sync semantics', () => {
      taskId: 2248339,
      schemeId: 10634556,
    }))).toEqual({
-      version: 3,
+      version: 4,
      workflowUpdatedAt: '2026-08-26 19:42:36',
      taskId: null,
      schemeId: null,
+      lastFullReconciledAt: null,
    })
  })

-  it('rejects malformed version 3 cursor ids and safely requests re-alignment', () => {
+  it('upgrades version 3 cursors and requests the first daily reconciliation', () => {
    expect(parseDingWorkflowSourceCursor(JSON.stringify({
      version: 3,
      workflowUpdatedAt: '2026-08-26 19:42:36',
      taskId: -1,
      schemeId: 'invalid',
    }))).toEqual({
-      version: 3,
+      version: 4,
      workflowUpdatedAt: '2026-08-26 19:42:36',
      taskId: null,
      schemeId: null,
+      lastFullReconciledAt: null,
    })
  })

+  it('runs a full reconciliation at least once every 24 hours', () => {
+    const now = new Date('2026-09-17T02:00:00.000Z')
+    expect(shouldRunDingWorkflowDailyReconciliation(null, now)).toBe(true)
+    expect(shouldRunDingWorkflowDailyReconciliation('2026-09-16T02:00:01.000Z', now)).toBe(false)
+    expect(shouldRunDingWorkflowDailyReconciliation('2026-09-16T02:00:00.000Z', now)).toBe(true)
+    expect(shouldRunDingWorkflowDailyReconciliation('invalid', now)).toBe(true)
+  })
+
+  it('selects the latest non-empty Scheme row across product ID aliases', () => {
+    expect(selectLatestDingWorkflowSchemeValue([
+      { sourceRowId: 101, name: '商品ID', value: '100000001' },
+      { sourceRowId: 109, name: '商品Id', value: '100000009' },
+      { sourceRowId: 110, name: '商品id', value: ' ' },
+      { sourceRowId: 108, name: '品牌', value: '亿家达' },
+    ], ['商品ID', '商品Id', '商品id'])).toBe('100000009')
+  })
+
  it('only queries ongoing workflows whose current assignee belongs to Brand Center', () => {
    expect(buildBrandCenterOngoingWorkflowWhere()).toEqual({
      isOngoing: true,
--- a/lib/tests/performance-conversion-workflow-filter.test.ts
+++ b/lib/tests/performance-conversion-workflow-filter.test.ts
@ -19,6 +19,11 @@ jest.mock('@/lib/prisma', () => ({
 jest.mock('@/lib/ding-workflow-sync', () => ({
  DING_WORKFLOW_ALLOWED_PROCESS_CODES: ['art-process'],
  DING_WORKFLOW_STARTED_AT_CUTOFF_ISO: '2026-03-31T16:00:00.000Z',
+  getDingWorkflowSyncMetadata: jest.fn().mockResolvedValue({
+    status: 'ready',
+    lastCompletedAt: '2026-09-17T01:30:00.000Z',
+    lastFullReconciledAt: '2026-09-17T00:00:00.000Z',
+  }),
  isDingWorkflowApprovedCompletion: jest.fn(() => true),
 }))

@ -70,6 +75,11 @@ describe('performance conversion completion date filters', () => {
    expect(result.invalidProductIdCount).toBe(1)
    expect(result.brands.find((item) => item.brand === '亿家达')?.count).toBe(2)
    expect(result.brands.find((item) => item.brand === '卓禾')?.count).toBe(1)
+    expect(result.workflowSync).toEqual({
+      status: 'ready',
+      lastCompletedAt: '2026-09-17T01:30:00.000Z',
+      lastFullReconciledAt: '2026-09-17T00:00:00.000Z',
+    })
  })

  it('rejects invalid or reversed date ranges before reading snapshots', async () => {
--- a/lib/ding-workflow-sync.ts
+++ b/lib/ding-workflow-sync.ts
@ -31,9 +31,9 @@ export type {
  TeamCapacityWorkflowRow,
 } from '@/lib/ding-workflow-team-capacity'

-// v4 forces one full refresh so historical snapshots also receive the newly
-// collected 商品ID/现商品ID fields used by the performance conversion view.
-const SYNC_STATE_ID = 'ding-workflow-hourly-v4-product-id'
+// Keep this persisted key stable: changing it would discard the production
+// cursor and force another baseline scan of the upstream tables.
+export const DING_WORKFLOW_SYNC_STATE_ID = 'ding-workflow-hourly-v4-product-id'
 const TERMINAL_WORKFLOW_STATUSES = new Set(['COMPLETED', 'TERMINATED', 'CANCELED'])
 const REJECTED_WORKFLOW_DECISIONS = new Set([
  'REFUSE',
@ -50,6 +50,7 @@ const REJECTED_WORKFLOW_DECISIONS = new Set([
 const LEASE_DURATION_MS = 20 * 60 * 1000
 const DEFAULT_SYNC_INTERVAL_MS = 60 * 60 * 1000
 const INITIAL_SYNC_DELAY_MS = 5_000
+const DAILY_RECONCILIATION_INTERVAL_MS = 24 * 60 * 60 * 1000
 const UPSERT_BATCH_SIZE = 750

 export const DING_WORKFLOW_STARTED_AT_CUTOFF_LOCAL = '2026-04-01 00:00:00'
@ -235,19 +236,24 @@ type SourceSchemeAggregate = {
  schemeFieldCount: unknown
 }

-type SourceSchemeField = {
-  processInstanceId: unknown
-  componentType: unknown
+export type DingWorkflowSchemeValueRow = {
  name: unknown
  value: unknown
+  sourceRowId?: unknown
+}
+
+type SourceSchemeField = DingWorkflowSchemeValueRow & {
+  processInstanceId: unknown
+  componentType: unknown
  extValue: unknown
 }

 type DingWorkflowSourceCursor = {
-  version: 3
+  version: 4
  workflowUpdatedAt: string | null
  taskId: number | null
  schemeId: number | null
+  lastFullReconciledAt: string | null
 }

 type SourceDeltaState = {
@ -642,13 +648,31 @@ function extractContactAssignments(fields: SourceSchemeField[]): ContactAssignme
  ])).values())
 }

+export function selectLatestDingWorkflowSchemeValue(
+  fields: DingWorkflowSchemeValueRow[],
+  names: string[],
+): string | null {
+  const acceptedNames = new Set(names)
+  const candidates = fields.flatMap((field, index) => {
+    const name = clean(field.name)
+    const value = clean(field.value)
+    if (!name || !acceptedNames.has(name) || !value) return []
+    const sourceRowId = normalizeSourceCursorId(field.sourceRowId)
+    return [{ value, sourceRowId, index }]
+  })
+  candidates.sort((left, right) => {
+    if (left.sourceRowId !== null && right.sourceRowId !== null && left.sourceRowId !== right.sourceRowId) {
+      return right.sourceRowId - left.sourceRowId
+    }
+    if (left.sourceRowId !== null) return -1
+    if (right.sourceRowId !== null) return 1
+    return left.index - right.index
+  })
+  return candidates[0]?.value || null
+}
+
 function firstSchemeValue(fields: SourceSchemeField[], names: string[]): string | null {
-  for (const name of names) {
-    const value = fields.find((field) => clean(field.name) === name && clean(field.value))?.value
-    const normalized = clean(value)
-    if (normalized) return normalized
-  }
-  return null
+  return selectLatestDingWorkflowSchemeValue(fields, names)
 }

 function collectSchemeDisplayValues(value: unknown): string[] {
@ -1309,43 +1333,75 @@ function normalizeSourceCursorId(value: unknown): number | null {
 /**
 * 兼容旧版只保存 ding_Workflow.updateAt 的游标。
 *
- * v3 新增交付时效所需表单字段；读取旧版 v2 时保留主表游标，但清空
- * Task / Scheme 游标触发一次全量对齐，之后继续按三表复合游标增量同步。
- * 后续每小时只读取主表变化和 Task / Scheme 新增版本所关联的流程。
+ * v4 在三表复合游标之外记录最近一次全量校准时间。小时增量继续负责
+ * 快速更新，24 小时校准负责修正上游原地修改但未推进任何游标的历史表单。
 */
 export function parseDingWorkflowSourceCursor(raw: string | null): DingWorkflowSourceCursor {
  const normalized = clean(raw)
  if (!normalized) {
-    return { version: 3, workflowUpdatedAt: null, taskId: null, schemeId: null }
+    return {
+      version: 4,
+      workflowUpdatedAt: null,
+      taskId: null,
+      schemeId: null,
+      lastFullReconciledAt: null,
+    }
  }
  try {
    const parsed = JSON.parse(normalized) as Record<string, unknown>
-    if (parsed.version === 3) {
+    if (parsed.version === 4) {
      return {
-        version: 3,
+        version: 4,
        workflowUpdatedAt: clean(parsed.workflowUpdatedAt),
        taskId: normalizeSourceCursorId(parsed.taskId),
        schemeId: normalizeSourceCursorId(parsed.schemeId),
+        lastFullReconciledAt: clean(parsed.lastFullReconciledAt),
+      }
+    }
+    if (parsed.version === 3) {
+      return {
+        version: 4,
+        workflowUpdatedAt: clean(parsed.workflowUpdatedAt),
+        taskId: normalizeSourceCursorId(parsed.taskId),
+        schemeId: normalizeSourceCursorId(parsed.schemeId),
+        lastFullReconciledAt: null,
      }
    }
    if (parsed.version === 2) {
      return {
-        version: 3,
+        version: 4,
        workflowUpdatedAt: clean(parsed.workflowUpdatedAt),
        taskId: null,
        schemeId: null,
+        lastFullReconciledAt: null,
      }
    }
  } catch {
    // Legacy cursor is a plain ding_Workflow.updateAt string.
  }
-  return { version: 3, workflowUpdatedAt: normalized, taskId: null, schemeId: null }
+  return {
+    version: 4,
+    workflowUpdatedAt: normalized,
+    taskId: null,
+    schemeId: null,
+    lastFullReconciledAt: null,
+  }
 }

 export function serializeDingWorkflowSourceCursor(cursor: DingWorkflowSourceCursor): string {
  return JSON.stringify(cursor)
 }

+export function shouldRunDingWorkflowDailyReconciliation(
+  lastFullReconciledAt: string | null,
+  now = new Date(),
+) {
+  if (!lastFullReconciledAt) return true
+  const previous = new Date(lastFullReconciledAt)
+  if (Number.isNaN(previous.getTime())) return true
+  return now.getTime() - previous.getTime() >= DAILY_RECONCILIATION_INTERVAL_MS
+}
+
 export function isReportWorkflowDbConfigured() {
  return Boolean(
    process.env.REPORT_DB_SERVER
@ -1605,6 +1661,7 @@ async function fetchWorkflowAggregates(pool: ConnectionPool, processInstanceIds:
    )
    SELECT
      CONVERT(nvarchar(255), scheme.processInstanceId) AS processInstanceId,
+      scheme.id AS sourceRowId,
      CONVERT(nvarchar(500), scheme.componentType) AS componentType,
      CONVERT(nvarchar(2000), scheme.name) AS name,
      CONVERT(nvarchar(max), scheme.value) AS value,
@ -1622,7 +1679,8 @@ async function fetchWorkflowAggregates(pool: ConnectionPool, processInstanceIds:
        '数量', '需求数量', '拍摄数量', '摄像数量', '摄影数量',
        '策划任务统计', '平面任务统计', '出图任务统计'
      )
-      OR CONVERT(nvarchar(500), scheme.componentType) IN ('InnerContactField', 'TableField');
+      OR CONVERT(nvarchar(500), scheme.componentType) IN ('InnerContactField', 'TableField')
+    ORDER BY scheme.processInstanceId, scheme.id DESC;
  `)

  const taskRows = (result.recordsets?.[0] || []) as SourceTaskAggregate[]
@ -1920,13 +1978,13 @@ async function upsertSnapshotBatch(rows: WorkflowSnapshotPayload[], syncedAt: Da

 async function acquireSyncLease(leaseOwner: string, startedAt: Date) {
  await prisma.dingWorkflowSyncState.upsert({
-    where: { id: SYNC_STATE_ID },
-    create: { id: SYNC_STATE_ID },
+    where: { id: DING_WORKFLOW_SYNC_STATE_ID },
+    create: { id: DING_WORKFLOW_SYNC_STATE_ID },
    update: {},
  })
  const claimed = await prisma.dingWorkflowSyncState.updateMany({
    where: {
-      id: SYNC_STATE_ID,
+      id: DING_WORKFLOW_SYNC_STATE_ID,
      OR: [
        { leaseExpiresAt: null },
        { leaseExpiresAt: { lt: startedAt } },
@ -1942,7 +2000,16 @@ async function acquireSyncLease(leaseOwner: string, startedAt: Date) {
    },
  })
  if (claimed.count !== 1) return null
-  return prisma.dingWorkflowSyncState.findUniqueOrThrow({ where: { id: SYNC_STATE_ID } })
+  return prisma.dingWorkflowSyncState.findUniqueOrThrow({ where: { id: DING_WORKFLOW_SYNC_STATE_ID } })
+}
+
+async function renewSyncLease(leaseOwner: string) {
+  const now = new Date()
+  const renewed = await prisma.dingWorkflowSyncState.updateMany({
+    where: { id: DING_WORKFLOW_SYNC_STATE_ID, leaseOwner },
+    data: { leaseExpiresAt: new Date(now.getTime() + LEASE_DURATION_MS) },
+  })
+  if (renewed.count !== 1) throw new Error('钉钉流程同步租约已失效，本轮停止写入')
 }

 export async function syncDingWorkflowSnapshots() {
@ -1960,6 +2027,8 @@ export async function syncDingWorkflowSnapshots() {
    await pool.connect()
    const sourceCursor = parseDingWorkflowSourceCursor(state.sourceCursorRaw)
    const sourceDelta = await fetchSourceDeltaState(pool, sourceCursor)
+    const performFullReconciliation = sourceDelta.requiresFullSync
+      || shouldRunDingWorkflowDailyReconciliation(sourceCursor.lastFullReconciledAt, startedAt)
    const knownOngoingRows = await prisma.dingWorkflowSnapshot.findMany({
      where: {
        isOngoing: true,
@ -1970,10 +2039,11 @@ export async function syncDingWorkflowSnapshots() {
    const knownOngoingWorkflowIds = knownOngoingRows.map((row) => row.processInstanceId)
    const sourceRows = await fetchSourceWorkflows(
      pool,
-      sourceDelta.requiresFullSync ? null : sourceCursor.workflowUpdatedAt,
+      performFullReconciliation ? null : sourceCursor.workflowUpdatedAt,
      sourceDelta.changedWorkflowIds,
      knownOngoingWorkflowIds,
    )
+    await renewSyncLease(leaseOwner)
    const workflowIds = sourceRows.flatMap((row) => {
      const id = clean(row.processInstanceId)
      return id ? [id] : []
@ -1984,6 +2054,7 @@ export async function syncDingWorkflowSnapshots() {
      schemeByWorkflow,
      schemeRowsByWorkflow,
    } = await fetchWorkflowAggregates(pool, workflowIds)
+    await renewSyncLease(leaseOwner)
    const employeeDirectory = await readEmployeeDirectory(sourceRows, taskRowsByWorkflow)
    const activityNodeNames = inferActivityNodeNames(sourceRows, taskRowsByWorkflow, schemeRowsByWorkflow)
    const rows = buildSnapshotPayload(
@ -1999,6 +2070,7 @@ export async function syncDingWorkflowSnapshots() {

    for (let index = 0; index < rows.length; index += UPSERT_BATCH_SIZE) {
      await upsertSnapshotBatch(rows.slice(index, index + UPSERT_BATCH_SIZE), syncedAt)
+      await renewSyncLease(leaseOwner)
    }

    const deleted = await prisma.dingWorkflowSnapshot.deleteMany({
@ -2015,13 +2087,16 @@ export async function syncDingWorkflowSnapshots() {
      where: buildBrandCenterOngoingWorkflowWhere(),
    })
    const sourceCursorRaw = serializeDingWorkflowSourceCursor({
-      version: 3,
+      version: 4,
      workflowUpdatedAt: maxSourceCursor(rows, sourceCursor.workflowUpdatedAt),
      taskId: sourceDelta.maxTaskId,
      schemeId: sourceDelta.maxSchemeId,
+      lastFullReconciledAt: performFullReconciliation
+        ? syncedAt.toISOString()
+        : sourceCursor.lastFullReconciledAt,
    })
    await prisma.dingWorkflowSyncState.updateMany({
-      where: { id: SYNC_STATE_ID, leaseOwner },
+      where: { id: DING_WORKFLOW_SYNC_STATE_ID, leaseOwner },
      data: {
        status: 'ready',
        sourceCursorRaw,
@ -2037,7 +2112,7 @@ export async function syncDingWorkflowSnapshots() {
  } catch (error) {
    const message = error instanceof Error ? error.message : String(error)
    await prisma.dingWorkflowSyncState.updateMany({
-      where: { id: SYNC_STATE_ID, leaseOwner },
+      where: { id: DING_WORKFLOW_SYNC_STATE_ID, leaseOwner },
      data: {
        status: 'error',
        lastError: message.slice(0, 2_000),
@ -2058,13 +2133,32 @@ function normalizeSyncStatus(state: { status: string; lastCompletedAt: Date | nu
  return 'ready'
 }

+export type DingWorkflowSyncMetadata = {
+  status: DingWorkflowSyncStatus
+  lastCompletedAt: string | null
+  lastFullReconciledAt: string | null
+}
+
+export async function getDingWorkflowSyncMetadata(): Promise<DingWorkflowSyncMetadata> {
+  const state = await prisma.dingWorkflowSyncState.findUnique({
+    where: { id: DING_WORKFLOW_SYNC_STATE_ID },
+    select: { status: true, lastCompletedAt: true, sourceCursorRaw: true },
+  })
+  const sourceCursor = parseDingWorkflowSourceCursor(state?.sourceCursorRaw || null)
+  return {
+    status: normalizeSyncStatus(state),
+    lastCompletedAt: state?.lastCompletedAt?.toISOString() || null,
+    lastFullReconciledAt: sourceCursor.lastFullReconciledAt,
+  }
+}
+
 async function readSummaryWithoutAccessCheck(): Promise<OngoingDingWorkflowSummary> {
  const [total, state] = await prisma.$transaction([
    prisma.dingWorkflowSnapshot.count({
      where: buildBrandCenterOngoingWorkflowWhere(),
    }),
    prisma.dingWorkflowSyncState.findUnique({
-      where: { id: SYNC_STATE_ID },
+      where: { id: DING_WORKFLOW_SYNC_STATE_ID },
      select: { status: true, lastCompletedAt: true },
    }),
  ])
@ -2333,7 +2427,7 @@ async function readBusinessTrendsInsightsWithoutAccessCheck(
      ],
    }),
    prisma.dingWorkflowSyncState.findUnique({
-      where: { id: SYNC_STATE_ID },
+      where: { id: DING_WORKFLOW_SYNC_STATE_ID },
      select: { status: true, lastCompletedAt: true },
    }),
  ])
@ -2392,7 +2486,7 @@ async function readTeamCapacityInsightsWithoutAccessCheck(
      ],
    }),
    prisma.dingWorkflowSyncState.findUnique({
-      where: { id: SYNC_STATE_ID },
+      where: { id: DING_WORKFLOW_SYNC_STATE_ID },
      select: { status: true, lastCompletedAt: true },
    }),
  ])
@ -2542,7 +2636,7 @@ async function readMonthlyLaunchInsightsWithoutAccessCheck(
      select: MONTHLY_LAUNCH_WORKFLOW_SELECT,
    }),
    prisma.dingWorkflowSyncState.findUnique({
-      where: { id: SYNC_STATE_ID },
+      where: { id: DING_WORKFLOW_SYNC_STATE_ID },
      select: { status: true, lastCompletedAt: true },
    }),
  ])
@ -2632,7 +2726,7 @@ async function readOnTimeDeliveryInsightsWithoutAccessCheck(
      ],
    }),
    prisma.dingWorkflowSyncState.findUnique({
-      where: { id: SYNC_STATE_ID },
+      where: { id: DING_WORKFLOW_SYNC_STATE_ID },
      select: { status: true, lastCompletedAt: true },
    }),
  ])
@ -2732,7 +2826,7 @@ export function ensureHourlyDingWorkflowSyncStarted() {

  const configuredHours = Number(process.env.DING_WORKFLOW_SYNC_INTERVAL_HOURS || 1)
  const intervalMs = Number.isFinite(configuredHours) && configuredHours > 0
-    ? Math.max(5 * 60 * 1000, configuredHours * 60 * 60 * 1000)
+    ? Math.min(DAILY_RECONCILIATION_INTERVAL_MS, Math.max(5 * 60 * 1000, configuredHours * 60 * 60 * 1000))
    : DEFAULT_SYNC_INTERVAL_MS
  const runOnce = () => {
    void syncDingWorkflowSnapshots()
@ -2750,5 +2844,5 @@ export function ensureHourlyDingWorkflowSyncStarted() {
  initialTimer.unref?.()
  const interval = setInterval(runOnce, intervalMs)
  interval.unref?.()
-  console.info(`[ding-workflow-sync] 已调度每小时增量同步，周期 ${Math.round(intervalMs / 60_000)} 分钟`)
+  console.info(`[ding-workflow-sync] 已调度流程增量同步，周期 ${Math.round(intervalMs / 60_000)} 分钟，并保证每 24 小时全量校准流程与商品 ID`)
 }
--- a/lib/performance-conversion.ts
+++ b/lib/performance-conversion.ts
@ -5,6 +5,7 @@ import { prisma } from '@/lib/prisma'
 import {
  DING_WORKFLOW_ALLOWED_PROCESS_CODES,
  DING_WORKFLOW_STARTED_AT_CUTOFF_ISO,
+  getDingWorkflowSyncMetadata,
  isDingWorkflowApprovedCompletion,
 } from '@/lib/ding-workflow-sync'
 import {
@ -49,7 +50,6 @@ type SnapshotRow = {
  brand: string | null
  sourceFinishedAtRaw: string | null
  formFields: unknown
-  syncedAt: Date
 }

 type SourceMetricRow = {
@ -138,7 +138,6 @@ async function readEligibleSnapshots(): Promise<SnapshotRow[]> {
      brand: true,
      sourceFinishedAtRaw: true,
      formFields: true,
-      syncedAt: true,
    },
    orderBy: [
      { sourceFinishedAtRaw: 'desc' },
@ -172,7 +171,10 @@ export async function getPerformanceConversionWorkflows(
  if (completedFrom && completedTo && compareDateKeys(completedFrom, completedTo) > 0) {
    throw new PerformanceConversionError('完结开始日期不能晚于结束日期')
  }
-  const snapshots = await readEligibleSnapshots()
+  const [snapshots, workflowSync] = await Promise.all([
+    readEligibleSnapshots(),
+    getDingWorkflowSyncMetadata(),
+  ])
  const rangeSnapshots = snapshots.filter((row) => {
    const completedDate = row.sourceFinishedAtRaw ? toShanghaiDateKey(row.sourceFinishedAtRaw) : null
    if (!completedDate) return false
@ -195,11 +197,6 @@ export async function getPerformanceConversionWorkflows(
    item.title,
  ].some((value) => String(value || '').toLocaleLowerCase('zh-CN').includes(query))))
  const offset = (page - 1) * pageSize
-  const latestSyncedAt = snapshots.reduce<Date | null>(
-    (latest, row) => !latest || row.syncedAt > latest ? row.syncedAt : latest,
-    null,
-  )
-
  return {
    brands: BUSINESS_TREND_BRANDS.map((item) => ({ brand: item, count: brandCounts.get(item) || 0 })),
    items: filtered.slice(offset, offset + pageSize),
@ -208,7 +205,7 @@ export async function getPerformanceConversionWorkflows(
    page,
    pageSize,
    hasMore: offset + pageSize < filtered.length,
-    lastWorkflowSyncedAt: latestSyncedAt?.toISOString() || null,
+    workflowSync,
  }
 }

--- a/prisma/schema.prisma
+++ b/prisma/schema.prisma
@ -1469,7 +1469,7 @@ model TaskDetailChangeNotification {
 ///
 /// 上游 rpt.ding_Workflow / ding_WorkflowTask / ding_WorkflowScheme 均为无索引大表，
 /// 部门看板不得在用户请求链路中直连扫描。后台任务首次建立全量基线，之后每小时
-/// 增量同步主流程，并为全部未完结流程刷新任务与表单字段汇总。
+/// 增量同步主流程，并每 24 小时全量校准流程与商品 ID 表单字段。
 model DingWorkflowSnapshot {
  processInstanceId   String   @id @map("process_instance_id")
  businessId          String?  @map("business_id")