forked from wangziqi/gongxue-base
465 lines
17 KiB
TypeScript
465 lines
17 KiB
TypeScript
import { ConflictException, Injectable, Logger, NotFoundException, ServiceUnavailableException } from '@nestjs/common';
|
||
import { InjectRepository } from '@nestjs/typeorm';
|
||
import { randomUUID } from 'node:crypto';
|
||
import { DataSource, In, Repository } from 'typeorm';
|
||
import { SyncLog, SyncState, Student, StudentDingMapping } from '../entities';
|
||
import { JinshujuMatchRule, type JinshujuFieldMapping } from '../entities/jinshuju-match-rule.entity';
|
||
import type { SyncPlatform, SyncStatus, SyncType } from '../entities/sync-log.entity';
|
||
import { AttendanceImportService } from '../attendance/attendance-import.service';
|
||
import { DingTalkService } from '../integration/dingtalk.service';
|
||
import { WeComService } from '../integration/wecom.service';
|
||
import { JinshujuService } from '../integration/jinshuju.service';
|
||
import { syncJinshujuStudents } from '../integration/jinshuju-student-sync';
|
||
import { ScheduleSyncService } from './schedule-sync.service';
|
||
|
||
@Injectable()
|
||
export class SyncService {
|
||
private readonly logger = new Logger(SyncService.name);
|
||
private static readonly LEASE_MS = 30 * 60 * 1000;
|
||
|
||
constructor(
|
||
@InjectRepository(SyncLog)
|
||
private readonly syncLogRepo: Repository<SyncLog>,
|
||
@InjectRepository(SyncState)
|
||
private readonly syncStateRepo: Repository<SyncState>,
|
||
@InjectRepository(StudentDingMapping)
|
||
private readonly studentDingMappingRepo: Repository<StudentDingMapping>,
|
||
@InjectRepository(JinshujuMatchRule)
|
||
private readonly matchRuleRepo: Repository<JinshujuMatchRule>,
|
||
private readonly dingTalkService: DingTalkService,
|
||
private readonly weComService: WeComService,
|
||
private readonly jinshujuService: JinshujuService,
|
||
private readonly attendanceImportService: AttendanceImportService,
|
||
private readonly scheduleSyncService: ScheduleSyncService,
|
||
private readonly dataSource: DataSource,
|
||
) {}
|
||
|
||
async syncDingTalkStudents(rootDeptId = 1): Promise<SyncLog> {
|
||
return this.runSync('dingtalk_students', async () => {
|
||
const result = await this.dingTalkService.syncAll(rootDeptId);
|
||
return {
|
||
recordsCount: result.created + result.updated,
|
||
status: result.conflicts.length ? 'partial' : 'success',
|
||
message: result.conflicts.length ? JSON.stringify(result.conflicts.slice(0, 20)) : undefined,
|
||
};
|
||
});
|
||
}
|
||
|
||
async syncDingTalkAttendance(): Promise<SyncLog> {
|
||
return this.runSync('dingtalk_attendance', async (lastSyncAt) => {
|
||
const endDate = new Date();
|
||
const startDate = lastSyncAt ? new Date(lastSyncAt) : new Date(endDate);
|
||
if (!lastSyncAt) startDate.setDate(startDate.getDate() - 7);
|
||
|
||
const mappings = await this.studentDingMappingRepo.find();
|
||
const userIds = [...new Set(mappings.map((mapping) => mapping.dingUserId).filter(Boolean))];
|
||
if (userIds.length === 0) {
|
||
throw new ServiceUnavailableException('没有可用于考勤导入的钉钉学生映射');
|
||
}
|
||
|
||
const result = await this.attendanceImportService.importFromDingTalk({
|
||
startDate: startDate.toISOString().slice(0, 10),
|
||
endDate: endDate.toISOString().slice(0, 10),
|
||
userIds,
|
||
autoMatch: true,
|
||
});
|
||
if (!result.success) {
|
||
throw new ServiceUnavailableException(result.errors.join('；') || '钉钉考勤导入失败');
|
||
}
|
||
return { recordsCount: result.imported, status: 'success' };
|
||
});
|
||
}
|
||
|
||
async syncWeCom(): Promise<SyncLog> {
|
||
return this.runSync('wecom', async () => {
|
||
const result = await this.weComService.syncAll();
|
||
return { recordsCount: result.userCount, status: 'success' };
|
||
});
|
||
}
|
||
async syncJinshuju(apiKey: string, apiSecret: string, formToken: string): Promise<SyncLog> {
|
||
return this.runSync('jinshuju', async () => {
|
||
const entries = await this.jinshujuService.fetchAllEntries(apiKey, apiSecret, formToken);
|
||
const result = await this.dataSource.transaction((manager) =>
|
||
syncJinshujuStudents(manager, entries),
|
||
);
|
||
return {
|
||
recordsCount: result.matched + result.created,
|
||
status: result.conflicts.length ? 'partial' : 'success',
|
||
message: result.conflicts.length
|
||
? JSON.stringify(result.conflicts.slice(0, 20))
|
||
: `匹配 ${result.matched} 人，新增 ${result.created} 人，跳过无姓名 ${result.skippedNoName} 条`,
|
||
};
|
||
});
|
||
}
|
||
|
||
/** Fetch Jinshuju entries and return with auto-suggested student matches (no writes). */
|
||
getJinshujuFormFields(apiKey: string, apiSecret: string, formToken: string) {
|
||
return this.jinshujuService.fetchFormFields(apiKey, apiSecret, formToken);
|
||
}
|
||
|
||
async previewJinshuju(apiKey: string, apiSecret: string, formToken: string, ruleId?: number) {
|
||
const entries = await this.jinshujuService.fetchAllEntries(apiKey, apiSecret, formToken);
|
||
const rule = ruleId ? await this.getMatchRule(ruleId, formToken) : null;
|
||
const map = rule?.mappings ?? { name: 'field_1', phone: 'field_2' };
|
||
|
||
const parsed = entries
|
||
.map((e) => ({
|
||
serialNumber: e.serial_number,
|
||
name: this.extractField(e, map.name),
|
||
phone: this.extractField(e, map.phone),
|
||
}))
|
||
.filter((p) => p.name);
|
||
|
||
const phones = [...new Set(parsed.filter((p) => p.phone).map((p) => p.phone))];
|
||
const names = [...new Set(parsed.map((p) => p.name))];
|
||
|
||
const [phoneStudents, nameStudents] = await Promise.all([
|
||
phones.length
|
||
? this.dataSource.getRepository(Student).find({ where: { phone: In(phones) } })
|
||
: ([] as Student[]),
|
||
names.length
|
||
? this.dataSource.getRepository(Student).find({ where: { name: In(names) } })
|
||
: ([] as Student[]),
|
||
]);
|
||
|
||
const studentByPhone = new Map(phoneStudents.map((s) => [s.phone, s]));
|
||
const studentByName = new Map<string, Student[]>();
|
||
for (const s of nameStudents) {
|
||
const list = studentByName.get(s.name) || [];
|
||
list.push(s);
|
||
studentByName.set(s.name, list);
|
||
}
|
||
|
||
const allStudents = await this.dataSource.getRepository(Student).find({
|
||
where: { status: 'active' },
|
||
order: { name: 'ASC' },
|
||
select: ['id', 'name', 'phone', 'studentNo'],
|
||
});
|
||
|
||
const rows = parsed.map((p) => {
|
||
const phoneMatch = p.phone ? studentByPhone.get(p.phone) : undefined;
|
||
const nameMatches = studentByName.get(p.name) || [];
|
||
const suggested = phoneMatch ?? nameMatches[0] ?? null;
|
||
return {
|
||
serialNumber: p.serialNumber,
|
||
name: p.name,
|
||
phone: p.phone || null,
|
||
suggestedStudent: suggested
|
||
? { id: suggested.id, name: suggested.name, phone: suggested.phone, studentNo: suggested.studentNo }
|
||
: null,
|
||
};
|
||
});
|
||
|
||
return { entries: rows, students: allStudents };
|
||
}
|
||
|
||
/** Apply user's matching decisions. */
|
||
async applyJinshuju(
|
||
apiKey: string,
|
||
apiSecret: string,
|
||
formToken: string,
|
||
decisions: Array<{
|
||
serialNumber: number;
|
||
action: 'match' | 'create' | 'skip';
|
||
matchStudentId?: number;
|
||
createName?: string;
|
||
createPhone?: string;
|
||
}>,
|
||
ruleId?: number,
|
||
): Promise<SyncLog> {
|
||
return this.runSync('jinshuju', async () => {
|
||
const rule = ruleId ? await this.getMatchRule(ruleId, formToken) : null;
|
||
const map = rule?.mappings ?? { name: 'field_1', phone: 'field_2' };
|
||
const entries = await this.jinshujuService.fetchAllEntries(apiKey, apiSecret, formToken);
|
||
const entryMap = new Map(entries.map((entry) => [entry.serial_number, entry]));
|
||
const decisionMap = new Map(decisions.map((decision) => [decision.serialNumber, decision]));
|
||
|
||
let matched = 0;
|
||
let created = 0;
|
||
await this.dataSource.transaction(async (manager) => {
|
||
const orgs = await manager.query(
|
||
'SELECT id FROM organizations WHERE is_host = 1 AND status = ? LIMIT 1',
|
||
['active'],
|
||
);
|
||
const orgId: number | undefined = orgs[0]?.id;
|
||
|
||
for (const [serial, entry] of entryMap) {
|
||
const decision = decisionMap.get(serial);
|
||
if (!decision || decision.action === 'skip') continue;
|
||
|
||
const mappedValues = Object.fromEntries(
|
||
Object.entries(map)
|
||
.map(([studentField, fieldKey]) => [studentField, this.extractField(entry, fieldKey)])
|
||
.filter(([, value]) => value),
|
||
);
|
||
|
||
if (decision.action === 'match' && decision.matchStudentId) {
|
||
await manager.update(Student, decision.matchStudentId, mappedValues);
|
||
matched++;
|
||
} else if (decision.action === 'create') {
|
||
const name = decision.createName || mappedValues.name;
|
||
if (!name) continue;
|
||
await manager.save(
|
||
manager.create(Student, {
|
||
...mappedValues,
|
||
name,
|
||
phone: decision.createPhone || mappedValues.phone || undefined,
|
||
organizationId: orgId,
|
||
}),
|
||
);
|
||
created++;
|
||
}
|
||
}
|
||
});
|
||
|
||
return {
|
||
recordsCount: matched + created,
|
||
status: 'success',
|
||
message: `匹配 ${matched} 人，新增 ${created} 人`,
|
||
};
|
||
});
|
||
}
|
||
|
||
async triggerSync(platform?: SyncPlatform, rootDeptId = 1): Promise<SyncLog[]> {
|
||
if (platform === 'dingtalk_students') return [await this.syncDingTalkStudents(rootDeptId)];
|
||
if (platform === 'dingtalk_attendance') return [await this.syncDingTalkAttendance()];
|
||
if (platform === 'wecom') return [await this.syncWeCom()];
|
||
return [
|
||
await this.syncDingTalkStudents(rootDeptId),
|
||
await this.syncDingTalkAttendance(),
|
||
await this.syncWeCom(),
|
||
];
|
||
}
|
||
|
||
async getDingTalkOrgTree(rootDeptId = 1) {
|
||
return this.dingTalkService.fetchOrgTree(rootDeptId);
|
||
}
|
||
|
||
async getDingTalkOrgTreeWithUsers(rootDeptId = 1) {
|
||
return this.dingTalkService.fetchOrgTreeWithUsers(rootDeptId);
|
||
}
|
||
|
||
async getDingTalkAttendanceGroups() {
|
||
return this.dingTalkService.queryAttendanceGroups();
|
||
}
|
||
|
||
async deleteAllDingTalkAttendanceGroups() {
|
||
const groups = await this.dingTalkService.queryAttendanceGroups();
|
||
const deleted: Array<{ groupId: number; groupName: string }> = [];
|
||
const failed: Array<{ groupId: number; groupName: string; error: string }> = [];
|
||
for (const group of groups) {
|
||
try {
|
||
await this.dingTalkService.deleteAttendanceGroup(group.group_id);
|
||
deleted.push({ groupId: group.group_id, groupName: group.group_name });
|
||
} catch (error: unknown) {
|
||
failed.push({
|
||
groupId: group.group_id,
|
||
groupName: group.group_name,
|
||
error: error instanceof Error ? error.message : String(error),
|
||
});
|
||
}
|
||
}
|
||
return { total: groups.length, deleted, failed };
|
||
}
|
||
|
||
async syncScheduleToDingTalk(
|
||
dateFrom?: string,
|
||
days = 30,
|
||
attendanceMachineOnly = false,
|
||
) {
|
||
return this.scheduleSyncService.syncAll(dateFrom, days, 'manager', attendanceMachineOnly);
|
||
}
|
||
|
||
async getScheduleSyncStatus(date?: string) {
|
||
return this.scheduleSyncService.getStatus(date || new Date().toISOString().slice(0, 10));
|
||
}
|
||
|
||
async getLogs(platform?: SyncPlatform, limit = 50): Promise<SyncLog[]> {
|
||
const where: Record<string, SyncPlatform> = {};
|
||
if (platform) where.platform = platform;
|
||
return this.syncLogRepo.find({ where, order: { createdAt: 'DESC' }, take: limit });
|
||
}
|
||
|
||
async getLastSync(platform: SyncPlatform | 'dingtalk'): Promise<SyncLog | null> {
|
||
if (platform === 'dingtalk') {
|
||
return this.syncLogRepo.findOne({
|
||
where: [{ platform: 'dingtalk_students' }, { platform: 'dingtalk_attendance' }],
|
||
order: { createdAt: 'DESC' },
|
||
});
|
||
}
|
||
return this.syncLogRepo.findOne({ where: { platform }, order: { createdAt: 'DESC' } });
|
||
}
|
||
|
||
private async runSync(
|
||
platform: SyncPlatform,
|
||
operation: (lastSyncAt: Date | null) => Promise<{
|
||
recordsCount: number;
|
||
status: Extract<SyncStatus, 'success' | 'partial'>;
|
||
message?: string;
|
||
}>,
|
||
): Promise<SyncLog> {
|
||
const runId = await this.acquireLease(platform);
|
||
let log: SyncLog | undefined;
|
||
try {
|
||
const lastSyncAt = await this.getLastSyncAt(platform);
|
||
log = await this.createSyncLog(platform, lastSyncAt ? 'incremental' : 'full', 'running');
|
||
const result = await operation(lastSyncAt);
|
||
await this.syncStateRepo.update({ platform }, { lastSyncAt: new Date() });
|
||
await this.finishSyncLog(log, result.status, result.recordsCount, result.message);
|
||
return log;
|
||
} catch (error: unknown) {
|
||
const message = error instanceof Error ? error.message : String(error);
|
||
if (log) await this.finishSyncLog(log, 'failed', 0, message);
|
||
this.logger.error(`${platform} sync failed: ${message}`, error instanceof Error ? error.stack : undefined);
|
||
throw error;
|
||
} finally {
|
||
await this.releaseLease(platform, runId);
|
||
}
|
||
}
|
||
|
||
private async acquireLease(platform: SyncPlatform): Promise<string> {
|
||
await this.syncStateRepo
|
||
.createQueryBuilder()
|
||
.insert()
|
||
.values({ platform, lastSyncAt: null, runId: null, runningSince: null })
|
||
.orIgnore()
|
||
.execute();
|
||
|
||
const runId = randomUUID();
|
||
const result = await this.syncStateRepo
|
||
.createQueryBuilder()
|
||
.update()
|
||
.set({ runId, runningSince: new Date() })
|
||
.where('platform = :platform', { platform })
|
||
.andWhere('(running_since IS NULL OR running_since < :staleBefore)', {
|
||
staleBefore: new Date(Date.now() - SyncService.LEASE_MS),
|
||
})
|
||
.execute();
|
||
if (result.affected !== 1) throw new ConflictException(`${platform} 同步正在进行中`);
|
||
return runId;
|
||
}
|
||
|
||
private async releaseLease(platform: SyncPlatform, runId: string): Promise<void> {
|
||
await this.syncStateRepo
|
||
.createQueryBuilder()
|
||
.update()
|
||
.set({ runId: null, runningSince: null })
|
||
.where('platform = :platform AND run_id = :runId', { platform, runId })
|
||
.execute();
|
||
}
|
||
|
||
private async getLastSyncAt(platform: SyncPlatform): Promise<Date | null> {
|
||
const state = await this.syncStateRepo.findOne({ where: { platform } });
|
||
return state?.lastSyncAt ?? null;
|
||
}
|
||
|
||
private async createSyncLog(
|
||
platform: SyncPlatform,
|
||
syncType: SyncType,
|
||
status: SyncStatus,
|
||
): Promise<SyncLog> {
|
||
return this.syncLogRepo.save(
|
||
this.syncLogRepo.create({
|
||
platform,
|
||
syncType,
|
||
status,
|
||
recordsCount: 0,
|
||
startedAt: new Date(),
|
||
}),
|
||
);
|
||
}
|
||
|
||
private async finishSyncLog(
|
||
log: SyncLog,
|
||
status: SyncStatus,
|
||
recordsCount: number,
|
||
errorMessage?: string,
|
||
): Promise<void> {
|
||
log.status = status;
|
||
log.recordsCount = recordsCount;
|
||
log.finishedAt = new Date();
|
||
log.errorMessage = errorMessage ?? null;
|
||
await this.syncLogRepo.save(log);
|
||
}
|
||
|
||
// ── Match Rules CRUD ──
|
||
|
||
async listMatchRules(): Promise<JinshujuMatchRule[]> {
|
||
return this.matchRuleRepo.find({ order: { updatedAt: 'DESC' } });
|
||
}
|
||
|
||
async createMatchRule(dto: {
|
||
name: string;
|
||
formToken: string;
|
||
mappings: JinshujuFieldMapping;
|
||
}): Promise<JinshujuMatchRule> {
|
||
this.validateMatchRule(dto.formToken, dto.mappings);
|
||
return this.matchRuleRepo.save(
|
||
this.matchRuleRepo.create({
|
||
...dto,
|
||
name: dto.name.trim(),
|
||
formToken: dto.formToken.trim(),
|
||
}),
|
||
);
|
||
}
|
||
|
||
async updateMatchRule(
|
||
id: number,
|
||
dto: { name?: string; mappings?: JinshujuFieldMapping },
|
||
): Promise<JinshujuMatchRule> {
|
||
const rule = await this.matchRuleRepo.findOne({ where: { id } });
|
||
if (!rule) throw new NotFoundException('规则不存在');
|
||
const mappings = dto.mappings ?? rule.mappings;
|
||
this.validateMatchRule(rule.formToken, mappings);
|
||
await this.matchRuleRepo.update(id, {
|
||
name: dto.name?.trim(),
|
||
mappings,
|
||
});
|
||
return this.matchRuleRepo.findOneOrFail({ where: { id } });
|
||
}
|
||
|
||
async deleteMatchRule(id: number): Promise<void> {
|
||
const result = await this.matchRuleRepo.delete(id);
|
||
if (!result.affected) throw new NotFoundException('规则不存在');
|
||
}
|
||
|
||
private async getMatchRule(id: number, formToken: string): Promise<JinshujuMatchRule> {
|
||
const rule = await this.matchRuleRepo.findOne({ where: { id } });
|
||
if (!rule) throw new NotFoundException('规则不存在');
|
||
if (rule.formToken !== formToken) {
|
||
throw new ConflictException('匹配规则不属于当前表单');
|
||
}
|
||
return rule;
|
||
}
|
||
|
||
private validateMatchRule(formToken: string, mappings: JinshujuFieldMapping): void {
|
||
if (!formToken.trim()) throw new ConflictException('表单 Token 不能为空');
|
||
if (!mappings.name) throw new ConflictException('匹配规则必须映射姓名字段');
|
||
const allowedStudentFields = new Set([
|
||
'name',
|
||
'studentNo',
|
||
'phone',
|
||
'idNumber',
|
||
'gender',
|
||
'ethnicity',
|
||
'emergencyContact',
|
||
'emergencyPhone',
|
||
]);
|
||
for (const [studentField, fieldKey] of Object.entries(mappings)) {
|
||
if (!allowedStudentFields.has(studentField)) {
|
||
throw new ConflictException(`不允许映射学生字段：${studentField}`);
|
||
}
|
||
if (fieldKey && !/^field_\d+$/.test(fieldKey)) {
|
||
throw new ConflictException(`无效的金数据字段：${fieldKey}`);
|
||
}
|
||
}
|
||
}
|
||
|
||
/** Extract value from a Jinshuju entry by field mapping. */
|
||
private extractField(entry: Record<string, unknown>, fieldKey: string | undefined): string {
|
||
if (!fieldKey) return '';
|
||
const val = entry[fieldKey];
|
||
return typeof val === 'string' ? val.trim() : '';
|
||
}
|
||
}
|