feat(import): HTML 报告与错误封顶;orchestrator 抽取

- report-html.ts:renderImportReportHtml(Plan 2 Task 6 模板纪律——全动态文本
  转义/静态骨架无 script/分组固定序渲染/封顶注记),import.html 与 import.json
  同 try 原子落盘(同 REPORT_WRITE_FAILED 失败口径)
- 报告 issues 封顶(Task 6 推迟项):REPORT_ISSUES_CAP=10k——flat 清单与
  issueGroups 各组同限截断 + issuesTruncated 计数;终态判定(succeeded_with_issues)
  在全量长度上做,截断不吞终态事实;counts/sanitize 始终完整
- orchestrator.ts 抽取(审查触发器 import.service 849 行):orchestratedPipeline/
  mergeOutcomes/sumCounts/trySyncFinish 迁出为纯函数 over ctx;契约类型单源
  import.service.ts 不动;既有测试全绿即证零行为变更
- 测试 +8:单元 7(恶意 payload 转义/结构/分组序/封顶注记/渲染不改输入)+
  集成 1(10_003 条注入:截断计数/JSON 原文保留/HTML 转义/终态判定);333 全绿

Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
浅念
2026-10-21 21:20:12 +08:00
parent 3776a0d6ad
commit a7177afde7
5 changed files with 622 additions and 198 deletions
+103 -198
View File
@@ -21,14 +21,14 @@
// site + wxr 快照 + 三语义版本(adapter/mapping/sanitizer——任一 bump 即新 job,对齐 // site + wxr 快照 + 三语义版本(adapter/mapping/sanitizer——任一 bump 即新 job,对齐
// spec §7.5 “版本变化可显式触发重新标准化”;parser 是诊断信息不进键)。 // spec §7.5 “版本变化可显式触发重新标准化”;parser 是诊断信息不进键)。
// //
// 管道注入:Task 6 起默认 = **编排管道**(四家写入器汇流,orchestratedPipeline)——真 // 管道注入:Task 6 起默认 = **编排管道**(四家写入器汇流)——orchestratedPipeline 自
// 事件流(runWxrAdapter 读 wxr 快照副本,mapConfig.locale 接 site.default_locale——前置 // Task 8 起抽取于 orchestrator.ts(纯函数 over ctx,零行为变更)。本文件负责:真事件流
// 事实 8)→ content/term/media/route 严格串行分派 → finish 硬序(media 先于 route)→ // 构造(runWxrAdapter 读 wxr 快照副本,mapConfig.locale 接 site.default_locale——前置
// counts 按键**求和**聚合(Task 6 钉死,见 ImportPipelineOutcome.counts 契约)。注入 // 事实 8)→ Outcome 汇合(mergeOutcomes)→ 报告固化。注入 pipeline 钩子供集成测试观察
// pipeline 钩子供集成测试观察中间态/失败路径(注入时缺省事件流为空——Task 1 语义保持)。 // 中间态/失败路径(注入时缺省事件流为空——Task 1 语义保持)。
// //
// stale 回收(Task 6 义务④):每次建 job 前把同 site 超过 24h 仍 queued/running 的 job // stale 回收(Task 6 义务④):每次建 job 前把同 site 超过 24h 仍 queued/running 的 job
// 置 stale(崩溃/中断残留终态化)。心跳列决策留 ADR-017——现阶段无心跳列,以 // 置 stale(崩溃/中断残留终态化)。心跳列决策见 ADR-017——现阶段无心跳列,以
// created_at 近似(job 为单进程就地推进,24h 远超任何真实执行时长)。 // created_at 近似(job 为单进程就地推进,24h 远超任何真实执行时长)。
import { createReadStream } from 'node:fs' import { createReadStream } from 'node:fs'
import { mkdir, rename, rm, writeFile } from 'node:fs/promises' import { mkdir, rename, rm, writeFile } from 'node:fs/promises'
@@ -48,16 +48,14 @@ import {
import { importJobs, sites, sourceConnections, sourceSnapshots } from '../../db/schema.js' import { importJobs, sites, sourceConnections, sourceSnapshots } from '../../db/schema.js'
import type * as schema from '../../db/schema.js' import type * as schema from '../../db/schema.js'
import { isUniqueViolation } from '../source/register.service.js' import { isUniqueViolation } from '../source/register.service.js'
import { createContentEventHandler } from './content-writer.js' import { mergeOutcomes, orchestratedPipeline } from './orchestrator.js'
import { createTermEventHandler } from './term-writer.js'
import { createMediaEventHandlerFromContext } from './media-writer.js'
import { import {
DEFAULT_NEWS_PATH, DEFAULT_NEWS_PATH,
DEFAULT_PRODUCTS_PATH, DEFAULT_PRODUCTS_PATH,
createRouteEventHandler,
isSafePath, isSafePath,
normalizeListPath, normalizeListPath,
} from './route-writer.js' } from './route-writer.js'
import { renderImportReportHtml } from './report-html.js'
/** 本文件位于 <repo>/apps/api/src/modules/import/ → 上五级是仓库根 */ /** 本文件位于 <repo>/apps/api/src/modules/import/ → 上五级是仓库根 */
const REPO_ROOT = resolve(dirname(fileURLToPath(import.meta.url)), '../../../../..') const REPO_ROOT = resolve(dirname(fileURLToPath(import.meta.url)), '../../../../..')
@@ -519,16 +517,27 @@ export async function createImportJob(
if (failed !== null) { if (failed !== null) {
finalStatus = 'failed' finalStatus = 'failed'
} else { } else {
// 终态判定在**全量** issues 长度上做(截断是报告固化层语义,不吞终态事实)
finalStatus = merged.issues.length > 0 ? 'succeeded_with_issues' : 'succeeded' 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 // 阶段 6:报告先行落盘(temp+rename 原子替换),终态落行在后——行内 report_path
// 出现即文件存在;写盘失败 → job failed(REPORT_WRITE_FAILED、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 reportRoot = resolve(input.reportRoot ?? join(REPO_ROOT, 'artifacts', 'import'))
const reportDir = join(reportRoot, jobId, 'reports') const reportDir = join(reportRoot, jobId, 'reports')
const reportPath = join(reportDir, 'import.json') const reportPath = join(reportDir, 'import.json')
const report = { const htmlPath = join(reportDir, 'import.html')
const report: ImportReport = {
jobId, jobId,
generatedAt: new Date().toISOString(), generatedAt: new Date().toISOString(),
status: finalStatus, status: finalStatus,
@@ -546,8 +555,9 @@ export async function createImportJob(
// 主列表路径(前置事实 9:CLI 参数来源记入报告——route-writer 实际消费的规范化值) // 主列表路径(前置事实 9:CLI 参数来源记入报告——route-writer 实际消费的规范化值)
paths: { productsPath, newsPath }, paths: { productsPath, newsPath },
counts: merged.counts, counts: merged.counts,
issues: merged.issues, issues: cappedIssues,
...(merged.issueGroups === undefined ? {} : { issueGroups: merged.issueGroups }), issuesTruncated,
...(cappedGroups === undefined ? {} : { issueGroups: cappedGroups }),
...(merged.sanitize === undefined ? {} : { sanitize: merged.sanitize }), ...(merged.sanitize === undefined ? {} : { sanitize: merged.sanitize }),
// 适配器分段计时(仅真流编排——Plan 2 报告纪律的 parse/sanitize 分段在导入侧续录) // 适配器分段计时(仅真流编排——Plan 2 报告纪律的 parse/sanitize 分段在导入侧续录)
...(adapterTimings === null ? {} : { adapterTiming: { parseMs: adapterTimings.parseMs, sanitizeMs: adapterTimings.sanitizeMs } }), ...(adapterTimings === null ? {} : { adapterTiming: { parseMs: adapterTimings.parseMs, sanitizeMs: adapterTimings.sanitizeMs } }),
@@ -557,14 +567,8 @@ export async function createImportJob(
let reportWritten = true let reportWritten = true
try { try {
await mkdir(reportDir, { recursive: true }) await mkdir(reportDir, { recursive: true })
const tmpPath = `${reportPath}.tmp` await writeAtomically(reportPath, `${JSON.stringify(report, null, 2)}\n`)
try { await writeAtomically(htmlPath, `${renderImportReportHtml(report)}\n`)
await writeFile(tmpPath, `${JSON.stringify(report, null, 2)}\n`, 'utf8')
await rename(tmpPath, reportPath)
} catch (err) {
await rm(tmpPath, { force: true })
throw err
}
} catch (err) { } catch (err) {
reportWritten = false reportWritten = false
failed = { failed = {
@@ -630,22 +634,95 @@ async function reuseExisting(
} }
/** 报告里的快照引用瘦身版:不含 metadata(media_bundle 清单可达数千条,报告只留身份锚点) */ /** 报告里的快照引用瘦身版:不含 metadata(media_bundle 清单可达数千条,报告只留身份锚点) */
function slimRef(ref: SnapshotRef): { export interface SnapshotRefSlim {
id: string id: string
kind: string kind: string
sha256: string sha256: string
byteSize: number byteSize: number
storagePath: string storagePath: string
} { }
function slimRef(ref: SnapshotRef): SnapshotRefSlim {
return { id: ref.id, kind: ref.kind, sha256: ref.sha256, byteSize: ref.byteSize, storagePath: ref.storagePath } 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<Record<ImportIssueGroup, string[]>> | undefined,
): Partial<Record<ImportIssueGroup, string[]>> | undefined {
if (groups === undefined) return undefined
const capped: Partial<Record<ImportIssueGroup, string[]>> = {}
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<string, number>
/** 行级问题清单(封顶 {@link REPORT_ISSUES_CAP} 条——超出只计数) */
issues: string[]
/** 超出 {@link REPORT_ISSUES_CAP} 被丢弃固化的条数(只计数;issueGroups 各组同限截断不另计) */
issuesTruncated: number
/** 可选:按来源分组的 issues 视图(各组同限截断) */
issueGroups?: Partial<Record<ImportIssueGroup, string[]>>
/** 可选:清洗统计(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<void> {
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 占位;每次调用产新迭代器) */ /** 缺省空事件流(input.events 缺席时的 ctx.events 占位;每次调用产新迭代器) */
async function* emptyEvents(): AsyncGenerator<AdapterEvent, void, void> {} async function* emptyEvents(): AsyncGenerator<AdapterEvent, void, void> {}
// ---- 编排管道(Task 6 汇流——Task 5 审查移交的契约底本在此兑现) ---- // ---- 事件流构造与 stale 回收(编排本体见 orchestrator.ts) ----
/** stale 回收阈值:queued/running 滞留超过此时长即宣判(毫秒) */ /** stale 回收阈值:queued/running 滞留超过此时长即宣判(毫秒)——决策与理由见 ADR-017 */
const STALE_AFTER_MS = 24 * 60 * 60 * 1000 const STALE_AFTER_MS = 24 * 60 * 60 * 1000
/** wxr 快照副本绝对路径(metadata.storageRoot + storagePath——登记期固化的存储根事实) */ /** wxr 快照副本绝对路径(metadata.storageRoot + storagePath——登记期固化的存储根事实) */
@@ -675,175 +752,3 @@ async function* realEventStream(
}) })
} }
/** counts 按键求和(Task 6 钉死的聚合策略——就地累加 target) */
function sumCounts(target: Record<string, number>, source: Record<string, number>): 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<string, number> = { ...acc.counts }
sumCounts(counts, ret.counts)
const issueGroups: Partial<Record<ImportIssueGroup, string[]>> = {}
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<string, number> = {}
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<string, number> = { 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
}
}
+196
View File
@@ -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<string, number>, source: Record<string, number>): 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<string, number> = { ...acc.counts }
sumCounts(counts, ret.counts)
const issueGroups: Partial<Record<ImportIssueGroup, string[]>> = {}
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<string, number> = {}
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<string, number> = { 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
}
}
+155
View File
@@ -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 形态,可含 `<script src="x">`),未转义渲染会使
// 报告本身成为存储型 XSS 载体(Plan 2 Task 5 审查移交纪律的导入侧续录);
// - 静态骨架无 script/事件属性;计数表键按字典序渲染(对象键序不保证,渲染须确定)。
//
// escapeHtml 与 adapter-wp/report.ts 的私有实现同形独立维护(17 行纯函数)——不为此扩
// @sf/adapter-wp 公共导出面;两处若任一改判集,由对抗测试各自护栏(本文件测试钉恶意
// payload 转义)。
import type { ImportReport, SnapshotRefSlim } from './import.service.js'
/** HTML 转义(渲染层唯一动态文本通道):<>&"' 全转义,& 最先避免二次转义陷阱 */
function escapeHtml(text: string): string {
return text.replace(/[&<>"']/g, (ch) => {
switch (ch) {
case '&':
return '&amp;'
case '<':
return '&lt;'
case '>':
return '&gt;'
case '"':
return '&quot;'
default:
return '&#39;'
}
})
}
const esc = escapeHtml
/** 分组渲染顺序(固定——对象键序不保证,人类视图须确定) */
const GROUP_ORDER = ['content', 'term', 'media', 'route', 'adapter', 'orchestrator'] as const
const GROUP_LABELS: Record<(typeof GROUP_ORDER)[number], string> = {
content: '内容(content-writer)',
term: '分类(term-writer)',
media: '媒体(media-writer)',
route: '路由(route-writer)',
adapter: '适配器(行级/unsupported/error)',
orchestrator: '编排器(orchestrator)',
}
function snapshotLine(label: string, ref: SnapshotRefSlim | null): string {
if (ref === null) return `<li>${label}:(无)</li>`
return `<li>${label}:${esc(ref.id)}(${esc(ref.kind)},sha256=${esc(ref.sha256.slice(0, 12))}…,${ref.byteSize} bytes)</li>`
}
/**
* 导入报告 → HTML 片段(结构:概览/快照/版本/路径/计数表/清洗统计/问题清单/错误/耗时)。
* 问题清单优先渲染分组视图(编排管道恒产 issueGroups——六组按固定序);分组缺席
* (注入管道的 Task 1 形态)回退 flat 清单。封顶溢出以 issuesTruncated 注记呈现。
*/
export function renderImportReportHtml(report: ImportReport): string {
const lines: string[] = []
lines.push('<h2>导入报告</h2>')
lines.push('<h3>概览</h3>')
lines.push(
`<ul><li>job: ${esc(report.jobId)}</li><li>状态: ${esc(report.status)}</li>` +
`<li>站点: ${esc(report.siteKey)}(${esc(report.siteId)})</li><li>connection: ${esc(report.connectionId)}</li>` +
`<li>幂等键: ${esc(report.idempotencyKey)}</li><li>attempt: ${report.attempt}</li>` +
`<li>生成时间: ${esc(report.generatedAt)}</li></ul>`,
)
lines.push('<h3>快照</h3>')
lines.push(
`<ul>${snapshotLine('wxr(主输入)', report.snapshots.wxr)}` +
`${snapshotLine('media_bundle', report.snapshots.media_bundle)}` +
`${snapshotLine('manual_url_list', report.snapshots.manual_url_list)}</ul>`,
)
lines.push('<h3>版本</h3>')
lines.push(
`<ul><li>adapter: ${esc(report.versions.adapter)}</li>` +
`<li>mapping: ${esc(report.versions.mapping)}</li>` +
`<li>sanitizer: ${esc(report.versions.sanitizer)}</li>` +
`<li>parser: ${esc(report.versions.parser)}</li></ul>`,
)
lines.push('<h3>路径</h3>')
lines.push(
`<ul><li>productsPath: ${esc(report.paths.productsPath)}</li>` +
`<li>newsPath: ${esc(report.paths.newsPath)}</li></ul>`,
)
lines.push('<h3>计数</h3>')
lines.push('<table><thead><tr><th>键</th><th>值</th></tr></thead><tbody>')
const keys = Object.keys(report.counts).sort()
if (keys.length === 0) {
lines.push('<tr><td colspan="2">(无)</td></tr>')
}
for (const key of keys) {
lines.push(`<tr><td>${esc(key)}</td><td>${report.counts[key]}</td></tr>`)
}
lines.push('</tbody></table>')
if (report.sanitize !== undefined) {
lines.push('<h3>清洗统计</h3>')
const diffTypes = Object.keys(report.sanitize.byDiffType).sort()
const diffs =
diffTypes.length === 0
? ''
: `<li>byDiffType: ${diffTypes.map((t) => `${esc(t)}=${report.sanitize!.byDiffType[t]}`).join('、')}</li>`
lines.push(`<ul><li>changed rows: ${report.sanitize.changedRows}</li>${diffs}</ul>`)
}
lines.push('<h3>问题清单</h3>')
const groups = report.issueGroups
if (groups !== undefined) {
for (const key of GROUP_ORDER) {
const groupIssues = groups[key]
if (groupIssues === undefined) continue
lines.push(`<h4>${esc(GROUP_LABELS[key])}(${groupIssues.length} 条)</h4>`)
if (groupIssues.length === 0) {
lines.push('<p>(无)</p>')
} else {
lines.push('<ul>')
for (const issue of groupIssues) {
lines.push(`<li>${esc(issue)}</li>`)
}
lines.push('</ul>')
}
}
} else if (report.issues.length === 0 && report.issuesTruncated === 0) {
lines.push('<p>(无)</p>')
} else {
lines.push('<ul>')
for (const issue of report.issues) {
lines.push(`<li>${esc(issue)}</li>`)
}
lines.push('</ul>')
}
if (report.issuesTruncated > 0) {
lines.push(`<p>(另有 ${report.issuesTruncated} 条超出固化上限,仅计数)</p>`)
}
if (report.error !== null) {
lines.push(
`<h3>错误</h3><ul><li>code: ${esc(report.error.code)}</li><li>summary: ${esc(report.error.summary)}</li></ul>`,
)
}
lines.push('<h3>耗时</h3>')
const adapterTiming =
report.adapterTiming === undefined
? ''
: `<li>adapter parse: ${report.adapterTiming.parseMs.toFixed(3)}ms</li>` +
`<li>adapter sanitize: ${report.adapterTiming.sanitizeMs.toFixed(3)}ms</li>`
lines.push(`<ul>${adapterTiming}<li>total: ${report.durationMs.toFixed(1)}ms</li></ul>`)
return lines.join('\n')
}
+44
View File
@@ -19,6 +19,9 @@
// unsupported/error 事件的 reason/issue 归档 adapter 组(Fix 4——此前被丢弃); // unsupported/error 事件的 reason/issue 归档 adapter 组(Fix 4——此前被丢弃);
// 8. 主列表路径解析期防线(Fix 3):--products-path/--news-path 含穿越段 → // 8. 主列表路径解析期防线(Fix 3):--products-path/--news-path 含穿越段 →
// INVALID_LIST_PATH 且零 job 行(建行前抛出——建行后再抛会留悬挂 running 行)。 // INVALID_LIST_PATH 且零 job 行(建行前抛出——建行后再抛会留悬挂 running 行)。
// 9. HTML 报告与 issues 封顶(Task 8 兑现 Task 6 推迟项):import.html 落盘且动态文本
// 全转义;issues 超上限(10k)→ 报告固化面截断 + issuesTruncated 计数;终态判定
// 用全量长度(截断不吞 succeeded_with_issues 事实);对象层(import.json)保留原文。
// 运行于 globalSetup 临时库(ADR-015);storage/报告根 os.tmpdir,finally 清理。 // 运行于 globalSetup 临时库(ADR-015);storage/报告根 os.tmpdir,finally 清理。
import { mkdir, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises' import { mkdir, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'
import { tmpdir } from 'node:os' import { tmpdir } from 'node:os'
@@ -601,4 +604,45 @@ describe('编排集成 + 导入报告(四家写入器汇流)', () => {
} }
}) })
}) })
it('HTML 报告与 issues 封顶(Task 8):import.html 落盘且转义;issues 超上限 → 截断 + issuesTruncated、终态判定用全量长度、JSON 对象层留原文', async () => {
await withTestDb(async (db) => {
const w = await makeWork('sf-full-html')
try {
const { key, siteId } = await registerWxrOnly(db, w)
try {
// 恶意 payload(清洗前 excerpt 形态)+ 超上限 issues(10_003 条)
const payload = '<script src="https://evil.example/x.js"></script>'
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('<script') // JSON 对象层保留原文(转义只在渲染层)
// HTML:落盘(与 import.json 同目录)+ 全动态文本转义 + 封顶注记
const html = await readFile(join(w.reportRoot, result.jobId, 'reports', 'import.html'), 'utf8')
expect(html).not.toContain('<script') // 骨架无 script,出现即 XSS 面
expect(html).toContain('&lt;script src=&quot;https://evil.example/x.js&quot;&gt;')
expect(html).toContain('(另有 3 条超出固化上限,仅计数)')
expect(html).toContain('悬空 attachment 引用:0')
} finally {
await cleanupSiteRows(db, siteId)
}
} finally {
await rm(w.work, { recursive: true, force: true })
}
})
})
}) })
+124
View File
@@ -0,0 +1,124 @@
// 导入报告 HTML 渲染测试(Plan 3 Task 8——Task 6 推迟项兑现):钉 Plan 2 Task 6 模板
// 纪律的导入侧镜像——①恶意 payload 全转义(issue/error/键名等所有动态文本通道,
// 报告不得成为存储型 XSS 载体)②结构(分组视图按固定序渲染、封顶注记、错误块)
// ③对象层/渲染层分离(renderImportReportHtml 不改输入——对象层保留原文)。
import { describe, expect, it } from 'vitest'
import type { ImportReport } from '../../apps/api/src/modules/import/import.service.js'
import { renderImportReportHtml } from '../../apps/api/src/modules/import/report-html.js'
const MALICIOUS = '<script src="https://evil.example/x.js"></script>'
const MALICIOUS_ESCAPED = '&lt;script src=&quot;https://evil.example/x.js&quot;&gt;&lt;/script&gt;'
const ATTR_ESCAPE = `"><img onerror=alert(1) src=x>`
const ATTR_ESCAPE_ESCAPED = '&quot;&gt;&lt;img onerror=alert(1) src=x&gt;'
function baseReport(overrides: Partial<ImportReport> = {}): 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('<script')
expect(html).not.toContain(MALICIOUS)
expect(html).not.toContain('<img') // 未转义标签形态(转义后是 &lt;img)
// 静态骨架纪律:真实标签上无事件属性(转义文本中的 onerror= 字样无害——标签语法已破)
expect(html).not.toMatch(/<[a-z][^>]*\son\w+=/i)
})
it('概览/计数/路径/版本块结构:计数表键字典序、快照缺席呈现(无)', () => {
const html = renderImportReportHtml(baseReport())
expect(html).toContain('<h2>导入报告</h2>')
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('<h3>计数</h3>'), html.indexOf('</table>'))
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('<p>(无)</p>') // adapter 空组
})
it('分组缺席(注入管道形态)回退 flat 清单;无问题时呈现(无)', () => {
const flat = renderImportReportHtml(baseReport({ issues: ['悬空 attachment 引用:2 条'] }))
expect(flat).toContain('<li>悬空 attachment 引用:2 条</li>')
const clean = renderImportReportHtml(baseReport({ status: 'succeeded' }))
expect(clean).toContain('<p>(无)</p>')
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)
})
})