diff --git a/apps/api/src/modules/import/import.service.ts b/apps/api/src/modules/import/import.service.ts index 884b84d..1a0b512 100644 --- a/apps/api/src/modules/import/import.service.ts +++ b/apps/api/src/modules/import/import.service.ts @@ -21,14 +21,14 @@ // site + wxr 快照 + 三语义版本（adapter/mapping/sanitizer——任一 bump 即新 job，对齐 // spec §7.5 “版本变化可显式触发重新标准化”；parser 是诊断信息不进键）。 // -// 管道注入：Task 6 起默认 = **编排管道**（四家写入器汇流，orchestratedPipeline）——真 -// 事件流（runWxrAdapter 读 wxr 快照副本，mapConfig.locale 接 site.default_locale——前置 -// 事实 8）→ content/term/media/route 严格串行分派 → finish 硬序（media 先于 route）→ -// counts 按键**求和**聚合（Task 6 钉死，见 ImportPipelineOutcome.counts 契约）。注入 -// pipeline 钩子供集成测试观察中间态/失败路径（注入时缺省事件流为空——Task 1 语义保持）。 +// 管道注入：Task 6 起默认 = **编排管道**（四家写入器汇流）——orchestratedPipeline 自 +// Task 8 起抽取于 orchestrator.ts（纯函数 over ctx，零行为变更）。本文件负责：真事件流 +// 构造（runWxrAdapter 读 wxr 快照副本，mapConfig.locale 接 site.default_locale——前置 +// 事实 8）→ Outcome 汇合（mergeOutcomes）→ 报告固化。注入 pipeline 钩子供集成测试观察 +// 中间态/失败路径（注入时缺省事件流为空——Task 1 语义保持）。 // // stale 回收（Task 6 义务④）：每次建 job 前把同 site 超过 24h 仍 queued/running 的 job -// 置 stale（崩溃/中断残留终态化）。心跳列决策留 ADR-017——现阶段无心跳列，以 +// 置 stale（崩溃/中断残留终态化）。心跳列决策见 ADR-017——现阶段无心跳列，以 // created_at 近似（job 为单进程就地推进，24h 远超任何真实执行时长）。 import { createReadStream } from 'node:fs' import { mkdir, rename, rm, writeFile } from 'node:fs/promises' @@ -48,16 +48,14 @@ import { import { importJobs, sites, sourceConnections, sourceSnapshots } from '../../db/schema.js' import type * as schema from '../../db/schema.js' import { isUniqueViolation } from '../source/register.service.js' -import { createContentEventHandler } from './content-writer.js' -import { createTermEventHandler } from './term-writer.js' -import { createMediaEventHandlerFromContext } from './media-writer.js' +import { mergeOutcomes, orchestratedPipeline } from './orchestrator.js' import { DEFAULT_NEWS_PATH, DEFAULT_PRODUCTS_PATH, - createRouteEventHandler, isSafePath, normalizeListPath, } from './route-writer.js' +import { renderImportReportHtml } from './report-html.js' /** 本文件位于 /apps/api/src/modules/import/ → 上五级是仓库根 */ const REPO_ROOT = resolve(dirname(fileURLToPath(import.meta.url)), '../../../../..') @@ -519,16 +517,27 @@ export async function createImportJob( if (failed !== null) { finalStatus = 'failed' } else { + // 终态判定在**全量** issues 长度上做（截断是报告固化层语义，不吞终态事实） finalStatus = merged.issues.length > 0 ? 'succeeded_with_issues' : 'succeeded' } + // 报告 issues 封顶（Task 8 兑现 Task 6 推迟项，对齐 Plan 2 ROW_ERRORS_CAP 纪律）： + // flat 清单与 issueGroups 各组按同一常量截断——对抗性输入（单条 item 上万行级问题） + // 不使报告体积失去上界；counts/sanitize 统计不受影响始终完整。分组是同一集合的 + // 视图，各组溢出 ≤ flat 溢出（分划性质），不单独入账。 + const cappedIssues = merged.issues.slice(0, REPORT_ISSUES_CAP) + const issuesTruncated = merged.issues.length - cappedIssues.length + const cappedGroups = capIssueGroups(merged.issueGroups) + // 阶段 6：报告先行落盘（temp+rename 原子替换），终态落行在后——行内 report_path // 出现即文件存在；写盘失败 → job failed（REPORT_WRITE_FAILED、report_path 不落， - // 不留“succeeded 但报告失踪”或悬挂路径） + // 不留“succeeded 但报告失踪”或悬挂路径）。import.json 为权威产物（机器消费）， + // import.html 为同数据的渲染视图（Task 8 兑现 Task 6 推迟项）——同 try 同失败口径。 const reportRoot = resolve(input.reportRoot ?? join(REPO_ROOT, 'artifacts', 'import')) const reportDir = join(reportRoot, jobId, 'reports') const reportPath = join(reportDir, 'import.json') - const report = { + const htmlPath = join(reportDir, 'import.html') + const report: ImportReport = { jobId, generatedAt: new Date().toISOString(), status: finalStatus, @@ -546,8 +555,9 @@ export async function createImportJob( // 主列表路径（前置事实 9：CLI 参数来源记入报告——route-writer 实际消费的规范化值） paths: { productsPath, newsPath }, counts: merged.counts, - issues: merged.issues, - ...(merged.issueGroups === undefined ? {} : { issueGroups: merged.issueGroups }), + issues: cappedIssues, + issuesTruncated, + ...(cappedGroups === undefined ? {} : { issueGroups: cappedGroups }), ...(merged.sanitize === undefined ? {} : { sanitize: merged.sanitize }), // 适配器分段计时（仅真流编排——Plan 2 报告纪律的 parse/sanitize 分段在导入侧续录） ...(adapterTimings === null ? {} : { adapterTiming: { parseMs: adapterTimings.parseMs, sanitizeMs: adapterTimings.sanitizeMs } }), @@ -557,14 +567,8 @@ export async function createImportJob( let reportWritten = true try { await mkdir(reportDir, { recursive: true }) - const tmpPath = `${reportPath}.tmp` - try { - await writeFile(tmpPath, `${JSON.stringify(report, null, 2)}\n`, 'utf8') - await rename(tmpPath, reportPath) - } catch (err) { - await rm(tmpPath, { force: true }) - throw err - } + await writeAtomically(reportPath, `${JSON.stringify(report, null, 2)}\n`) + await writeAtomically(htmlPath, `${renderImportReportHtml(report)}\n`) } catch (err) { reportWritten = false failed = { @@ -630,22 +634,95 @@ async function reuseExisting( } /** 报告里的快照引用瘦身版：不含 metadata（media_bundle 清单可达数千条，报告只留身份锚点） */ -function slimRef(ref: SnapshotRef): { +export interface SnapshotRefSlim { id: string kind: string sha256: string byteSize: number storagePath: string -} { +} + +function slimRef(ref: SnapshotRef): SnapshotRefSlim { return { id: ref.id, kind: ref.kind, sha256: ref.sha256, byteSize: ref.byteSize, storagePath: ref.storagePath } } +/** + * 报告 issues 固化上限（条）：flat 清单与 issueGroups 各组按此截断，超出只计数 + * （issuesTruncated——见 {@link ImportReport.issuesTruncated}）。Plan 2 Task 6 + * ROW_ERRORS_CAP 纪律的导入侧镜像：对抗性输入下单条 item 可产出上万行级问题， + * 固化全部会使报告体积失去上界；统计维度（counts/sanitize）不受影响始终完整。 + */ +const REPORT_ISSUES_CAP = 10_000 + +/** issueGroups 各组按 {@link REPORT_ISSUES_CAP} 截断（组溢出不单独入账——分划性质下 ≤ flat 溢出） */ +function capIssueGroups( + groups: Partial> | undefined, +): Partial> | undefined { + if (groups === undefined) return undefined + const capped: Partial> = {} + for (const [key, value] of Object.entries(groups) as Array<[ImportIssueGroup, string[] | undefined]>) { + if (value === undefined) continue + capped[key] = value.slice(0, REPORT_ISSUES_CAP) + } + return capped +} + +/** + * 导入报告对象（import.json 的固化形状——renderImportReportHtml 的输入契约）。 + * issues/issueGroups 已按 {@link REPORT_ISSUES_CAP} 截断（issuesTruncated 记 flat + * 溢出条数）；对象层保留**原文**（未转义——JSON 归档/机器消费的语义层），HTML + * 渲染层（report-html.ts）负责转义。 + */ +export interface ImportReport { + jobId: string + generatedAt: string + status: ImportJobStatus + siteKey: string + siteId: string + connectionId: string + idempotencyKey: string + attempt: number + snapshots: { + wxr: SnapshotRefSlim + media_bundle: SnapshotRefSlim | null + manual_url_list: SnapshotRefSlim | null + } + versions: ImportVersionBlock + /** 主列表路径（前置事实 9：CLI 参数来源——route-writer 实际消费的规范化值） */ + paths: { productsPath: string; newsPath: string } + counts: Record + /** 行级问题清单（封顶 {@link REPORT_ISSUES_CAP} 条——超出只计数） */ + issues: string[] + /** 超出 {@link REPORT_ISSUES_CAP} 被丢弃固化的条数（只计数；issueGroups 各组同限截断不另计） */ + issuesTruncated: number + /** 可选：按来源分组的 issues 视图（各组同限截断） */ + issueGroups?: Partial> + /** 可选：清洗统计（row 事件 sanitizeDiffs/changed 的聚合——编排管道产出） */ + sanitize?: ImportSanitizeStats + /** 可选：适配器分段计时（仅真流编排） */ + adapterTiming?: { parseMs: number; sanitizeMs: number } + error: { code: ImportErrorCode; summary: string } | null + durationMs: number +} + +/** temp+rename 原子写（失败清走 tmp 不留半成品——报告落盘纪律的共用形状） */ +async function writeAtomically(path: string, content: string): Promise { + const tmpPath = `${path}.tmp` + try { + await writeFile(tmpPath, content, 'utf8') + await rename(tmpPath, path) + } catch (err) { + await rm(tmpPath, { force: true }) + throw err + } +} + /** 缺省空事件流（input.events 缺席时的 ctx.events 占位；每次调用产新迭代器） */ async function* emptyEvents(): AsyncGenerator {} -// ---- 编排管道（Task 6 汇流——Task 5 审查移交的契约底本在此兑现） ---- +// ---- 事件流构造与 stale 回收（编排本体见 orchestrator.ts） ---- -/** stale 回收阈值：queued/running 滞留超过此时长即宣判（毫秒） */ +/** stale 回收阈值：queued/running 滞留超过此时长即宣判（毫秒）——决策与理由见 ADR-017 */ const STALE_AFTER_MS = 24 * 60 * 60 * 1000 /** wxr 快照副本绝对路径（metadata.storageRoot + storagePath——登记期固化的存储根事实） */ @@ -675,175 +752,3 @@ async function* realEventStream( }) } -/** counts 按键求和（Task 6 钉死的聚合策略——就地累加 target） */ -function sumCounts(target: Record, source: Record): void { - for (const [key, value] of Object.entries(source)) { - target[key] = (target[key] ?? 0) + value - } -} - -/** Outcome 汇合（ImportPipelineOutcome JSDoc 契约）：counts 求和、issues/issueGroups 串联、sanitize 相加 */ -function mergeOutcomes(acc: ImportPipelineOutcome, ret: ImportPipelineOutcome): ImportPipelineOutcome { - const counts: Record = { ...acc.counts } - sumCounts(counts, ret.counts) - const issueGroups: Partial> = {} - let hasGroups = false - for (const key of Object.keys(acc.issueGroups ?? {}) as ImportIssueGroup[]) { - issueGroups[key] = [...(acc.issueGroups![key] ?? [])] - hasGroups = true - } - for (const key of Object.keys(ret.issueGroups ?? {}) as ImportIssueGroup[]) { - issueGroups[key] = [...(issueGroups[key] ?? []), ...(ret.issueGroups![key] ?? [])] - hasGroups = true - } - const sanitize: ImportSanitizeStats | undefined = - acc.sanitize === undefined && ret.sanitize === undefined - ? undefined - : { - changedRows: (acc.sanitize?.changedRows ?? 0) + (ret.sanitize?.changedRows ?? 0), - byDiffType: {}, - } - if (sanitize !== undefined) { - for (const src of [acc.sanitize?.byDiffType, ret.sanitize?.byDiffType]) { - if (src !== undefined) sumCounts(sanitize.byDiffType, src) - } - } - return { - counts, - issues: [...acc.issues, ...ret.issues], - ...(hasGroups ? { issueGroups } : {}), - ...(sanitize === undefined ? {} : { sanitize }), - } -} - -/** - * 编排管道（Task 6 汇流契约底本——Task 5 审查移交钉死）： - * - 事件期**严格串行 await** 分派 content → term → media → route：三家（term/media/ - * route）的 entry 锚点解析依赖 content 对**同事件**的落库产物（source_records. - * content_entry_id / entry_locales），禁 Promise.all——并行下锚点查询可能抢在 content - * 写入前发生； - * - finish 唯一硬序：term/content（同步、无库依赖）任意 → `await media.finish()` 先于 - * `await route.finish()`（attachment 301 反查与 asset 全表都读 media_assets——资产须已齐）； - * - 分派规则（Task 3 审查移交⑥）：wp:attachment 行的媒体域归 media-writer - * （media_assets 唯一写入方），content-writer 对其只留 source_records record-only 足迹 - * （前置事实 11）——两家各写各域，单一 owner； - * - counts 按键求和聚合（安全网——四家键前缀互斥）；清洗统计与 row 事件行级问题在 - * 分派循环收集（RowResult 契约：issues 逐条收录为行级错误）；unsupported/error 事件 - * 的 reason/issue 同归 adapter 组（审查 Fix 4——不静默丢弃诊断信息）； - * - **失败报告部分冲刷**（审查 Fix 1）：分派/finish 中途抛错时，把可安全取得的部分产出 - * 冲入 ctx.report 累加器后重抛——term/content 的 finish 是纯返回（无库依赖，就地补算）； - * media/route 的 finish 依赖库与先序产物，失败路径**不调用**（其 issueGroups 为空数组——未到该 finish 阶段，部分数据不可安全取得）； - * 进度以 orchestrator 键 eventsDispatched 落 counts（仅失败路径出现——成功路径不占键）。 - */ -function orchestratedPipeline(paths: { productsPath: string; newsPath: string }): ImportPipeline { - return async (ctx) => { - const content = createContentEventHandler({ - db: ctx.db, - siteId: ctx.siteId, - connectionId: ctx.connectionId, - snapshotId: ctx.wxrSnapshot.id, - jobId: ctx.jobId, - defaultLocale: ctx.defaultLocale, - versions: ctx.versions, - }) - const term = createTermEventHandler({ - db: ctx.db, - siteId: ctx.siteId, - connectionId: ctx.connectionId, - }) - const media = await createMediaEventHandlerFromContext(ctx) - const route = createRouteEventHandler({ - db: ctx.db, - siteId: ctx.siteId, - connectionId: ctx.connectionId, - snapshotId: ctx.wxrSnapshot.id, - defaultLocale: ctx.defaultLocale, - productsPath: paths.productsPath, - newsPath: paths.newsPath, - }) - - const sanitize: ImportSanitizeStats = { changedRows: 0, byDiffType: {} } - const adapterIssues: string[] = [] - let eventsDispatched = 0 - // 已算成的纯返回 finish（media/route finish 抛错时直接复用，不重调） - let finished: { content: ImportPipelineOutcome; term: ImportPipelineOutcome } | null = null - try { - for await (const ev of ctx.events) { - if (ev.kind === 'row') { - if (ev.changed) sanitize.changedRows++ - for (const diff of ev.sanitizeDiffs) { - sanitize.byDiffType[diff.type] = (sanitize.byDiffType[diff.type] ?? 0) + 1 - } - for (const issue of ev.issues) { - adapterIssues.push(`行级问题（${ev.row.sourceType}/${ev.row.sourceId}）：${issue}`) - } - } else if (ev.kind === 'unsupported') { - adapterIssues.push(`不支持对象（${ev.sourceType}/${ev.postId ?? '?'}, ${ev.rawLocation}）：${ev.reason}`) - } else if (ev.kind === 'error') { - adapterIssues.push(`适配器错误（${ev.sourceType}/${ev.postId ?? '?'}, ${ev.rawLocation}）：${ev.issue}`) - } - await content.handleEvent(ev) - await term.handleEvent(ev) - await media.handleEvent(ev) - await route.handleEvent(ev) - eventsDispatched++ - } - const t = term.finish() - const c = content.finish() - finished = { content: c, term: t } - const m = await media.finish() - const r = await route.finish() - - const counts: Record = {} - sumCounts(counts, c.counts) - sumCounts(counts, t.counts) - sumCounts(counts, m.counts) - sumCounts(counts, r.counts) - return { - counts, - issues: [...c.issues, ...t.issues, ...m.issues, ...r.issues, ...adapterIssues], - issueGroups: { - content: c.issues, - term: t.issues, - media: m.issues, - route: r.issues, - adapter: adapterIssues, - orchestrator: [], - }, - sanitize, - } - } catch (err) { - // 失败报告部分冲刷（审查 Fix 1）：重抛前把可审计的部分产出落入累加器—— - // createImportJob 对抛错管道取 ctx.report 现值进报告，失败诊断不再只剩 error 块。 - const c = finished?.content ?? trySyncFinish(content) - const t = finished?.term ?? trySyncFinish(term) - const partial: Record = { eventsDispatched } - if (c !== null) sumCounts(partial, c.counts) - if (t !== null) sumCounts(partial, t.counts) - sumCounts(ctx.report.counts, partial) - const partialIssues = [...(c?.issues ?? []), ...(t?.issues ?? []), ...adapterIssues] - ctx.report.issues.push(...partialIssues) - ctx.report.issueGroups = { - content: c?.issues ?? [], - term: t?.issues ?? [], - media: [], - route: [], - adapter: adapterIssues, - orchestrator: [], - } - ctx.report.sanitize = sanitize - throw err - } - } -} - -/** 失败路径的安全 finish：content/term 的 finish 是纯返回（无库依赖）——已算成则由调用方 - * 复用，此处只对未算成的就地补算；finish 本身抛错则放弃该部分（不掩盖原错误） */ -function trySyncFinish(handler: { finish(): ImportPipelineOutcome }): ImportPipelineOutcome | null { - try { - return handler.finish() - } catch { - return null - } -} - diff --git a/apps/api/src/modules/import/orchestrator.ts b/apps/api/src/modules/import/orchestrator.ts new file mode 100644 index 0000000..07ecb1c --- /dev/null +++ b/apps/api/src/modules/import/orchestrator.ts @@ -0,0 +1,196 @@ +// 编排管道（Plan 3 Task 6 汇流，Task 8 自 import.service.ts 抽取——纯函数 over ctx， +// 零行为变更，既有测试全绿即证）：四家写入器（content/term/media/route）汇流为缺省 +// ImportPipeline，外加 Outcome 汇合（mergeOutcomes）与 counts 求和（sumCounts）。 +// +// 本文件与 import.service.ts 同纪律保持 **零装饰器**：被 CLI（scripts/sf.ts 经 tsx） +// 与集成测试直接 import；契约类型（ImportPipeline/ImportPipelineOutcome 等）单源于 +// import.service.ts（type-only import，无运行时环）。 +// +// 编排契约（Task 5 审查移交、Task 6 钉死——细节 JSDoc 见各函数）： +// - 事件期**严格串行 await** 分派 content → term → media → route（锚点依赖推导）； +// - finish 唯一硬序：media 先于 route；term/content 同步任意； +// - wp:attachment 行媒体域归 media-writer（单一 owner）； +// - counts 按键求和聚合；清洗统计与 adapter 行级问题在分派循环收集； +// - 失败报告部分冲刷（term/content 纯 finish 补算 + eventsDispatched 进度计数）。 +import type { + ImportIssueGroup, + ImportPipeline, + ImportPipelineOutcome, + ImportSanitizeStats, +} from './import.service.js' +import { createContentEventHandler } from './content-writer.js' +import { createTermEventHandler } from './term-writer.js' +import { createMediaEventHandlerFromContext } from './media-writer.js' +import { createRouteEventHandler } from './route-writer.js' + +/** counts 按键求和（Task 6 钉死的聚合策略——就地累加 target） */ +export function sumCounts(target: Record, source: Record): void { + for (const [key, value] of Object.entries(source)) { + target[key] = (target[key] ?? 0) + value + } +} + +/** Outcome 汇合（ImportPipelineOutcome JSDoc 契约）：counts 求和、issues/issueGroups 串联、sanitize 相加 */ +export function mergeOutcomes(acc: ImportPipelineOutcome, ret: ImportPipelineOutcome): ImportPipelineOutcome { + const counts: Record = { ...acc.counts } + sumCounts(counts, ret.counts) + const issueGroups: Partial> = {} + let hasGroups = false + for (const key of Object.keys(acc.issueGroups ?? {}) as ImportIssueGroup[]) { + issueGroups[key] = [...(acc.issueGroups![key] ?? [])] + hasGroups = true + } + for (const key of Object.keys(ret.issueGroups ?? {}) as ImportIssueGroup[]) { + issueGroups[key] = [...(issueGroups[key] ?? []), ...(ret.issueGroups![key] ?? [])] + hasGroups = true + } + const sanitize: ImportSanitizeStats | undefined = + acc.sanitize === undefined && ret.sanitize === undefined + ? undefined + : { + changedRows: (acc.sanitize?.changedRows ?? 0) + (ret.sanitize?.changedRows ?? 0), + byDiffType: {}, + } + if (sanitize !== undefined) { + for (const src of [acc.sanitize?.byDiffType, ret.sanitize?.byDiffType]) { + if (src !== undefined) sumCounts(sanitize.byDiffType, src) + } + } + return { + counts, + issues: [...acc.issues, ...ret.issues], + ...(hasGroups ? { issueGroups } : {}), + ...(sanitize === undefined ? {} : { sanitize }), + } +} + +/** + * 编排管道（Task 6 汇流契约底本——Task 5 审查移交钉死）： + * - 事件期**严格串行 await** 分派 content → term → media → route：三家（term/media/ + * route）的 entry 锚点解析依赖 content 对**同事件**的落库产物（source_records. + * content_entry_id / entry_locales），禁 Promise.all——并行下锚点查询可能抢在 content + * 写入前发生； + * - finish 唯一硬序：term/content（同步、无库依赖）任意 → `await media.finish()` 先于 + * `await route.finish()`（attachment 301 反查与 asset 全表都读 media_assets——资产须已齐）； + * - 分派规则（Task 3 审查移交⑥）：wp:attachment 行的媒体域归 media-writer + * （media_assets 唯一写入方），content-writer 对其只留 source_records record-only 足迹 + * （前置事实 11）——两家各写各域，单一 owner； + * - counts 按键求和聚合（安全网——四家键前缀互斥）；清洗统计与 row 事件行级问题在 + * 分派循环收集（RowResult 契约：issues 逐条收录为行级错误）；unsupported/error 事件 + * 的 reason/issue 同归 adapter 组（审查 Fix 4——不静默丢弃诊断信息）； + * - **失败报告部分冲刷**（审查 Fix 1）：分派/finish 中途抛错时，把可安全取得的部分产出 + * 冲入 ctx.report 累加器后重抛——term/content 的 finish 是纯返回（无库依赖，就地补算）； + * media/route 的 finish 依赖库与先序产物，失败路径**不调用**（其 issueGroups 为空数组——未到该 finish 阶段，部分数据不可安全取得）； + * 进度以 orchestrator 键 eventsDispatched 落 counts（仅失败路径出现——成功路径不占键）。 + */ +export function orchestratedPipeline(paths: { productsPath: string; newsPath: string }): ImportPipeline { + return async (ctx) => { + const content = createContentEventHandler({ + db: ctx.db, + siteId: ctx.siteId, + connectionId: ctx.connectionId, + snapshotId: ctx.wxrSnapshot.id, + jobId: ctx.jobId, + defaultLocale: ctx.defaultLocale, + versions: ctx.versions, + }) + const term = createTermEventHandler({ + db: ctx.db, + siteId: ctx.siteId, + connectionId: ctx.connectionId, + }) + const media = await createMediaEventHandlerFromContext(ctx) + const route = createRouteEventHandler({ + db: ctx.db, + siteId: ctx.siteId, + connectionId: ctx.connectionId, + snapshotId: ctx.wxrSnapshot.id, + defaultLocale: ctx.defaultLocale, + productsPath: paths.productsPath, + newsPath: paths.newsPath, + }) + + const sanitize: ImportSanitizeStats = { changedRows: 0, byDiffType: {} } + const adapterIssues: string[] = [] + let eventsDispatched = 0 + // 已算成的纯返回 finish（media/route finish 抛错时直接复用，不重调） + let finished: { content: ImportPipelineOutcome; term: ImportPipelineOutcome } | null = null + try { + for await (const ev of ctx.events) { + if (ev.kind === 'row') { + if (ev.changed) sanitize.changedRows++ + for (const diff of ev.sanitizeDiffs) { + sanitize.byDiffType[diff.type] = (sanitize.byDiffType[diff.type] ?? 0) + 1 + } + for (const issue of ev.issues) { + adapterIssues.push(`行级问题（${ev.row.sourceType}/${ev.row.sourceId}）：${issue}`) + } + } else if (ev.kind === 'unsupported') { + adapterIssues.push(`不支持对象（${ev.sourceType}/${ev.postId ?? '?'}, ${ev.rawLocation}）：${ev.reason}`) + } else if (ev.kind === 'error') { + adapterIssues.push(`适配器错误（${ev.sourceType}/${ev.postId ?? '?'}, ${ev.rawLocation}）：${ev.issue}`) + } + await content.handleEvent(ev) + await term.handleEvent(ev) + await media.handleEvent(ev) + await route.handleEvent(ev) + eventsDispatched++ + } + const t = term.finish() + const c = content.finish() + finished = { content: c, term: t } + const m = await media.finish() + const r = await route.finish() + + const counts: Record = {} + sumCounts(counts, c.counts) + sumCounts(counts, t.counts) + sumCounts(counts, m.counts) + sumCounts(counts, r.counts) + return { + counts, + issues: [...c.issues, ...t.issues, ...m.issues, ...r.issues, ...adapterIssues], + issueGroups: { + content: c.issues, + term: t.issues, + media: m.issues, + route: r.issues, + adapter: adapterIssues, + orchestrator: [], + }, + sanitize, + } + } catch (err) { + // 失败报告部分冲刷（审查 Fix 1）：重抛前把可审计的部分产出落入累加器—— + // createImportJob 对抛错管道取 ctx.report 现值进报告，失败诊断不再只剩 error 块。 + const c = finished?.content ?? trySyncFinish(content) + const t = finished?.term ?? trySyncFinish(term) + const partial: Record = { eventsDispatched } + if (c !== null) sumCounts(partial, c.counts) + if (t !== null) sumCounts(partial, t.counts) + sumCounts(ctx.report.counts, partial) + const partialIssues = [...(c?.issues ?? []), ...(t?.issues ?? []), ...adapterIssues] + ctx.report.issues.push(...partialIssues) + ctx.report.issueGroups = { + content: c?.issues ?? [], + term: t?.issues ?? [], + media: [], + route: [], + adapter: adapterIssues, + orchestrator: [], + } + ctx.report.sanitize = sanitize + throw err + } + } +} + +/** 失败路径的安全 finish：content/term 的 finish 是纯返回（无库依赖）——已算成则由调用方 + * 复用，此处只对未算成的就地补算；finish 本身抛错则放弃该部分（不掩盖原错误） */ +function trySyncFinish(handler: { finish(): ImportPipelineOutcome }): ImportPipelineOutcome | null { + try { + return handler.finish() + } catch { + return null + } +} diff --git a/apps/api/src/modules/import/report-html.ts b/apps/api/src/modules/import/report-html.ts new file mode 100644 index 0000000..af0c4ff --- /dev/null +++ b/apps/api/src/modules/import/report-html.ts @@ -0,0 +1,155 @@ +// 导入报告 HTML 渲染（Plan 3 Task 8——Task 6 推迟项兑现）：沿用 Plan 2 Task 6 模板纪律 +// （@sf/adapter-wp 的 renderReportHtml 同款）： +// - 对象层（import.json）保留**原文**（issue 字符串未转义）——JSON 归档/机器消费的语义层； +// - 渲染层对**所有**动态文本经 escapeHtml（<>&"' 全转义）——报告的动态文本来自不可信 +// WXR 内容（行级问题携带清洗前 excerpt 形态，可含 `' + const total = 10_003 + const issues = Array.from({ length: total }, (_, i) => + i === 0 ? `${payload}悬空 attachment 引用：${i}` : `悬空 attachment 引用：${i}`, + ) + const result = await createImportJob(db, { + siteKey: key, + reportRoot: w.reportRoot, + pipeline: async () => ({ counts: { contentEntries: 0 }, issues, issueGroups: { media: issues } }), + }) + // 终态判定在截断前的全量长度上做（截断是报告固化层语义，不吞终态事实） + expect(result.status).toBe('succeeded_with_issues') + + const report = await readReport(w, result.jobId) + expect(report.issues).toHaveLength(10_000) + expect(report.issuesTruncated).toBe(3) + expect(report.issueGroups.media).toHaveLength(10_000) // 各组同限截断 + expect(report.issues[0]).toContain('` +const ATTR_ESCAPE_ESCAPED = '"><img onerror=alert(1) src=x>' + +function baseReport(overrides: Partial = {}): ImportReport { + return { + jobId: 'job-0001', + generatedAt: '2026-10-21T00:00:00.000Z', + status: 'succeeded_with_issues', + siteKey: 'demo', + siteId: '11111111-1111-1111-1111-111111111111', + connectionId: '22222222-2222-2222-2222-222222222222', + idempotencyKey: 'import:demo:snap:v1/v1/v1', + attempt: 1, + snapshots: { + wxr: { id: '33333333-3333-3333-3333-333333333333', kind: 'wxr', sha256: 'a'.repeat(64), byteSize: 1024, storagePath: 'incoming/a.xml' }, + media_bundle: null, + manual_url_list: null, + }, + versions: { adapter: 'a1', mapping: 'm1', sanitizer: 's1', parser: 'p1' }, + paths: { productsPath: '/products/', newsPath: '/news/' }, + counts: { contentEntries: 3, termsProcessed: 2 }, + issues: [], + issuesTruncated: 0, + error: null, + durationMs: 1234.5, + ...overrides, + } +} + +describe('renderImportReportHtml（Task 8：转义/结构/封顶注记）', () => { + it('所有动态文本通道全转义：issue/错误 summary/快照 id 与 sha——恶意 payload 不以原文出现', () => { + const report = baseReport({ + issues: [MALICIOUS, ATTR_ESCAPE], + issueGroups: { content: [MALICIOUS], adapter: [ATTR_ESCAPE] }, + error: { code: 'PIPELINE_FAILED', summary: `[pipeline] Error: ${MALICIOUS}` }, + }) + const html = renderImportReportHtml(report) + + // 转义形出现在分组视图与错误块（<>&"' 全集） + expect(html).toContain(MALICIOUS_ESCAPED) + expect(html).toContain(ATTR_ESCAPE_ESCAPED) + expect(html).toContain(`[pipeline] Error: ${MALICIOUS_ESCAPED}`) + // 原文 payload 零出现（骨架无 script 标签——出现即 XSS 面） + expect(html).not.toContain(']*\son\w+=/i) + }) + + it('概览/计数/路径/版本块结构：计数表键字典序、快照缺席呈现（无）', () => { + const html = renderImportReportHtml(baseReport()) + expect(html).toContain('

导入报告

') + expect(html).toContain('状态: succeeded_with_issues') + expect(html).toContain('media_bundle：（无）') + expect(html).toContain('manual_url_list：（无）') + // 计数表键字典序（contentEntries < termsProcessed） + const table = html.slice(html.indexOf('

计数

'), html.indexOf('')) + expect(table.indexOf('contentEntries')).toBeLessThan(table.indexOf('termsProcessed')) + expect(html).toContain('productsPath: /products/') + expect(html).toContain('adapter: a1') + }) + + it('分组视图按固定序渲染（content→term→media→route→adapter→orchestrator），组内条数入标题', () => { + const html = renderImportReportHtml( + baseReport({ + issues: ['a1', 'r1'], + issueGroups: { route: ['r1'], content: ['a1'], adapter: [] }, + }), + ) + expect(html).toContain('内容（content-writer）（1 条）') + expect(html).toContain('路由（route-writer）（1 条）') + expect(html).toContain('适配器（行级/unsupported/error）（0 条）') + expect(html.indexOf('内容（content-writer）')).toBeLessThan(html.indexOf('路由（route-writer）')) + expect(html.indexOf('路由（route-writer）')).toBeLessThan(html.indexOf('适配器（行级/unsupported/error）')) + expect(html).toContain('

（无）

') // adapter 空组 + }) + + it('分组缺席（注入管道形态）回退 flat 清单；无问题时呈现（无）', () => { + const flat = renderImportReportHtml(baseReport({ issues: ['悬空 attachment 引用：2 条'] })) + expect(flat).toContain('
  • 悬空 attachment 引用：2 条
    • ') + const clean = renderImportReportHtml(baseReport({ status: 'succeeded' })) + expect(clean).toContain('

    （无）

    ') + expect(clean).not.toContain('另有') + }) + + it('封顶注记：issuesTruncated>0 时呈现溢出条数（仅计数）', () => { + const html = renderImportReportHtml( + baseReport({ issues: ['i1'], issuesTruncated: 10_003 }), + ) + expect(html).toContain('（另有 10003 条超出固化上限，仅计数）') + }) + + it('清洗统计与适配器分段计时块（出席才渲染）', () => { + const withBlocks = renderImportReportHtml( + baseReport({ + sanitize: { changedRows: 7, byDiffType: { 'removed-tag': 5, 'removed-attr': 2 } }, + adapterTiming: { parseMs: 11.2, sanitizeMs: 3.4 }, + }), + ) + expect(withBlocks).toContain('changed rows: 7') + expect(withBlocks).toContain('removed-attr=2、removed-tag=5') // byDiffType 键也字典序 + expect(withBlocks).toContain('adapter parse: 11.200ms') + const without = renderImportReportHtml(baseReport()) + expect(without).not.toContain('清洗统计') + expect(without).not.toContain('adapter parse') + }) + + it('渲染不改输入对象（对象层保留原文——JSON 归档语义层不动）', () => { + const report = baseReport({ issues: [MALICIOUS] }) + renderImportReportHtml(report) + expect(report.issues[0]).toBe(MALICIOUS) + }) +})