import { Hono } from "hono"; import { streamSSE } from "hono/streaming"; import { bodyLimit } from "hono/body-limit"; import type { AuthUser } from "../../auth/types.js"; import { canManageFeedback } from "../../auth/types.js"; import { vehicleRepository } from "../vehicles/repository.js"; import { visibleVehicles } from "./model.js"; import { initialPlan, planConversation } from "./planner.js"; import { execute, applyAction } from "./execute.js"; import { executeComposition } from "./composition.js"; import { generatePresentation } from "./generative.js"; import { applyGeneratedAction } from "./generated-action.js"; import { getSession, getNode, addNode, trail, deleteSession, acquire, conversationContext, } from "./session.js"; import { resultScope, followupSuggestions } from "./presentation.js"; import { gapStore } from "./gaps.js"; import { questionLog, questionOutcome, finishQuestionLog } from "./question-log.js"; import { CATEGORY_LABELS, GAP_STATUSES, GROUP_LABELS, type AssistantRequest, type AssistantResult, type AnalysisQuery, type GapReason, type GapStatus, } from "../../../shared/assets/assistant.js"; const router = new Hono<{ Variables: { user: AuthUser } }>(); router.use("*", bodyLimit({ maxSize: 16000 })); router.use("*", async (c, next) => { if (!c.get("user")) return c.json({ error: "请先登录。" }, 401); c.header("Cache-Control", "no-store"); await next(); }); router.get("/capabilities", (c) => { const enabled = !!process.env.DEEPSEEK_API_KEY && process.env.ASSET_AI_EXTERNAL_ENABLED !== "0"; return c.json({ engine: enabled ? "deepseek" : "local", model: enabled ? "deepseek-flash" : null, message: enabled ? "支持自然语言理解、连续追问和多层分析" : "当前使用本地基础问数;配置 DeepSeek 后可开启自然语言分析", }); }); router.get("/gaps", async (c) => { const user = c.get("user"), manage = canManageFeedback(user.roles); if (c.req.query("scope") === "all" && !manage) return c.json({ error: "无权限。" }, 403); try { return c.json({ items: await gapStore.list( c.req.query("scope") === "all" ? null : user.userId, { readOnly: !!user.preview }, ), canManage: manage, }); } catch { return c.json({ error: "需求记录暂不可用,请检查内部存储配置。" }, 503); } }); router.patch("/gaps/:id", async (c) => { if (!canManageFeedback(c.get("user").roles)) return c.json({ error: "无权限。" }, 403); const body = await c.req.json().catch(() => null), id = Number(c.req.param("id")); if ( !Number.isSafeInteger(id) || id <= 0 || !body || !Object.hasOwn(GAP_STATUSES, body.status) ) return c.json({ error: "参数无效。" }, 400); try { await gapStore.update(id, body.status as GapStatus); return c.json({ ok: true }); } catch { return c.json({ error: "保存失败,请重试。" }, 503); } }); router.post("/feedback", async (c) => { const body = await c.req.json().catch(() => null); if ( !body || typeof body.sessionId !== "string" || typeof body.resultId !== "string" || (body.subject !== null && typeof body.subject !== "string") ) return c.json({ error: "参数无效。" }, 400); try { const session = getSession(c.get("user"), body.subject, body.sessionId), node = getNode(session, body.resultId)!; if (!node.flagged) { const logId = await questionLog.start(c.get("user"), { message: node.question || node.demand, subject: body.subject, sessionId: session.id, baseResultId: node.id }, "feedback"); const saved = await finishQuestionLog(logId, { outcome: "feedback", reason: "unsatisfied", resultId: node.id }); if (!saved) return c.json({ error: "反馈已接收,处理状态暂未保存,请稍后重试。" }, 503); await gapStore.record(c.get("user").userId, node.demand, "unsatisfied"); node.flagged = true; } return c.json({ ok: true }); } catch { return c.json({ error: "记录失败或会话已过期,请重试。" }, 400); } }); router.delete("/session/:id", async (c) => { const subject = c.req.query("subject") || null; try { deleteSession(c.get("user"), subject, c.req.param("id")); return c.json({ ok: true }); } catch { return c.json({ error: "会话已过期。" }, 404); } }); router.post("/export", async (c) => { const body = await c.req.json().catch(() => null); if ( !body || typeof body.sessionId !== "string" || typeof body.resultId !== "string" || (body.subject !== null && typeof body.subject !== "string") ) return c.json({ error: "参数无效。" }, 400); const release = acquire(c.get("user").userId); if (!release) return c.json({ error: "操作过于频繁。" }, 429); try { const session = getSession(c.get("user"), body.subject, body.sessionId), node = getNode(session, body.resultId)!; if (node.plan.kind === "clarify") return c.json({ error: "暂无可导出结果。" }, 400); const vehicles = visibleVehicles( await vehicleRepository.getVehicles(), c.get("user"), body.subject, ); const result = await execute(vehicles, node.plan, c.get("user")); const records = [ [ "车牌", "车型", "型号", "区域", "城市", "部门", "运营状态", "异动类型", "预计完成时间", "日期", "业务类型", ], ...result.rows.map((r) => [ r.plate, r.type, r.model, r.region, r.city, r.department, r.status, r.abnormalType || "", r.targetCompletionTime || "", r.date || "", r.event || "", ]), ]; const csv = records .map((row) => row .map( (v) => '"' + String(v) .replace(/^[=+@\-\t\r\n]/, "'$&") .replaceAll('"', '""') + '"', ) .join(","), ) .join("\r\n"); c.header("Content-Type", "text/csv; charset=utf-8"); c.header( "Content-Disposition", 'attachment; filename="asset-analysis.csv"', ); return c.body("\uFEFF" + csv); } catch { return c.json({ error: "导出失败或权限已变化,请重新查询。" }, 400); } finally { release(); } }); router.post("/ask", async (c) => { const user = c.get("user"); const raw = await c.req.json().catch(() => null); if ( !raw || typeof raw !== "object" || Array.isArray(raw) || Object.keys(raw).some( (k) => !["message", "subject", "sessionId", "baseResultId", "action"].includes( k, ), ) || (raw.subject != null && (typeof raw.subject !== "string" || raw.subject.length > 150)) || (raw.sessionId != null && (typeof raw.sessionId !== "string" || raw.sessionId.length > 64)) || (raw.baseResultId != null && (typeof raw.baseResultId !== "string" || !raw.sessionId)) || !!raw.action === !!raw.message || (raw.message != null && (typeof raw.message !== "string" || !raw.message.trim() || raw.message.length > 2000)) || (raw.action && (typeof raw.action !== "object" || !raw.baseResultId)) ) return c.json({ error: "请输入有效问题或选择分析操作。" }, 400); const input: AssistantRequest = { ...raw, subject: raw.subject || null }; let logId: string; try { logId = await questionLog.start(user, input); } catch { return c.json({ error: "提问记录暂不可用,本次查询未执行,请稍后重试。" }, 503); } const release = acquire(user.userId); if (!release) { await finishQuestionLog(logId, { outcome: "rejected", reason: "busy" }); return c.json({ error: "上一条查询仍在执行,请稍后重试。" }, 429); } let session, base; try { session = getSession(user, input.subject, input.sessionId); base = getNode(session, input.baseResultId); } catch { release(); await finishQuestionLog(logId, { outcome: "rejected", reason: "session_expired" }); return c.json({ error: "会话已过期或范围已变化,请新建对话。" }, 409); } const currentSession = session, baseNode = base; return streamSSE(c, async (stream) => { const controller = new AbortController(); stream.onAbort(() => controller.abort()); const emit = (event: object) => stream.writeSSE({ data: JSON.stringify(event) }); let logFinished = false; let demand = "查询执行", reason: GapReason | undefined; try { await emit({ type: "step", label: input.action ? "延续当前分析条件" : "理解问题与对话上下文", }); const vehicles = visibleVehicles( await vehicleRepository.getVehicles(), user, input.subject, ); if (controller.signal.aborted) return; const context = conversationContext(currentSession, baseNode); const previous = context.previous; let plan = previous || initialPlan(); let queries: AnalysisQuery[] | undefined; let warning: string | undefined; let reply: string | undefined; let answerKind: "answer" | "clarify" | undefined; let suggestions: string[] | undefined; let unanswered: AssistantResult["unanswered"]; let correction: string | undefined; let engine: "deepseek" | "local" = "local"; let parent = baseNode; if (input.action) { if (!previous) throw new Error("请先完成一次统计。"); const currentData = await execute(vehicles, previous, user); if (input.action.type === "explore") { const exploration = applyGeneratedAction(previous, input.action, currentData, baseNode?.presentation); plan = exploration.plan; demand = exploration.label; } else { plan = applyAction(previous, input.action, currentData); demand = `${CATEGORY_LABELS[previous.category]} ${GROUP_LABELS[previous.groupBy]} ${input.action.type}`; } } else { demand = input .message!.trim() .replace(/sk-[a-z0-9]{8,}/gi, "[凭据已隐藏]") .slice(0, 600); const decision = await planConversation({ message: input.message!, previous, history: context.history, resultContext: context.resultContext, candidates: context.candidates, vehicles, signal: controller.signal, }); plan = decision.plan; queries = decision.queries; reason = decision.reason; warning = decision.warning; reply = decision.reply; answerKind = decision.answerKind; suggestions = decision.suggestions; unanswered = decision.unanswered; correction = decision.correction; engine = decision.engine; if (decision.baseResultId) parent = getNode(currentSession, decision.baseResultId); } if (controller.signal.aborted) return; await emit({ type: "step", label: plan.kind === "clarify" ? "组织回答与下一步建议" : "查询数据并计算统计结果", }); // Reserve enough session slots before executing or publishing any part of an answer. if (currentSession.nodes.size + 1 + (queries?.length || 0) > 100) throw new Error("本次对话分析步骤已满,请新建对话。"); let result: AssistantResult = queries?.length ? await executeComposition(vehicles, queries, input.subject, controller.signal, user) : plan.kind === "clarify" ? { plan, title: answerKind === "answer" ? "资产问数助手" : reason ? "这项分析还需要补充能力" : "继续完善分析", summary: reply || plan.clarification || "请补充想看的指标与范围。", source: "资产问数助手", fetchedAt: new Date().toISOString(), groups: [], rows: [], metrics: [], answerKind: answerKind === "answer" ? "help" : "clarification", } : await execute(vehicles, plan, user); if (controller.signal.aborted) return; plan = result.plan; if (result.unanswered?.length) { unanswered = [...(unanswered || []), ...result.unanswered]; reason ||= "generation"; } result.engine = engine; if (plan.kind === "clarify" && suggestions?.length) result.suggestions = suggestions; result.scope = resultScope(plan, input.subject); result.suggestions = followupSuggestions(result); if (plan.kind === "clarify" && !result.sections) result.outcome = reason === "unsupported" ? "unsupported" : answerKind === "answer" ? "help" : "clarification"; if (unanswered?.length) { result.unanswered = unanswered; if (plan.kind !== "clarify") result.outcome = "partial"; } if (correction) result.correction = { message: correction, resultId: parent ? trail(currentSession, parent).at(-1)?.resultId : undefined }; if (warning) result.warning = warning; if (plan.kind !== "clarify") { const includes = Object.entries(plan.filters).map( ([key, values]) => `${GROUP_LABELS[key as keyof typeof GROUP_LABELS]}:${values.join("、")}`, ); const excludes = Object.entries(plan.excludeFilters || {}).map( ([key, values]) => `排除${GROUP_LABELS[key as keyof typeof GROUP_LABELS]}:${values.join("、")}`, ); result.interpretation = [ plan.kind === "flow" ? "交退换车" : CATEGORY_LABELS[plan.category], ...includes, ...excludes, ...(plan.exception ? [ { overdue: "已超预计完成时间", dueSoon: "未来 7 天到期", missingDueDate: "未填写预计完成时间", }[plan.exception], ] : []), ...(plan.vehicleIds ? [`指定 ${plan.vehicleIds.length} 台车辆`] : []), ...(plan.kind === "flow" ? [`${plan.start} 至 ${plan.end}`] : []), ].join(" · "); if (parent) { const baseLabel = trail(currentSession, parent).at(-1)?.label; if (baseLabel) result.contextNote = `承接「${baseLabel}」${parent.id !== baseNode?.id ? "的历史分析" : ""}`; } } const label = plan.kind === "clarify" ? "对话补充" : `${CATEGORY_LABELS[plan.category]} · ${Object.values(plan.filters).flat().join(" / ") || (plan.kind === "flow" ? `${plan.start}—${plan.end}` : GROUP_LABELS[plan.groupBy])}`; if (parent) { const sourcePlan = conversationContext(currentSession, parent).previous; if (sourcePlan) { const scope = [ CATEGORY_LABELS[sourcePlan.category], ...Object.values(sourcePlan.filters).flat(), ...Object.values(sourcePlan.excludeFilters || {}) .flat() .map((v) => `排除${v}`), ...(sourcePlan.kind === "flow" ? [`交退换 ${sourcePlan.start}—${sourcePlan.end}`] : []), ...(sourcePlan.exception ? [ { overdue: "已超期", dueSoon: "未来 7 天到期", missingDueDate: "缺预计完成时间", }[sourcePlan.exception], ] : []), ].join(" · "); demand = `${demand.slice(0, 400)}(承接:${scope})`.slice(0, 600); } } if (result.plan.kind !== "clarify" || result.sections?.some(section => section.result)) { await emit({ type: "step", label: "根据结果设计展示与下钻路径" }); await generatePresentation(result, input.message || `${baseNode?.question || "当前分析"};${demand}`, controller.signal); if (result.presentationStatus === "generated" || result.sections?.some(section => section.result?.presentationStatus === "generated")) { engine = "deepseek"; result.engine = engine; } } if (controller.signal.aborted) return; for (const item of result.sections ? result.sections.flatMap(section => section.result ? [section.result] : []) : [result]) { if (!item.presentation && item.presentationWarning) item.warning = [item.warning, item.presentationWarning].filter(Boolean).join(" "); } const missingPresentation = result.sections ? result.sections.filter(section => section.result?.presentationStatus === "fallback").map(section => ({ question: `${section.question}(定制展示与下钻)`, reason: section.result!.presentationWarning || "展示方案未能生成,已保留统计结果。" })) : result.presentationStatus === "fallback" ? [{ question: `${input.message || demand}(定制展示与下钻)`, reason: result.presentationWarning || "展示方案未能生成,已保留统计结果。" }] : []; if (missingPresentation.length) { unanswered = [...(unanswered || []), ...missingPresentation]; result.unanswered = unanswered; result.outcome = "partial"; reason ||= "generation"; } const node = addNode(currentSession, { plan, parent: parent?.id, label, demand, reason, question: input.message || demand, answer: result.summary, container: !!result.sections, presentation: result.presentation, resultContext: plan.kind === "clarify" ? undefined : { groups: result.groups.slice(0, 100), inventoryBreakdown: result.inventoryBreakdown, metrics: result.metrics, insights: result.insights, summary: result.summary, analyses: parent?.resultContext?.analyses, }, }); result = { ...result, sessionId: currentSession.id, resultId: node.id, breadcrumbs: trail(currentSession, node), }; if (!result.sections && node.resultContext?.analyses) { node.resultContext.analyses = node.resultContext.analyses.map(analysis => analysis.id === parent?.id ? { id: node.id, title: result.title, plan, summary: result.summary, metrics: result.metrics, } : analysis); result.analyses = node.resultContext.analyses; } if (result.sections) { for (const section of result.sections) { if (!section.result) continue; const child = section.result; const childNode = addNode(currentSession, { plan: child.plan, parent: node.id, label: section.title, demand: `${section.question}(原问题:${demand})`.slice(0, 600), question: section.question, answer: child.summary, presentation: child.presentation, resultContext: { groups: child.groups.slice(0, 100), metrics: child.metrics, insights: child.insights, summary: child.summary, inventoryBreakdown: child.inventoryBreakdown, }, }); child.engine = engine; child.sessionId = currentSession.id; child.resultId = childNode.id; child.breadcrumbs = trail(currentSession, childNode); } result.analyses = result.sections.flatMap(section => section.result ? [{ id: section.result.resultId!, title: section.title, plan: section.result.plan, summary: section.result.summary, metrics: section.result.metrics, }] : []); if (node.resultContext) node.resultContext.analyses = result.analyses; for (const section of result.sections) { if (section.result?.resultId) { const childNode = getNode(currentSession, section.result.resultId)!; if (childNode.resultContext) childNode.resultContext.analyses = result.analyses; } } } const gapReason = reason || (unanswered?.length ? "unsupported" : undefined); if (gapReason) { if (unanswered?.length) demand = `${unanswered.map(item => `${item.question}:${item.reason}`).join(";")}(原问题:${demand})`.slice(0, 600); let recorded = false; try { if (!user.preview) { await gapStore.record(user.userId, demand, gapReason); recorded = true; } } catch { /* Report failure explicitly, never pretend it was persisted. */ } result.gap = { recorded, reason: gapReason, demand }; } await emit({ type: "step", label: "分析完成", }); logFinished = true; const saved = await finishQuestionLog(logId, { outcome: questionOutcome(result, gapReason), reason: gapReason, sessionId: currentSession.id, result }); if (!saved) result.warning = [result.warning, "提问已记录,处理结果暂未保存。"].filter(Boolean).join(" "); await emit({ type: "result", result }); } catch { if (!logFinished) { logFinished = true; await finishQuestionLog(logId, { outcome: controller.signal.aborted ? "cancelled" : "failed", reason: controller.signal.aborted ? "client_aborted" : "generation", sessionId: currentSession.id }); } if (!controller.signal.aborted) { let recorded = false; try { if (!user.preview) { await gapStore.record(user.userId, demand, "generation"); recorded = true; } } catch { /* No raw error or query payload in logs. */ } await emit({ type: "error", message: `查询未完成,请检查条件后重试。${recorded ? "已记录本次未完成的需求。" : "需求记录暂未保存。"}`, }); } } finally { if (!logFinished) await finishQuestionLog(logId, { outcome: controller.signal.aborted ? "cancelled" : "failed", reason: controller.signal.aborted ? "client_aborted" : "unfinished", sessionId: currentSession.id }); release(); } }); }); export default router;