feat(iit): Complete CRA Agent V3.0 P0 milestone - autonomous QC pipeline
P0-1: Variable list sync from REDCap metadata P0-2: QC rule configuration with JSON Logic + AI suggestion P0-3: Scheduled QC + report generation + eQuery closed loop P0-4: Unified dashboard + AI stream timeline + critical events Backend: - Add IitEquery, IitCriticalEvent Prisma models + migration - Add cronEnabled/cronExpression to IitProject - Implement eQuery service/controller/routes (CRUD + respond/review/close) - Implement DailyQcOrchestrator (report -> eQuery -> critical events -> notify) - Add AI rule suggestion service - Register daily QC cron worker and eQuery auto-review worker - Extend QC cockpit with timeline, trend, critical events APIs - Fix timeline issues field compat (object vs array format) Frontend: - Create IIT business module with 6 pages (Dashboard, AI Stream, eQuery, Reports, Variable List + project config pages) - Migrate IIT config from admin panel to business module - Implement health score, risk heatmap, trend chart, critical event alerts - Register IIT module in App router and top navigation Testing: - Add E2E API test script covering 7 modules (46 assertions, all passing) Tested: E2E API tests 46/46 passed, backend and frontend verified Made-with: Cursor
This commit is contained in:
@@ -110,7 +110,7 @@ import { tenantRoutes, moduleRoutes } from './modules/admin/routes/tenantRoutes.
|
||||
import { userRoutes } from './modules/admin/routes/userRoutes.js';
|
||||
import { statsRoutes, userOverviewRoute } from './modules/admin/routes/statsRoutes.js';
|
||||
import { systemKbRoutes } from './modules/admin/system-kb/index.js';
|
||||
import { iitProjectRoutes, iitQcRuleRoutes, iitUserMappingRoutes, iitBatchRoutes, iitQcCockpitRoutes } from './modules/admin/iit-projects/index.js';
|
||||
import { iitProjectRoutes, iitQcRuleRoutes, iitUserMappingRoutes, iitBatchRoutes, iitQcCockpitRoutes, iitEqueryRoutes } from './modules/admin/iit-projects/index.js';
|
||||
await fastify.register(tenantRoutes, { prefix: '/api/admin/tenants' });
|
||||
await fastify.register(moduleRoutes, { prefix: '/api/admin/modules' });
|
||||
await fastify.register(userRoutes, { prefix: '/api/admin/users' });
|
||||
@@ -122,6 +122,7 @@ await fastify.register(iitQcRuleRoutes, { prefix: '/api/v1/admin/iit-projects' }
|
||||
await fastify.register(iitUserMappingRoutes, { prefix: '/api/v1/admin/iit-projects' });
|
||||
await fastify.register(iitBatchRoutes, { prefix: '/api/v1/admin/iit-projects' }); // 一键全量质控/汇总
|
||||
await fastify.register(iitQcCockpitRoutes, { prefix: '/api/v1/admin/iit-projects' }); // 质控驾驶舱
|
||||
await fastify.register(iitEqueryRoutes, { prefix: '/api/v1/admin/iit-projects' }); // eQuery 闭环
|
||||
logger.info('✅ 运营管理路由已注册: /api/admin/tenants, /api/admin/modules, /api/admin/users, /api/admin/stats, /api/v1/admin/system-kb, /api/v1/admin/iit-projects');
|
||||
|
||||
// ============================================
|
||||
|
||||
138
backend/src/modules/admin/iit-projects/iitEqueryController.ts
Normal file
138
backend/src/modules/admin/iit-projects/iitEqueryController.ts
Normal file
@@ -0,0 +1,138 @@
|
||||
/**
|
||||
* eQuery 控制器
|
||||
*/
|
||||
|
||||
import { FastifyRequest, FastifyReply } from 'fastify';
|
||||
import { getIitEqueryService } from './iitEqueryService.js';
|
||||
import { prisma } from '../../../config/database.js';
|
||||
import { logger } from '../../../common/logging/index.js';
|
||||
|
||||
interface ProjectIdParams {
|
||||
projectId: string;
|
||||
}
|
||||
|
||||
interface EqueryIdParams {
|
||||
projectId: string;
|
||||
equeryId: string;
|
||||
}
|
||||
|
||||
export async function listEqueries(
|
||||
request: FastifyRequest<{
|
||||
Params: ProjectIdParams;
|
||||
Querystring: { status?: string; recordId?: string; severity?: string; page?: string; pageSize?: string };
|
||||
}>,
|
||||
reply: FastifyReply
|
||||
) {
|
||||
try {
|
||||
const { projectId } = request.params;
|
||||
const query = request.query as any;
|
||||
const service = getIitEqueryService(prisma);
|
||||
const result = await service.list({
|
||||
projectId,
|
||||
status: query.status,
|
||||
recordId: query.recordId,
|
||||
severity: query.severity,
|
||||
page: query.page ? parseInt(query.page) : 1,
|
||||
pageSize: query.pageSize ? parseInt(query.pageSize) : 50,
|
||||
});
|
||||
return reply.send({ success: true, data: result });
|
||||
} catch (error) {
|
||||
const msg = error instanceof Error ? error.message : String(error);
|
||||
logger.error('获取 eQuery 列表失败', { error: msg });
|
||||
return reply.status(500).send({ success: false, error: msg });
|
||||
}
|
||||
}
|
||||
|
||||
export async function getEquery(
|
||||
request: FastifyRequest<{ Params: EqueryIdParams }>,
|
||||
reply: FastifyReply
|
||||
) {
|
||||
try {
|
||||
const { equeryId } = request.params;
|
||||
const service = getIitEqueryService(prisma);
|
||||
const equery = await service.getById(equeryId);
|
||||
if (!equery) return reply.status(404).send({ success: false, error: 'eQuery 不存在' });
|
||||
return reply.send({ success: true, data: equery });
|
||||
} catch (error) {
|
||||
const msg = error instanceof Error ? error.message : String(error);
|
||||
return reply.status(500).send({ success: false, error: msg });
|
||||
}
|
||||
}
|
||||
|
||||
export async function getEqueryStats(
|
||||
request: FastifyRequest<{ Params: ProjectIdParams }>,
|
||||
reply: FastifyReply
|
||||
) {
|
||||
try {
|
||||
const { projectId } = request.params;
|
||||
const service = getIitEqueryService(prisma);
|
||||
const stats = await service.getStats(projectId);
|
||||
return reply.send({ success: true, data: stats });
|
||||
} catch (error) {
|
||||
const msg = error instanceof Error ? error.message : String(error);
|
||||
return reply.status(500).send({ success: false, error: msg });
|
||||
}
|
||||
}
|
||||
|
||||
export async function respondEquery(
|
||||
request: FastifyRequest<{
|
||||
Params: EqueryIdParams;
|
||||
Body: { responseText: string; responseData?: Record<string, unknown> };
|
||||
}>,
|
||||
reply: FastifyReply
|
||||
) {
|
||||
try {
|
||||
const { equeryId } = request.params;
|
||||
const { responseText, responseData } = request.body;
|
||||
if (!responseText) {
|
||||
return reply.status(400).send({ success: false, error: '请提供回复内容' });
|
||||
}
|
||||
const service = getIitEqueryService(prisma);
|
||||
const updated = await service.respond(equeryId, { responseText, responseData });
|
||||
return reply.send({ success: true, data: updated });
|
||||
} catch (error) {
|
||||
const msg = error instanceof Error ? error.message : String(error);
|
||||
logger.error('eQuery 回复失败', { error: msg });
|
||||
return reply.status(400).send({ success: false, error: msg });
|
||||
}
|
||||
}
|
||||
|
||||
export async function reviewEquery(
|
||||
request: FastifyRequest<{
|
||||
Params: EqueryIdParams;
|
||||
Body: { passed: boolean; reviewNote?: string };
|
||||
}>,
|
||||
reply: FastifyReply
|
||||
) {
|
||||
try {
|
||||
const { equeryId } = request.params;
|
||||
const { passed, reviewNote } = request.body;
|
||||
const service = getIitEqueryService(prisma);
|
||||
const updated = await service.review(equeryId, { passed, reviewNote });
|
||||
return reply.send({ success: true, data: updated });
|
||||
} catch (error) {
|
||||
const msg = error instanceof Error ? error.message : String(error);
|
||||
logger.error('eQuery 复核失败', { error: msg });
|
||||
return reply.status(400).send({ success: false, error: msg });
|
||||
}
|
||||
}
|
||||
|
||||
export async function closeEquery(
|
||||
request: FastifyRequest<{
|
||||
Params: EqueryIdParams;
|
||||
Body: { closedBy: string; resolution?: string };
|
||||
}>,
|
||||
reply: FastifyReply
|
||||
) {
|
||||
try {
|
||||
const { equeryId } = request.params;
|
||||
const { closedBy, resolution } = request.body;
|
||||
const service = getIitEqueryService(prisma);
|
||||
const updated = await service.close(equeryId, closedBy || 'manual', resolution);
|
||||
return reply.send({ success: true, data: updated });
|
||||
} catch (error) {
|
||||
const msg = error instanceof Error ? error.message : String(error);
|
||||
logger.error('eQuery 关闭失败', { error: msg });
|
||||
return reply.status(400).send({ success: false, error: msg });
|
||||
}
|
||||
}
|
||||
26
backend/src/modules/admin/iit-projects/iitEqueryRoutes.ts
Normal file
26
backend/src/modules/admin/iit-projects/iitEqueryRoutes.ts
Normal file
@@ -0,0 +1,26 @@
|
||||
/**
|
||||
* eQuery 路由
|
||||
*/
|
||||
|
||||
import { FastifyInstance } from 'fastify';
|
||||
import * as controller from './iitEqueryController.js';
|
||||
|
||||
export async function iitEqueryRoutes(fastify: FastifyInstance) {
|
||||
// 获取项目的 eQuery 列表
|
||||
fastify.get('/:projectId/equeries', controller.listEqueries);
|
||||
|
||||
// 获取 eQuery 统计
|
||||
fastify.get('/:projectId/equeries/stats', controller.getEqueryStats);
|
||||
|
||||
// 获取单条 eQuery
|
||||
fastify.get('/:projectId/equeries/:equeryId', controller.getEquery);
|
||||
|
||||
// CRC 回复 eQuery
|
||||
fastify.post('/:projectId/equeries/:equeryId/respond', controller.respondEquery);
|
||||
|
||||
// AI 复核 eQuery
|
||||
fastify.post('/:projectId/equeries/:equeryId/review', controller.reviewEquery);
|
||||
|
||||
// 手动关闭 eQuery
|
||||
fastify.post('/:projectId/equeries/:equeryId/close', controller.closeEquery);
|
||||
}
|
||||
308
backend/src/modules/admin/iit-projects/iitEqueryService.ts
Normal file
308
backend/src/modules/admin/iit-projects/iitEqueryService.ts
Normal file
@@ -0,0 +1,308 @@
|
||||
/**
|
||||
* eQuery 闭环服务
|
||||
*
|
||||
* 状态机: pending → responded → reviewing → closed / reopened
|
||||
* AI 自动生成 Query → CRC 回复 → AI 复核 → 关闭 / 重开
|
||||
*/
|
||||
|
||||
import { PrismaClient } from '@prisma/client';
|
||||
import { logger } from '../../../common/logging/index.js';
|
||||
import { jobQueue } from '../../../common/jobs/index.js';
|
||||
|
||||
// ==================== Types ====================
|
||||
|
||||
export type EqueryStatus = 'pending' | 'responded' | 'reviewing' | 'closed' | 'reopened';
|
||||
|
||||
export interface CreateEqueryInput {
|
||||
projectId: string;
|
||||
recordId: string;
|
||||
eventId?: string;
|
||||
formName?: string;
|
||||
fieldName?: string;
|
||||
qcLogId?: string;
|
||||
reportId?: string;
|
||||
queryText: string;
|
||||
expectedAction?: string;
|
||||
severity?: string;
|
||||
category?: string;
|
||||
assignedTo?: string;
|
||||
}
|
||||
|
||||
export interface RespondEqueryInput {
|
||||
responseText: string;
|
||||
responseData?: Record<string, unknown>;
|
||||
}
|
||||
|
||||
export interface ReviewEqueryInput {
|
||||
passed: boolean;
|
||||
reviewNote?: string;
|
||||
}
|
||||
|
||||
export interface EqueryListParams {
|
||||
projectId: string;
|
||||
status?: string;
|
||||
recordId?: string;
|
||||
severity?: string;
|
||||
assignedTo?: string;
|
||||
page?: number;
|
||||
pageSize?: number;
|
||||
}
|
||||
|
||||
export interface EqueryStats {
|
||||
total: number;
|
||||
pending: number;
|
||||
responded: number;
|
||||
reviewing: number;
|
||||
closed: number;
|
||||
reopened: number;
|
||||
avgResolutionHours: number | null;
|
||||
}
|
||||
|
||||
// ==================== Service ====================
|
||||
|
||||
export class IitEqueryService {
|
||||
constructor(private prisma: PrismaClient) {}
|
||||
|
||||
/**
|
||||
* 创建 eQuery（通常由 AI 质控后自动调用）
|
||||
*/
|
||||
async create(input: CreateEqueryInput) {
|
||||
const equery = await this.prisma.iitEquery.create({
|
||||
data: {
|
||||
projectId: input.projectId,
|
||||
recordId: input.recordId,
|
||||
eventId: input.eventId,
|
||||
formName: input.formName,
|
||||
fieldName: input.fieldName,
|
||||
qcLogId: input.qcLogId,
|
||||
reportId: input.reportId,
|
||||
queryText: input.queryText,
|
||||
expectedAction: input.expectedAction,
|
||||
severity: input.severity || 'warning',
|
||||
category: input.category,
|
||||
status: 'pending',
|
||||
assignedTo: input.assignedTo,
|
||||
},
|
||||
});
|
||||
|
||||
logger.info('eQuery created', {
|
||||
id: equery.id,
|
||||
projectId: input.projectId,
|
||||
recordId: input.recordId,
|
||||
severity: input.severity,
|
||||
});
|
||||
|
||||
return equery;
|
||||
}
|
||||
|
||||
/**
|
||||
* 批量创建 eQuery（报告生成后一次性派发）
|
||||
*/
|
||||
async createBatch(inputs: CreateEqueryInput[]) {
|
||||
if (inputs.length === 0) return { count: 0 };
|
||||
|
||||
const data = inputs.map((input) => ({
|
||||
projectId: input.projectId,
|
||||
recordId: input.recordId,
|
||||
eventId: input.eventId,
|
||||
formName: input.formName,
|
||||
fieldName: input.fieldName,
|
||||
qcLogId: input.qcLogId,
|
||||
reportId: input.reportId,
|
||||
queryText: input.queryText,
|
||||
expectedAction: input.expectedAction,
|
||||
severity: input.severity || 'warning',
|
||||
category: input.category,
|
||||
status: 'pending' as const,
|
||||
assignedTo: input.assignedTo,
|
||||
}));
|
||||
|
||||
const result = await this.prisma.iitEquery.createMany({ data });
|
||||
|
||||
logger.info('eQuery batch created', {
|
||||
projectId: inputs[0].projectId,
|
||||
count: result.count,
|
||||
});
|
||||
|
||||
return { count: result.count };
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取 eQuery 列表
|
||||
*/
|
||||
async list(params: EqueryListParams) {
|
||||
const { projectId, status, recordId, severity, assignedTo, page = 1, pageSize = 50 } = params;
|
||||
|
||||
const where: any = { projectId };
|
||||
if (status) where.status = status;
|
||||
if (recordId) where.recordId = recordId;
|
||||
if (severity) where.severity = severity;
|
||||
if (assignedTo) where.assignedTo = assignedTo;
|
||||
|
||||
const [items, total] = await Promise.all([
|
||||
this.prisma.iitEquery.findMany({
|
||||
where,
|
||||
orderBy: [{ status: 'asc' }, { createdAt: 'desc' }],
|
||||
skip: (page - 1) * pageSize,
|
||||
take: pageSize,
|
||||
}),
|
||||
this.prisma.iitEquery.count({ where }),
|
||||
]);
|
||||
|
||||
return { items, total, page, pageSize };
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取单条 eQuery
|
||||
*/
|
||||
async getById(id: string) {
|
||||
return this.prisma.iitEquery.findUnique({ where: { id } });
|
||||
}
|
||||
|
||||
/**
|
||||
* CRC 回复 eQuery（pending / reopened → responded）
|
||||
*/
|
||||
async respond(id: string, input: RespondEqueryInput) {
|
||||
const equery = await this.prisma.iitEquery.findUnique({ where: { id } });
|
||||
if (!equery) throw new Error('eQuery 不存在');
|
||||
if (!['pending', 'reopened'].includes(equery.status)) {
|
||||
throw new Error(`当前状态 ${equery.status} 不允许回复`);
|
||||
}
|
||||
|
||||
const updated = await this.prisma.iitEquery.update({
|
||||
where: { id },
|
||||
data: {
|
||||
status: 'responded',
|
||||
responseText: input.responseText,
|
||||
responseData: input.responseData as any,
|
||||
respondedAt: new Date(),
|
||||
},
|
||||
});
|
||||
|
||||
logger.info('eQuery responded', { id, recordId: equery.recordId });
|
||||
|
||||
// Trigger async AI review
|
||||
try {
|
||||
await jobQueue.push('iit_equery_review', {
|
||||
equeryId: id,
|
||||
projectId: equery.projectId,
|
||||
recordId: equery.recordId,
|
||||
fieldName: equery.fieldName,
|
||||
responseText: input.responseText,
|
||||
});
|
||||
logger.info('eQuery AI review job queued', { id });
|
||||
} catch (err) {
|
||||
logger.warn('Failed to queue eQuery review job (non-fatal)', { id, error: String(err) });
|
||||
}
|
||||
|
||||
return updated;
|
||||
}
|
||||
|
||||
/**
|
||||
* AI 复核 eQuery（responded → reviewing → closed / reopened）
|
||||
*/
|
||||
async review(id: string, input: ReviewEqueryInput) {
|
||||
const equery = await this.prisma.iitEquery.findUnique({ where: { id } });
|
||||
if (!equery) throw new Error('eQuery 不存在');
|
||||
if (equery.status !== 'responded') {
|
||||
throw new Error(`当前状态 ${equery.status} 不允许复核`);
|
||||
}
|
||||
|
||||
const now = new Date();
|
||||
const newStatus: EqueryStatus = input.passed ? 'closed' : 'reopened';
|
||||
|
||||
const updated = await this.prisma.iitEquery.update({
|
||||
where: { id },
|
||||
data: {
|
||||
status: newStatus,
|
||||
reviewResult: input.passed ? 'passed' : 'failed',
|
||||
reviewNote: input.reviewNote,
|
||||
reviewedAt: now,
|
||||
...(input.passed ? { closedAt: now, closedBy: 'ai_review' } : {}),
|
||||
},
|
||||
});
|
||||
|
||||
logger.info('eQuery reviewed', {
|
||||
id,
|
||||
recordId: equery.recordId,
|
||||
passed: input.passed,
|
||||
newStatus,
|
||||
});
|
||||
return updated;
|
||||
}
|
||||
|
||||
/**
|
||||
* 手动关闭 eQuery
|
||||
*/
|
||||
async close(id: string, closedBy: string, resolution?: string) {
|
||||
const equery = await this.prisma.iitEquery.findUnique({ where: { id } });
|
||||
if (!equery) throw new Error('eQuery 不存在');
|
||||
if (equery.status === 'closed') throw new Error('eQuery 已关闭');
|
||||
|
||||
const updated = await this.prisma.iitEquery.update({
|
||||
where: { id },
|
||||
data: {
|
||||
status: 'closed',
|
||||
closedAt: new Date(),
|
||||
closedBy,
|
||||
resolution,
|
||||
},
|
||||
});
|
||||
|
||||
logger.info('eQuery manually closed', { id, closedBy });
|
||||
return updated;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取统计
|
||||
*/
|
||||
async getStats(projectId: string): Promise<EqueryStats> {
|
||||
const counts = await this.prisma.iitEquery.groupBy({
|
||||
by: ['status'],
|
||||
where: { projectId },
|
||||
_count: true,
|
||||
});
|
||||
|
||||
const statusMap: Record<string, number> = {};
|
||||
let total = 0;
|
||||
for (const c of counts) {
|
||||
statusMap[c.status] = c._count;
|
||||
total += c._count;
|
||||
}
|
||||
|
||||
// Average resolution time (hours) for closed equeries
|
||||
let avgResolutionHours: number | null = null;
|
||||
const closedEqueries = await this.prisma.iitEquery.findMany({
|
||||
where: { projectId, status: 'closed', closedAt: { not: null } },
|
||||
select: { createdAt: true, closedAt: true },
|
||||
take: 100,
|
||||
orderBy: { closedAt: 'desc' },
|
||||
});
|
||||
if (closedEqueries.length > 0) {
|
||||
const totalHours = closedEqueries.reduce((sum: number, eq: { createdAt: Date; closedAt: Date | null }) => {
|
||||
const diff = eq.closedAt!.getTime() - eq.createdAt.getTime();
|
||||
return sum + diff / (1000 * 60 * 60);
|
||||
}, 0);
|
||||
avgResolutionHours = Math.round((totalHours / closedEqueries.length) * 10) / 10;
|
||||
}
|
||||
|
||||
return {
|
||||
total,
|
||||
pending: statusMap['pending'] || 0,
|
||||
responded: statusMap['responded'] || 0,
|
||||
reviewing: statusMap['reviewing'] || 0,
|
||||
closed: statusMap['closed'] || 0,
|
||||
reopened: statusMap['reopened'] || 0,
|
||||
avgResolutionHours,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
let serviceInstance: IitEqueryService | null = null;
|
||||
|
||||
export function getIitEqueryService(prisma: PrismaClient): IitEqueryService {
|
||||
if (!serviceInstance) {
|
||||
serviceInstance = new IitEqueryService(prisma);
|
||||
}
|
||||
return serviceInstance;
|
||||
}
|
||||
@@ -286,6 +286,64 @@ export async function syncMetadata(
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取字段元数据列表
|
||||
*/
|
||||
export async function listFieldMetadata(
|
||||
request: FastifyRequest<{
|
||||
Params: ProjectIdParams;
|
||||
Querystring: { formName?: string; search?: string };
|
||||
}>,
|
||||
reply: FastifyReply
|
||||
) {
|
||||
try {
|
||||
const { id } = request.params;
|
||||
const { formName, search } = request.query as { formName?: string; search?: string };
|
||||
|
||||
const where: any = { projectId: id };
|
||||
if (formName) {
|
||||
where.formName = formName;
|
||||
}
|
||||
if (search) {
|
||||
where.OR = [
|
||||
{ fieldName: { contains: search, mode: 'insensitive' } },
|
||||
{ fieldLabel: { contains: search, mode: 'insensitive' } },
|
||||
];
|
||||
}
|
||||
|
||||
const [fields, total] = await Promise.all([
|
||||
prisma.iitFieldMetadata.findMany({
|
||||
where,
|
||||
orderBy: [{ formName: 'asc' }, { fieldName: 'asc' }],
|
||||
}),
|
||||
prisma.iitFieldMetadata.count({ where }),
|
||||
]);
|
||||
|
||||
const forms = await prisma.iitFieldMetadata.findMany({
|
||||
where: { projectId: id },
|
||||
select: { formName: true },
|
||||
distinct: ['formName'],
|
||||
orderBy: { formName: 'asc' },
|
||||
});
|
||||
|
||||
return reply.send({
|
||||
success: true,
|
||||
data: {
|
||||
fields,
|
||||
total,
|
||||
forms: forms.map(f => f.formName),
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
logger.error('获取字段元数据失败', { error: message });
|
||||
return reply.status(500).send({
|
||||
success: false,
|
||||
error: message,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 关联知识库
|
||||
*/
|
||||
|
||||
@@ -34,6 +34,9 @@ export async function iitProjectRoutes(fastify: FastifyInstance) {
|
||||
// 同步 REDCap 元数据
|
||||
fastify.post('/:id/sync-metadata', controller.syncMetadata);
|
||||
|
||||
// 获取字段元数据列表
|
||||
fastify.get('/:id/field-metadata', controller.listFieldMetadata);
|
||||
|
||||
// ==================== 知识库关联 ====================
|
||||
|
||||
// 关联知识库
|
||||
|
||||
@@ -12,6 +12,7 @@ import { FastifyRequest, FastifyReply } from 'fastify';
|
||||
import { iitQcCockpitService } from './iitQcCockpitService.js';
|
||||
import { QcReportService } from '../../iit-manager/services/QcReportService.js';
|
||||
import { logger } from '../../../common/logging/index.js';
|
||||
import { prisma } from '../../../config/database.js';
|
||||
|
||||
class IitQcCockpitController {
|
||||
/**
|
||||
@@ -185,6 +186,185 @@ class IitQcCockpitController {
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取 AI 工作时间线（QC 日志 + Agent Trace 合并）
|
||||
*/
|
||||
async getTimeline(
|
||||
request: FastifyRequest<{
|
||||
Params: { projectId: string };
|
||||
Querystring: { page?: string; pageSize?: string; date?: string };
|
||||
}>,
|
||||
reply: FastifyReply
|
||||
) {
|
||||
const { projectId } = request.params;
|
||||
const query = request.query as any;
|
||||
const page = query.page ? parseInt(query.page) : 1;
|
||||
const pageSize = query.pageSize ? parseInt(query.pageSize) : 50;
|
||||
const dateFilter = query.date;
|
||||
|
||||
try {
|
||||
const dateWhere: any = {};
|
||||
if (dateFilter) {
|
||||
const start = new Date(dateFilter);
|
||||
const end = new Date(dateFilter);
|
||||
end.setDate(end.getDate() + 1);
|
||||
dateWhere.createdAt = { gte: start, lt: end };
|
||||
}
|
||||
|
||||
const [qcLogs, totalLogs] = await Promise.all([
|
||||
prisma.iitQcLog.findMany({
|
||||
where: { projectId, ...dateWhere },
|
||||
orderBy: { createdAt: 'desc' },
|
||||
skip: (page - 1) * pageSize,
|
||||
take: pageSize,
|
||||
select: {
|
||||
id: true,
|
||||
recordId: true,
|
||||
eventId: true,
|
||||
qcType: true,
|
||||
formName: true,
|
||||
status: true,
|
||||
issues: true,
|
||||
rulesEvaluated: true,
|
||||
rulesPassed: true,
|
||||
rulesFailed: true,
|
||||
triggeredBy: true,
|
||||
createdAt: true,
|
||||
},
|
||||
}),
|
||||
prisma.iitQcLog.count({ where: { projectId, ...dateWhere } }),
|
||||
]);
|
||||
|
||||
// Transform to timeline items
|
||||
const items = qcLogs.map((log) => {
|
||||
const rawIssues = log.issues as any;
|
||||
const issues: any[] = Array.isArray(rawIssues) ? rawIssues : (rawIssues?.items || []);
|
||||
const redCount = issues.filter((i: any) => i.level === 'RED').length;
|
||||
const yellowCount = issues.filter((i: any) => i.level === 'YELLOW').length;
|
||||
|
||||
let description = `扫描受试者 ${log.recordId}`;
|
||||
if (log.formName) description += ` [${log.formName}]`;
|
||||
description += ` → 执行 ${log.rulesEvaluated} 条规则 (${log.rulesPassed} 通过`;
|
||||
if (log.rulesFailed > 0) description += `, ${log.rulesFailed} 失败`;
|
||||
description += ')';
|
||||
if (redCount > 0) description += ` → 发现 ${redCount} 个严重问题`;
|
||||
if (yellowCount > 0) description += `, ${yellowCount} 个警告`;
|
||||
|
||||
return {
|
||||
id: log.id,
|
||||
type: 'qc_check' as const,
|
||||
time: log.createdAt,
|
||||
recordId: log.recordId,
|
||||
formName: log.formName,
|
||||
status: log.status,
|
||||
triggeredBy: log.triggeredBy,
|
||||
description,
|
||||
details: {
|
||||
rulesEvaluated: log.rulesEvaluated,
|
||||
rulesPassed: log.rulesPassed,
|
||||
rulesFailed: log.rulesFailed,
|
||||
issuesSummary: { red: redCount, yellow: yellowCount },
|
||||
},
|
||||
};
|
||||
});
|
||||
|
||||
return reply.send({
|
||||
success: true,
|
||||
data: { items, total: totalLogs, page, pageSize },
|
||||
});
|
||||
} catch (error: any) {
|
||||
logger.error('[QcCockpitController] 获取时间线失败', { projectId, error: error.message });
|
||||
return reply.status(500).send({ success: false, error: error.message });
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取重大事件列表
|
||||
*/
|
||||
async getCriticalEvents(
|
||||
request: FastifyRequest<{
|
||||
Params: { projectId: string };
|
||||
Querystring: { status?: string; eventType?: string; page?: string; pageSize?: string };
|
||||
}>,
|
||||
reply: FastifyReply
|
||||
) {
|
||||
const { projectId } = request.params;
|
||||
const query = request.query as any;
|
||||
const page = query.page ? parseInt(query.page) : 1;
|
||||
const pageSize = query.pageSize ? parseInt(query.pageSize) : 50;
|
||||
|
||||
try {
|
||||
const where: any = { projectId };
|
||||
if (query.status) where.status = query.status;
|
||||
if (query.eventType) where.eventType = query.eventType;
|
||||
|
||||
const [items, total] = await Promise.all([
|
||||
prisma.iitCriticalEvent.findMany({
|
||||
where,
|
||||
orderBy: { createdAt: 'desc' },
|
||||
skip: (page - 1) * pageSize,
|
||||
take: pageSize,
|
||||
}),
|
||||
prisma.iitCriticalEvent.count({ where }),
|
||||
]);
|
||||
|
||||
return reply.send({
|
||||
success: true,
|
||||
data: { items, total, page, pageSize },
|
||||
});
|
||||
} catch (error: any) {
|
||||
logger.error('[QcCockpitController] 获取重大事件失败', { projectId, error: error.message });
|
||||
return reply.status(500).send({ success: false, error: error.message });
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取质控趋势数据（近30天每日通过率）
|
||||
*/
|
||||
async getTrend(
|
||||
request: FastifyRequest<{
|
||||
Params: { projectId: string };
|
||||
Querystring: { days?: string };
|
||||
}>,
|
||||
reply: FastifyReply
|
||||
) {
|
||||
const { projectId } = request.params;
|
||||
const days = parseInt((request.query as any).days || '30');
|
||||
|
||||
try {
|
||||
const since = new Date();
|
||||
since.setDate(since.getDate() - days);
|
||||
|
||||
const logs = await prisma.iitQcLog.findMany({
|
||||
where: { projectId, createdAt: { gte: since } },
|
||||
select: { createdAt: true, status: true },
|
||||
orderBy: { createdAt: 'asc' },
|
||||
});
|
||||
|
||||
// Group by date
|
||||
const dailyMap = new Map<string, { total: number; passed: number }>();
|
||||
for (const log of logs) {
|
||||
const dateKey = log.createdAt.toISOString().split('T')[0];
|
||||
const entry = dailyMap.get(dateKey) || { total: 0, passed: 0 };
|
||||
entry.total++;
|
||||
if (log.status === 'PASS') entry.passed++;
|
||||
dailyMap.set(dateKey, entry);
|
||||
}
|
||||
|
||||
const trend = Array.from(dailyMap.entries()).map(([date, { total, passed }]) => ({
|
||||
date,
|
||||
total,
|
||||
passed,
|
||||
passRate: total > 0 ? Math.round((passed / total) * 100) : 0,
|
||||
}));
|
||||
|
||||
return reply.send({ success: true, data: trend });
|
||||
} catch (error: any) {
|
||||
logger.error('[QcCockpitController] 获取趋势失败', { projectId, error: error.message });
|
||||
return reply.status(500).send({ success: false, error: error.message });
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export const iitQcCockpitController = new IitQcCockpitController();
|
||||
|
||||
@@ -242,4 +242,13 @@ export async function iitQcCockpitRoutes(fastify: FastifyInstance) {
|
||||
},
|
||||
},
|
||||
}, iitQcCockpitController.refreshReport.bind(iitQcCockpitController));
|
||||
|
||||
// AI 工作时间线
|
||||
fastify.get('/:projectId/qc-cockpit/timeline', iitQcCockpitController.getTimeline.bind(iitQcCockpitController));
|
||||
|
||||
// 重大事件列表
|
||||
fastify.get('/:projectId/qc-cockpit/critical-events', iitQcCockpitController.getCriticalEvents.bind(iitQcCockpitController));
|
||||
|
||||
// 质控趋势（近N天通过率折线）
|
||||
fastify.get('/:projectId/qc-cockpit/trend', iitQcCockpitController.getTrend.bind(iitQcCockpitController));
|
||||
}
|
||||
|
||||
@@ -4,6 +4,7 @@
|
||||
|
||||
import { FastifyRequest, FastifyReply } from 'fastify';
|
||||
import { getIitQcRuleService, CreateRuleInput, UpdateRuleInput, TestRuleInput } from './iitQcRuleService.js';
|
||||
import { getIitRuleSuggestionService } from './iitRuleSuggestionService.js';
|
||||
import { prisma } from '../../../config/database.js';
|
||||
import { logger } from '../../../common/logging/index.js';
|
||||
|
||||
@@ -286,3 +287,29 @@ export async function getRuleStats(
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* AI 规则建议
|
||||
*/
|
||||
export async function suggestRules(
|
||||
request: FastifyRequest<{ Params: ProjectIdParams }>,
|
||||
reply: FastifyReply
|
||||
) {
|
||||
try {
|
||||
const { projectId } = request.params;
|
||||
const service = getIitRuleSuggestionService(prisma);
|
||||
const suggestions = await service.suggestRules(projectId);
|
||||
|
||||
return reply.send({
|
||||
success: true,
|
||||
data: suggestions,
|
||||
});
|
||||
} catch (error) {
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
logger.error('AI 规则建议生成失败', { error: message });
|
||||
return reply.status(500).send({
|
||||
success: false,
|
||||
error: message,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -27,6 +27,9 @@ export async function iitQcRuleRoutes(fastify: FastifyInstance) {
|
||||
// 批量导入规则
|
||||
fastify.post('/:projectId/rules/import', controller.importRules);
|
||||
|
||||
// AI 规则建议
|
||||
fastify.post('/:projectId/rules/suggest', controller.suggestRules);
|
||||
|
||||
// 测试规则逻辑（不需要项目 ID）
|
||||
fastify.post('/rules/test', controller.testRule);
|
||||
}
|
||||
|
||||
@@ -0,0 +1,161 @@
|
||||
/**
|
||||
* AI 规则建议服务
|
||||
*
|
||||
* 读取项目的变量元数据和知识库文档，调用 LLM 生成质控规则建议。
|
||||
*/
|
||||
|
||||
import { PrismaClient } from '@prisma/client';
|
||||
import { LLMFactory } from '../../../common/llm/adapters/LLMFactory.js';
|
||||
import type { Message } from '../../../common/llm/adapters/types.js';
|
||||
import { logger } from '../../../common/logging/index.js';
|
||||
|
||||
export interface RuleSuggestion {
|
||||
name: string;
|
||||
field: string | string[];
|
||||
logic: Record<string, unknown>;
|
||||
message: string;
|
||||
severity: 'error' | 'warning' | 'info';
|
||||
category: string;
|
||||
}
|
||||
|
||||
export class IitRuleSuggestionService {
|
||||
constructor(private prisma: PrismaClient) {}
|
||||
|
||||
async suggestRules(projectId: string): Promise<RuleSuggestion[]> {
|
||||
const project = await this.prisma.iitProject.findFirst({
|
||||
where: { id: projectId, deletedAt: null },
|
||||
});
|
||||
if (!project) throw new Error('项目不存在');
|
||||
|
||||
// 1. Gather variable metadata
|
||||
const fields = await this.prisma.iitFieldMetadata.findMany({
|
||||
where: { projectId },
|
||||
orderBy: [{ formName: 'asc' }, { fieldName: 'asc' }],
|
||||
});
|
||||
|
||||
if (fields.length === 0) {
|
||||
throw new Error('请先从 REDCap 同步变量元数据');
|
||||
}
|
||||
|
||||
// 2. Gather knowledge base context (protocol summary)
|
||||
let protocolContext = '';
|
||||
if (project.knowledgeBaseId) {
|
||||
try {
|
||||
const docs = await this.prisma.ekbDocument.findMany({
|
||||
where: { kbId: project.knowledgeBaseId, status: 'completed' },
|
||||
select: { filename: true, summary: true, extractedText: true },
|
||||
take: 3,
|
||||
orderBy: { createdAt: 'desc' },
|
||||
});
|
||||
for (const doc of docs) {
|
||||
const text = doc.summary || doc.extractedText?.substring(0, 3000) || '';
|
||||
if (text) {
|
||||
protocolContext += `\n### ${doc.filename}\n${text}\n`;
|
||||
}
|
||||
}
|
||||
} catch (err) {
|
||||
logger.warn('读取知识库失败，仅基于变量生成规则', { error: String(err) });
|
||||
}
|
||||
}
|
||||
|
||||
// 3. Build variable summary for LLM
|
||||
const variableSummary = fields.map((f) => {
|
||||
const parts = [`${f.fieldName} (${f.fieldLabel}): type=${f.fieldType}, form=${f.formName}`];
|
||||
if (f.validation) parts.push(`validation=${f.validation}`);
|
||||
if (f.validationMin || f.validationMax) parts.push(`range=[${f.validationMin || ''},${f.validationMax || ''}]`);
|
||||
if (f.choices) parts.push(`choices=${f.choices.substring(0, 100)}`);
|
||||
if (f.required) parts.push('required=true');
|
||||
return parts.join(', ');
|
||||
}).join('\n');
|
||||
|
||||
// 4. Call LLM
|
||||
const systemPrompt = `You are an expert clinical research data manager. You generate quality control (QC) rules for clinical trial data captured in REDCap.
|
||||
|
||||
Rules must be in JSON Logic format (https://jsonlogic.com). Each rule checks one or more fields.
|
||||
|
||||
Available categories:
|
||||
- variable_qc: Field-level checks (range, required, format, enum)
|
||||
- inclusion: Inclusion criteria checks
|
||||
- exclusion: Exclusion criteria checks
|
||||
- lab_values: Lab value range checks
|
||||
- logic_check: Cross-field logic checks
|
||||
- protocol_deviation: Visit window / time constraint checks
|
||||
- ae_monitoring: AE reporting timeline checks
|
||||
|
||||
Severity levels: error (blocking), warning (review needed), info (informational)
|
||||
|
||||
Respond ONLY with a JSON array of rule objects. Each object must have these fields:
|
||||
- name (string): short descriptive name in Chinese
|
||||
- field (string or string[]): REDCap field name(s)
|
||||
- logic (object): JSON Logic expression
|
||||
- message (string): error message in Chinese
|
||||
- severity: "error" | "warning" | "info"
|
||||
- category: one of the categories listed above
|
||||
|
||||
Generate 5-10 practical rules. Focus on:
|
||||
1. Required field checks for key variables
|
||||
2. Range checks for numeric fields that have validation ranges
|
||||
3. Logical consistency checks between related fields
|
||||
4. Date field checks (visit windows, timelines)
|
||||
Do NOT include explanations, only the JSON array.`;
|
||||
|
||||
const userPrompt = `Project: ${project.name}
|
||||
|
||||
Variable List (${fields.length} fields):
|
||||
${variableSummary}
|
||||
${protocolContext ? `\nProtocol / Study Document Context:\n${protocolContext}` : ''}
|
||||
|
||||
Generate QC rules for this project:`;
|
||||
|
||||
const messages: Message[] = [
|
||||
{ role: 'system', content: systemPrompt },
|
||||
{ role: 'user', content: userPrompt },
|
||||
];
|
||||
|
||||
try {
|
||||
const llm = LLMFactory.getAdapter('deepseek-v3');
|
||||
const response = await llm.chat(messages, {
|
||||
temperature: 0.3,
|
||||
maxTokens: 4000,
|
||||
});
|
||||
|
||||
const content = response.content.trim();
|
||||
// Extract JSON array from response (handle markdown code fences)
|
||||
const jsonMatch = content.match(/\[[\s\S]*\]/);
|
||||
if (!jsonMatch) {
|
||||
logger.error('LLM 返回非 JSON 格式', { content: content.substring(0, 200) });
|
||||
throw new Error('AI 返回格式异常，请重试');
|
||||
}
|
||||
|
||||
const rules: RuleSuggestion[] = JSON.parse(jsonMatch[0]);
|
||||
|
||||
// Validate structure
|
||||
const validRules = rules.filter(
|
||||
(r) => r.name && r.field && r.logic && r.message && r.severity && r.category
|
||||
);
|
||||
|
||||
logger.info('AI 规则建议生成成功', {
|
||||
projectId,
|
||||
total: rules.length,
|
||||
valid: validRules.length,
|
||||
model: response.model,
|
||||
});
|
||||
|
||||
return validRules;
|
||||
} catch (err) {
|
||||
if (err instanceof SyntaxError) {
|
||||
throw new Error('AI 返回格式解析失败，请重试');
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let serviceInstance: IitRuleSuggestionService | null = null;
|
||||
|
||||
export function getIitRuleSuggestionService(prisma: PrismaClient): IitRuleSuggestionService {
|
||||
if (!serviceInstance) {
|
||||
serviceInstance = new IitRuleSuggestionService(prisma);
|
||||
}
|
||||
return serviceInstance;
|
||||
}
|
||||
@@ -7,12 +7,15 @@ export { iitQcRuleRoutes } from './iitQcRuleRoutes.js';
|
||||
export { iitUserMappingRoutes } from './iitUserMappingRoutes.js';
|
||||
export { iitBatchRoutes } from './iitBatchRoutes.js';
|
||||
export { iitQcCockpitRoutes } from './iitQcCockpitRoutes.js';
|
||||
export { iitEqueryRoutes } from './iitEqueryRoutes.js';
|
||||
export { IitProjectService, getIitProjectService } from './iitProjectService.js';
|
||||
export { IitQcRuleService, getIitQcRuleService } from './iitQcRuleService.js';
|
||||
export { IitUserMappingService, getIitUserMappingService } from './iitUserMappingService.js';
|
||||
export { iitQcCockpitService, IitQcCockpitService } from './iitQcCockpitService.js';
|
||||
export { IitEqueryService, getIitEqueryService } from './iitEqueryService.js';
|
||||
export * from './iitProjectController.js';
|
||||
export * from './iitQcRuleController.js';
|
||||
export * from './iitUserMappingController.js';
|
||||
export * from './iitBatchController.js';
|
||||
export * from './iitQcCockpitController.js';
|
||||
export * from './iitEqueryController.js';
|
||||
|
||||
@@ -18,6 +18,7 @@ import { PrismaClient } from '@prisma/client';
|
||||
import { logger } from '../../common/logging/index.js';
|
||||
import { createHardRuleEngine, QCResult } from './engines/HardRuleEngine.js';
|
||||
import { RedcapAdapter } from './adapters/RedcapAdapter.js';
|
||||
import { dailyQcOrchestrator } from './services/DailyQcOrchestrator.js';
|
||||
|
||||
// 初始化 Prisma Client
|
||||
const prisma = new PrismaClient();
|
||||
@@ -51,10 +52,22 @@ export async function initIitManager(): Promise<void> {
|
||||
// 1. 注册定时轮询任务（每5分钟）
|
||||
// =============================================
|
||||
// ⏸️ 暂时禁用定时轮询（MVP阶段，Webhook已足够）
|
||||
// TODO: Phase 2 - 实现定时轮询作为补充机制
|
||||
// await syncManager.initScheduledJob();
|
||||
|
||||
logger.info('IIT Manager: Scheduled job registration skipped (using Webhook only for MVP)');
|
||||
// =============================================
|
||||
// 1b. 注册定时全量质控任务（每天 08:00 UTC+8）
|
||||
// =============================================
|
||||
try {
|
||||
await (jobQueue as any).schedule(
|
||||
'iit_daily_qc',
|
||||
'0 0 * * *', // UTC 00:00 = 北京时间 08:00
|
||||
{},
|
||||
{ tz: 'Asia/Shanghai' }
|
||||
);
|
||||
logger.info('IIT Manager: Daily QC cron registered (08:00 CST)');
|
||||
} catch (cronErr: any) {
|
||||
logger.warn('IIT Manager: Failed to register daily QC cron (non-fatal)', { error: cronErr.message });
|
||||
}
|
||||
|
||||
// =============================================
|
||||
// 2. 注册Worker：处理定时轮询任务
|
||||
@@ -85,6 +98,162 @@ export async function initIitManager(): Promise<void> {
|
||||
|
||||
logger.info('IIT Manager: Worker registered - iit_redcap_poll');
|
||||
|
||||
// =============================================
|
||||
// 2b. 注册Worker：定时全量质控（每日 Agent 自主巡查）
|
||||
// =============================================
|
||||
jobQueue.process('iit_daily_qc', async (job: any) => {
|
||||
const startTime = Date.now();
|
||||
logger.info('Worker: iit_daily_qc started', { jobId: job.id });
|
||||
|
||||
try {
|
||||
const activeProjects = await prisma.iitProject.findMany({
|
||||
where: { status: 'active', deletedAt: null, cronEnabled: true },
|
||||
select: { id: true, name: true, redcapUrl: true, redcapApiToken: true },
|
||||
});
|
||||
|
||||
logger.info(`Daily QC: found ${activeProjects.length} active projects`);
|
||||
|
||||
for (const project of activeProjects) {
|
||||
try {
|
||||
const adapter = new RedcapAdapter(project.redcapUrl, project.redcapApiToken);
|
||||
const records = await adapter.exportRecords({ rawOrLabel: 'raw' });
|
||||
const recordIds = [...new Set(records.map((r: any) => r.record_id || r[Object.keys(r)[0]]))];
|
||||
|
||||
// Run HardRuleEngine for each record
|
||||
const ruleEngine = await createHardRuleEngine(project.id);
|
||||
let totalErrors = 0;
|
||||
let totalWarnings = 0;
|
||||
|
||||
for (const recordId of recordIds) {
|
||||
try {
|
||||
const recordData = records.filter((r: any) => (r.record_id || r[Object.keys(r)[0]]) === recordId);
|
||||
const flat = recordData.reduce((acc: any, row: any) => ({ ...acc, ...row }), {});
|
||||
const qcResult = await ruleEngine.execute(String(recordId), flat);
|
||||
|
||||
totalErrors += qcResult.summary.failed;
|
||||
totalWarnings += qcResult.summary.warnings;
|
||||
|
||||
// Save QC log
|
||||
await prisma.iitQcLog.create({
|
||||
data: {
|
||||
projectId: project.id,
|
||||
recordId: String(recordId),
|
||||
qcType: 'full',
|
||||
status: qcResult.overallStatus,
|
||||
issues: [
|
||||
...qcResult.errors.map((r: any) => ({ field: r.field, rule: r.ruleName, level: 'RED', message: r.errorMessage || r.message })),
|
||||
...qcResult.warnings.map((r: any) => ({ field: r.field, rule: r.ruleName, level: 'YELLOW', message: r.errorMessage || r.message })),
|
||||
],
|
||||
rulesEvaluated: qcResult.summary.totalRules,
|
||||
rulesSkipped: 0,
|
||||
rulesPassed: qcResult.summary.passed,
|
||||
rulesFailed: qcResult.summary.failed,
|
||||
ruleVersion: new Date().toISOString().split('T')[0],
|
||||
triggeredBy: 'cron',
|
||||
},
|
||||
});
|
||||
} catch (recErr: any) {
|
||||
logger.warn('Daily QC: record check failed', { projectId: project.id, recordId, error: recErr.message });
|
||||
}
|
||||
}
|
||||
|
||||
logger.info('Daily QC: project QC completed, starting orchestration', {
|
||||
projectId: project.id,
|
||||
projectName: project.name,
|
||||
records: recordIds.length,
|
||||
totalErrors,
|
||||
totalWarnings,
|
||||
});
|
||||
|
||||
// Orchestrate: report → eQuery → critical events → push
|
||||
try {
|
||||
const orchResult = await dailyQcOrchestrator.orchestrate(project.id);
|
||||
logger.info('Daily QC: orchestration completed', {
|
||||
projectId: project.id,
|
||||
...orchResult,
|
||||
});
|
||||
} catch (orchErr: any) {
|
||||
logger.error('Daily QC: orchestration failed (non-fatal)', {
|
||||
projectId: project.id,
|
||||
error: orchErr.message,
|
||||
});
|
||||
}
|
||||
} catch (projErr: any) {
|
||||
logger.error('Daily QC: project failed', { projectId: project.id, error: projErr.message });
|
||||
}
|
||||
}
|
||||
|
||||
logger.info('Worker: iit_daily_qc completed', {
|
||||
jobId: job.id,
|
||||
projects: activeProjects.length,
|
||||
durationMs: Date.now() - startTime,
|
||||
});
|
||||
return { success: true, projects: activeProjects.length };
|
||||
} catch (error: any) {
|
||||
logger.error('Worker: iit_daily_qc failed', { jobId: job.id, error: error.message });
|
||||
throw error;
|
||||
}
|
||||
});
|
||||
|
||||
logger.info('IIT Manager: Worker registered - iit_daily_qc');
|
||||
|
||||
// =============================================
|
||||
// 2c. 注册Worker：eQuery AI 自动复核
|
||||
// =============================================
|
||||
jobQueue.process('iit_equery_review', async (job: { id: string; data: { equeryId: string; projectId: string; recordId: string; fieldName?: string; responseText: string } }) => {
|
||||
const { equeryId, projectId, recordId, fieldName, responseText } = job.data;
|
||||
logger.info('Worker: iit_equery_review started', { jobId: job.id, equeryId });
|
||||
|
||||
try {
|
||||
// Re-check: get the latest record data and re-run relevant rules
|
||||
const projectConfig = await prisma.iitProject.findUnique({ where: { id: projectId } });
|
||||
if (!projectConfig) {
|
||||
logger.warn('eQuery review: project not found', { projectId });
|
||||
return { status: 'project_not_found' };
|
||||
}
|
||||
|
||||
const adapter = new RedcapAdapter(projectConfig.redcapUrl, projectConfig.redcapApiToken);
|
||||
const recordData = await adapter.getRecordById(recordId);
|
||||
|
||||
let passed = true;
|
||||
let reviewNote = 'AI 复核：数据已修正，问题已解决。';
|
||||
|
||||
if (recordData && fieldName) {
|
||||
// Re-run rules for the specific field
|
||||
const ruleEngine = await createHardRuleEngine(projectId);
|
||||
const flat = Array.isArray(recordData) ? recordData.reduce((acc: any, row: any) => ({ ...acc, ...row }), {}) : recordData;
|
||||
const qcResult = await ruleEngine.execute(recordId, flat);
|
||||
|
||||
// Check if the specific field still has issues
|
||||
const allIssues = [...qcResult.errors, ...qcResult.warnings];
|
||||
const fieldStillFailing = allIssues.some((issue: any) => issue.field === fieldName);
|
||||
|
||||
if (fieldStillFailing) {
|
||||
passed = false;
|
||||
const failingIssue = allIssues.find((issue: any) => issue.field === fieldName);
|
||||
reviewNote = `AI 复核：字段 ${fieldName} 仍存在问题 — ${failingIssue?.errorMessage || failingIssue?.message || '规则未通过'}。CRC 回复: "${responseText}"`;
|
||||
}
|
||||
}
|
||||
|
||||
// Update eQuery via the service
|
||||
const { getIitEqueryService } = await import('../admin/iit-projects/iitEqueryService.js');
|
||||
const equeryService = getIitEqueryService(prisma);
|
||||
await equeryService.review(equeryId, { passed, reviewNote });
|
||||
|
||||
logger.info('Worker: iit_equery_review completed', {
|
||||
jobId: job.id,
|
||||
equeryId,
|
||||
passed,
|
||||
});
|
||||
return { status: 'success', passed };
|
||||
} catch (error: any) {
|
||||
logger.error('Worker: iit_equery_review failed', { jobId: job.id, equeryId, error: error.message });
|
||||
throw error;
|
||||
}
|
||||
});
|
||||
|
||||
logger.info('IIT Manager: Worker registered - iit_equery_review');
|
||||
|
||||
// =============================================
|
||||
// 3. 注册Worker：处理质控任务 + 双产出（质控日志 + 录入汇总）
|
||||
// =============================================
|
||||
|
||||
279
backend/src/modules/iit-manager/services/DailyQcOrchestrator.ts
Normal file
279
backend/src/modules/iit-manager/services/DailyQcOrchestrator.ts
Normal file
@@ -0,0 +1,279 @@
|
||||
/**
|
||||
* DailyQcOrchestrator - 每日全量质控编排器
|
||||
*
|
||||
* 在 SkillRunner / HardRuleEngine 执行完质控后：
|
||||
* 1. 调用 QcReportService 生成完整报告
|
||||
* 2. 从报告中提取严重/警告问题 → 批量创建 eQuery
|
||||
* 3. 将 SAE / 重大方案偏离写入 iit_critical_events
|
||||
* 4. 与上次报告对比（新增 / 已解决）
|
||||
* 5. 通过企微推送摘要
|
||||
*/
|
||||
|
||||
import { PrismaClient } from '@prisma/client';
|
||||
import { logger } from '../../../common/logging/index.js';
|
||||
import { QcReportService, type QcReport, type ReportIssue } from './QcReportService.js';
|
||||
import { getIitEqueryService, type CreateEqueryInput } from '../../admin/iit-projects/iitEqueryService.js';
|
||||
import { wechatService } from './WechatService.js';
|
||||
|
||||
const prisma = new PrismaClient();
|
||||
|
||||
export interface OrchestratorResult {
|
||||
reportId: string;
|
||||
equeriesCreated: number;
|
||||
criticalEventsArchived: number;
|
||||
newIssues: number;
|
||||
resolvedIssues: number;
|
||||
pushSent: boolean;
|
||||
}
|
||||
|
||||
class DailyQcOrchestratorClass {
|
||||
/**
|
||||
* 主方法：质控后编排全部后续动作
|
||||
*/
|
||||
async orchestrate(projectId: string): Promise<OrchestratorResult> {
|
||||
const startTime = Date.now();
|
||||
logger.info('[DailyQcOrchestrator] Starting orchestration', { projectId });
|
||||
|
||||
// Step 1: Generate report
|
||||
const report = await QcReportService.getReport(projectId, {
|
||||
forceRefresh: true,
|
||||
reportType: 'daily',
|
||||
expirationHours: 24,
|
||||
});
|
||||
|
||||
// Save report to DB and get ID
|
||||
const savedReport = await prisma.iitQcReport.create({
|
||||
data: {
|
||||
projectId,
|
||||
reportType: 'daily',
|
||||
summary: report.summary as any,
|
||||
issues: JSON.parse(JSON.stringify({
|
||||
critical: report.criticalIssues,
|
||||
warning: report.warningIssues,
|
||||
topIssues: report.topIssues,
|
||||
grouped: report.groupedIssues,
|
||||
})),
|
||||
llmReport: report.llmFriendlyXml,
|
||||
},
|
||||
});
|
||||
|
||||
// Step 2: Create eQueries from issues
|
||||
const equeriesCreated = await this.dispatchEqueries(projectId, report, savedReport.id);
|
||||
|
||||
// Step 3: Archive critical events
|
||||
const criticalEventsArchived = await this.archiveCriticalEvents(projectId, report);
|
||||
|
||||
// Step 4: Compare with previous report
|
||||
const { newIssues, resolvedIssues } = await this.compareWithPrevious(projectId, report);
|
||||
|
||||
// Step 5: Push notification
|
||||
const pushSent = await this.pushNotification(projectId, report, equeriesCreated, newIssues, resolvedIssues);
|
||||
|
||||
const duration = Date.now() - startTime;
|
||||
logger.info('[DailyQcOrchestrator] Orchestration completed', {
|
||||
projectId,
|
||||
reportId: savedReport.id,
|
||||
equeriesCreated,
|
||||
criticalEventsArchived,
|
||||
newIssues,
|
||||
resolvedIssues,
|
||||
pushSent,
|
||||
durationMs: duration,
|
||||
});
|
||||
|
||||
return {
|
||||
reportId: savedReport.id,
|
||||
equeriesCreated,
|
||||
criticalEventsArchived,
|
||||
newIssues,
|
||||
resolvedIssues,
|
||||
pushSent,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Step 2: 从报告中的严重/警告问题自动派发 eQuery
|
||||
*/
|
||||
private async dispatchEqueries(projectId: string, report: QcReport, reportId: string): Promise<number> {
|
||||
const issues = [...report.criticalIssues, ...report.warningIssues];
|
||||
if (issues.length === 0) return 0;
|
||||
|
||||
// Deduplicate: skip if there's already an open eQuery for the same record+field
|
||||
const existingEqueries = await prisma.iitEquery.findMany({
|
||||
where: {
|
||||
projectId,
|
||||
status: { in: ['pending', 'responded', 'reviewing', 'reopened'] },
|
||||
},
|
||||
select: { recordId: true, fieldName: true },
|
||||
});
|
||||
const existingKeys = new Set(existingEqueries.map((e: { recordId: string; fieldName: string | null }) => `${e.recordId}:${e.fieldName || ''}`));
|
||||
|
||||
const newEqueries: CreateEqueryInput[] = [];
|
||||
for (const issue of issues) {
|
||||
const key = `${issue.recordId}:${issue.field || ''}`;
|
||||
if (existingKeys.has(key)) continue;
|
||||
existingKeys.add(key);
|
||||
|
||||
newEqueries.push({
|
||||
projectId,
|
||||
recordId: issue.recordId,
|
||||
fieldName: issue.field,
|
||||
reportId,
|
||||
queryText: `[${issue.severity === 'critical' ? '严重' : '警告'}] ${issue.message}`,
|
||||
expectedAction: `请核实受试者 ${issue.recordId} 的 ${issue.field || '相关'} 数据并修正或提供说明`,
|
||||
severity: issue.severity === 'critical' ? 'error' : 'warning',
|
||||
category: issue.ruleName,
|
||||
});
|
||||
}
|
||||
|
||||
if (newEqueries.length === 0) return 0;
|
||||
|
||||
const equeryService = getIitEqueryService(prisma);
|
||||
const result = await equeryService.createBatch(newEqueries);
|
||||
return result.count;
|
||||
}
|
||||
|
||||
/**
|
||||
* Step 3: 将 SAE / 重大方案偏离写入 critical_events
|
||||
*/
|
||||
private async archiveCriticalEvents(projectId: string, report: QcReport): Promise<number> {
|
||||
const criticalIssues = report.criticalIssues.filter((issue) => {
|
||||
const lowerMsg = (issue.message || '').toLowerCase();
|
||||
const lowerRule = (issue.ruleName || '').toLowerCase();
|
||||
return (
|
||||
lowerMsg.includes('sae') ||
|
||||
lowerMsg.includes('严重不良') ||
|
||||
lowerMsg.includes('方案偏离') ||
|
||||
lowerRule.includes('ae_monitoring') ||
|
||||
lowerRule.includes('protocol_deviation')
|
||||
);
|
||||
});
|
||||
|
||||
if (criticalIssues.length === 0) return 0;
|
||||
|
||||
// Deduplicate against existing events
|
||||
const existing = await prisma.iitCriticalEvent.findMany({
|
||||
where: { projectId, status: 'open' },
|
||||
select: { recordId: true, title: true },
|
||||
});
|
||||
const existingKeys = new Set(existing.map((e: { recordId: string; title: string }) => `${e.recordId}:${e.title}`));
|
||||
|
||||
let archived = 0;
|
||||
for (const issue of criticalIssues) {
|
||||
const title = `${issue.ruleName}: ${issue.message}`.substring(0, 200);
|
||||
const key = `${issue.recordId}:${title}`;
|
||||
if (existingKeys.has(key)) continue;
|
||||
|
||||
await prisma.iitCriticalEvent.create({
|
||||
data: {
|
||||
projectId,
|
||||
recordId: issue.recordId,
|
||||
eventType: issue.ruleName?.includes('ae') ? 'SAE' : 'PROTOCOL_DEVIATION',
|
||||
severity: 'critical',
|
||||
title,
|
||||
description: `受试者 ${issue.recordId}: ${issue.message}. 字段: ${issue.field || 'N/A'}`,
|
||||
detectedAt: new Date(issue.detectedAt),
|
||||
detectedBy: 'ai',
|
||||
sourceQcLogId: issue.ruleId,
|
||||
status: 'open',
|
||||
},
|
||||
});
|
||||
archived++;
|
||||
}
|
||||
|
||||
return archived;
|
||||
}
|
||||
|
||||
/**
|
||||
* Step 4: 与上一份报告对比
|
||||
*/
|
||||
private async compareWithPrevious(
|
||||
projectId: string,
|
||||
currentReport: QcReport
|
||||
): Promise<{ newIssues: number; resolvedIssues: number }> {
|
||||
const previousReports = await prisma.iitQcReport.findMany({
|
||||
where: { projectId, reportType: 'daily' },
|
||||
orderBy: { generatedAt: 'desc' },
|
||||
take: 2,
|
||||
select: { issues: true },
|
||||
});
|
||||
|
||||
if (previousReports.length < 2) {
|
||||
return { newIssues: currentReport.criticalIssues.length + currentReport.warningIssues.length, resolvedIssues: 0 };
|
||||
}
|
||||
|
||||
const prevIssues = previousReports[1].issues as any;
|
||||
const prevKeys = new Set<string>();
|
||||
for (const issue of [...(prevIssues?.critical || []), ...(prevIssues?.warning || [])]) {
|
||||
prevKeys.add(`${issue.recordId}:${issue.field || ''}:${issue.ruleName || ''}`);
|
||||
}
|
||||
|
||||
const currentIssues = [...currentReport.criticalIssues, ...currentReport.warningIssues];
|
||||
const currentKeys = new Set<string>();
|
||||
let newIssues = 0;
|
||||
for (const issue of currentIssues) {
|
||||
const key = `${issue.recordId}:${issue.field || ''}:${issue.ruleName || ''}`;
|
||||
currentKeys.add(key);
|
||||
if (!prevKeys.has(key)) newIssues++;
|
||||
}
|
||||
|
||||
let resolvedIssues = 0;
|
||||
for (const key of prevKeys) {
|
||||
if (!currentKeys.has(key)) resolvedIssues++;
|
||||
}
|
||||
|
||||
return { newIssues, resolvedIssues };
|
||||
}
|
||||
|
||||
/**
|
||||
* Step 5: 企微推送摘要
|
||||
*/
|
||||
private async pushNotification(
|
||||
projectId: string,
|
||||
report: QcReport,
|
||||
equeriesCreated: number,
|
||||
newIssues: number,
|
||||
resolvedIssues: number
|
||||
): Promise<boolean> {
|
||||
try {
|
||||
const project = await prisma.iitProject.findUnique({
|
||||
where: { id: projectId },
|
||||
select: { name: true },
|
||||
});
|
||||
|
||||
const { summary } = report;
|
||||
const now = new Date().toLocaleString('zh-CN');
|
||||
|
||||
const markdown = [
|
||||
`# AI 监查日报 - ${project?.name || projectId}`,
|
||||
`> ${now}`,
|
||||
'',
|
||||
`**质控通过率**: ${summary.passRate}%`,
|
||||
`**总受试者**: ${summary.totalRecords} | **已完成**: ${summary.completedRecords}`,
|
||||
`**严重问题**: ${summary.criticalIssues} | **警告**: ${summary.warningIssues}`,
|
||||
'',
|
||||
`📊 **对比上期**: 新增 ${newIssues} 个问题，已解决 ${resolvedIssues} 个`,
|
||||
`📋 **新派发 eQuery**: ${equeriesCreated} 条`,
|
||||
'',
|
||||
summary.criticalIssues > 0
|
||||
? `⚠️ 有 ${summary.criticalIssues} 个严重问题需要关注，请登录平台查看详情`
|
||||
: '✅ 无严重问题',
|
||||
'',
|
||||
'[查看详情](点击进入 CRA 质控平台)',
|
||||
].join('\n');
|
||||
|
||||
const piUserId = process.env.WECHAT_TEST_USER_ID || 'FengZhiBo';
|
||||
await wechatService.sendMarkdownMessage(piUserId, markdown);
|
||||
|
||||
return true;
|
||||
} catch (err) {
|
||||
logger.warn('[DailyQcOrchestrator] Push notification failed (non-fatal)', {
|
||||
projectId,
|
||||
error: String(err),
|
||||
});
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export const dailyQcOrchestrator = new DailyQcOrchestratorClass();
|
||||
Reference in New Issue
Block a user