gongxiangp2/lib/organization-sync.ts

646 lines
22 KiB
TypeScript
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

/**
* 组织架构与在职状态同步
*
* 直接镜像上游 HR 系统(rpt schema),取代此前两处靠猜的做法:
*
* 1. 组织架构——此前部门树是遍历已同步员工的 departmentPath 字符串现场拼出来的,
* 因此只能看到"有人的部门",同名部门会被合并成一行,父子归属取决于谁先被遍历到。
* 上游 hr_groups 实际是一棵完整单根树(单一根节点、零孤儿节点),直接同步即可。
*
* 2. 在职状态——上游口径混杂:数字码(2 试用 / 3 正式 / 5 已离职)、中文(正式/离职/试用)
* 与空值并存。此前用「离职/停用/退休」做关键词包含匹配,对数字码和空值完全失效,
* 已核实 216 条已离职档案中有 215 条被判定为在职。现综合离职表、花名册原始状态与
* 注册表状态码,结果写入规范化的 employmentState 字段,原始值保留作审计。
*
* 本模块独立于既有的 performance-survey-report-sync,不改动那条链路。
*/
import mssql from 'mssql'
import { prisma } from '@/lib/prisma'
import { EMPLOYMENT_STATE, resolveEmploymentState } from '@/lib/employment-state'
export { EMPLOYMENT_STATE, resolveEmploymentState } from '@/lib/employment-state'
const GROUP_TABLE = {
schema: process.env.REPORT_DB_GROUP_SCHEMA || 'rpt',
table: process.env.REPORT_DB_GROUP_TABLE || 'hr_groups',
}
const DIMISSION_TABLE = {
schema: process.env.REPORT_DB_DIMISSION_SCHEMA || 'rpt',
table: process.env.REPORT_DB_DIMISSION_TABLE || 'hr_employee_dimission',
}
const REGISTER_TABLE = {
schema: process.env.REPORT_DB_REGISTER_SCHEMA || 'rpt',
table: process.env.REPORT_DB_REGISTER_TABLE || 'hr_employee_register',
}
function quoteIdentifier(value: string) {
return '[' + String(value).replace(/]/g, ']]') + ']'
}
function tableRef(t: { schema: string; table: string }) {
return quoteIdentifier(t.schema) + '.' + quoteIdentifier(t.table)
}
export function isReportDbConfigured() {
return Boolean(
process.env.REPORT_DB_SERVER &&
process.env.REPORT_DB_DATABASE &&
process.env.REPORT_DB_USER &&
process.env.REPORT_DB_PASSWORD,
)
}
function buildConfig() {
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: 20000,
requestTimeout: 120000,
pool: { max: 4, min: 0, idleTimeoutMillis: 30000 },
}
}
/** 上游 bigint 时间戳(秒或毫秒均可能)转 Date */
function toDate(value: unknown): Date | null {
if (value === null || value === undefined) return null
const n = Number(value)
if (!Number.isFinite(n) || n <= 0) return null
const ms = n > 1e12 ? n : n * 1000
const d = new Date(ms)
return Number.isNaN(d.getTime()) ? null : d
}
function clean(value: unknown): string | null {
if (value === null || value === undefined) return null
const s = String(value).trim()
return s === '' ? null : s
}
// ============================================================
// 一、组织架构同步
// ============================================================
export interface OrganizationSyncResult {
fetched: number
created: number
updated: number
deactivated: number
rootName: string | null
maxDepth: number
dryRun: boolean
}
interface RawGroup {
Id: number
parent_id: number | null
name: string | null
}
/**
* 在内存中把上游扁平部门列表整理成树,算出每个节点的全路径与层级。
*
* 采用逐层展开而非递归,避免异常数据造成的深递归;父节点缺失或成环的节点
* 会留在 unresolved 中,由调用方决定如何处理,不会静默丢失。
*/
function buildTree(rows: RawGroup[]) {
const byId = new Map<number, RawGroup>()
rows.forEach((r) => byId.set(r.Id, r))
const resolved = new Map<number, { path: string; level: number }>()
const roots = rows.filter((r) => r.parent_id === null || r.parent_id === 0 || r.parent_id === r.Id)
const displayName = (r: RawGroup) => clean(r.name) ?? `部门#${r.Id}`
const queue: RawGroup[] = []
roots.forEach((r) => {
resolved.set(r.Id, { path: displayName(r), level: 1 })
queue.push(r)
})
const childrenOf = new Map<number, RawGroup[]>()
rows.forEach((r) => {
if (r.parent_id === null || r.parent_id === 0 || r.parent_id === r.Id) return
const list = childrenOf.get(r.parent_id) ?? []
list.push(r)
childrenOf.set(r.parent_id, list)
})
while (queue.length > 0) {
const current = queue.shift() as RawGroup
const parentInfo = resolved.get(current.Id)
if (!parentInfo) continue
for (const child of childrenOf.get(current.Id) ?? []) {
if (resolved.has(child.Id)) continue // 防成环
resolved.set(child.Id, {
path: parentInfo.path + '/' + displayName(child),
level: parentInfo.level + 1,
})
queue.push(child)
}
}
const unresolved = rows.filter((r) => !resolved.has(r.Id))
return { resolved, unresolved, roots, byId }
}
/**
* 全量同步组织架构。
*
* 分两阶段写入:先落所有节点(此时不设 parentId),再回填父子关系——
* 因为父节点的本地主键要等它自己写完才存在。
* 上游已不存在的部门标记为 isActive=false 而非物理删除,避免历史引用断裂。
*/
export async function syncOrganizationUnits(options: { dryRun?: boolean } = {}): Promise<OrganizationSyncResult> {
const { dryRun = false } = options
if (!isReportDbConfigured()) {
throw new Error('未配置上游报表库(REPORT_DB_*),无法同步组织架构')
}
const pool = await mssql.connect(buildConfig())
let rows: RawGroup[]
try {
const result = await pool.request().query(
`SELECT Id, parent_id, name FROM ${tableRef(GROUP_TABLE)}`,
)
rows = result.recordset as RawGroup[]
} finally {
await pool.close()
}
const { resolved, unresolved, roots } = buildTree(rows)
const maxDepth = Math.max(1, ...Array.from(resolved.values()).map((v) => v.level))
const rootName = roots.length > 0 ? clean(roots[0].name) : null
if (unresolved.length > 0) {
console.warn(
`[organization-sync] ${unresolved.length} 个部门无法定位到根(父部门缺失或成环),本次跳过`,
)
}
if (dryRun) {
return {
fetched: rows.length,
created: 0,
updated: 0,
deactivated: 0,
rootName,
maxDepth,
dryRun: true,
}
}
const now = new Date()
const existing = await prisma.organizationUnit.findMany({
select: { id: true, sourceDeptId: true },
})
const existingBySource = new Map(existing.map((e) => [e.sourceDeptId, e.id]))
const payload = rows
.map((row) => {
const info = resolved.get(row.Id)
if (!info) return null
return {
sourceDeptId: row.Id,
name: clean(row.name) ?? `部门#${row.Id}`,
sourceParentId: row.parent_id ?? null,
path: info.path,
level: info.level,
}
})
.filter((v): v is NonNullable<typeof v> => v !== null)
const toCreate = payload.filter((p) => !existingBySource.has(p.sourceDeptId))
const toUpdate = payload.filter((p) => existingBySource.has(p.sourceDeptId))
// 新节点批量插入。父子关系稍后统一回填,此处不设 parentId
if (toCreate.length > 0) {
await prisma.organizationUnit.createMany({
data: toCreate.map((p) => ({ ...p, isActive: true, syncedAt: now })),
skipDuplicates: true,
})
}
// 已有节点分批更新。逐条 await 在远程库上要一次次往返,这里用 $transaction
// 把一批操作打包成一次往返,既保持类型安全,也不必手工拼接 SQL 字符串。
const BATCH = 200
for (let i = 0; i < toUpdate.length; i += BATCH) {
const batch = toUpdate.slice(i, i + BATCH)
await prisma.$transaction(
batch.map((p) =>
prisma.organizationUnit.update({
where: { sourceDeptId: p.sourceDeptId },
data: {
name: p.name,
sourceParentId: p.sourceParentId,
path: p.path,
level: p.level,
isActive: true,
syncedAt: now,
},
}),
),
)
}
const created = toCreate.length
const updated = toUpdate.length
// 父子关系整体回填:按上游的 source_parent_id 关联到父节点的本地主键,
// 一条语句取代逐节点更新。SQL 中没有任何外部字符串拼接。
await prisma.$executeRaw`
UPDATE organization_units AS child
SET parent_id = parent.id, updated_at = NOW()
FROM organization_units AS parent
WHERE child.source_parent_id = parent.source_dept_id
AND child.source_parent_id IS NOT NULL
AND child.source_parent_id <> 0
AND child.source_dept_id <> child.source_parent_id
AND child.parent_id IS DISTINCT FROM parent.id
`
// 根节点保持 parent_id 为空
await prisma.$executeRaw`
UPDATE organization_units
SET parent_id = NULL, updated_at = NOW()
WHERE (source_parent_id IS NULL OR source_parent_id = 0 OR source_parent_id = source_dept_id)
AND parent_id IS NOT NULL
`
// 上游已消失的部门下线,不物理删除
const upstreamIds = new Set(rows.map((r) => r.Id))
const stale = existing.filter((e) => !upstreamIds.has(e.sourceDeptId))
let deactivated = 0
if (stale.length > 0) {
const result = await prisma.organizationUnit.updateMany({
where: { id: { in: stale.map((s) => s.id) }, isActive: true },
data: { isActive: false, syncedAt: now },
})
deactivated = result.count
}
return { fetched: rows.length, created, updated, deactivated, rootName, maxDepth, dryRun: false }
}
// ============================================================
// 二、在职状态同步
// ============================================================
export interface EmploymentSyncResult {
upstreamResigned: number
upstreamRegistered: number
localProfiles: number
markedResigned: number
markedActive: number
markedUnknown: number
dryRun: boolean
}
/**
* 依据上游离职表与员工状态码,校正本地档案的在职状态。
*
* 判定优先级:
* 1. 出现在离职表中 → RESIGNED(同时落最后工作日与交接人)
* 2. 花名册原始状态明确离职 → RESIGNED(包括缺少钉钉 userid 的档案)
* 3. 状态码 5 → RESIGNED
* 4. 状态码 2 / 3 → ACTIVE
* 5. 其余 → UNKNOWN(不臆断,交由人工核对)
*/
export async function syncEmploymentStates(options: { dryRun?: boolean } = {}): Promise<EmploymentSyncResult> {
const { dryRun = false } = options
if (!isReportDbConfigured()) {
throw new Error('未配置上游报表库(REPORT_DB_*),无法同步在职状态')
}
const pool = await mssql.connect(buildConfig())
let resignedRows: Array<{ userid: string; last_work_day: any; handover_userid: string | null }>
let registerRows: Array<{ userid: string; status: any }>
try {
const r1 = await pool.request().query(
`SELECT userid, last_work_day, handover_userid
FROM ${tableRef(DIMISSION_TABLE)} WHERE userid IS NOT NULL`,
)
resignedRows = r1.recordset as any[]
const r2 = await pool.request().query(
`SELECT userid, [sys01-employeeStatus] AS status
FROM ${tableRef(REGISTER_TABLE)} WHERE userid IS NOT NULL`,
)
registerRows = r2.recordset as any[]
} finally {
await pool.close()
}
const resignedMap = new Map<string, { at: Date | null; handover: string | null }>()
resignedRows.forEach((r) => {
const key = clean(r.userid)
if (!key) return
resignedMap.set(key, {
at: toDate(r.last_work_day),
handover: clean(r.handover_userid),
})
})
const statusMap = new Map<string, number | null>()
registerRows.forEach((r) => {
const key = clean(r.userid)
if (!key) return
const n = Number(clean(r.status))
statusMap.set(key, Number.isFinite(n) ? n : null)
})
const profiles = await prisma.performanceSurveyEmployeeProfile.findMany({
select: { id: true, dingtalkUserId: true, employmentStatus: true, employmentState: true },
})
let markedResigned = 0
let markedActive = 0
let markedUnknown = 0
for (const profile of profiles) {
const ding = clean(profile.dingtalkUserId)
let nextState: string = EMPLOYMENT_STATE.UNKNOWN
let resignedAt: Date | null = null
let handover: string | null = null
let hasResignationRecord = false
let registeredStatusCode: number | null = null
if (ding) {
const resigned = resignedMap.get(ding)
if (resigned) {
hasResignationRecord = true
resignedAt = resigned.at
handover = resigned.handover
} else {
registeredStatusCode = statusMap.get(ding) ?? null
}
}
nextState = resolveEmploymentState({
hasResignationRecord,
rawEmploymentStatus: profile.employmentStatus,
registeredStatusCode,
})
if (nextState === EMPLOYMENT_STATE.RESIGNED) markedResigned += 1
else if (nextState === EMPLOYMENT_STATE.ACTIVE) markedActive += 1
else markedUnknown += 1
if (!dryRun && profile.employmentState !== nextState) {
await prisma.performanceSurveyEmployeeProfile.update({
where: { id: profile.id },
data: {
employmentState: nextState,
resignedAt,
handoverDingtalkUserId: handover,
},
})
}
}
return {
upstreamResigned: resignedMap.size,
upstreamRegistered: statusMap.size,
localProfiles: profiles.length,
markedResigned,
markedActive,
markedUnknown,
dryRun,
}
}
/**
* 在职人员的查询条件,供替换此前基于关键词包含匹配的写法。
*
* UNKNOWN 视为在职:这些人上游查不到明确状态,宁可多留也不误删——
* 判定错误导致的是"看不见的人",比"多出来的人"更难被发现。
*/
export function getActiveEmployeeWhere() {
return {
OR: [
{ employmentState: EMPLOYMENT_STATE.ACTIVE },
{ employmentState: EMPLOYMENT_STATE.UNKNOWN },
{ employmentState: null },
],
}
}
// ============================================================
// 三、离职联动
// ============================================================
export interface ResignationEffectResult {
/** 权威人员已离职、素材库账号却仍活跃的总人数 */
candidates: number
/** 满足处理条件、实际纳入本次联动的人数 */
pending: number
archivedAccounts: number
custodiedMaterials: number
details: Array<{
displayName: string
username: string
department: string | null
materials: number
resignedAt: Date | null
}>
failures: Array<{ username: string; reason: string }>
dryRun: boolean
}
/**
* 依据权威人员的离职状态,联动处理素材库侧。
*
* 上游标记离职后,此前需要人工发现并逐个处理:账号一直可登录、素材一直挂在
* 离职者名下无人维护。这里把两件事自动化:
* 1. 归档素材库账号(软归档,历史记录全部保留)
* 2. 把其名下素材托管给所在部门,userId 与 authorName 保持不变
*
* 仅处理已通过钉钉唯一值建立正式绑定的账号,不按姓名猜测。
*
* 人事已确认:花名册标记离职即为事实离职,因此离职即托管+归档,
* 不再设天数缓冲;离职日期缺失者同样处理(日期仅作展示,不作门槛)。
*/
export async function applyResignationEffects(
options: { dryRun?: boolean } = {},
): Promise<ResignationEffectResult> {
const { dryRun = false } = options
// 通过正式绑定关系找出"人已离职、账号还活着"的情况
const bindings = await prisma.userEmployeeBinding.findMany({
where: {
status: 'ACTIVE',
employeeProfile: { employmentState: EMPLOYMENT_STATE.RESIGNED },
user: { archivedAt: null },
},
select: {
userId: true,
employeeProfile: { select: { displayName: true, resignedAt: true } },
user: { select: { username: true, department: true, isSuperAdmin: true, role: true } },
},
})
const result: ResignationEffectResult = {
candidates: bindings.length,
pending: 0,
archivedAccounts: 0,
custodiedMaterials: 0,
details: [],
failures: [],
dryRun,
}
for (const binding of bindings) {
const username = binding.user.username
const displayName = binding.employeeProfile.displayName
const resignedAt = binding.employeeProfile.resignedAt
const materialCount = await prisma.material.count({
where: { userId: binding.userId, deletedAt: null },
})
result.pending += 1
result.details.push({
displayName,
username,
department: binding.user.department || null,
materials: materialCount,
resignedAt,
})
if (dryRun) continue
try {
// 超级管理员不由自动流程降权归档,避免误伤关键账号,交人工确认
if (binding.user.isSuperAdmin) {
result.failures.push({ username, reason: '该账号是超级管理员,需人工确认后处理' })
continue
}
const { custodyMaterialsToDepartment } = await import('@/lib/material-transfer')
if (materialCount > 0) {
const custody = await custodyMaterialsToDepartment({ fromUserId: binding.userId })
result.custodiedMaterials += custody.custodied
}
await prisma.user.update({
where: { id: binding.userId },
data: {
isActive: false,
status: 'archived',
archivedAt: new Date(),
statusChangedAt: new Date(),
employmentNote: '上游花名册标记离职,系统自动归档',
},
})
result.archivedAccounts += 1
} catch (error) {
result.failures.push({
username,
reason: error instanceof Error ? error.message : String(error),
})
}
}
return result
}
// ============================================================
// 四、定时同步
// ============================================================
let dailySyncStarted = false
/**
* 启动组织架构与人事状态的每日同步。
*
* 由 instrumentation.ts 在应用启动时调用,不再依赖"有人访问考核调研"懒启动,
* 容器重启后自动恢复。首次执行定在下一个 Asia/Shanghai 的 02:00,
* 之后每 24 小时一次;定时器 unref,不会阻止进程退出。
* 多实例部署时各实例会各自执行,同步本身是幂等的,重复执行只会得到相同结果。
*
* 三件事按顺序跑,任一失败只记录日志、不影响后续周期:
* 组织架构 → 在职状态 → 离职联动
*/
export function ensureDailyOrganizationSyncStarted() {
if (dailySyncStarted) return
dailySyncStarted = true
if (!isReportDbConfigured()) {
console.info('[organization-sync] 未配置上游报表库,跳过每日同步')
return
}
const syncHour = Math.min(23, Math.max(0, Number(process.env.ORGANIZATION_SYNC_HOUR || 2)))
const runOnce = async () => {
try {
const org = await syncOrganizationUnits()
console.info(
`[organization-sync] 组织架构:上游 ${org.fetched} 个部门,新建 ${org.created},更新 ${org.updated},下线 ${org.deactivated}`,
)
} catch (error) {
console.warn('[organization-sync] 组织架构同步失败:', error)
}
try {
const emp = await syncEmploymentStates()
console.info(
`[organization-sync] 在职状态:在职 ${emp.markedActive},离职 ${emp.markedResigned},未知 ${emp.markedUnknown}`,
)
} catch (error) {
// 状态没更新就不做离职联动,避免依据陈旧数据归档账号
console.warn('[organization-sync] 在职状态同步失败,本轮跳过离职联动:', error)
return
}
try {
const effect = await applyResignationEffects()
if (effect.archivedAccounts > 0 || effect.custodiedMaterials > 0) {
console.info(
`[organization-sync] 离职联动:归档 ${effect.archivedAccounts} 个账号,托管 ${effect.custodiedMaterials} 条素材`,
)
}
if (effect.failures.length > 0) {
console.warn(
`[organization-sync] 离职联动有 ${effect.failures.length} 人需人工处理:` +
effect.failures.map((f) => f.username + '(' + f.reason + ')').join('、'),
)
}
} catch (error) {
console.warn('[organization-sync] 离职联动失败:', error)
}
}
// 首次执行时间:下一个 Asia/Shanghai 的 syncHour:00(默认 02:00)。
// 用 Intl 按上海时区取当前"小时+分钟",换算成当天第几分钟,
// 计算到目标小时的分钟差,避免依赖服务器本地时区,也避免 hour 格式四舍五入误差。
// 取严格"下一个"整点:当前分钟数已过目标点时(如上海 02:00~02:59),
// 说明今天的目标点已过,应等到明天,即 24 小时。
const now = new Date()
const shanghaiParts = new Intl.DateTimeFormat('en-US', {
timeZone: 'Asia/Shanghai',
hour: 'numeric',
minute: 'numeric',
hourCycle: 'h23',
}).formatToParts(now)
const partOf = (type: string) => Number(shanghaiParts.find((p) => p.type === type)?.value ?? '0')
const minutesNow = partOf('hour') * 60 + partOf('minute')
const minutesUntilTarget = ((syncHour * 60 - minutesNow + 24 * 60 - 1) % (24 * 60)) + 1
const secondsUntilTarget = minutesUntilTarget * 60
const starter = setTimeout(() => {
void runOnce()
const timer = setInterval(() => void runOnce(), 24 * 60 * 60 * 1000)
timer.unref?.()
}, secondsUntilTarget * 1000)
starter.unref?.()
console.info(`[organization-sync] 每日同步已调度:首个执行点在 Asia/Shanghai ${syncHour}:00(约 ${secondsUntilTarget / 3600} 小时后),之后每 24 小时`)
}