2078 lines
77 KiB
TypeScript
2078 lines
77 KiB
TypeScript
import { randomUUID } from 'node:crypto'
|
||
import { hostname } from 'node:os'
|
||
import type { Session } from 'next-auth'
|
||
import mssql, { type ConnectionPool } from 'mssql'
|
||
import { resolveBrandCenterAccessForUser } from '@/lib/brand-center-access'
|
||
import { isBrandCenterOrganizationPath } from '@/lib/brand-center-access-policy'
|
||
import {
|
||
assessDingWorkflowDeliveryBatch,
|
||
type DingWorkflowDeliveryAssessment,
|
||
} from '@/lib/ding-workflow-delivery'
|
||
import { prisma } from '@/lib/prisma'
|
||
|
||
const SYNC_STATE_ID = 'ding-workflow-hourly'
|
||
const TERMINAL_WORKFLOW_STATUSES = new Set(['COMPLETED', 'TERMINATED', 'CANCELED'])
|
||
const REJECTED_WORKFLOW_DECISIONS = new Set([
|
||
'REFUSE',
|
||
'REFUSED',
|
||
'REJECT',
|
||
'REJECTED',
|
||
'DISAGREE',
|
||
'DENY',
|
||
'DENIED',
|
||
'拒绝',
|
||
'已拒绝',
|
||
'不同意',
|
||
])
|
||
const LEASE_DURATION_MS = 20 * 60 * 1000
|
||
const DEFAULT_SYNC_INTERVAL_MS = 60 * 60 * 1000
|
||
const INITIAL_SYNC_DELAY_MS = 5_000
|
||
const UPSERT_BATCH_SIZE = 750
|
||
|
||
export const DING_WORKFLOW_STARTED_AT_CUTOFF_LOCAL = '2026-04-01 00:00:00'
|
||
export const DING_WORKFLOW_STARTED_AT_CUTOFF_ISO = '2026-03-31T16:00:00.000Z'
|
||
|
||
export const DING_WORKFLOW_ALLOWED_PROCESS_CODES = [
|
||
'PROC-3571C915-0506-4490-A91A-8AA7EA33F838',
|
||
'PROC-AE9FF3E1-A442-4E6A-85CF-0F2B86409673',
|
||
] as const
|
||
export const DING_WORKFLOW_BRANDS = ['亿家达', '卓禾', '欧意朗', '顾家', '索菲亚', '朴美'] as const
|
||
export const DING_WORKFLOW_NEW_PRODUCT_DEMAND_TYPE = '新品'
|
||
const ALLOWED_PROCESS_CODE_VALUES: string[] = [...DING_WORKFLOW_ALLOWED_PROCESS_CODES]
|
||
const ALLOWED_PROCESS_CODE_SET = new Set<string>(DING_WORKFLOW_ALLOWED_PROCESS_CODES)
|
||
const BRAND_CENTER_ACTIVE_TASK_JSON = [{ isBrandCenter: true, isActive: true }]
|
||
// Review milestones come only from the confirmed activityId mapping. A
|
||
// completed leadership task with AGREE is not sufficient to infer a review:
|
||
// content flows contain several supervisor assignment and production tasks.
|
||
const CONFIRMED_REVIEW_NODE_NAMES = new Set([
|
||
'品牌策划主管审核',
|
||
'编导审核',
|
||
'内容主管审核',
|
||
])
|
||
const PROCESS_TYPE_LABELS: Record<string, string> = {
|
||
[DING_WORKFLOW_ALLOWED_PROCESS_CODES[0]]: '美术设计流程',
|
||
[DING_WORKFLOW_ALLOWED_PROCESS_CODES[1]]: '内容视频和图片拍摄',
|
||
}
|
||
const ACTIVITY_NODE_NAMES_BY_PROCESS_CODE: Readonly<Record<string, Readonly<Record<string, string>>>> = {
|
||
[DING_WORKFLOW_ALLOWED_PROCESS_CODES[0]]: {
|
||
'sid-startevent': '发起人',
|
||
'1de3_de55': '部门负责人审核',
|
||
'94c1_1f0d': '部门总监审核',
|
||
'3cb8_8096': '品牌策划总监审批',
|
||
'38da_544a': '品牌策划主管分配任务',
|
||
'31cf_5b23': '出策划案',
|
||
'a9e1_5be0': '出图',
|
||
'de46_708d': '平面',
|
||
'2f9e_36f0': 'AI视频',
|
||
'b306_c849': '出图',
|
||
'f073_738b': '平面',
|
||
'6864_5429': '出图',
|
||
'162b_23ea': '平面',
|
||
'c9fe_a6b8': '出图/平面/AI视频',
|
||
'c255_84b3': '品牌策划主管审核',
|
||
'ebb9_213a': '运营接收图片并评分',
|
||
'b426_1125': '运营接收图片并评分',
|
||
'3e66_c105': '运营接收图片并评分',
|
||
'64bb_b63a': '运营接收图片并评分',
|
||
'2748_c5f9': '运营接收图片并评分',
|
||
'45f4_e095': '运营接收图片并评分',
|
||
'a8aa_ce41': '部门负责确认',
|
||
'00d9_b57f': '抄送部门总监',
|
||
},
|
||
[DING_WORKFLOW_ALLOWED_PROCESS_CODES[1]]: {
|
||
'sid-startevent': '发起人',
|
||
'cc59_3e5a': '内容负责人分配任务',
|
||
'9fb8_32e7': '内容主管分配任务',
|
||
'd742_3645': '内容主管分配任务',
|
||
'b9a4_c03a': '内容主管分配任务',
|
||
'9f14_7ed4': '内容主管分配任务',
|
||
'436a_c54c': '内容主管分配任务',
|
||
'fdcc_6e23': '内容主管分配任务',
|
||
'b596_d441': '内容主管分配任务',
|
||
'660b_13c2': '内容主管分配任务',
|
||
'0392_2965': '编导',
|
||
'2d94_f5f4': '编导',
|
||
'61b4_a6df': '编导',
|
||
'f4fb_5337': '编导',
|
||
'b7e6_434c': '编导',
|
||
'b454_f597': '编导',
|
||
'a25b_3158': '编导',
|
||
'3f4a_3e4a': '编导',
|
||
'ed38_970c': '编导',
|
||
'd3e1_85ec': '编导',
|
||
'27d5_92a3': '编导',
|
||
'375c_e8b0': '摄像',
|
||
'1d43_056a': '摄像',
|
||
'3078_2143': '摄像',
|
||
'f9fb_94bb': '摄像',
|
||
'003f_37a0': '摄像',
|
||
'be4c_5d3d': '摄像',
|
||
'6d09_69b9': '摄像',
|
||
'562b_2040': '摄像',
|
||
'9804_36c4': '摄像',
|
||
'663f_b63b': '摄像',
|
||
'7b9b_d818': '剪辑',
|
||
'3bde_6d5d': '剪辑',
|
||
'c372_4bc2': '剪辑',
|
||
'70e5_5548': '剪辑',
|
||
'538d_60ee': '剪辑',
|
||
'0768_9fc0': '剪辑',
|
||
'a8b6_f915': '剪辑',
|
||
'72f0_2fae': '剪辑',
|
||
'2fdb_dd98': '二次剪辑',
|
||
'238a_ba93': '二次剪辑',
|
||
'4151_7ec0': '二次剪辑',
|
||
'fe6d_4f56': '二次剪辑',
|
||
'7358_3df5': '封面',
|
||
'd027_fe1f': '封面',
|
||
'fe3d_054a': '封面',
|
||
'8a65_ac0b': '封面',
|
||
'7060_f10d': '封面',
|
||
'7190_0bdb': '封面',
|
||
'ca0a_c52d': '封面',
|
||
'3dfd_6eed': '修图',
|
||
'6b40_cb6e': '修图',
|
||
'45bd_56ad': '修图',
|
||
'7e09_6fad': '修图',
|
||
'e56b_4e05': 'AI视频制作',
|
||
'a021_1e2e': '编导审核',
|
||
'39ba_efb1': '编导审核',
|
||
'ce0a_7681': '编导审核',
|
||
'a3bb_f3fc': '编导审核',
|
||
'2e58_b875': '编导审核',
|
||
'6c43_575f': '内容主管审核',
|
||
'0abf_690f': '内容主管审核',
|
||
'36a0_2698': '内容主管审核',
|
||
'9a79_9db8': '运营验收',
|
||
'cdec_df3c': '运营验收',
|
||
'f7ff_0cd4': '运营验收',
|
||
'4073_b7af': '运营验收',
|
||
'cddb_3c2d': '运营验收',
|
||
'a03e_07f7': '策划验收',
|
||
'7f14_5f28': '策划验收',
|
||
'fa9d_aab6': '客服验收',
|
||
'5e28_391d': '预约客服安装',
|
||
},
|
||
}
|
||
|
||
const WORKFLOW_TABLE = {
|
||
schema: process.env.REPORT_DB_WORKFLOW_SCHEMA || 'rpt',
|
||
table: process.env.REPORT_DB_WORKFLOW_TABLE || 'ding_Workflow',
|
||
}
|
||
const WORKFLOW_SCHEME_TABLE = {
|
||
schema: process.env.REPORT_DB_WORKFLOW_SCHEMA || 'rpt',
|
||
table: process.env.REPORT_DB_WORKFLOW_SCHEME_TABLE || 'ding_WorkflowScheme',
|
||
}
|
||
const WORKFLOW_TASK_TABLE = {
|
||
schema: process.env.REPORT_DB_WORKFLOW_SCHEMA || 'rpt',
|
||
table: process.env.REPORT_DB_WORKFLOW_TASK_TABLE || 'ding_WorkflowTask',
|
||
}
|
||
|
||
type SourceWorkflowRow = {
|
||
processInstanceId: unknown
|
||
businessId: unknown
|
||
title: unknown
|
||
result: unknown
|
||
originatorUserId: unknown
|
||
originatorDeptId: unknown
|
||
originatorDeptName: unknown
|
||
status: unknown
|
||
bizAction: unknown
|
||
processCode: unknown
|
||
flowType: unknown
|
||
joinedAtRaw: unknown
|
||
sourceCreatedAtRaw: unknown
|
||
sourceUpdatedAtRaw: unknown
|
||
sourceFinishedAtRaw: unknown
|
||
}
|
||
|
||
type SourceTaskAggregate = {
|
||
processInstanceId: unknown
|
||
taskCount: unknown
|
||
activeTaskCount: unknown
|
||
latestTaskStatus: unknown
|
||
latestTaskAtRaw: unknown
|
||
}
|
||
|
||
type SourceTaskRow = {
|
||
processInstanceId: unknown
|
||
taskId: unknown
|
||
activityId: unknown
|
||
userId: unknown
|
||
status: unknown
|
||
result: unknown
|
||
createTime: unknown
|
||
finishTime: unknown
|
||
}
|
||
|
||
type SourceSchemeAggregate = {
|
||
processInstanceId: unknown
|
||
schemeFieldCount: unknown
|
||
}
|
||
|
||
type SourceSchemeField = {
|
||
processInstanceId: unknown
|
||
componentType: unknown
|
||
name: unknown
|
||
value: unknown
|
||
extValue: unknown
|
||
}
|
||
|
||
type DingWorkflowSourceCursor = {
|
||
version: 3
|
||
workflowUpdatedAt: string | null
|
||
taskId: number | null
|
||
schemeId: number | null
|
||
}
|
||
|
||
type SourceDeltaState = {
|
||
changedWorkflowIds: string[]
|
||
maxTaskId: number | null
|
||
maxSchemeId: number | null
|
||
requiresFullSync: boolean
|
||
}
|
||
|
||
type ContactAssignment = {
|
||
userId: string
|
||
name: string | null
|
||
nodeName: string
|
||
}
|
||
|
||
export type OngoingDingWorkflowActiveTask = {
|
||
taskId: string
|
||
activityId: string | null
|
||
nodeName: string
|
||
assigneeUserId: string | null
|
||
assigneeName: string
|
||
assigneeDepartment: string | null
|
||
isBrandCenter: boolean
|
||
isLeadership: boolean
|
||
isActive: boolean
|
||
reviewPassed: boolean
|
||
result: string | null
|
||
createdAt: string | null
|
||
finishedAt: string | null
|
||
}
|
||
|
||
export type DingWorkflowDetailFormFields = {
|
||
platform: string | null
|
||
brand: string | null
|
||
category: string | null
|
||
spu: string | null
|
||
demandType: string | null
|
||
standardOutputDeadline: string | null
|
||
sourceType: string | null
|
||
taskType: string | null
|
||
installVideo: string | null
|
||
quantity: string | null
|
||
cameraQuantity: string | null
|
||
planningTaskStats: string | null
|
||
graphicTaskStats: string | null
|
||
imageTaskStats: string | null
|
||
}
|
||
|
||
export type DingWorkflowDetailMilestones = {
|
||
assignmentNodeLabel: string
|
||
assignmentStartedAt: string | null
|
||
reviewNodeLabel: string
|
||
reviewFinishedAt: string | null
|
||
}
|
||
|
||
type WorkflowSnapshotPayload = {
|
||
processInstanceId: string
|
||
businessId: string | null
|
||
title: string | null
|
||
result: string | null
|
||
originatorUserId: string | null
|
||
originatorDeptId: string | null
|
||
originatorDeptName: string | null
|
||
status: string
|
||
bizAction: string | null
|
||
processCode: string | null
|
||
flowType: string | null
|
||
joinedAtRaw: string | null
|
||
sourceCreatedAtRaw: string | null
|
||
sourceUpdatedAtRaw: string | null
|
||
sourceFinishedAtRaw: string | null
|
||
isOngoing: boolean
|
||
schemeFieldCount: number
|
||
taskCount: number
|
||
activeTaskCount: number
|
||
latestTaskStatus: string | null
|
||
latestTaskAtRaw: string | null
|
||
brand: string
|
||
demandType: string | null
|
||
requesterName: string
|
||
requesterDepartment: string
|
||
projectType: string
|
||
formFields: DingWorkflowDetailFormFields
|
||
activeTasks: OngoingDingWorkflowActiveTask[]
|
||
}
|
||
|
||
export type DingWorkflowSyncStatus = 'pending' | 'syncing' | 'ready' | 'error'
|
||
|
||
export type OngoingDingWorkflowSummary = {
|
||
total: number
|
||
lastSyncedAt: string | null
|
||
syncStatus: DingWorkflowSyncStatus
|
||
}
|
||
|
||
export type OngoingDingWorkflowListItem = {
|
||
id: string
|
||
businessId: string | null
|
||
title: string
|
||
status: string
|
||
originatorUserId: string | null
|
||
originatorDeptName: string | null
|
||
sourceCreatedAt: string | null
|
||
sourceUpdatedAt: string | null
|
||
activeTaskCount: number
|
||
taskCount: number
|
||
schemeFieldCount: number
|
||
latestTaskStatus: string | null
|
||
processCode: string
|
||
projectType: string
|
||
brand: string
|
||
requesterName: string
|
||
requesterDepartment: string
|
||
initiatorName: string
|
||
initiatorDepartment: string
|
||
formFields: DingWorkflowDetailFormFields
|
||
assignmentNodeLabel: string
|
||
assignmentStartedAt: string | null
|
||
reviewNodeLabel: string
|
||
reviewFinishedAt: string | null
|
||
activeTasks: OngoingDingWorkflowActiveTask[]
|
||
}
|
||
|
||
export type OngoingDingWorkflowPage = {
|
||
items: OngoingDingWorkflowListItem[]
|
||
pagination: {
|
||
page: number
|
||
pageSize: number
|
||
total: number
|
||
totalPages: number
|
||
}
|
||
sync: OngoingDingWorkflowSummary
|
||
}
|
||
|
||
export type OngoingDingWorkflowInsights = {
|
||
items: OngoingDingWorkflowListItem[]
|
||
total: number
|
||
sync: OngoingDingWorkflowSummary
|
||
}
|
||
|
||
export type MonthlyLaunchWorkflowStatus = 'completed' | 'ongoing'
|
||
|
||
export type MonthlyLaunchWorkflowItem = OngoingDingWorkflowListItem & {
|
||
launchStatus: MonthlyLaunchWorkflowStatus
|
||
result: string | null
|
||
sourceFinishedAt: string | null
|
||
}
|
||
|
||
export type MonthlyLaunchBrandSummary = {
|
||
brand: typeof DING_WORKFLOW_BRANDS[number]
|
||
completed: number
|
||
ongoing: number
|
||
total: number
|
||
}
|
||
|
||
export type MonthlyLaunchSummary = {
|
||
completed: number
|
||
ongoing: number
|
||
total: number
|
||
monthStart: string
|
||
monthEnd: string
|
||
lastSyncedAt: string | null
|
||
syncStatus: DingWorkflowSyncStatus
|
||
}
|
||
|
||
export type MonthlyLaunchInsights = {
|
||
items: MonthlyLaunchWorkflowItem[]
|
||
brands: MonthlyLaunchBrandSummary[]
|
||
summary: MonthlyLaunchSummary
|
||
}
|
||
|
||
export type OnTimeDeliveryWorkflowItem = OngoingDingWorkflowListItem & {
|
||
result: string | null
|
||
sourceFinishedAt: string | null
|
||
assessment: DingWorkflowDeliveryAssessment
|
||
}
|
||
|
||
export type OnTimeDeliverySummary = {
|
||
onTime: number
|
||
overtime: number
|
||
pending: number
|
||
total: number
|
||
rate: number | null
|
||
monthStart: string
|
||
monthEnd: string
|
||
lastSyncedAt: string | null
|
||
syncStatus: DingWorkflowSyncStatus
|
||
}
|
||
|
||
export type OnTimeDeliveryInsights = {
|
||
items: OnTimeDeliveryWorkflowItem[]
|
||
summary: OnTimeDeliverySummary
|
||
}
|
||
|
||
export class DingWorkflowMetricApiError extends Error {
|
||
status: number
|
||
|
||
constructor(message: string, status = 400) {
|
||
super(message)
|
||
this.status = status
|
||
}
|
||
}
|
||
|
||
function clean(value: unknown): string | null {
|
||
if (value === null || value === undefined) return null
|
||
const normalized = String(value).trim()
|
||
return normalized || null
|
||
}
|
||
|
||
export function normalizeDingWorkflowProcessCode(value: unknown): string | null {
|
||
return clean(value)?.toUpperCase() || null
|
||
}
|
||
|
||
export function isAllowedDingWorkflowProcessCode(value: unknown): boolean {
|
||
const processCode = normalizeDingWorkflowProcessCode(value)
|
||
return Boolean(processCode && ALLOWED_PROCESS_CODE_SET.has(processCode))
|
||
}
|
||
|
||
export function getDingWorkflowProjectType(value: unknown): string {
|
||
const processCode = normalizeDingWorkflowProcessCode(value)
|
||
return processCode ? PROCESS_TYPE_LABELS[processCode] || '其他流程' : '其他流程'
|
||
}
|
||
|
||
export function getDingWorkflowActivityNodeName(
|
||
processCodeValue: unknown,
|
||
activityIdValue: unknown,
|
||
): string | null {
|
||
const processCode = normalizeDingWorkflowProcessCode(processCodeValue)
|
||
const activityId = clean(activityIdValue)
|
||
if (!processCode || !activityId) return null
|
||
return ACTIVITY_NODE_NAMES_BY_PROCESS_CODE[processCode]?.[activityId] || null
|
||
}
|
||
|
||
export function normalizeDingWorkflowBrand(value: unknown): string {
|
||
const normalized = clean(value)
|
||
if (!normalized) return '其他品牌'
|
||
return DING_WORKFLOW_BRANDS.find((brand) => normalized.includes(brand)) || '其他品牌'
|
||
}
|
||
|
||
function collectDingWorkflowDepartmentNames(value: unknown): string[] {
|
||
if (value === null || value === undefined) return []
|
||
if (Array.isArray(value)) {
|
||
return value.flatMap(collectDingWorkflowDepartmentNames)
|
||
}
|
||
if (typeof value === 'object') {
|
||
const record = value as Record<string, unknown>
|
||
return ['name', 'label', 'value', 'text', 'title']
|
||
.flatMap((key) => collectDingWorkflowDepartmentNames(record[key]))
|
||
}
|
||
|
||
const normalized = clean(value)
|
||
if (!normalized) return []
|
||
if (normalized.startsWith('[') || normalized.startsWith('{')) {
|
||
try {
|
||
return collectDingWorkflowDepartmentNames(JSON.parse(normalized))
|
||
} catch {
|
||
// Keep non-JSON department names unchanged.
|
||
}
|
||
}
|
||
return [normalized]
|
||
}
|
||
|
||
export function normalizeDingWorkflowDepartment(value: unknown): string | null {
|
||
const names = Array.from(new Set(collectDingWorkflowDepartmentNames(value)))
|
||
return names.length > 0 ? names.join(' / ') : null
|
||
}
|
||
|
||
function parseJson(value: unknown): unknown {
|
||
const normalized = clean(value)
|
||
if (!normalized) return null
|
||
try {
|
||
return JSON.parse(normalized)
|
||
} catch {
|
||
return null
|
||
}
|
||
}
|
||
|
||
function collectContactAssignments(value: unknown, fallbackNodeName: string): ContactAssignment[] {
|
||
if (!value) return []
|
||
if (Array.isArray(value)) {
|
||
return value.flatMap((entry) => collectContactAssignments(entry, fallbackNodeName))
|
||
}
|
||
if (typeof value !== 'object') return []
|
||
|
||
const record = value as Record<string, unknown>
|
||
const userId = clean(record.emplId) || clean(record.userId)
|
||
const contactName = clean(record.name) || clean(record.value)
|
||
const currentNodeName = clean(record.label) || fallbackNodeName
|
||
const direct = userId ? [{ userId, name: contactName, nodeName: currentNodeName }] : []
|
||
const rowValues = Array.isArray(record.rowValue)
|
||
? record.rowValue.flatMap((entry) => {
|
||
if (!entry || typeof entry !== 'object') return []
|
||
const cell = entry as Record<string, unknown>
|
||
const nodeName = clean(cell.label) || fallbackNodeName
|
||
return [cell.extendValue, cell.value]
|
||
.flatMap((candidate) => collectContactAssignments(candidate, nodeName))
|
||
})
|
||
: []
|
||
return [...direct, ...rowValues]
|
||
}
|
||
|
||
function extractContactAssignments(fields: SourceSchemeField[]): ContactAssignment[] {
|
||
const assignments = fields.flatMap((field) => {
|
||
const componentType = clean(field.componentType)
|
||
const nodeName = clean(field.name) || '当前处理节点'
|
||
if (componentType === 'InnerContactField') {
|
||
return [parseJson(field.extValue), parseJson(field.value)]
|
||
.flatMap((value) => collectContactAssignments(value, nodeName))
|
||
}
|
||
if (componentType === 'TableField') {
|
||
return collectContactAssignments(parseJson(field.value), nodeName)
|
||
}
|
||
return []
|
||
})
|
||
|
||
return Array.from(new Map(assignments.map((assignment) => [
|
||
`${assignment.userId}:${assignment.nodeName}`,
|
||
assignment,
|
||
])).values())
|
||
}
|
||
|
||
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
|
||
}
|
||
|
||
function collectSchemeDisplayValues(value: unknown): string[] {
|
||
if (value === null || value === undefined) return []
|
||
if (Array.isArray(value)) return value.flatMap(collectSchemeDisplayValues)
|
||
if (typeof value === 'object') {
|
||
const record = value as Record<string, unknown>
|
||
return ['name', 'label', 'value', 'text', 'title']
|
||
.flatMap((key) => collectSchemeDisplayValues(record[key]))
|
||
}
|
||
|
||
const normalized = clean(value)
|
||
if (!normalized) return []
|
||
if (normalized.startsWith('[') || normalized.startsWith('{')) {
|
||
try {
|
||
return collectSchemeDisplayValues(JSON.parse(normalized))
|
||
} catch {
|
||
// Keep non-JSON form values unchanged.
|
||
}
|
||
}
|
||
return [normalized]
|
||
}
|
||
|
||
export function normalizeDingWorkflowDetailValue(value: unknown): string | null {
|
||
const values = Array.from(new Set(collectSchemeDisplayValues(value)))
|
||
return values.length > 0 ? values.join(' / ') : null
|
||
}
|
||
|
||
function buildDingWorkflowDetailFormFields(fields: SourceSchemeField[]): DingWorkflowDetailFormFields {
|
||
return {
|
||
platform: normalizeDingWorkflowDetailValue(firstSchemeValue(fields, ['平台'])),
|
||
brand: normalizeDingWorkflowDetailValue(firstSchemeValue(fields, ['品牌'])),
|
||
category: normalizeDingWorkflowDetailValue(firstSchemeValue(fields, ['类目'])),
|
||
spu: normalizeDingWorkflowDetailValue(firstSchemeValue(fields, ['SPU', 'Spu', 'spu'])),
|
||
demandType: normalizeDingWorkflowDetailValue(firstSchemeValue(fields, ['需求类型'])),
|
||
standardOutputDeadline: normalizeDingWorkflowDetailValue(firstSchemeValue(fields, ['标准出图最长时效限制'])),
|
||
sourceType: normalizeDingWorkflowDetailValue(firstSchemeValue(fields, ['原图需求', '原图类型'])),
|
||
taskType: normalizeDingWorkflowDetailValue(firstSchemeValue(fields, ['任务类型'])),
|
||
installVideo: normalizeDingWorkflowDetailValue(firstSchemeValue(fields, [
|
||
'安装视频', '安装视频分类', '安装图纸/安装视频', '【安装图纸/视频】标准',
|
||
])),
|
||
quantity: normalizeDingWorkflowDetailValue(firstSchemeValue(fields, ['数量', '需求数量', '拍摄数量'])),
|
||
cameraQuantity: normalizeDingWorkflowDetailValue(firstSchemeValue(fields, ['摄像数量', '摄影数量'])),
|
||
planningTaskStats: firstSchemeValue(fields, ['策划任务统计']),
|
||
graphicTaskStats: firstSchemeValue(fields, ['平面任务统计']),
|
||
imageTaskStats: firstSchemeValue(fields, ['出图任务统计']),
|
||
}
|
||
}
|
||
|
||
function titleOriginatorName(value: unknown): string | null {
|
||
return clean(value)?.match(/^(.+?)提交的/)?.[1]?.trim() || null
|
||
}
|
||
|
||
function normalizeWorkflowTasks(value: unknown): OngoingDingWorkflowActiveTask[] {
|
||
if (!Array.isArray(value)) return []
|
||
return value.flatMap((task) => {
|
||
if (!task || typeof task !== 'object') return []
|
||
const row = task as Record<string, unknown>
|
||
const taskId = clean(row.taskId)
|
||
if (!taskId) return []
|
||
return [{
|
||
taskId,
|
||
activityId: clean(row.activityId),
|
||
nodeName: clean(row.nodeName) || '当前处理节点',
|
||
assigneeUserId: clean(row.assigneeUserId),
|
||
assigneeName: clean(row.assigneeName) || '待同步人员',
|
||
assigneeDepartment: clean(row.assigneeDepartment),
|
||
isBrandCenter: row.isBrandCenter === true,
|
||
isLeadership: row.isLeadership === true,
|
||
isActive: row.isActive !== false,
|
||
reviewPassed: row.reviewPassed === true,
|
||
result: clean(row.result)?.toUpperCase() || null,
|
||
createdAt: clean(row.createdAt),
|
||
finishedAt: clean(row.finishedAt),
|
||
}]
|
||
})
|
||
}
|
||
|
||
function normalizeDetailFormFields(value: unknown): DingWorkflowDetailFormFields {
|
||
const record = value && typeof value === 'object' && !Array.isArray(value)
|
||
? value as Record<string, unknown>
|
||
: {}
|
||
return {
|
||
platform: normalizeDingWorkflowDetailValue(record.platform),
|
||
brand: normalizeDingWorkflowDetailValue(record.brand),
|
||
category: normalizeDingWorkflowDetailValue(record.category),
|
||
spu: normalizeDingWorkflowDetailValue(record.spu),
|
||
demandType: normalizeDingWorkflowDetailValue(record.demandType),
|
||
standardOutputDeadline: normalizeDingWorkflowDetailValue(record.standardOutputDeadline),
|
||
sourceType: normalizeDingWorkflowDetailValue(record.sourceType),
|
||
taskType: normalizeDingWorkflowDetailValue(record.taskType),
|
||
installVideo: normalizeDingWorkflowDetailValue(record.installVideo),
|
||
quantity: normalizeDingWorkflowDetailValue(record.quantity),
|
||
cameraQuantity: normalizeDingWorkflowDetailValue(record.cameraQuantity),
|
||
planningTaskStats: clean(record.planningTaskStats),
|
||
graphicTaskStats: clean(record.graphicTaskStats),
|
||
imageTaskStats: clean(record.imageTaskStats),
|
||
}
|
||
}
|
||
|
||
const DETAIL_MILESTONE_NODES: Readonly<Record<string, {
|
||
assignmentNodeLabel: string
|
||
assignmentNodeNames: readonly string[]
|
||
reviewNodes: readonly {
|
||
label: string
|
||
names: readonly string[]
|
||
}[]
|
||
}>> = {
|
||
[DING_WORKFLOW_ALLOWED_PROCESS_CODES[0]]: {
|
||
assignmentNodeLabel: '品牌策划主管分配任务',
|
||
assignmentNodeNames: ['品牌策划主管分配任务'],
|
||
reviewNodes: [{
|
||
label: '品牌策划主管审核',
|
||
names: ['品牌策划主管审核'],
|
||
}],
|
||
},
|
||
[DING_WORKFLOW_ALLOWED_PROCESS_CODES[1]]: {
|
||
assignmentNodeLabel: '内容主管分配任务',
|
||
assignmentNodeNames: ['内容主管分配任务', '内容负责人分配任务'],
|
||
// 进行中项目只展示内容主管审核已完成的时间；编导审核是制作分支
|
||
// 的过程节点，不能在主管尚未审核时被误显示为“审核结束时间”。
|
||
reviewNodes: [{
|
||
label: '内容主管审核',
|
||
names: ['内容主管审核'],
|
||
}],
|
||
},
|
||
}
|
||
|
||
function compareTaskTimestamps(valueA: string | null, valueB: string | null) {
|
||
return String(valueA || '').localeCompare(String(valueB || ''))
|
||
}
|
||
|
||
export function resolveDingWorkflowDetailMilestones(
|
||
processCodeValue: unknown,
|
||
tasks: OngoingDingWorkflowActiveTask[],
|
||
): DingWorkflowDetailMilestones {
|
||
const processCode = normalizeDingWorkflowProcessCode(processCodeValue) || ''
|
||
const config = DETAIL_MILESTONE_NODES[processCode] || {
|
||
assignmentNodeLabel: '分配任务',
|
||
assignmentNodeNames: ['分配任务'],
|
||
reviewNodes: [{ label: '审核', names: ['审核'] }],
|
||
}
|
||
const assignmentStartedAt = tasks
|
||
.filter((task) => config.assignmentNodeNames.includes(task.nodeName) && task.createdAt)
|
||
.map((task) => task.createdAt)
|
||
.sort(compareTaskTimestamps)[0] || null
|
||
const completedReviews = config.reviewNodes.flatMap((reviewNode) => tasks
|
||
// A source task may carry a finish timestamp while the approval node is
|
||
// still active (for example, during an upstream mirror update). An active
|
||
// or unconfirmed review node is not an end milestone, even when
|
||
// finishTime is present.
|
||
.filter((task) => {
|
||
if (
|
||
!reviewNode.names.includes(task.nodeName)
|
||
|| !task.finishedAt
|
||
|| task.isActive
|
||
|| task.reviewPassed !== true
|
||
) return false
|
||
// When an activity id is present, require it to resolve to the same
|
||
// canonical review label. This protects old snapshots from arbitrary
|
||
// labels attached by the previous leadership-task fallback.
|
||
const mappedName = task.activityId
|
||
? getDingWorkflowActivityNodeName(processCode, task.activityId)
|
||
: task.nodeName
|
||
return mappedName === task.nodeName
|
||
})
|
||
.map((task) => ({
|
||
label: reviewNode.label,
|
||
finishedAt: task.finishedAt as string,
|
||
})))
|
||
.sort((reviewA, reviewB) => compareTaskTimestamps(reviewA.finishedAt, reviewB.finishedAt))
|
||
const completedReview = completedReviews.at(-1)
|
||
return {
|
||
assignmentNodeLabel: config.assignmentNodeLabel,
|
||
assignmentStartedAt,
|
||
reviewNodeLabel: completedReview?.label || config.reviewNodes[0].label,
|
||
reviewFinishedAt: completedReview?.finishedAt || null,
|
||
}
|
||
}
|
||
|
||
function nonNegativeInteger(value: unknown): number {
|
||
const number = Number(value)
|
||
return Number.isFinite(number) && number > 0 ? Math.floor(number) : 0
|
||
}
|
||
|
||
function quoteIdentifier(value: string) {
|
||
return '[' + String(value).replace(/]/g, ']]') + ']'
|
||
}
|
||
|
||
function tableRef(table: { schema: string; table: string }) {
|
||
return `${quoteIdentifier(table.schema)}.${quoteIdentifier(table.table)}`
|
||
}
|
||
|
||
function sourceTimestampSql(value: string) {
|
||
return `TRY_CONVERT(datetime2, REPLACE(REPLACE(REPLACE(NULLIF(CONVERT(nvarchar(64), ${value}), ''), 'T', ' '), 'Z', ''), 'z', ''))`
|
||
}
|
||
|
||
export function isDingWorkflowOngoing(status: unknown, finishedAt: unknown): boolean {
|
||
const normalizedStatus = String(status || '').trim().toUpperCase()
|
||
return Boolean(normalizedStatus)
|
||
&& !TERMINAL_WORKFLOW_STATUSES.has(normalizedStatus)
|
||
&& !clean(finishedAt)
|
||
}
|
||
|
||
export function sourceWorkflowTimestampToIso(value: unknown): string | null {
|
||
const normalized = clean(value)
|
||
if (!normalized) return null
|
||
// 报表库三张钉钉表中的时间是上海本地墙上时间；SQL Server 驱动序列化
|
||
// datetime2 时会附加一个并不代表 UTC 的 `Z`。先移除这个伪时区标记，
|
||
// 再按 Asia/Shanghai 解释，避免页面展示和 SLA 计算整体多出 8 小时。
|
||
const workflowClockValue = normalized.replace(/Z$/i, '')
|
||
const hasNumericTimeZone = /[+-]\d{2}:?\d{2}$/.test(workflowClockValue)
|
||
const candidate = hasNumericTimeZone
|
||
? workflowClockValue
|
||
: `${workflowClockValue.replace(' ', 'T')}+08:00`
|
||
const parsed = new Date(candidate)
|
||
return Number.isNaN(parsed.getTime()) ? null : parsed.toISOString()
|
||
}
|
||
|
||
export function storedWorkflowTimestampToIso(value: unknown): string | null {
|
||
const normalized = clean(value)
|
||
if (!normalized) return null
|
||
const hasTimeZone = /(?:Z|[+-]\d{2}:?\d{2})$/i.test(normalized)
|
||
if (!hasTimeZone) return sourceWorkflowTimestampToIso(normalized)
|
||
const parsed = new Date(normalized)
|
||
return Number.isNaN(parsed.getTime()) ? null : parsed.toISOString()
|
||
}
|
||
|
||
export function getShanghaiMonthRange(now = new Date()) {
|
||
const parts = new Intl.DateTimeFormat('en-US', {
|
||
timeZone: 'Asia/Shanghai',
|
||
year: 'numeric',
|
||
month: 'numeric',
|
||
}).formatToParts(now)
|
||
const year = Number(parts.find((part) => part.type === 'year')?.value)
|
||
const month = Number(parts.find((part) => part.type === 'month')?.value)
|
||
const shanghaiOffsetMs = 8 * 60 * 60 * 1000
|
||
return {
|
||
start: new Date(Date.UTC(year, month - 1, 1) - shanghaiOffsetMs),
|
||
end: new Date(Date.UTC(year, month, 1) - shanghaiOffsetMs),
|
||
}
|
||
}
|
||
|
||
export function isDingWorkflowCompletedInRange(
|
||
status: unknown,
|
||
finishedAt: unknown,
|
||
range: { start: Date; end: Date },
|
||
) {
|
||
if (String(status || '').trim().toUpperCase() !== 'COMPLETED') return false
|
||
const finishedAtIso = storedWorkflowTimestampToIso(finishedAt)
|
||
if (!finishedAtIso) return false
|
||
const finishedAtMs = new Date(finishedAtIso).getTime()
|
||
return finishedAtMs >= range.start.getTime() && finishedAtMs < range.end.getTime()
|
||
}
|
||
|
||
/**
|
||
* 按时交付只统计已经完成时效判定的流程：待计算（缺少标准时效或关键节点）
|
||
* 不进入分母。月份边界以时效实际结束节点（主管审核或该内容规则指定的审核
|
||
* 节点）为准，而不是审批实例的最终结束时间。
|
||
*/
|
||
export function isDingWorkflowDeliveryAssessmentInRange(
|
||
assessment: Pick<DingWorkflowDeliveryAssessment, 'status' | 'finishedAt'>,
|
||
range: { start: Date; end: Date },
|
||
) {
|
||
if (assessment.status !== 'on_time' && assessment.status !== 'overtime') return false
|
||
return isDingWorkflowCompletedInRange('COMPLETED', assessment.finishedAt, range)
|
||
}
|
||
|
||
/**
|
||
* 钉钉会把审批结束状态和审批结论分开保存：被拒绝的流程也可能处于 COMPLETED。
|
||
* 按时交付只统计已完结且没有明确拒绝结论的流程，避免拒绝单进入分母。
|
||
*/
|
||
export function isDingWorkflowApprovedCompletion(
|
||
status: unknown,
|
||
result: unknown,
|
||
bizAction: unknown,
|
||
) {
|
||
if (String(status || '').trim().toUpperCase() !== 'COMPLETED') return false
|
||
return ![result, bizAction].some((value) => {
|
||
const decision = String(value || '').trim().toUpperCase()
|
||
return decision ? REJECTED_WORKFLOW_DECISIONS.has(decision) : false
|
||
})
|
||
}
|
||
|
||
export function isDingWorkflowStartedOnOrAfterCutoff(value: unknown): boolean {
|
||
const startedAt = sourceWorkflowTimestampToIso(value)
|
||
return Boolean(startedAt && startedAt >= DING_WORKFLOW_STARTED_AT_CUTOFF_ISO)
|
||
}
|
||
|
||
/**
|
||
* 品牌中心“进行中项目”的统一查询口径。
|
||
*
|
||
* 一个流程可能同时存在多个当前任务；只要至少一个当前节点处理人属于品牌中心，
|
||
* 该流程就计入。快照 JSON 同时保存当前任务和审核历史，所以查询必须显式要求
|
||
* isActive=true，避免把历史品牌中心处理人误当作当前负责人。
|
||
*/
|
||
export function buildBrandCenterOngoingWorkflowWhere() {
|
||
return {
|
||
isOngoing: true,
|
||
processCode: { in: ALLOWED_PROCESS_CODE_VALUES },
|
||
sourceCreatedAtRaw: { gte: DING_WORKFLOW_STARTED_AT_CUTOFF_ISO },
|
||
activeTasks: { array_contains: BRAND_CENTER_ACTIVE_TASK_JSON },
|
||
}
|
||
}
|
||
|
||
/**
|
||
* “进行中项目”中美术设计流程的统一子集。
|
||
*
|
||
* 本月上新的进行中数量必须直接复用这个口径，避免同一条在办美术流程
|
||
* 在两个指标中因状态、时间或当前处理人条件不同而出现不一致。
|
||
*/
|
||
export function buildBrandCenterOngoingArtWorkflowWhere() {
|
||
return {
|
||
...buildBrandCenterOngoingWorkflowWhere(),
|
||
processCode: DING_WORKFLOW_ALLOWED_PROCESS_CODES[0],
|
||
}
|
||
}
|
||
|
||
function maxSourceCursor(rows: WorkflowSnapshotPayload[], current: string | null) {
|
||
return rows.reduce<string | null>((latest, row) => {
|
||
const candidate = row.sourceUpdatedAtRaw
|
||
if (!candidate) return latest
|
||
return !latest || candidate > latest ? candidate : latest
|
||
}, current)
|
||
}
|
||
|
||
function normalizeSourceCursorId(value: unknown): number | null {
|
||
if (value === null || value === undefined || value === '') return null
|
||
const numeric = Number(value)
|
||
return Number.isSafeInteger(numeric) && numeric >= 0 ? numeric : null
|
||
}
|
||
|
||
/**
|
||
* 兼容旧版只保存 ding_Workflow.updateAt 的游标。
|
||
*
|
||
* v3 新增交付时效所需表单字段；读取旧版 v2 时保留主表游标，但清空
|
||
* Task / Scheme 游标触发一次全量对齐，之后继续按三表复合游标增量同步。
|
||
* 后续每小时只读取主表变化和 Task / Scheme 新增版本所关联的流程。
|
||
*/
|
||
export function parseDingWorkflowSourceCursor(raw: string | null): DingWorkflowSourceCursor {
|
||
const normalized = clean(raw)
|
||
if (!normalized) {
|
||
return { version: 3, workflowUpdatedAt: null, taskId: null, schemeId: null }
|
||
}
|
||
try {
|
||
const parsed = JSON.parse(normalized) as Record<string, unknown>
|
||
if (parsed.version === 3) {
|
||
return {
|
||
version: 3,
|
||
workflowUpdatedAt: clean(parsed.workflowUpdatedAt),
|
||
taskId: normalizeSourceCursorId(parsed.taskId),
|
||
schemeId: normalizeSourceCursorId(parsed.schemeId),
|
||
}
|
||
}
|
||
if (parsed.version === 2) {
|
||
return {
|
||
version: 3,
|
||
workflowUpdatedAt: clean(parsed.workflowUpdatedAt),
|
||
taskId: null,
|
||
schemeId: null,
|
||
}
|
||
}
|
||
} catch {
|
||
// Legacy cursor is a plain ding_Workflow.updateAt string.
|
||
}
|
||
return { version: 3, workflowUpdatedAt: normalized, taskId: null, schemeId: null }
|
||
}
|
||
|
||
export function serializeDingWorkflowSourceCursor(cursor: DingWorkflowSourceCursor): string {
|
||
return JSON.stringify(cursor)
|
||
}
|
||
|
||
export function isReportWorkflowDbConfigured() {
|
||
return Boolean(
|
||
process.env.REPORT_DB_SERVER
|
||
&& process.env.REPORT_DB_DATABASE
|
||
&& process.env.REPORT_DB_USER
|
||
&& process.env.REPORT_DB_PASSWORD,
|
||
)
|
||
}
|
||
|
||
function buildSourceConfig() {
|
||
return {
|
||
server: process.env.REPORT_DB_SERVER as string,
|
||
database: process.env.REPORT_DB_DATABASE as string,
|
||
user: process.env.REPORT_DB_USER as string,
|
||
password: process.env.REPORT_DB_PASSWORD as string,
|
||
port: Number(process.env.REPORT_DB_PORT || 1433),
|
||
options: {
|
||
encrypt: process.env.REPORT_DB_ENCRYPT === 'true',
|
||
trustServerCertificate: process.env.REPORT_DB_TRUST_SERVER_CERTIFICATE !== 'false',
|
||
},
|
||
connectionTimeout: 20_000,
|
||
requestTimeout: 300_000,
|
||
pool: { max: 3, min: 0, idleTimeoutMillis: 30_000 },
|
||
}
|
||
}
|
||
|
||
async function assertDashboardAccess(session: Session | null) {
|
||
const userId = session?.user?.id
|
||
if (!userId) throw new DingWorkflowMetricApiError('请先登录', 401)
|
||
|
||
const user = await prisma.user.findUnique({
|
||
where: { id: userId },
|
||
select: { role: true, isSuperAdmin: true, isActive: true, status: true },
|
||
})
|
||
if (!user || !user.isActive || user.status === 'archived' || user.status === 'suspended') {
|
||
throw new DingWorkflowMetricApiError('账号不可用，请重新登录', 401)
|
||
}
|
||
|
||
const allowed = await resolveBrandCenterAccessForUser(userId, {
|
||
role: user.role,
|
||
isSuperAdmin: user.isSuperAdmin,
|
||
})
|
||
if (!allowed) throw new DingWorkflowMetricApiError('无权查看品牌中心流程指标', 403)
|
||
}
|
||
|
||
async function fetchSourceDeltaState(
|
||
pool: ConnectionPool,
|
||
cursor: DingWorkflowSourceCursor,
|
||
): Promise<SourceDeltaState> {
|
||
const request = pool.request()
|
||
request.multiple = true
|
||
request.input('taskCursorId', mssql.Int, cursor.taskId)
|
||
request.input('schemeCursorId', mssql.Int, cursor.schemeId)
|
||
request.input('startedAtCutoff', mssql.NVarChar(64), DING_WORKFLOW_STARTED_AT_CUTOFF_LOCAL)
|
||
DING_WORKFLOW_ALLOWED_PROCESS_CODES.forEach((processCode, index) => {
|
||
request.input(`allowedProcessCode${index}`, mssql.NVarChar(255), processCode)
|
||
})
|
||
const result = await request.query(`
|
||
SELECT MAX(id) AS maxTaskId
|
||
FROM ${tableRef(WORKFLOW_TASK_TABLE)};
|
||
|
||
SELECT MAX(id) AS maxSchemeId
|
||
FROM ${tableRef(WORKFLOW_SCHEME_TABLE)};
|
||
|
||
;WITH changed_workflow_ids AS (
|
||
SELECT CONVERT(nvarchar(255), processInstanceId) AS processInstanceId
|
||
FROM ${tableRef(WORKFLOW_TASK_TABLE)}
|
||
WHERE @taskCursorId IS NOT NULL AND id > @taskCursorId
|
||
UNION
|
||
SELECT CONVERT(nvarchar(255), processInstanceId) AS processInstanceId
|
||
FROM ${tableRef(WORKFLOW_SCHEME_TABLE)}
|
||
WHERE @schemeCursorId IS NOT NULL AND id > @schemeCursorId
|
||
)
|
||
SELECT DISTINCT changed.processInstanceId
|
||
FROM changed_workflow_ids changed
|
||
INNER JOIN ${tableRef(WORKFLOW_TABLE)} workflow
|
||
ON CONVERT(nvarchar(255), workflow.processInstanceId) = changed.processInstanceId
|
||
WHERE
|
||
UPPER(LTRIM(RTRIM(COALESCE(CONVERT(nvarchar(255), workflow.processCode), '')))) IN (
|
||
@allowedProcessCode0,
|
||
@allowedProcessCode1
|
||
)
|
||
AND ${sourceTimestampSql('workflow.createAt')} >=
|
||
TRY_CONVERT(datetime2, @startedAtCutoff);
|
||
`)
|
||
const taskMaxRows = (result.recordsets?.[0] || []) as Array<{ maxTaskId: unknown }>
|
||
const schemeMaxRows = (result.recordsets?.[1] || []) as Array<{ maxSchemeId: unknown }>
|
||
const changedRows = (result.recordsets?.[2] || []) as Array<{ processInstanceId: unknown }>
|
||
return {
|
||
changedWorkflowIds: changedRows.flatMap((row) => {
|
||
const processInstanceId = clean(row.processInstanceId)
|
||
return processInstanceId ? [processInstanceId] : []
|
||
}),
|
||
maxTaskId: normalizeSourceCursorId(taskMaxRows[0]?.maxTaskId),
|
||
maxSchemeId: normalizeSourceCursorId(schemeMaxRows[0]?.maxSchemeId),
|
||
requiresFullSync: cursor.taskId === null || cursor.schemeId === null,
|
||
}
|
||
}
|
||
|
||
async function fetchSourceWorkflows(
|
||
pool: ConnectionPool,
|
||
cursor: string | null,
|
||
changedWorkflowIds: string[] = [],
|
||
) {
|
||
const request = pool.request()
|
||
const currentMonth = getShanghaiMonthRange()
|
||
const currentMonthStart = new Intl.DateTimeFormat('sv-SE', {
|
||
timeZone: 'Asia/Shanghai',
|
||
year: 'numeric',
|
||
month: '2-digit',
|
||
day: '2-digit',
|
||
hour: '2-digit',
|
||
minute: '2-digit',
|
||
second: '2-digit',
|
||
hour12: false,
|
||
}).format(currentMonth.start)
|
||
request.input('cursor', mssql.NVarChar(64), cursor)
|
||
request.input('changedWorkflowIds', mssql.NVarChar(mssql.MAX), JSON.stringify(changedWorkflowIds))
|
||
request.input('startedAtCutoff', mssql.NVarChar(64), DING_WORKFLOW_STARTED_AT_CUTOFF_LOCAL)
|
||
request.input('currentMonthStart', mssql.NVarChar(64), currentMonthStart)
|
||
DING_WORKFLOW_ALLOWED_PROCESS_CODES.forEach((processCode, index) => {
|
||
request.input(`allowedProcessCode${index}`, mssql.NVarChar(255), processCode)
|
||
})
|
||
const result = await request.query<SourceWorkflowRow>(`
|
||
SELECT
|
||
CONVERT(nvarchar(255), processInstanceId) AS processInstanceId,
|
||
CONVERT(nvarchar(255), businessId) AS businessId,
|
||
CONVERT(nvarchar(max), title) AS title,
|
||
CONVERT(nvarchar(100), result) AS result,
|
||
CONVERT(nvarchar(255), originatorUserId) AS originatorUserId,
|
||
CONVERT(nvarchar(255), originatorDeptId) AS originatorDeptId,
|
||
CONVERT(nvarchar(500), originatorDeptName) AS originatorDeptName,
|
||
CONVERT(nvarchar(100), status) AS status,
|
||
CONVERT(nvarchar(100), bizAction) AS bizAction,
|
||
CONVERT(nvarchar(255), processCode) AS processCode,
|
||
CONVERT(nvarchar(100), flowType) AS flowType,
|
||
CONVERT(nvarchar(64), joinAt) AS joinedAtRaw,
|
||
CONVERT(nvarchar(64), createAt) AS sourceCreatedAtRaw,
|
||
CONVERT(nvarchar(64), updateAt) AS sourceUpdatedAtRaw,
|
||
CONVERT(nvarchar(64), finishTime) AS sourceFinishedAtRaw
|
||
FROM ${tableRef(WORKFLOW_TABLE)}
|
||
WHERE
|
||
UPPER(LTRIM(RTRIM(COALESCE(CONVERT(nvarchar(255), processCode), '')))) IN (
|
||
@allowedProcessCode0,
|
||
@allowedProcessCode1
|
||
)
|
||
AND ${sourceTimestampSql('createAt')} >=
|
||
TRY_CONVERT(datetime2, @startedAtCutoff)
|
||
AND (
|
||
@cursor IS NULL
|
||
OR CONVERT(nvarchar(255), processInstanceId) IN (
|
||
SELECT CONVERT(nvarchar(255), [value])
|
||
FROM OPENJSON(@changedWorkflowIds)
|
||
)
|
||
OR UPPER(LTRIM(RTRIM(COALESCE(CONVERT(nvarchar(100), status), '')))) NOT IN ('COMPLETED', 'TERMINATED', 'CANCELED')
|
||
OR (
|
||
UPPER(LTRIM(RTRIM(COALESCE(CONVERT(nvarchar(100), status), '')))) IN ('COMPLETED', 'TERMINATED', 'CANCELED')
|
||
AND ${sourceTimestampSql('finishTime')} >=
|
||
TRY_CONVERT(datetime2, @currentMonthStart)
|
||
)
|
||
OR ${sourceTimestampSql('updateAt')} >=
|
||
DATEADD(minute, -5, ${sourceTimestampSql('@cursor')})
|
||
)
|
||
`)
|
||
return result.recordset
|
||
}
|
||
|
||
async function fetchWorkflowAggregates(pool: ConnectionPool, processInstanceIds: string[]) {
|
||
if (processInstanceIds.length === 0) {
|
||
return {
|
||
taskByWorkflow: new Map<string, SourceTaskAggregate>(),
|
||
taskRowsByWorkflow: new Map<string, SourceTaskRow[]>(),
|
||
schemeByWorkflow: new Map<string, SourceSchemeAggregate>(),
|
||
schemeRowsByWorkflow: new Map<string, SourceSchemeField[]>(),
|
||
}
|
||
}
|
||
|
||
const request = pool.request()
|
||
request.multiple = true
|
||
request.input('workflowIds', mssql.NVarChar(mssql.MAX), JSON.stringify(processInstanceIds))
|
||
const result = await request.query(`
|
||
;WITH workflow_ids AS (
|
||
SELECT CONVERT(nvarchar(255), [value]) AS processInstanceId
|
||
FROM OPENJSON(@workflowIds)
|
||
)
|
||
SELECT
|
||
CONVERT(nvarchar(255), task.processInstanceId) AS processInstanceId,
|
||
CONVERT(nvarchar(255), task.taskId) AS taskId,
|
||
CONVERT(nvarchar(255), task.activityId) AS activityId,
|
||
CONVERT(nvarchar(255), task.userId) AS userId,
|
||
CONVERT(nvarchar(100), task.status) AS status,
|
||
CONVERT(nvarchar(100), task.result) AS result,
|
||
CONVERT(nvarchar(64), task.createTime) AS createTime,
|
||
CONVERT(nvarchar(64), task.finishTime) AS finishTime,
|
||
ROW_NUMBER() OVER (
|
||
PARTITION BY task.processInstanceId, task.taskId
|
||
ORDER BY task.id DESC
|
||
) AS duplicateRow,
|
||
task.id AS sourceRowId
|
||
INTO #workflow_task_rows
|
||
FROM ${tableRef(WORKFLOW_TASK_TABLE)} task
|
||
INNER JOIN workflow_ids ids
|
||
ON ids.processInstanceId = CONVERT(nvarchar(255), task.processInstanceId)
|
||
;
|
||
|
||
;WITH ranked_tasks AS (
|
||
SELECT *, ROW_NUMBER() OVER (
|
||
PARTITION BY processInstanceId
|
||
ORDER BY ${sourceTimestampSql('createTime')} DESC, sourceRowId DESC
|
||
) AS latestRow
|
||
FROM #workflow_task_rows
|
||
WHERE duplicateRow = 1
|
||
)
|
||
SELECT
|
||
processInstanceId,
|
||
COUNT_BIG(*) AS taskCount,
|
||
SUM(CASE
|
||
WHEN UPPER(LTRIM(RTRIM(COALESCE(status, '')))) NOT IN ('COMPLETED', 'TERMINATED', 'CANCELED')
|
||
AND NULLIF(LTRIM(RTRIM(COALESCE(finishTime, ''))), '') IS NULL
|
||
THEN 1 ELSE 0
|
||
END) AS activeTaskCount,
|
||
MAX(CASE WHEN latestRow = 1 THEN status END) AS latestTaskStatus,
|
||
MAX(CASE WHEN latestRow = 1 THEN createTime END) AS latestTaskAtRaw
|
||
FROM ranked_tasks
|
||
GROUP BY processInstanceId;
|
||
|
||
SELECT processInstanceId, taskId, activityId, userId, status, result, createTime, finishTime
|
||
FROM #workflow_task_rows
|
||
WHERE duplicateRow = 1;
|
||
|
||
;WITH workflow_ids AS (
|
||
SELECT CONVERT(nvarchar(255), [value]) AS processInstanceId
|
||
FROM OPENJSON(@workflowIds)
|
||
)
|
||
SELECT
|
||
CONVERT(nvarchar(255), scheme.processInstanceId) AS processInstanceId,
|
||
COUNT_BIG(*) AS schemeFieldCount
|
||
FROM ${tableRef(WORKFLOW_SCHEME_TABLE)} scheme
|
||
INNER JOIN workflow_ids ids
|
||
ON ids.processInstanceId = CONVERT(nvarchar(255), scheme.processInstanceId)
|
||
GROUP BY scheme.processInstanceId;
|
||
|
||
;WITH workflow_ids AS (
|
||
SELECT CONVERT(nvarchar(255), [value]) AS processInstanceId
|
||
FROM OPENJSON(@workflowIds)
|
||
)
|
||
SELECT
|
||
CONVERT(nvarchar(255), scheme.processInstanceId) AS processInstanceId,
|
||
CONVERT(nvarchar(500), scheme.componentType) AS componentType,
|
||
CONVERT(nvarchar(2000), scheme.name) AS name,
|
||
CONVERT(nvarchar(max), scheme.value) AS value,
|
||
CONVERT(nvarchar(max), scheme.extValue) AS extValue
|
||
FROM ${tableRef(WORKFLOW_SCHEME_TABLE)} scheme
|
||
INNER JOIN workflow_ids ids
|
||
ON ids.processInstanceId = CONVERT(nvarchar(255), scheme.processInstanceId)
|
||
WHERE
|
||
LTRIM(RTRIM(COALESCE(CONVERT(nvarchar(2000), scheme.name), ''))) IN (
|
||
'平台', '品牌', '类目', 'SPU', 'Spu', 'spu', '需求类型', '标准出图最长时效限制',
|
||
'需求部门', '需求人', '任务类型', '原图需求', '原图类型',
|
||
'安装视频', '安装视频分类', '安装图纸/安装视频', '【安装图纸/视频】标准',
|
||
'数量', '需求数量', '拍摄数量', '摄像数量', '摄影数量',
|
||
'策划任务统计', '平面任务统计', '出图任务统计'
|
||
)
|
||
OR CONVERT(nvarchar(500), scheme.componentType) IN ('InnerContactField', 'TableField');
|
||
`)
|
||
|
||
const taskRows = (result.recordsets?.[0] || []) as SourceTaskAggregate[]
|
||
const activeTaskRows = (result.recordsets?.[1] || []) as SourceTaskRow[]
|
||
const schemeRows = (result.recordsets?.[2] || []) as SourceSchemeAggregate[]
|
||
const schemeFieldRows = (result.recordsets?.[3] || []) as SourceSchemeField[]
|
||
const groupByWorkflow = <T extends { processInstanceId: unknown }>(rows: T[]) => {
|
||
const grouped = new Map<string, T[]>()
|
||
rows.forEach((row) => {
|
||
const processInstanceId = clean(row.processInstanceId)
|
||
if (!processInstanceId) return
|
||
grouped.set(processInstanceId, [...(grouped.get(processInstanceId) || []), row])
|
||
})
|
||
return grouped
|
||
}
|
||
return {
|
||
taskByWorkflow: new Map(taskRows.map((row) => [String(row.processInstanceId), row])),
|
||
taskRowsByWorkflow: groupByWorkflow(activeTaskRows),
|
||
schemeByWorkflow: new Map(schemeRows.map((row) => [String(row.processInstanceId), row])),
|
||
schemeRowsByWorkflow: groupByWorkflow(schemeFieldRows),
|
||
}
|
||
}
|
||
|
||
export function isDingWorkflowTaskOngoing(status: unknown, finishTime: unknown): boolean {
|
||
// The external dashboard treats a task with a finish timestamp as history,
|
||
// even when the mirrored status field still says RUNNING. This is important
|
||
// for assignment nodes that have already handed work to the next node.
|
||
if (clean(finishTime)) return false
|
||
const normalized = clean(status)?.toUpperCase()
|
||
return Boolean(normalized && !TERMINAL_WORKFLOW_STATUSES.has(normalized))
|
||
}
|
||
|
||
type EmployeeDirectoryEntry = {
|
||
name: string
|
||
departmentName: string | null
|
||
isBrandCenter: boolean
|
||
isLeadership: boolean
|
||
}
|
||
|
||
async function readEmployeeDirectory(
|
||
sourceRows: SourceWorkflowRow[],
|
||
taskRowsByWorkflow: Map<string, SourceTaskRow[]>,
|
||
) {
|
||
const ids = Array.from(new Set([
|
||
...sourceRows.map((row) => clean(row.originatorUserId)),
|
||
...Array.from(taskRowsByWorkflow.values()).flatMap((rows) => rows.map((row) => clean(row.userId))),
|
||
].filter((value): value is string => Boolean(value) && value !== 'bpms_system')))
|
||
if (ids.length === 0) return new Map<string, EmployeeDirectoryEntry>()
|
||
const profiles = await prisma.performanceSurveyEmployeeProfile.findMany({
|
||
where: { dingtalkUserId: { in: ids } },
|
||
select: {
|
||
dingtalkUserId: true,
|
||
displayName: true,
|
||
departmentName: true,
|
||
departmentPath: true,
|
||
isLeadership: true,
|
||
levelType: true,
|
||
positionLevel: true,
|
||
employmentState: true,
|
||
departmentMemberships: { select: { departmentPath: true } },
|
||
},
|
||
})
|
||
return new Map(profiles.flatMap((profile) => (
|
||
profile.dingtalkUserId ? [[profile.dingtalkUserId, {
|
||
name: profile.displayName,
|
||
departmentName: profile.departmentName,
|
||
isLeadership: profile.isLeadership
|
||
|| profile.levelType?.toUpperCase() === 'M'
|
||
|| profile.positionLevel?.toUpperCase().startsWith('M') === true,
|
||
isBrandCenter: profile.employmentState !== 'RESIGNED' && [
|
||
profile.departmentPath,
|
||
...profile.departmentMemberships.map((membership) => membership.departmentPath),
|
||
].some(isBrandCenterOrganizationPath),
|
||
}] as const] : []
|
||
)))
|
||
}
|
||
|
||
function inferActivityNodeNames(
|
||
sourceRows: SourceWorkflowRow[],
|
||
taskRowsByWorkflow: Map<string, SourceTaskRow[]>,
|
||
schemeRowsByWorkflow: Map<string, SourceSchemeField[]>,
|
||
) {
|
||
const processCodeByWorkflow = new Map(sourceRows.flatMap((row) => {
|
||
const id = clean(row.processInstanceId)
|
||
const code = normalizeDingWorkflowProcessCode(row.processCode)
|
||
return id && code ? [[id, code] as const] : []
|
||
}))
|
||
const votes = new Map<string, Map<string, number>>()
|
||
|
||
taskRowsByWorkflow.forEach((taskRows, processInstanceId) => {
|
||
const processCode = processCodeByWorkflow.get(processInstanceId)
|
||
if (!processCode) return
|
||
const assignments = extractContactAssignments(schemeRowsByWorkflow.get(processInstanceId) || [])
|
||
taskRows.forEach((task) => {
|
||
const activityId = clean(task.activityId)
|
||
const userId = clean(task.userId)
|
||
if (!activityId || !userId) return
|
||
const nodeNames = Array.from(new Set(
|
||
assignments.filter((assignment) => assignment.userId === userId).map((assignment) => assignment.nodeName),
|
||
))
|
||
if (nodeNames.length !== 1) return
|
||
const key = `${processCode}:${activityId}`
|
||
const activityVotes = votes.get(key) || new Map<string, number>()
|
||
activityVotes.set(nodeNames[0], (activityVotes.get(nodeNames[0]) || 0) + 1)
|
||
votes.set(key, activityVotes)
|
||
})
|
||
})
|
||
|
||
return new Map(Array.from(votes.entries()).map(([key, activityVotes]) => {
|
||
const [nodeName] = Array.from(activityVotes.entries()).sort((a, b) => b[1] - a[1])[0]
|
||
return [key, nodeName]
|
||
}))
|
||
}
|
||
|
||
function buildSnapshotPayload(
|
||
sourceRows: SourceWorkflowRow[],
|
||
taskByWorkflow: Map<string, SourceTaskAggregate>,
|
||
taskRowsByWorkflow: Map<string, SourceTaskRow[]>,
|
||
schemeByWorkflow: Map<string, SourceSchemeAggregate>,
|
||
schemeRowsByWorkflow: Map<string, SourceSchemeField[]>,
|
||
employeeDirectory: Map<string, EmployeeDirectoryEntry>,
|
||
activityNodeNames: Map<string, string>,
|
||
): WorkflowSnapshotPayload[] {
|
||
return sourceRows.flatMap((row) => {
|
||
const processInstanceId = clean(row.processInstanceId)
|
||
const status = clean(row.status)
|
||
const sourceCreatedAtRaw = sourceWorkflowTimestampToIso(row.sourceCreatedAtRaw)
|
||
if (!processInstanceId || !status || !isDingWorkflowStartedOnOrAfterCutoff(row.sourceCreatedAtRaw)) return []
|
||
const ongoing = isDingWorkflowOngoing(status, row.sourceFinishedAtRaw)
|
||
const task = taskByWorkflow.get(processInstanceId)
|
||
const scheme = schemeByWorkflow.get(processInstanceId)
|
||
const processCode = normalizeDingWorkflowProcessCode(row.processCode)
|
||
const schemeFields = schemeRowsByWorkflow.get(processInstanceId) || []
|
||
const assignments = extractContactAssignments(schemeFields)
|
||
const brand = normalizeDingWorkflowBrand(firstSchemeValue(schemeFields, ['品牌']))
|
||
const demandType = firstSchemeValue(schemeFields, ['需求类型'])
|
||
const requesterDepartment = normalizeDingWorkflowDepartment(firstSchemeValue(schemeFields, ['需求部门']))
|
||
|| normalizeDingWorkflowDepartment(row.originatorDeptName)
|
||
|| '未记录部门'
|
||
const originatorUserId = clean(row.originatorUserId)
|
||
const requesterName = firstSchemeValue(schemeFields, ['需求人'])
|
||
|| (originatorUserId ? employeeDirectory.get(originatorUserId)?.name : null)
|
||
|| titleOriginatorName(row.title)
|
||
|| '未记录'
|
||
const workflowTasks = (taskRowsByWorkflow.get(processInstanceId) || []).flatMap((sourceTask) => {
|
||
const taskId = clean(sourceTask.taskId)
|
||
if (!taskId) return []
|
||
const activityId = clean(sourceTask.activityId)
|
||
const assigneeUserId = clean(sourceTask.userId)
|
||
const taskStatus = clean(sourceTask.status)?.toUpperCase() || null
|
||
const taskResult = clean(sourceTask.result)?.toUpperCase() || null
|
||
const isActive = ongoing && isDingWorkflowTaskOngoing(sourceTask.status, sourceTask.finishTime)
|
||
const matchingAssignments = assigneeUserId
|
||
? assignments.filter((assignment) => assignment.userId === assigneeUserId)
|
||
: []
|
||
const employee = assigneeUserId ? employeeDirectory.get(assigneeUserId) : null
|
||
const uniqueAssignmentNodeNames = Array.from(new Set(matchingAssignments.map((assignment) => assignment.nodeName)))
|
||
const configuredNodeName = getDingWorkflowActivityNodeName(processCode, activityId)
|
||
const inferredNodeName = processCode && activityId
|
||
? activityNodeNames.get(`${processCode}:${activityId}`)
|
||
: null
|
||
// Inference can label otherwise unnamed production tasks, but it must
|
||
// never manufacture a review milestone. Review nodes are identified by
|
||
// the explicit activityId mapping above.
|
||
const safeInferredNodeName = inferredNodeName && !CONFIRMED_REVIEW_NODE_NAMES.has(inferredNodeName)
|
||
? inferredNodeName
|
||
: null
|
||
const safeAssignmentNodeName = uniqueAssignmentNodeNames.find(
|
||
(nodeName) => !CONFIRMED_REVIEW_NODE_NAMES.has(nodeName),
|
||
) || null
|
||
const nodeName = configuredNodeName
|
||
|| safeInferredNodeName
|
||
|| safeAssignmentNodeName
|
||
|| (isActive ? '当前处理节点' : '历史处理节点')
|
||
// Once the activity id resolves to a confirmed review node, completion
|
||
// is determined by the task decision itself. The employee directory is
|
||
// intentionally not part of this gate: a stale/missing roster entry
|
||
// must not erase a valid review timestamp from the dashboard.
|
||
const reviewPassed = CONFIRMED_REVIEW_NODE_NAMES.has(nodeName)
|
||
&& taskStatus === 'COMPLETED'
|
||
&& taskResult === 'AGREE'
|
||
return [{
|
||
taskId,
|
||
activityId,
|
||
nodeName,
|
||
assigneeUserId,
|
||
assigneeName: matchingAssignments.find((assignment) => assignment.name)?.name
|
||
|| employee?.name
|
||
|| (assigneeUserId === 'bpms_system' ? '系统处理' : '待同步人员'),
|
||
assigneeDepartment: employee?.departmentName || null,
|
||
isBrandCenter: employee?.isBrandCenter === true,
|
||
isLeadership: employee?.isLeadership === true,
|
||
isActive,
|
||
reviewPassed,
|
||
result: taskResult,
|
||
createdAt: sourceWorkflowTimestampToIso(sourceTask.createTime),
|
||
finishedAt: sourceWorkflowTimestampToIso(sourceTask.finishTime),
|
||
}]
|
||
})
|
||
|
||
return [{
|
||
processInstanceId,
|
||
businessId: clean(row.businessId),
|
||
title: clean(row.title),
|
||
result: clean(row.result),
|
||
originatorUserId,
|
||
originatorDeptId: clean(row.originatorDeptId),
|
||
originatorDeptName: clean(row.originatorDeptName),
|
||
status: status.toUpperCase(),
|
||
bizAction: clean(row.bizAction),
|
||
processCode,
|
||
flowType: clean(row.flowType),
|
||
joinedAtRaw: clean(row.joinedAtRaw),
|
||
sourceCreatedAtRaw,
|
||
sourceUpdatedAtRaw: sourceWorkflowTimestampToIso(row.sourceUpdatedAtRaw),
|
||
sourceFinishedAtRaw: sourceWorkflowTimestampToIso(row.sourceFinishedAtRaw),
|
||
isOngoing: ongoing,
|
||
schemeFieldCount: nonNegativeInteger(scheme?.schemeFieldCount),
|
||
taskCount: nonNegativeInteger(task?.taskCount),
|
||
activeTaskCount: ongoing ? nonNegativeInteger(task?.activeTaskCount) : 0,
|
||
latestTaskStatus: clean(task?.latestTaskStatus)?.toUpperCase() || null,
|
||
latestTaskAtRaw: sourceWorkflowTimestampToIso(task?.latestTaskAtRaw),
|
||
brand,
|
||
demandType,
|
||
requesterName,
|
||
requesterDepartment,
|
||
projectType: getDingWorkflowProjectType(processCode),
|
||
formFields: buildDingWorkflowDetailFormFields(schemeFields),
|
||
activeTasks: workflowTasks,
|
||
}]
|
||
})
|
||
}
|
||
|
||
async function upsertSnapshotBatch(rows: WorkflowSnapshotPayload[], syncedAt: Date) {
|
||
if (rows.length === 0) return
|
||
await prisma.$executeRawUnsafe(`
|
||
INSERT INTO "ding_workflow_snapshots" (
|
||
"process_instance_id", "business_id", "title", "result",
|
||
"originator_user_id", "originator_dept_id", "originator_dept_name",
|
||
"status", "biz_action", "process_code", "flow_type", "joined_at_raw",
|
||
"source_created_at_raw", "source_updated_at_raw", "source_finished_at_raw",
|
||
"is_ongoing", "scheme_field_count", "task_count", "active_task_count",
|
||
"latest_task_status", "latest_task_at_raw", "brand", "demand_type", "requester_name",
|
||
"requester_department", "project_type", "form_fields", "active_tasks", "synced_at", "updated_at"
|
||
)
|
||
SELECT
|
||
incoming."processInstanceId", incoming."businessId", incoming."title", incoming."result",
|
||
incoming."originatorUserId", incoming."originatorDeptId", incoming."originatorDeptName",
|
||
incoming."status", incoming."bizAction", incoming."processCode", incoming."flowType", incoming."joinedAtRaw",
|
||
incoming."sourceCreatedAtRaw", incoming."sourceUpdatedAtRaw", incoming."sourceFinishedAtRaw",
|
||
incoming."isOngoing", incoming."schemeFieldCount", incoming."taskCount", incoming."activeTaskCount",
|
||
incoming."latestTaskStatus", incoming."latestTaskAtRaw", incoming."brand", incoming."demandType", incoming."requesterName",
|
||
incoming."requesterDepartment", incoming."projectType", incoming."formFields", incoming."activeTasks", $2::timestamp, $2::timestamp
|
||
FROM jsonb_to_recordset($1::jsonb) AS incoming(
|
||
"processInstanceId" text, "businessId" text, "title" text, "result" text,
|
||
"originatorUserId" text, "originatorDeptId" text, "originatorDeptName" text,
|
||
"status" text, "bizAction" text, "processCode" text, "flowType" text, "joinedAtRaw" text,
|
||
"sourceCreatedAtRaw" text, "sourceUpdatedAtRaw" text, "sourceFinishedAtRaw" text,
|
||
"isOngoing" boolean, "schemeFieldCount" integer, "taskCount" integer, "activeTaskCount" integer,
|
||
"latestTaskStatus" text, "latestTaskAtRaw" text, "brand" text, "demandType" text, "requesterName" text,
|
||
"requesterDepartment" text, "projectType" text, "formFields" jsonb, "activeTasks" jsonb
|
||
)
|
||
ON CONFLICT ("process_instance_id") DO UPDATE SET
|
||
"business_id" = EXCLUDED."business_id",
|
||
"title" = EXCLUDED."title",
|
||
"result" = EXCLUDED."result",
|
||
"originator_user_id" = EXCLUDED."originator_user_id",
|
||
"originator_dept_id" = EXCLUDED."originator_dept_id",
|
||
"originator_dept_name" = EXCLUDED."originator_dept_name",
|
||
"status" = EXCLUDED."status",
|
||
"biz_action" = EXCLUDED."biz_action",
|
||
"process_code" = EXCLUDED."process_code",
|
||
"flow_type" = EXCLUDED."flow_type",
|
||
"joined_at_raw" = EXCLUDED."joined_at_raw",
|
||
"source_created_at_raw" = EXCLUDED."source_created_at_raw",
|
||
"source_updated_at_raw" = EXCLUDED."source_updated_at_raw",
|
||
"source_finished_at_raw" = EXCLUDED."source_finished_at_raw",
|
||
"is_ongoing" = EXCLUDED."is_ongoing",
|
||
"scheme_field_count" = EXCLUDED."scheme_field_count",
|
||
"task_count" = EXCLUDED."task_count",
|
||
"active_task_count" = EXCLUDED."active_task_count",
|
||
"latest_task_status" = EXCLUDED."latest_task_status",
|
||
"latest_task_at_raw" = EXCLUDED."latest_task_at_raw",
|
||
"brand" = EXCLUDED."brand",
|
||
"demand_type" = EXCLUDED."demand_type",
|
||
"requester_name" = EXCLUDED."requester_name",
|
||
"requester_department" = EXCLUDED."requester_department",
|
||
"project_type" = EXCLUDED."project_type",
|
||
"form_fields" = EXCLUDED."form_fields",
|
||
"active_tasks" = EXCLUDED."active_tasks",
|
||
"synced_at" = EXCLUDED."synced_at",
|
||
"updated_at" = EXCLUDED."updated_at"
|
||
`, JSON.stringify(rows), syncedAt)
|
||
}
|
||
|
||
async function acquireSyncLease(leaseOwner: string, startedAt: Date) {
|
||
await prisma.dingWorkflowSyncState.upsert({
|
||
where: { id: SYNC_STATE_ID },
|
||
create: { id: SYNC_STATE_ID },
|
||
update: {},
|
||
})
|
||
const claimed = await prisma.dingWorkflowSyncState.updateMany({
|
||
where: {
|
||
id: SYNC_STATE_ID,
|
||
OR: [
|
||
{ leaseExpiresAt: null },
|
||
{ leaseExpiresAt: { lt: startedAt } },
|
||
{ leaseOwner },
|
||
],
|
||
},
|
||
data: {
|
||
status: 'syncing',
|
||
lastStartedAt: startedAt,
|
||
lastError: null,
|
||
leaseOwner,
|
||
leaseExpiresAt: new Date(startedAt.getTime() + LEASE_DURATION_MS),
|
||
},
|
||
})
|
||
if (claimed.count !== 1) return null
|
||
return prisma.dingWorkflowSyncState.findUniqueOrThrow({ where: { id: SYNC_STATE_ID } })
|
||
}
|
||
|
||
export async function syncDingWorkflowSnapshots() {
|
||
if (!isReportWorkflowDbConfigured()) {
|
||
throw new Error('未配置上游报表库（REPORT_DB_*），无法同步钉钉流程')
|
||
}
|
||
|
||
const startedAt = new Date()
|
||
const leaseOwner = `${hostname()}:${process.pid}:${randomUUID()}`
|
||
const state = await acquireSyncLease(leaseOwner, startedAt)
|
||
if (!state) return { skipped: true, rowsUpserted: 0, rowsDeleted: 0, ongoingCount: null }
|
||
|
||
const pool = new mssql.ConnectionPool(buildSourceConfig())
|
||
try {
|
||
await pool.connect()
|
||
const sourceCursor = parseDingWorkflowSourceCursor(state.sourceCursorRaw)
|
||
const sourceDelta = await fetchSourceDeltaState(pool, sourceCursor)
|
||
const sourceRows = await fetchSourceWorkflows(
|
||
pool,
|
||
sourceDelta.requiresFullSync ? null : sourceCursor.workflowUpdatedAt,
|
||
sourceDelta.changedWorkflowIds,
|
||
)
|
||
const workflowIds = sourceRows.flatMap((row) => {
|
||
const id = clean(row.processInstanceId)
|
||
return id ? [id] : []
|
||
})
|
||
const {
|
||
taskByWorkflow,
|
||
taskRowsByWorkflow,
|
||
schemeByWorkflow,
|
||
schemeRowsByWorkflow,
|
||
} = await fetchWorkflowAggregates(pool, workflowIds)
|
||
const employeeDirectory = await readEmployeeDirectory(sourceRows, taskRowsByWorkflow)
|
||
const activityNodeNames = inferActivityNodeNames(sourceRows, taskRowsByWorkflow, schemeRowsByWorkflow)
|
||
const rows = buildSnapshotPayload(
|
||
sourceRows,
|
||
taskByWorkflow,
|
||
taskRowsByWorkflow,
|
||
schemeByWorkflow,
|
||
schemeRowsByWorkflow,
|
||
employeeDirectory,
|
||
activityNodeNames,
|
||
)
|
||
const syncedAt = new Date()
|
||
|
||
for (let index = 0; index < rows.length; index += UPSERT_BATCH_SIZE) {
|
||
await upsertSnapshotBatch(rows.slice(index, index + UPSERT_BATCH_SIZE), syncedAt)
|
||
}
|
||
|
||
const deleted = await prisma.dingWorkflowSnapshot.deleteMany({
|
||
where: {
|
||
OR: [
|
||
{ processCode: null },
|
||
{ processCode: { notIn: ALLOWED_PROCESS_CODE_VALUES } },
|
||
{ sourceCreatedAtRaw: null },
|
||
{ sourceCreatedAtRaw: { lt: DING_WORKFLOW_STARTED_AT_CUTOFF_ISO } },
|
||
],
|
||
},
|
||
})
|
||
const ongoingCount = await prisma.dingWorkflowSnapshot.count({
|
||
where: buildBrandCenterOngoingWorkflowWhere(),
|
||
})
|
||
const sourceCursorRaw = serializeDingWorkflowSourceCursor({
|
||
version: 3,
|
||
workflowUpdatedAt: maxSourceCursor(rows, sourceCursor.workflowUpdatedAt),
|
||
taskId: sourceDelta.maxTaskId,
|
||
schemeId: sourceDelta.maxSchemeId,
|
||
})
|
||
await prisma.dingWorkflowSyncState.updateMany({
|
||
where: { id: SYNC_STATE_ID, leaseOwner },
|
||
data: {
|
||
status: 'ready',
|
||
sourceCursorRaw,
|
||
lastCompletedAt: syncedAt,
|
||
lastError: null,
|
||
leaseOwner: null,
|
||
leaseExpiresAt: null,
|
||
rowsUpserted: rows.length,
|
||
currentOngoingCount: ongoingCount,
|
||
},
|
||
})
|
||
return { skipped: false, rowsUpserted: rows.length, rowsDeleted: deleted.count, ongoingCount }
|
||
} catch (error) {
|
||
const message = error instanceof Error ? error.message : String(error)
|
||
await prisma.dingWorkflowSyncState.updateMany({
|
||
where: { id: SYNC_STATE_ID, leaseOwner },
|
||
data: {
|
||
status: 'error',
|
||
lastError: message.slice(0, 2_000),
|
||
leaseOwner: null,
|
||
leaseExpiresAt: null,
|
||
},
|
||
}).catch(() => undefined)
|
||
throw error
|
||
} finally {
|
||
await pool.close().catch(() => undefined)
|
||
}
|
||
}
|
||
|
||
function normalizeSyncStatus(state: { status: string; lastCompletedAt: Date | null } | null): DingWorkflowSyncStatus {
|
||
if (!state?.lastCompletedAt) return state?.status === 'syncing' ? 'syncing' : state?.status === 'error' ? 'error' : 'pending'
|
||
if (state.status === 'syncing') return 'syncing'
|
||
if (state.status === 'error') return 'error'
|
||
return 'ready'
|
||
}
|
||
|
||
async function readSummaryWithoutAccessCheck(): Promise<OngoingDingWorkflowSummary> {
|
||
const [total, state] = await prisma.$transaction([
|
||
prisma.dingWorkflowSnapshot.count({
|
||
where: buildBrandCenterOngoingWorkflowWhere(),
|
||
}),
|
||
prisma.dingWorkflowSyncState.findUnique({
|
||
where: { id: SYNC_STATE_ID },
|
||
select: { status: true, lastCompletedAt: true },
|
||
}),
|
||
])
|
||
return {
|
||
total,
|
||
lastSyncedAt: state?.lastCompletedAt?.toISOString() || null,
|
||
syncStatus: normalizeSyncStatus(state),
|
||
}
|
||
}
|
||
|
||
export async function getOngoingDingWorkflowSummary(session: Session | null) {
|
||
await assertDashboardAccess(session)
|
||
return readSummaryWithoutAccessCheck()
|
||
}
|
||
|
||
const ONGOING_WORKFLOW_SELECT = {
|
||
processInstanceId: true,
|
||
businessId: true,
|
||
title: true,
|
||
status: true,
|
||
processCode: true,
|
||
projectType: true,
|
||
brand: true,
|
||
requesterName: true,
|
||
requesterDepartment: true,
|
||
formFields: true,
|
||
activeTasks: true,
|
||
originatorUserId: true,
|
||
originatorDeptName: true,
|
||
sourceCreatedAtRaw: true,
|
||
sourceUpdatedAtRaw: true,
|
||
activeTaskCount: true,
|
||
taskCount: true,
|
||
schemeFieldCount: true,
|
||
latestTaskStatus: true,
|
||
} as const
|
||
|
||
const MONTHLY_LAUNCH_WORKFLOW_SELECT = {
|
||
...ONGOING_WORKFLOW_SELECT,
|
||
result: true,
|
||
bizAction: true,
|
||
sourceFinishedAtRaw: true,
|
||
} as const
|
||
|
||
type DingWorkflowInitiatorProfile = {
|
||
displayName: string | null
|
||
departmentName: string | null
|
||
departmentPath: string | null
|
||
}
|
||
|
||
const COMPANY_ORGANIZATION_ROOT = '江苏简尚品牌管理有限公司'
|
||
|
||
function formatInitiatorDepartmentPath(value: string | null | undefined) {
|
||
const parts = String(value || '')
|
||
.split('/')
|
||
.map((part) => part.trim())
|
||
.filter(Boolean)
|
||
if (parts[0] === COMPANY_ORGANIZATION_ROOT) parts.shift()
|
||
return parts.join('-')
|
||
}
|
||
|
||
export function resolveDingWorkflowInitiator(
|
||
profile: DingWorkflowInitiatorProfile | null | undefined,
|
||
fallback: { name: string | null | undefined; department: string | null | undefined },
|
||
) {
|
||
return {
|
||
name: profile?.displayName?.trim() || fallback.name?.trim() || '未记录',
|
||
department: formatInitiatorDepartmentPath(profile?.departmentPath)
|
||
|| profile?.departmentName?.trim()
|
||
|| normalizeDingWorkflowDepartment(fallback.department)
|
||
|| '未记录部门',
|
||
}
|
||
}
|
||
|
||
async function readInitiatorDirectory(rows: Array<{ originatorUserId: string | null }>) {
|
||
const dingtalkUserIds = Array.from(new Set(
|
||
rows.map((row) => row.originatorUserId?.trim()).filter((value): value is string => Boolean(value)),
|
||
))
|
||
if (dingtalkUserIds.length === 0) return new Map<string, DingWorkflowInitiatorProfile>()
|
||
|
||
const profiles = await prisma.performanceSurveyEmployeeProfile.findMany({
|
||
where: { dingtalkUserId: { in: dingtalkUserIds } },
|
||
select: {
|
||
dingtalkUserId: true,
|
||
displayName: true,
|
||
departmentName: true,
|
||
departmentPath: true,
|
||
},
|
||
})
|
||
return new Map(profiles.flatMap((profile) => (
|
||
profile.dingtalkUserId ? [[profile.dingtalkUserId, profile] as const] : []
|
||
)))
|
||
}
|
||
|
||
function mapSnapshotListItem(row: {
|
||
processInstanceId: string
|
||
businessId: string | null
|
||
title: string | null
|
||
status: string
|
||
processCode: string | null
|
||
projectType: string | null
|
||
brand: string | null
|
||
requesterName: string | null
|
||
requesterDepartment: string | null
|
||
formFields: unknown
|
||
activeTasks: unknown
|
||
originatorUserId: string | null
|
||
originatorDeptName: string | null
|
||
sourceCreatedAtRaw: string | null
|
||
sourceUpdatedAtRaw: string | null
|
||
activeTaskCount: number
|
||
taskCount: number
|
||
schemeFieldCount: number
|
||
latestTaskStatus: string | null
|
||
}, initiatorProfile?: DingWorkflowInitiatorProfile | null): OngoingDingWorkflowListItem {
|
||
const processCode = normalizeDingWorkflowProcessCode(row.processCode) || 'UNKNOWN'
|
||
const initiator = resolveDingWorkflowInitiator(initiatorProfile, {
|
||
name: row.requesterName,
|
||
department: row.originatorDeptName || row.requesterDepartment,
|
||
})
|
||
const tasks = normalizeWorkflowTasks(row.activeTasks)
|
||
const milestones = resolveDingWorkflowDetailMilestones(processCode, tasks)
|
||
return {
|
||
id: row.processInstanceId,
|
||
businessId: row.businessId,
|
||
title: row.title || '未命名流程',
|
||
status: row.status,
|
||
processCode,
|
||
projectType: row.projectType || getDingWorkflowProjectType(processCode),
|
||
brand: row.brand || '其他品牌',
|
||
requesterName: row.requesterName || '未记录',
|
||
requesterDepartment: normalizeDingWorkflowDepartment(row.requesterDepartment)
|
||
|| normalizeDingWorkflowDepartment(row.originatorDeptName)
|
||
|| '未记录部门',
|
||
initiatorName: initiator.name,
|
||
initiatorDepartment: initiator.department,
|
||
formFields: normalizeDetailFormFields(row.formFields),
|
||
...milestones,
|
||
activeTasks: tasks.filter((task) => task.isActive),
|
||
originatorUserId: row.originatorUserId,
|
||
originatorDeptName: row.originatorDeptName,
|
||
sourceCreatedAt: storedWorkflowTimestampToIso(row.sourceCreatedAtRaw),
|
||
sourceUpdatedAt: storedWorkflowTimestampToIso(row.sourceUpdatedAtRaw),
|
||
activeTaskCount: row.activeTaskCount,
|
||
taskCount: row.taskCount,
|
||
schemeFieldCount: row.schemeFieldCount,
|
||
latestTaskStatus: row.latestTaskStatus,
|
||
}
|
||
}
|
||
|
||
export async function getOngoingDingWorkflowInsights(session: Session | null): Promise<OngoingDingWorkflowInsights> {
|
||
await assertDashboardAccess(session)
|
||
const where = buildBrandCenterOngoingWorkflowWhere()
|
||
const [rows, sync] = await Promise.all([
|
||
prisma.dingWorkflowSnapshot.findMany({
|
||
where,
|
||
select: ONGOING_WORKFLOW_SELECT,
|
||
orderBy: [
|
||
{ sourceUpdatedAtRaw: { sort: 'desc', nulls: 'last' } },
|
||
{ processInstanceId: 'desc' },
|
||
],
|
||
}),
|
||
readSummaryWithoutAccessCheck(),
|
||
])
|
||
const initiatorDirectory = await readInitiatorDirectory(rows)
|
||
return {
|
||
items: rows.map((row) => mapSnapshotListItem(
|
||
row,
|
||
row.originatorUserId ? initiatorDirectory.get(row.originatorUserId.trim()) : null,
|
||
)),
|
||
total: rows.length,
|
||
sync,
|
||
}
|
||
}
|
||
|
||
export async function getOngoingDingWorkflowProjects(
|
||
session: Session | null,
|
||
params?: { page?: number; pageSize?: number; query?: string },
|
||
): Promise<OngoingDingWorkflowPage> {
|
||
await assertDashboardAccess(session)
|
||
const page = Math.max(1, Number(params?.page || 1) || 1)
|
||
const pageSize = Math.min(50, Math.max(1, Number(params?.pageSize || 20) || 20))
|
||
const query = clean(params?.query)
|
||
const where = {
|
||
...buildBrandCenterOngoingWorkflowWhere(),
|
||
...(query ? {
|
||
OR: [
|
||
{ title: { contains: query, mode: 'insensitive' as const } },
|
||
{ businessId: { contains: query, mode: 'insensitive' as const } },
|
||
{ processInstanceId: { contains: query, mode: 'insensitive' as const } },
|
||
{ originatorDeptName: { contains: query, mode: 'insensitive' as const } },
|
||
{ originatorUserId: { contains: query, mode: 'insensitive' as const } },
|
||
],
|
||
} : {}),
|
||
}
|
||
|
||
const [total, rows, sync] = await Promise.all([
|
||
prisma.dingWorkflowSnapshot.count({ where }),
|
||
prisma.dingWorkflowSnapshot.findMany({
|
||
where,
|
||
select: ONGOING_WORKFLOW_SELECT,
|
||
orderBy: [
|
||
{ sourceUpdatedAtRaw: { sort: 'desc', nulls: 'last' } },
|
||
{ processInstanceId: 'desc' },
|
||
],
|
||
skip: (page - 1) * pageSize,
|
||
take: pageSize,
|
||
}),
|
||
readSummaryWithoutAccessCheck(),
|
||
])
|
||
const initiatorDirectory = await readInitiatorDirectory(rows)
|
||
|
||
return {
|
||
items: rows.map((row) => mapSnapshotListItem(
|
||
row,
|
||
row.originatorUserId ? initiatorDirectory.get(row.originatorUserId.trim()) : null,
|
||
)),
|
||
pagination: {
|
||
page,
|
||
pageSize,
|
||
total,
|
||
totalPages: Math.max(1, Math.ceil(total / pageSize)),
|
||
},
|
||
sync,
|
||
}
|
||
}
|
||
|
||
export function buildMonthlyLaunchWorkflowWhere() {
|
||
return {
|
||
processCode: DING_WORKFLOW_ALLOWED_PROCESS_CODES[0],
|
||
sourceCreatedAtRaw: { gte: DING_WORKFLOW_STARTED_AT_CUTOFF_ISO },
|
||
brand: { in: [...DING_WORKFLOW_BRANDS] },
|
||
demandType: DING_WORKFLOW_NEW_PRODUCT_DEMAND_TYPE,
|
||
}
|
||
}
|
||
|
||
export function buildMonthlyLaunchOngoingWorkflowWhere() {
|
||
return {
|
||
...buildBrandCenterOngoingArtWorkflowWhere(),
|
||
brand: { in: [...DING_WORKFLOW_BRANDS] },
|
||
demandType: DING_WORKFLOW_NEW_PRODUCT_DEMAND_TYPE,
|
||
}
|
||
}
|
||
|
||
async function readMonthlyLaunchInsightsWithoutAccessCheck(now = new Date()): Promise<MonthlyLaunchInsights> {
|
||
const range = getShanghaiMonthRange(now)
|
||
const baseWhere = buildMonthlyLaunchWorkflowWhere()
|
||
const [completedCandidates, ongoingRows, state] = await Promise.all([
|
||
prisma.dingWorkflowSnapshot.findMany({
|
||
where: {
|
||
...baseWhere,
|
||
status: 'COMPLETED',
|
||
sourceFinishedAtRaw: {
|
||
gte: range.start.toISOString(),
|
||
lt: range.end.toISOString(),
|
||
},
|
||
},
|
||
select: MONTHLY_LAUNCH_WORKFLOW_SELECT,
|
||
}),
|
||
prisma.dingWorkflowSnapshot.findMany({
|
||
where: buildMonthlyLaunchOngoingWorkflowWhere(),
|
||
select: MONTHLY_LAUNCH_WORKFLOW_SELECT,
|
||
}),
|
||
prisma.dingWorkflowSyncState.findUnique({
|
||
where: { id: SYNC_STATE_ID },
|
||
select: { status: true, lastCompletedAt: true },
|
||
}),
|
||
])
|
||
const completedRows = completedCandidates.filter((row) => (
|
||
isDingWorkflowCompletedInRange(row.status, row.sourceFinishedAtRaw, range)
|
||
))
|
||
const completedIds = new Set(completedRows.map((row) => row.processInstanceId))
|
||
const uniqueOngoingRows = ongoingRows.filter((row) => !completedIds.has(row.processInstanceId))
|
||
const allRows = [...completedRows, ...uniqueOngoingRows]
|
||
const initiatorDirectory = await readInitiatorDirectory(allRows)
|
||
const mapMonthlyItem = (
|
||
row: typeof allRows[number],
|
||
launchStatus: MonthlyLaunchWorkflowStatus,
|
||
): MonthlyLaunchWorkflowItem => ({
|
||
...mapSnapshotListItem(
|
||
row,
|
||
row.originatorUserId ? initiatorDirectory.get(row.originatorUserId.trim()) : null,
|
||
),
|
||
launchStatus,
|
||
result: row.result,
|
||
sourceFinishedAt: storedWorkflowTimestampToIso(row.sourceFinishedAtRaw),
|
||
})
|
||
const items = [
|
||
...completedRows.map((row) => mapMonthlyItem(row, 'completed')),
|
||
...uniqueOngoingRows.map((row) => mapMonthlyItem(row, 'ongoing')),
|
||
].sort((a, b) => {
|
||
const aTime = a.launchStatus === 'completed' ? a.sourceFinishedAt : a.sourceUpdatedAt
|
||
const bTime = b.launchStatus === 'completed' ? b.sourceFinishedAt : b.sourceUpdatedAt
|
||
return String(bTime || '').localeCompare(String(aTime || '')) || b.id.localeCompare(a.id)
|
||
})
|
||
const brands = DING_WORKFLOW_BRANDS.map((brand) => {
|
||
const brandItems = items.filter((item) => item.brand === brand)
|
||
const completed = brandItems.filter((item) => item.launchStatus === 'completed').length
|
||
const ongoing = brandItems.filter((item) => item.launchStatus === 'ongoing').length
|
||
return { brand, completed, ongoing, total: completed + ongoing }
|
||
})
|
||
const completed = completedRows.length
|
||
const ongoing = uniqueOngoingRows.length
|
||
|
||
return {
|
||
items,
|
||
brands,
|
||
summary: {
|
||
completed,
|
||
ongoing,
|
||
total: completed + ongoing,
|
||
monthStart: range.start.toISOString(),
|
||
monthEnd: range.end.toISOString(),
|
||
lastSyncedAt: state?.lastCompletedAt?.toISOString() || null,
|
||
syncStatus: normalizeSyncStatus(state),
|
||
},
|
||
}
|
||
}
|
||
|
||
export async function getMonthlyLaunchSummary(session: Session | null): Promise<MonthlyLaunchSummary> {
|
||
await assertDashboardAccess(session)
|
||
return (await readMonthlyLaunchInsightsWithoutAccessCheck()).summary
|
||
}
|
||
|
||
export async function getMonthlyLaunchInsights(session: Session | null): Promise<MonthlyLaunchInsights> {
|
||
await assertDashboardAccess(session)
|
||
return readMonthlyLaunchInsightsWithoutAccessCheck()
|
||
}
|
||
|
||
async function readOnTimeDeliveryInsightsWithoutAccessCheck(now = new Date()): Promise<OnTimeDeliveryInsights> {
|
||
const range = getShanghaiMonthRange(now)
|
||
const [candidates, state] = await Promise.all([
|
||
prisma.dingWorkflowSnapshot.findMany({
|
||
where: {
|
||
processCode: { in: ALLOWED_PROCESS_CODE_VALUES },
|
||
sourceCreatedAtRaw: { gte: DING_WORKFLOW_STARTED_AT_CUTOFF_ISO },
|
||
},
|
||
select: MONTHLY_LAUNCH_WORKFLOW_SELECT,
|
||
orderBy: [
|
||
{ sourceFinishedAtRaw: { sort: 'desc', nulls: 'last' } },
|
||
{ processInstanceId: 'desc' },
|
||
],
|
||
}),
|
||
prisma.dingWorkflowSyncState.findUnique({
|
||
where: { id: SYNC_STATE_ID },
|
||
select: { status: true, lastCompletedAt: true },
|
||
}),
|
||
])
|
||
// Keep in-flight rows while calculating the shared content queue: the
|
||
// external dashboard lets earlier in-flight assignments consume capacity,
|
||
// even though only approved COMPLETED rows enter this month's denominator.
|
||
const rows = candidates.filter((row) => (
|
||
String(row.status || '').trim().toUpperCase() !== 'COMPLETED'
|
||
|| isDingWorkflowApprovedCompletion(row.status, row.result, row.bizAction)
|
||
))
|
||
const initiatorDirectory = await readInitiatorDirectory(rows)
|
||
const assessmentInputs = rows.map((row) => {
|
||
const processCode = normalizeDingWorkflowProcessCode(row.processCode) || 'UNKNOWN'
|
||
const allTasks = normalizeWorkflowTasks(row.activeTasks)
|
||
const sourceCreatedAt = storedWorkflowTimestampToIso(row.sourceCreatedAtRaw)
|
||
const sourceFinishedAt = storedWorkflowTimestampToIso(row.sourceFinishedAtRaw)
|
||
return {
|
||
processCode,
|
||
sourceCreatedAt,
|
||
sourceFinishedAt,
|
||
formFields: normalizeDetailFormFields(row.formFields),
|
||
tasks: allTasks,
|
||
businessId: clean(row.businessId),
|
||
workflowId: clean(row.processInstanceId),
|
||
workflowStatus: clean(row.status),
|
||
}
|
||
})
|
||
const assessments = assessDingWorkflowDeliveryBatch(assessmentInputs)
|
||
const assessedItems = rows.map((row, index): OnTimeDeliveryWorkflowItem => {
|
||
const sourceFinishedAt = storedWorkflowTimestampToIso(row.sourceFinishedAtRaw)
|
||
return {
|
||
...mapSnapshotListItem(
|
||
row,
|
||
row.originatorUserId ? initiatorDirectory.get(row.originatorUserId.trim()) : null,
|
||
),
|
||
result: row.result,
|
||
sourceFinishedAt,
|
||
assessment: assessments[index],
|
||
}
|
||
})
|
||
const items = assessedItems.filter((item) => (
|
||
String(item.status || '').trim().toUpperCase() === 'COMPLETED'
|
||
&& isDingWorkflowDeliveryAssessmentInRange(item.assessment, range)
|
||
))
|
||
const onTime = items.filter((item) => item.assessment.status === 'on_time').length
|
||
const overtime = items.filter((item) => item.assessment.status === 'overtime').length
|
||
const pending = items.filter((item) => item.assessment.status === 'pending').length
|
||
const total = items.length
|
||
|
||
return {
|
||
items,
|
||
summary: {
|
||
onTime,
|
||
overtime,
|
||
pending,
|
||
total,
|
||
rate: total > 0 ? Math.round((onTime / total) * 1_000) / 10 : null,
|
||
monthStart: range.start.toISOString(),
|
||
monthEnd: range.end.toISOString(),
|
||
lastSyncedAt: state?.lastCompletedAt?.toISOString() || null,
|
||
syncStatus: normalizeSyncStatus(state),
|
||
},
|
||
}
|
||
}
|
||
|
||
export async function getOnTimeDeliverySummary(session: Session | null): Promise<OnTimeDeliverySummary> {
|
||
await assertDashboardAccess(session)
|
||
return (await readOnTimeDeliveryInsightsWithoutAccessCheck()).summary
|
||
}
|
||
|
||
export async function getOnTimeDeliveryInsights(session: Session | null): Promise<OnTimeDeliveryInsights> {
|
||
await assertDashboardAccess(session)
|
||
return readOnTimeDeliveryInsightsWithoutAccessCheck()
|
||
}
|
||
|
||
type DingWorkflowSchedulerGlobal = typeof globalThis & {
|
||
__dingWorkflowHourlySyncStarted?: boolean
|
||
}
|
||
|
||
export function ensureHourlyDingWorkflowSyncStarted() {
|
||
const schedulerGlobal = globalThis as DingWorkflowSchedulerGlobal
|
||
if (schedulerGlobal.__dingWorkflowHourlySyncStarted) return
|
||
schedulerGlobal.__dingWorkflowHourlySyncStarted = true
|
||
|
||
if (!isReportWorkflowDbConfigured()) {
|
||
console.info('[ding-workflow-sync] 未配置上游报表库，跳过每小时同步')
|
||
return
|
||
}
|
||
|
||
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)
|
||
: DEFAULT_SYNC_INTERVAL_MS
|
||
const runOnce = () => {
|
||
void syncDingWorkflowSnapshots()
|
||
.then((result) => {
|
||
if (result.skipped) {
|
||
console.info('[ding-workflow-sync] 另一实例正在同步，本轮跳过')
|
||
return
|
||
}
|
||
console.info(`[ding-workflow-sync] 同步完成：更新 ${result.rowsUpserted} 条，清理 ${result.rowsDeleted} 条，进行中 ${result.ongoingCount} 条`)
|
||
})
|
||
.catch((error) => console.warn('[ding-workflow-sync] 同步失败：', error))
|
||
}
|
||
|
||
const initialTimer = setTimeout(runOnce, INITIAL_SYNC_DELAY_MS)
|
||
initialTimer.unref?.()
|
||
const interval = setInterval(runOnce, intervalMs)
|
||
interval.unref?.()
|
||
console.info(`[ding-workflow-sync] 已调度每小时增量同步，周期 ${Math.round(intervalMs / 60_000)} 分钟`)
|
||
}
|