gongxiangp2/lib/organization-sync.ts
chenpei 7b0ee5efb4 feat: 人员架构统一到钉钉唯一值,接入上游组织架构动态同步
素材库与考核调研此前各有一套人事数据,彼此没有外键,靠姓名、邮箱在运行时
逐层软匹配。本次把身份判定收敛到钉钉唯一值,并让组织架构与人员状态跟随
上游 HR 系统自动更新。

人员身份
- 新增账号与权威人员档案的正式绑定表,活跃绑定双向唯一
- users.dingtalkUserId 补上唯一约束,数据库层面杜绝一人多号
- 移除钉钉登录的同名自动合并,同名冲突一律转人工核对
- 看板删除特判由比对用户名改为比对钉钉唯一值,避免同名越权

组织架构
- 新增 OrganizationUnit,直接镜像上游 HR 部门树
- 不再从员工部门路径倒推,同名部门不会再被合并
- 上游已删除的部门下线而非物理删除

人员状态
- 新增规范化的 employmentState,取代对混杂原始值做关键词匹配
- 接入上游离职表,离职判定不再依赖中文关键词
- 离职联动:归档账号并把素材托管给所在部门,默认只处理离职满 30 天者

素材归属
- 归属变更纳入正式转移,记录最初上传者、转移流水与理由
- 新增素材托管部门,归属与署名分离:托管只改维护责任,不动署名
- 编辑素材的权限判定统一到 canManageMaterial
- 修复任务-详情上传把邮箱误存为作者名的问题

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-08-14 09:53:48 +08:00

660 lines
22 KiB
TypeScript
Raw 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'
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',
}
/** 上游员工状态码。5 已离职,2 试用、3 正式均视为在职 */
const STATUS_CODE_RESIGNED = 5
const STATUS_CODE_ACTIVE = [2, 3]
export const EMPLOYMENT_STATE = {
ACTIVE: 'ACTIVE',
RESIGNED: 'RESIGNED',
UNKNOWN: 'UNKNOWN',
} as const
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. 状态码 5 → RESIGNED
* 3. 状态码 2 / 3 → ACTIVE
* 4. 其余 → 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, 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
if (ding) {
const resigned = resignedMap.get(ding)
if (resigned) {
nextState = EMPLOYMENT_STATE.RESIGNED
resignedAt = resigned.at
handover = resigned.handover
} else {
const code = statusMap.get(ding)
if (code === STATUS_CODE_RESIGNED) nextState = EMPLOYMENT_STATE.RESIGNED
else if (code !== null && code !== undefined && STATUS_CODE_ACTIVE.includes(code)) {
nextState = EMPLOYMENT_STATE.ACTIVE
}
}
}
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
/** 因离职时间不足或日期缺失而暂缓的人数 */
deferred: number
archivedAccounts: number
custodiedMaterials: number
details: Array<{
displayName: string
username: string
department: string | null
materials: number
resignedAt: Date | null
}>
deferredDetails: Array<{ displayName: string; reason: string }>
failures: Array<{ username: string; reason: string }>
dryRun: boolean
}
/**
* 依据权威人员的离职状态,联动处理素材库侧。
*
* 上游标记离职后,此前需要人工发现并逐个处理:账号一直可登录、素材一直挂在
* 离职者名下无人维护。这里把两件事自动化:
* 1. 归档素材库账号(软归档,历史记录全部保留)
* 2. 把其名下素材托管给所在部门,userId 与 authorName 保持不变
*
* 仅处理已通过钉钉唯一值建立正式绑定的账号,不按姓名猜测。
*
* `minDaysSinceResignation` 用于避开交接期、返聘与上游数据延迟:只处理离职已满
* 指定天数的人;离职日期缺失者一并暂缓,交人工核实,不做推断。
*/
export async function applyResignationEffects(
options: { dryRun?: boolean; minDaysSinceResignation?: number } = {},
): Promise<ResignationEffectResult> {
const { dryRun = false, minDaysSinceResignation } = 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,
deferred: 0,
archivedAccounts: 0,
custodiedMaterials: 0,
details: [],
deferredDetails: [],
failures: [],
dryRun,
}
const cutoff =
typeof minDaysSinceResignation === 'number'
? new Date(Date.now() - minDaysSinceResignation * 24 * 60 * 60 * 1000)
: null
for (const binding of bindings) {
const username = binding.user.username
const displayName = binding.employeeProfile.displayName
const resignedAt = binding.employeeProfile.resignedAt
if (cutoff) {
if (!resignedAt) {
result.deferred += 1
result.deferredDetails.push({ displayName, reason: '离职日期缺失,需人工核实' })
continue
}
if (resignedAt > cutoff) {
result.deferred += 1
result.deferredDetails.push({
displayName,
reason: '离职未满 ' + minDaysSinceResignation + ' 天(' + resignedAt.toLocaleDateString('zh-CN') + ')',
})
continue
}
}
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
/**
* 启动组织架构与人事状态的每日同步。
*
* 沿用项目既有的进程内定时器模式:首次有人访问相关功能时懒启动,定时器全部
* unref,不会阻止进程退出。多实例部署时各实例会各自执行,同步本身是幂等的,
* 重复执行只会得到相同结果。
*
* 三件事按顺序跑,任一失败只记录日志、不影响后续周期:
* 组织架构 → 在职状态 → 离职联动
*/
export function ensureDailyOrganizationSyncStarted() {
if (dailySyncStarted) return
dailySyncStarted = true
if (!isReportDbConfigured()) {
console.info('[organization-sync] 未配置上游报表库,跳过每日同步')
return
}
const intervalHours = Math.max(1, Number(process.env.ORGANIZATION_SYNC_INTERVAL_HOURS || 24))
const initialDelaySeconds = Math.max(
30,
Number(process.env.ORGANIZATION_SYNC_INITIAL_DELAY_SECONDS || 600),
)
const minDays = Math.max(0, Number(process.env.RESIGNATION_MIN_DAYS || 30))
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({ minDaysSinceResignation: minDays })
if (effect.archivedAccounts > 0 || effect.custodiedMaterials > 0) {
console.info(
`[organization-sync] 离职联动:归档 ${effect.archivedAccounts} 个账号,托管 ${effect.custodiedMaterials} 条素材,暂缓 ${effect.deferred} 人`,
)
}
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)
}
}
const starter = setTimeout(() => {
void runOnce()
const timer = setInterval(() => void runOnce(), intervalHours * 60 * 60 * 1000)
timer.unref?.()
}, initialDelaySeconds * 1000)
starter.unref?.()
}