Files
cms-website/scripts/sf.ts
T
浅念 0e8b2b5356 feat(import): job 骨架与生命周期
Plan 3 Task 1:sf import 命令骨架 + import_jobs 生命周期(空管道,Task 6 接真实管道)。

- import.service.ts(零装饰器):createImportJob——站点/快照解析(缺省按 (connection,
  kind='wxr') 选最新;三快照绑定按前置事实 13 记入报告,job 行不扩列)、幂等建行
  (默认 key=site+wxr快照+三语义版本;重复提交复用既有 job 不重跑,并发撞唯一约束
  重查收敛)、状态推进 queued→running→succeeded/failed(attempt/started_at/finished_at/
  error_code/error_summary)、报告落 artifacts/import/<jobId>/reports/import.json
  (全四版本块:adapter/mapping/sanitizer/parser;job 行落 mapping_version)
- 可预期错误带 code:SITE_NOT_FOUND / SNAPSHOT_NOT_FOUND / NO_WXR_SNAPSHOT /
  PIPELINE_FAILED——解析期失败零 job 行
- adapter 事件加法式补 postId(前置事实 12):unsupported/error 事件可选携带
  String(postId),缺失时键缺席不硬造 source_id;row 事件与报告聚合零影响、不触发版本 bump
- canonicalJson 提取至 apps/api/src/common/canonical.ts 共享导出(前置事实 6),
  register 侧改 import 零行为变更(既有清单哈希测试守护)
- sf CLI + import 命令(--site/--snapshot/--key/--connection),输出 job 摘要与版本块
- 测试:tests/integration/import-job.test.ts(5 例:生命周期观察/最新快照解析/幂等复用
  与并发收敛/可预期失败零落行/failed 路径)+ adapter postId 3 例 + canonical 单测 6 例
  (265→279 全绿);typecheck 四段链/lint 绿;临时库 CLI 冒烟(register→import→复用)通过

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

216 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=${result.siteKey} 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`)
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
})