357 lines
13 KiB
TypeScript
357 lines
13 KiB
TypeScript
import { BadRequestException, Injectable, Logger } from '@nestjs/common';
|
|
import { InjectRepository } from '@nestjs/typeorm';
|
|
import { Repository, In } from 'typeorm';
|
|
import { Subject, Observable } from 'rxjs';
|
|
import {
|
|
DingAttendanceRaw,
|
|
Student,
|
|
StudentDingMapping,
|
|
} from '../entities';
|
|
import { DingTalkService, DingTalkAttendanceResult } from '../integration/dingtalk.service';
|
|
import { AttendanceService } from './attendance.service';
|
|
import type { ImportProgressEvent, ImportResult } from './dto/dingtalk-import.dto';
|
|
|
|
/**
|
|
* Service for importing DingTalk attendance data into the system.
|
|
*
|
|
* ## Stream Architecture
|
|
* Uses RxJS Subjects to emit progress events during the import pipeline:
|
|
* fetch → parse → deduplicate → save → auto-match
|
|
*
|
|
* Progress is exposed as an Observable so the controller can relay it via SSE.
|
|
* Only ONE import can run at a time (guarded by `isRunning`).
|
|
*/
|
|
@Injectable()
|
|
export class AttendanceImportService {
|
|
private readonly logger = new Logger(AttendanceImportService.name);
|
|
|
|
/** RxJS Subject emitting live progress during import */
|
|
private progressSubject = new Subject<ImportProgressEvent>();
|
|
private isRunning = false;
|
|
/** ID of the user who triggered the current import (for SSE scoping) */
|
|
private importingUserId?: number;
|
|
constructor(
|
|
@InjectRepository(DingAttendanceRaw)
|
|
private readonly dingRawRepo: Repository<DingAttendanceRaw>,
|
|
@InjectRepository(Student)
|
|
private readonly studentRepo: Repository<Student>,
|
|
@InjectRepository(StudentDingMapping)
|
|
private readonly studentDingMappingRepo: Repository<StudentDingMapping>,
|
|
private readonly dingTalkService: DingTalkService,
|
|
private readonly attendanceService: AttendanceService,
|
|
) {}
|
|
|
|
/**
|
|
* Expose progress as a read-only Observable for SSE.
|
|
*/
|
|
get progress$(): Observable<ImportProgressEvent> {
|
|
return this.progressSubject.asObservable();
|
|
}
|
|
|
|
/**
|
|
* Whether an import is currently in progress.
|
|
*/
|
|
get running(): boolean {
|
|
return this.isRunning;
|
|
}
|
|
|
|
/**
|
|
* Run the DingTalk attendance import pipeline.
|
|
*
|
|
* Pipeline stages:
|
|
* 1. Fetch attendance results from DingTalk (paginated)
|
|
* 2. Parse and validate each record
|
|
* 3. Deduplicate by `dingId` (unique in DB)
|
|
* 5. Optionally auto-match to students by name
|
|
*/
|
|
async importFromDingTalk(params: {
|
|
startDate: string;
|
|
endDate: string;
|
|
userIds?: string[];
|
|
autoMatch?: boolean;
|
|
/** ID of the HTTP user triggering the import (for SSE event scoping) */
|
|
userId?: number;
|
|
}): Promise<ImportResult> {
|
|
if (this.isRunning) {
|
|
throw new Error('An import is already in progress');
|
|
}
|
|
|
|
const startedAt = Date.now();
|
|
this.isRunning = true;
|
|
this.importingUserId = params.userId;
|
|
|
|
// Safety timeout: auto-reset isRunning after 30 minutes in case of
|
|
// an unhandled exception that bypasses the finally block (extremely rare).
|
|
const SAFETY_TIMEOUT_MS = 30 * 60 * 1000;
|
|
const safetyTimer = setTimeout(() => {
|
|
if (this.isRunning) {
|
|
this.logger.error('Import safety timeout triggered — force-resetting isRunning');
|
|
this.isRunning = false;
|
|
}
|
|
}, SAFETY_TIMEOUT_MS);
|
|
|
|
const errors: string[] = [];
|
|
let imported = 0;
|
|
let skipped = 0;
|
|
let matched = 0;
|
|
|
|
try {
|
|
// Stage 1: Fetch
|
|
this.emit('fetching', 0, 0, 'Fetching attendance results from DingTalk...');
|
|
const rawResults = await this.fetchAllPages(params);
|
|
const total = rawResults.length;
|
|
this.emit('fetching', total, total, `Fetched ${total} raw attendance records`);
|
|
|
|
// Stage 2: Parse & deduplicate
|
|
this.emit('parsing', 0, total, `Parsing ${total} records...`);
|
|
const existingByDingId = await this.getExistingRecordsByDingId(rawResults);
|
|
const newRecords = rawResults.filter((r) => !existingByDingId.has(r.checkId));
|
|
const duplicateRecords = rawResults.filter((r) => existingByDingId.has(r.checkId));
|
|
await this.refreshDuplicatePunchMetadata(duplicateRecords, existingByDingId);
|
|
skipped = rawResults.length - newRecords.length;
|
|
this.emit('parsing', newRecords.length, total, `${newRecords.length} new records, ${skipped} duplicates skipped`);
|
|
|
|
if (newRecords.length === 0) {
|
|
if (params.autoMatch) matched = await this.autoMatchUnmatched();
|
|
this.emit('complete', imported + skipped, total, 'Nothing new to import');
|
|
return { success: true, imported, skipped, matched, errors, duration: Date.now() - startedAt };
|
|
}
|
|
|
|
// Stage 3: Batch save
|
|
this.emit('saving', 0, newRecords.length, `Saving ${newRecords.length} records...`);
|
|
const batchSize = 100;
|
|
for (let i = 0; i < newRecords.length; i += batchSize) {
|
|
const batch = newRecords.slice(i, i + batchSize);
|
|
const entities = await Promise.all(batch.map((record) => this.mapToEntity(record)));
|
|
try {
|
|
await this.dingRawRepo.save(entities, { chunk: 50 });
|
|
imported += entities.length;
|
|
this.emit('saving', imported, newRecords.length, `Saved ${imported}/${newRecords.length}`);
|
|
} catch (err: unknown) {
|
|
const msg = err instanceof Error ? err.message : String(err);
|
|
errors.push(`Batch save error at offset ${i}: ${msg}`);
|
|
this.logger.error(`Batch save error: ${msg}`);
|
|
}
|
|
}
|
|
|
|
// Stage 4: Auto-match (optional)
|
|
if (params.autoMatch && imported > 0) {
|
|
this.emit('matching', 0, imported, 'Auto-matching records to students...');
|
|
matched = await this.autoMatchUnmatched();
|
|
this.emit('matching', matched, imported, `Matched ${matched} records to students`);
|
|
}
|
|
|
|
const duration = Date.now() - startedAt;
|
|
this.emit('complete', imported, rawResults.length, `Import complete: ${imported} new, ${skipped} skipped, ${matched} matched (${duration}ms)`);
|
|
this.logger.log(`DingTalk attendance import done: ${imported} imported, ${skipped} skipped, ${matched} matched`);
|
|
return { success: true, imported, skipped, matched, errors, duration };
|
|
} catch (err: unknown) {
|
|
const msg = err instanceof Error ? err.message : String(err);
|
|
errors.push(msg);
|
|
this.emit('error', imported, 0, `Import failed: ${msg}`, msg);
|
|
this.logger.error(`DingTalk attendance import failed: ${msg}`);
|
|
return { success: false, imported, skipped, matched, errors, duration: Date.now() - startedAt };
|
|
} finally {
|
|
clearTimeout(safetyTimer);
|
|
this.isRunning = false;
|
|
this.importingUserId = undefined;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* DingTalk requires userIds, accepts at most 50 users per request, and
|
|
* allows a maximum inclusive date range of 7 calendar days.
|
|
*/
|
|
private async fetchAllPages(params: {
|
|
startDate: string;
|
|
endDate: string;
|
|
userIds?: string[];
|
|
}): Promise<DingTalkAttendanceResult[]> {
|
|
const userIds = [...new Set((params.userIds ?? []).filter(Boolean))];
|
|
if (userIds.length === 0) {
|
|
throw new BadRequestException('拉取钉钉考勤必须指定人员范围');
|
|
}
|
|
if (params.startDate > params.endDate) {
|
|
throw new BadRequestException('开始日期不能晚于结束日期');
|
|
}
|
|
|
|
const allResults: DingTalkAttendanceResult[] = [];
|
|
const userBatches = this.chunk(userIds, 50);
|
|
const dateRanges = this.splitDateRanges(params.startDate, params.endDate, 7);
|
|
const totalRequests = userBatches.length * dateRanges.length;
|
|
let completedRequests = 0;
|
|
|
|
for (const range of dateRanges) {
|
|
for (const users of userBatches) {
|
|
const batch = await this.dingTalkService.fetchAttendanceResults({
|
|
startDate: range.startDate,
|
|
endDate: range.endDate,
|
|
userIds: users,
|
|
});
|
|
allResults.push(...batch);
|
|
completedRequests++;
|
|
this.emit(
|
|
'fetching',
|
|
completedRequests,
|
|
totalRequests,
|
|
`已完成 ${completedRequests}/${totalRequests} 批，获取 ${allResults.length} 条记录`,
|
|
);
|
|
}
|
|
}
|
|
|
|
return allResults;
|
|
}
|
|
|
|
private chunk<T>(items: T[], size: number): T[][] {
|
|
const result: T[][] = [];
|
|
for (let index = 0; index < items.length; index += size) {
|
|
result.push(items.slice(index, index + size));
|
|
}
|
|
return result;
|
|
}
|
|
|
|
private splitDateRanges(
|
|
startDate: string,
|
|
endDate: string,
|
|
maxDays: number,
|
|
): Array<{ startDate: string; endDate: string }> {
|
|
const ranges: Array<{ startDate: string; endDate: string }> = [];
|
|
let cursor = this.parseDate(startDate);
|
|
const end = this.parseDate(endDate);
|
|
|
|
while (cursor.getTime() <= end.getTime()) {
|
|
const rangeEnd = new Date(cursor);
|
|
rangeEnd.setUTCDate(rangeEnd.getUTCDate() + maxDays - 1);
|
|
if (rangeEnd.getTime() > end.getTime()) rangeEnd.setTime(end.getTime());
|
|
ranges.push({
|
|
startDate: this.formatDate(cursor),
|
|
endDate: this.formatDate(rangeEnd),
|
|
});
|
|
cursor = new Date(rangeEnd);
|
|
cursor.setUTCDate(cursor.getUTCDate() + 1);
|
|
}
|
|
return ranges;
|
|
}
|
|
|
|
private parseDate(value: string): Date {
|
|
const date = new Date(`${value}T00:00:00.000Z`);
|
|
if (Number.isNaN(date.getTime())) {
|
|
throw new BadRequestException(`无效日期: ${value}`);
|
|
}
|
|
return date;
|
|
}
|
|
|
|
private formatDate(value: Date): string {
|
|
return value.toISOString().slice(0, 10);
|
|
}
|
|
|
|
/**
|
|
* Query which dingIds already exist to skip duplicates.
|
|
*/
|
|
private async getExistingRecordsByDingId(
|
|
results: DingTalkAttendanceResult[],
|
|
): Promise<Map<string, DingAttendanceRaw>> {
|
|
const dingIds = results.map((r) => r.checkId).filter(Boolean);
|
|
if (dingIds.length === 0) return new Map();
|
|
|
|
const existing = await this.dingRawRepo.find({
|
|
where: { dingId: In(dingIds) },
|
|
});
|
|
return new Map(existing.map((entity) => [entity.dingId, entity]));
|
|
}
|
|
|
|
private async refreshDuplicatePunchMetadata(
|
|
results: DingTalkAttendanceResult[],
|
|
existingByDingId: Map<string, DingAttendanceRaw>,
|
|
): Promise<void> {
|
|
const changed: DingAttendanceRaw[] = [];
|
|
for (const result of results) {
|
|
const entity = existingByDingId.get(result.checkId);
|
|
if (!entity) continue;
|
|
const punchSource = result.sourceType || entity.punchSource || null;
|
|
const punchDeviceName = result.deviceName || entity.punchDeviceName || null;
|
|
const punchDeviceId = result.deviceId || entity.punchDeviceId || null;
|
|
if (
|
|
entity.punchSource === punchSource &&
|
|
entity.punchDeviceName === punchDeviceName &&
|
|
entity.punchDeviceId === punchDeviceId
|
|
) {
|
|
continue;
|
|
}
|
|
entity.punchSource = punchSource;
|
|
entity.punchDeviceName = punchDeviceName;
|
|
entity.punchDeviceId = punchDeviceId;
|
|
entity.rawData = JSON.stringify(result);
|
|
changed.push(entity);
|
|
}
|
|
if (changed.length > 0) {
|
|
await this.dingRawRepo.save(changed, { chunk: 50 });
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Map a DingTalk API result to a DingAttendanceRaw entity.
|
|
*/
|
|
private async mapToEntity(r: DingTalkAttendanceResult): Promise<DingAttendanceRaw> {
|
|
const entity = new DingAttendanceRaw();
|
|
entity.dingUserId = r.userId;
|
|
entity.userName = r.userName || await this.resolveStudentName(r.userId);
|
|
entity.attendanceDate = r.workDate;
|
|
entity.dingId = r.checkId;
|
|
entity.attendanceType = r.checkType || 'OnDuty';
|
|
entity.timeResult = r.timeResult;
|
|
entity.locationResult = r.locationResult || '';
|
|
entity.punchSource = r.sourceType || null;
|
|
entity.punchDeviceName = r.deviceName || null;
|
|
entity.punchDeviceId = r.deviceId || null;
|
|
|
|
// Parse check-in/out times
|
|
if (r.actualCheckTime) {
|
|
const dt = new Date(r.actualCheckTime);
|
|
if (!isNaN(dt.getTime())) {
|
|
if (r.checkType === 'OnDuty') {
|
|
entity.checkInTime = dt;
|
|
} else {
|
|
entity.checkOutTime = dt;
|
|
}
|
|
}
|
|
}
|
|
|
|
entity.matchStatus = 'unmatched';
|
|
entity.rawData = JSON.stringify(r);
|
|
return entity;
|
|
}
|
|
|
|
private async resolveStudentName(dingUserId: string): Promise<string> {
|
|
const mapping = await this.studentDingMappingRepo.findOne({
|
|
where: { dingUserId },
|
|
});
|
|
if (!mapping) return '';
|
|
const student = await this.studentRepo.findOne({ where: { id: mapping.studentId } });
|
|
return student?.name || '';
|
|
}
|
|
|
|
/**
|
|
* Auto-match unmatched records to students via the dingUserId → userId mapping chain.
|
|
*
|
|
* Delegates to {@link AttendanceService.autoMatchDingRecords} to avoid duplicate logic.
|
|
*/
|
|
private async autoMatchUnmatched(): Promise<number> {
|
|
const result = await this.attendanceService.autoMatchDingRecords();
|
|
return result.matched;
|
|
}
|
|
|
|
/**
|
|
* Emit a progress event to the stream.
|
|
*/
|
|
private emit(
|
|
phase: ImportProgressEvent['phase'],
|
|
current: number,
|
|
total: number,
|
|
message: string,
|
|
error?: string,
|
|
): void {
|
|
this.progressSubject.next({ phase, current, total, message, error, userId: this.importingUserId });
|
|
}
|
|
}
|