Files
cms-website/scripts/sf.ts
T
浅念 eb188da5c2 fix(import): 复用口径忠实化;管道 Outcome 契约与 succeeded_with_issues;报告写入时序;stale 警告
评审修复(Plan 3 Task 1 复审):
- 复用路径只返回 job 锚定字段(jobId/status/reused/siteId/mapping_version/
  report_path),快照/版本块仅在非复用时解析返回——复用结果不再回显编造数据
- ImportPipeline 契约钉死:返回 Promise<ImportPipelineOutcome>
  ({ counts, issues });服务按"累加器 ⊕ 返回值"合并(counts 同键返回值
  覆盖、issues 累加器在前),issues 非空 → succeeded_with_issues;
  ctx.report 即累加器(管道抛错时部分产出仍入报告)
- 报告时序:mkdir + tmp 写入 + rename 先于终态行更新——report_path 落库
  即文件存在;写失败 → failed(REPORT_WRITE_FAILED),不挂悬空路径
- CLI 复用命中非终态(queued/running)→ 崩溃残留警告(重试需显式换 --key)
- 管道抛 ImportError 保留原 code 入 error_code,summary 冠 [pipeline]
- 确定性 23505 探针:未提交孪生 job 行事务内持唯一键触发真实 PG 唯一
  冲突 → 败者复用既有 job
- 报告补 idempotencyKey / attempt 字段
- 计划文档 Task 6 增补 stale 回收决策项(ADR-017)

测试 282 通过(279 + 3:Outcome 合并/ImportError 前缀/23505 探针);
typecheck 四段链 + lint 全绿。

Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-10-21 15:56:00 +08:00

226 lines
12 KiB
JavaScript
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env node
// sf CLI(Plan 2 Task 7 + Plan 3 Task 1):commander 三命令——
// - wxr:probe:只读诊断(任意可读路径;Task 8 直读 cms 用此通道),出 JSON/HTML 报告不入库;
// - source:register:调 registerSource 登记 content-addressed 快照(路径白名单校验在服务内);
// - import:调 createImportJob 建 import_jobs 并推进生命周期(Task 1 空管道骨架;Task 6 接真实管道)。
// 执行机制(Plan 2 评审钉死):根 devDep tsx 直吃 TS 源码(含 apps/api 的 Nest 源码与
// packages 的装饰器/TS 源文件),@sf/* 经根/包 tsconfig paths 直连 src——无需预构建 dist。
// 注意:CLI 只 import register.service.ts / import.service.ts(零装饰器纯函数文件),
// 不拉起 Nest 模块——反射元数据依赖由此避免;module 文件留给 standalone context。
import { createReadStream } from 'node:fs'
import { mkdir, writeFile } from 'node:fs/promises'
import { dirname, join, resolve } from 'node:path'
import { fileURLToPath } from 'node:url'
import { Command } from 'commander'
import { renderReportHtml, runWxrAdapterReport, type WxrAdapterReport } from '@sf/adapter-wp'
import { getDb } from '../apps/api/src/db/client.js'
import {
ImportError,
createImportJob,
type CreateImportJobInput,
} from '../apps/api/src/modules/import/import.service.js'
import {
makeRunId,
registerSource,
SourceRegisterError,
type RegisterInput,
} from '../apps/api/src/modules/source/register.service.js'
const REPO_ROOT = resolve(dirname(fileURLToPath(import.meta.url)), '..')
/** 探测报告摘要(stdout):版本块之外的计数/清洗/耗时 + 致命错误 */
function printProbeSummary(report: WxrAdapterReport): void {
console.log(` rows=${report.totals.rows} terms=${report.totals.terms} unsupported=${report.totals.unsupported} failed=${report.totals.failed}`)
console.log(` changedRows=${report.sanitize.changedRows} removed-tag=${report.sanitize.byDiffType['removed-tag']} removed-attr=${report.sanitize.byDiffType['removed-attr']}`)
console.log(` timing: parse=${report.timing.parseMs.toFixed(1)}ms sanitize=${report.timing.sanitizeMs.toFixed(1)}ms total=${report.timing.totalMs.toFixed(1)}ms`)
for (const type of Object.keys(report.counts).sort()) {
const c = report.counts[type]!
console.log(` ${type}: created=${c.created} skipped=${c.skipped} failed=${c.failed} unsupported=${c.unsupported}`)
}
if (report.fatalError !== null) {
console.error(` fatalError: ${report.fatalError}`)
}
}
const program = new Command()
program
.name('sf')
.description('SiteForge CLI(阶段 3:来源输入登记、WXR 探测与导入 job 生命周期)')
.version('0.1.0')
program
.command('wxr:probe')
.description('只读诊断:WXR 全链解析(parse→map→sanitize)并出报告;不入库、不写业务存储,接受任意可读路径')
.requiredOption('--wxr <path>', 'WXR 文件路径(任意可读路径;损坏文件出报告不崩溃)')
.option('--out <dir>', '报告输出根目录', join(REPO_ROOT, 'artifacts', 'probe'))
.action(async (opts: { wxr: string; out: string }) => {
const wxrPath = resolve(opts.wxr)
const runId = makeRunId()
// Task 6 fatalError 契约:IO/损坏文件错误被 runWxrAdapterReport 收敛进报告——
// 这里永远能产出报告文件(createReadStream 的 ENOENT 同样走 fatalError 分支)
const report = await runWxrAdapterReport(createReadStream(wxrPath))
const dir = join(resolve(opts.out), runId)
await mkdir(dir, { recursive: true })
await writeFile(join(dir, 'report.json'), `${JSON.stringify(report, null, 2)}\n`, 'utf8')
await writeFile(join(dir, 'report.html'), `${renderReportHtml(report)}\n`, 'utf8')
console.log(`[wxr:probe] runId=${runId}`)
console.log(` input: ${wxrPath}`)
printProbeSummary(report)
console.log(` report: ${join(dir, 'report.json')}`)
if (report.fatalError !== null) process.exitCode = 1
})
program
.command('source:register')
.description('登记来源输入为 content-addressed 快照(WXR / sitemap / 媒体目录 / 媒体映射侧车 JSON,可多选)')
.requiredOption('--site <key>', '站点 key(不存在则自动创建)')
.option('--wxr <path>', 'WXR 文件路径')
.option('--sitemap <path>', 'sitemap 文件路径')
.option('--media <dir>', '媒体目录路径(已解包目录,递归登记)')
.option('--media-map <path>', '媒体映射侧车 JSON 路径([{sourceUrl, filePath}] 数组,登记为 manual_url_list 快照)')
.option('--allow <dir...>', '额外输入白名单根目录(可多个;默认含 storage/incoming 与当前目录)', [])
.option('--name <name>', '新建站点时的显示名(缺省 = site key)')
.option('--locale <locale>', '新建站点时的默认 locale(缺省 en)')
.option('--origin <origin>', '新建站点时的生产域名(缺省 https://<key>.example)')
.action(async (opts: {
site: string
wxr?: string
sitemap?: string
media?: string
mediaMap?: string
allow: string[]
name?: string
locale?: string
origin?: string
}) => {
if (
opts.wxr === undefined &&
opts.sitemap === undefined &&
opts.media === undefined &&
opts.mediaMap === undefined
) {
throw new SourceRegisterError('NO_INPUTS', '至少提供 --wxr / --sitemap / --media / --media-map 之一')
}
// fail fast:建站默认列在服务端只在"站点新建"时生效——先做形状校验,避免跑完摄取才报错
if (opts.origin !== undefined) {
let parsed: URL
try {
parsed = new URL(opts.origin)
} catch {
throw new Error(`--origin 不是合法 URL:${opts.origin}`)
}
if (parsed.protocol !== 'https:' && parsed.protocol !== 'http:') {
throw new Error(`--origin 必须是 http(s) URL:${opts.origin}`)
}
}
if (opts.locale !== undefined && !/^[A-Za-z]{2,8}(-[A-Za-z0-9]{1,8})*$/.test(opts.locale)) {
throw new Error(`--locale 不符合语言标签形状(如 en、zh-CN):${opts.locale}`)
}
// CLI 是可信操作者通道:白名单 = storage/incoming(常规通道)+ 操作者所在 cwd(显式
// 于此运行的动作本身即放行依据)+ --allow 追加根。服务内的白名单校验(PATH_NOT_ALLOWED)
// 防的是未来程序化/REST 通道的越界读取。生效白名单先行打印(审计:路径放行决策可回溯)。
const allowedRoots = [join(REPO_ROOT, 'storage', 'incoming'), process.cwd(), ...opts.allow.map((d) => resolve(d))]
const input: RegisterInput = {
siteKey: opts.site,
allowedRoots,
storageRoot: join(REPO_ROOT, 'storage'),
reportRoot: join(REPO_ROOT, 'artifacts', 'register'),
}
if (opts.wxr !== undefined) input.wxrPath = opts.wxr
if (opts.sitemap !== undefined) input.sitemapPath = opts.sitemap
if (opts.media !== undefined) input.mediaDir = opts.media
if (opts.mediaMap !== undefined) input.mediaMapPath = opts.mediaMap
const defaultsPassed = opts.name !== undefined || opts.locale !== undefined || opts.origin !== undefined
if (defaultsPassed) {
const defaults: NonNullable<RegisterInput['siteDefaults']> = {}
if (opts.name !== undefined) defaults.name = opts.name
if (opts.locale !== undefined) defaults.defaultLocale = opts.locale
if (opts.origin !== undefined) defaults.productionOrigin = opts.origin
input.siteDefaults = defaults
}
console.log(`[source:register] 允许目录:${allowedRoots.join(' ; ')}`)
// getDb 走 DATABASE_URL_TEST ?? DATABASE_URL 回退链——对临时库冒烟只需 env 覆盖,无需改代码
const db = getDb()
try {
const result = await registerSource(db, input)
console.log(`[source:register] runId=${result.runId} site=${result.siteKey} (${result.siteId}${result.siteCreated ? ' 新建' : ' 已存在'})`)
console.log(
` connection=${result.connectionId}${result.connectionCreated ? ' (created)' : ' (existing)'}`,
)
for (const s of result.snapshots) {
console.log(
` ${s.kind}: sha256=${s.sha256.slice(0, 12)}… bytes=${s.byteSize} ${s.reused ? 'snapshot复用' : 'snapshot新建'}${s.copied ? '' : ' 副本已存在'} (${s.durationMs.toFixed(1)}ms)`,
)
}
if (!result.siteCreated && defaultsPassed) {
console.error(' 注意:站点已存在,--name/--locale/--origin 未生效')
}
console.log(` report: ${result.reportPath} (total ${result.durationMs.toFixed(1)}ms)`)
} finally {
await (db as { $client: { end: () => Promise<void> } }).$client.end()
}
})
program
.command('import')
.description('导入已登记的 wxr 快照(Task 1 空管道骨架:job 生命周期 + 报告;真实管道 Task 6 接入)')
.requiredOption('--site <key>', '站点 key')
.option('--snapshot <id>', 'wxr 快照 id(缺省按 connection+kind=wxr 选最新)')
.option('--key <key>', '幂等键(缺省 = site+snapshot+版本块 自动派生;重复提交同 key 复用既有 job 不重跑)')
.option('--connection <name>', '连接名(快照解析的 connection 侧,缺省 default)', 'default')
.action(async (opts: { site: string; snapshot?: string; key?: string; connection: string }) => {
// getDb 走 DATABASE_URL_TEST ?? DATABASE_URL 回退链——临时库冒烟只需 env 覆盖
const db = getDb()
// exactOptionalPropertyTypes:缺席键不传 undefined 值(与 source:register 同款)
const input: CreateImportJobInput = { siteKey: opts.site, connectionName: opts.connection }
if (opts.snapshot !== undefined) input.snapshotId = opts.snapshot
if (opts.key !== undefined) input.idempotencyKey = opts.key
try {
const result = await createImportJob(db, input)
console.log(
`[import] job=${result.jobId} status=${result.status}${result.reused ? '(复用既有 job,未重跑)' : ''}`,
)
console.log(` site=${opts.site} mapping_version=${result.mappingVersion}`)
// 复用口径忠实化:快照/版本块只在新跑时解析返回——复用时权威记录在既有 job 报告。
// 非复用路径结构性填满这些可选字段(服务侧不变量),此处非空断言表达该约定
if (!result.reused) {
console.log(` connection=${result.connectionId!}`)
console.log(
` wxr 快照: ${result.wxrSnapshot!.sha256.slice(0, 12)}… (${result.wxrSnapshot!.id})`,
)
console.log(
` media_bundle: ${result.mediaBundleSnapshot! === null ? '(无)' : result.mediaBundleSnapshot!.id} / manual_url_list: ${result.manualUrlListSnapshot! === null ? '(无)' : result.manualUrlListSnapshot!.id}`,
)
console.log(
` 版本: adapter=${result.versions!.adapter} mapping=${result.versions!.mapping} sanitizer=${result.versions!.sanitizer} parser=${result.versions!.parser}`,
)
}
if (result.reportPath !== null) console.log(` report: ${result.reportPath}`)
console.log(` total ${result.durationMs.toFixed(1)}ms`)
// 崩溃残留哨兵:复用命中非终态(queued/running)= 上一次执行未走完终态更新——
// 本命令不自动重跑(幂等纪律),操作者显式换 --key 才发起新执行
if (result.reused && (result.status === 'queued' || result.status === 'running')) {
console.error(` 警告:job 处于 ${result.status}——可能为崩溃残留,重试请显式换 --key`)
}
if (result.status === 'failed') process.exitCode = 1
} finally {
await (db as { $client: { end: () => Promise<void> } }).$client.end()
}
})
// parseAsync:action 内抛错走此 catch(commander 默认会 process.exit(1) 并打印堆栈,
// 这里收敛为单行错误信息 + exitCode——SourceRegisterError/ImportError 打 code,其余原样呈现)
program.parseAsync(process.argv).catch((err: unknown) => {
if (err instanceof SourceRegisterError) {
console.error(`[sf] ${err.code}: ${err.message}`)
if (err.code === 'PATH_NOT_ALLOWED') {
console.error(' 提示:可用 --allow <dir> 放行额外目录,或将文件移入上述目录')
}
} else if (err instanceof ImportError) {
console.error(`[sf] ${err.code}: ${err.message}`)
} else {
console.error(`[sf] ${err instanceof Error ? err.message : String(err)}`)
}
process.exitCode = 1
})