feat(source): 媒体映射侧车导出与登记

Plan 3 Task 0(评审 Issue 1):旧 cms 库 media_assets 的 source_url→file_path
映射只读导出为侧车 JSON(859 条,覆盖率对照 WXR 855 attachment_url:精确命中
836 / 归一命中 0 / 未命中 19,97.8%),register 新增 --media-map 输入通道
(kind=manual_url_list 快照,形状校验 INVALID_MEDIA_MAP,metadata 含条目数,
幂等复用)。

Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
浅念
2026-10-21 14:42:32 +08:00
parent ceb6106025
commit a227cfac49
5 changed files with 515 additions and 14 deletions
+85 -10
View File
@@ -1,5 +1,6 @@
// source:register 输入登记服务(Plan 2 Task 7,spec §7.1 / ADR-004/008):
// 把 WXR / sitemap / 媒体目录三类输入登记为 content-addressed 快照——
// 把 WXR / sitemap / 媒体目录 / 媒体映射侧车(manual_url_list,Plan 3 Task 0)四类输入
// 登记为 content-addressed 快照——
// 副本落 storage/snapshots/<kind>/<sha256前2>/<sha256>(文件)或 <sha256>/(媒体整树),
// 源输入文件从此不再被后续任务读取(导入只寻址副本)。
//
@@ -25,7 +26,7 @@
// 新 uuid 重写)但不再被本服务清理——手工清理策略:CLI 未运行时整目录删除 .tmp/ 即可。
import { createHash, randomUUID } from 'node:crypto'
import { createReadStream, createWriteStream, existsSync } from 'node:fs'
import { lstat, mkdir, readdir, realpath, rename, rm, writeFile } from 'node:fs/promises'
import { lstat, mkdir, readdir, readFile, realpath, rename, rm, writeFile } from 'node:fs/promises'
import { basename, dirname, isAbsolute, join, relative, resolve } from 'node:path'
import { fileURLToPath } from 'node:url'
import { pipeline } from 'node:stream/promises'
@@ -49,13 +50,14 @@ export type SourceDb = NodePgDatabase<typeof schema>
*/
type SourceDbWriter = Pick<SourceDb, 'select' | 'insert' | 'execute'>
export type SourceSnapshotKind = 'wxr' | 'sitemap' | 'media_bundle'
export type SourceSnapshotKind = 'wxr' | 'sitemap' | 'media_bundle' | 'manual_url_list'
/** 各 kind 快照的 adapter_version:wxr 用适配器包单源版本;sitemap/media 尚无解析器,占位钉死 */
/** 各 kind 快照的 adapter_version:wxr 用适配器包单源版本;sitemap/media/media_map 尚无解析器,占位钉死 */
const SNAPSHOT_ADAPTER_VERSION: Record<SourceSnapshotKind, string> = {
wxr: `adapter-wp@${ADAPTER_VERSION}`,
sitemap: 'sitemap@0',
media_bundle: 'media-bundle@1',
manual_url_list: 'manual-url-list@1',
}
export type SourceRegisterErrorCode =
@@ -66,6 +68,7 @@ export type SourceRegisterErrorCode =
| 'INPUT_NOT_DIR'
| 'EMPTY_MEDIA_DIR'
| 'INPUT_READ_FAILED'
| 'INVALID_MEDIA_MAP'
| 'COPY_FAILED'
/** 登记失败(可预期错误):CLI 以 code 区分呈现;不裹程序 bug(那些原样上抛) */
@@ -89,6 +92,12 @@ export interface RegisterInput {
wxrPath?: string
sitemapPath?: string
mediaDir?: string
/**
* 媒体映射侧车 JSON(Plan 3 Task 0,评审 Issue 1):[{sourceUrl, filePath}] 数组——
* 旧 cms 库 media_assets 导出的 URL→扁平文件名映射;登记为 kind=manual_url_list 快照。
* 导入管道对拍三级中的第二级数据源(①清单 uploads 路径 → ②本侧车 → ③missing 报告)。
*/
mediaMapPath?: string
/** 输入路径白名单(realpath 后必须位于其中之一);缺省 [repoRoot/storage/incoming] */
allowedRoots?: string[]
/** 快照存储根(副本落 <storageRoot>/snapshots/...);缺省 repoRoot/storage */
@@ -256,9 +265,49 @@ interface Ingested {
durationMs: number
}
/** 文件输入(wxr/sitemap):temp 副本上单遍 copy+hash → 按 sha 定名落位(已存在则命中复用) */
/**
* 校验 media-map 侧车 JSON 形状并数条目:[{sourceUrl: string, filePath: string}] 数组,
* 每条两字段均为非空字符串。形状即契约(Plan 3 对拍二级的数据源)——脏输入在登记期
* 拒绝(快照不可变,落库后再发现无法补救)。返回条目数入 metadata。
*/
async function validateMediaMap(path: string): Promise<number> {
let parsed: unknown
try {
parsed = JSON.parse(await readFile(path, 'utf8'))
} catch (err) {
throw new SourceRegisterError(
'INVALID_MEDIA_MAP',
`media-map 不是合法 JSON:${path}(${err instanceof Error ? err.message : String(err)})`,
)
}
if (!Array.isArray(parsed)) {
throw new SourceRegisterError(
'INVALID_MEDIA_MAP',
`media-map 顶层必须是数组([{sourceUrl, filePath}]):${path},实际 ${parsed === null ? 'null' : typeof parsed}`,
)
}
for (let i = 0; i < parsed.length; i++) {
const entry = parsed[i] as unknown
const ok =
entry !== null &&
typeof entry === 'object' &&
typeof (entry as { sourceUrl?: unknown }).sourceUrl === 'string' &&
(entry as { sourceUrl: string }).sourceUrl !== '' &&
typeof (entry as { filePath?: unknown }).filePath === 'string' &&
(entry as { filePath: string }).filePath !== ''
if (!ok) {
throw new SourceRegisterError(
'INVALID_MEDIA_MAP',
`media-map 第 ${i} 条不满足形状 {sourceUrl: string, filePath: string}:${path}`,
)
}
}
return parsed.length
}
/** 文件输入(wxr/sitemap/media-map):temp 副本上单遍 copy+hash → 按 sha 定名落位(已存在则命中复用) */
async function ingestFile(
kind: 'wxr' | 'sitemap',
kind: 'wxr' | 'sitemap' | 'manual_url_list',
src: string,
storageRoot: string,
): Promise<Ingested> {
@@ -276,6 +325,17 @@ async function ingestFile(
`${kind} 输入读取失败:${src}(${err instanceof Error ? err.message : String(err)})`,
)
}
// media-map:形状校验在副本 rename 落位**之前**——脏输入不产 content-addressed 副本
//(校验读 temp 副本而非源,登记窗口内源被改写不会使校验结果与哈希脱钩)
let entryCount: number | undefined
if (kind === 'manual_url_list') {
try {
entryCount = await validateMediaMap(tmp)
} catch (err) {
await rm(tmp, { force: true })
throw err
}
}
const rel = `snapshots/${kind}/${sha256.slice(0, 2)}/${sha256}`
let copied: boolean
try {
@@ -286,6 +346,8 @@ async function ingestFile(
`${kind} 快照副本落位失败:${join(storageRoot, rel)}(${err instanceof Error ? err.message : String(err)})`,
)
}
const metadata: Record<string, unknown> = { sourceName: basename(src) }
if (entryCount !== undefined) metadata.entryCount = entryCount
return {
kind,
sourcePath: src,
@@ -293,7 +355,7 @@ async function ingestFile(
byteSize: bytes,
storagePath: rel,
copied,
metadata: { sourceName: basename(src) },
metadata,
durationMs: performance.now() - t0,
}
}
@@ -578,20 +640,33 @@ export async function registerSource(db: SourceDb, input: RegisterInput): Promis
let wxrReal: string | undefined
let sitemapReal: string | undefined
let mediaReal: string | undefined
let mediaMapReal: string | undefined
if (input.wxrPath !== undefined) wxrReal = await validateInputPath(input.wxrPath, 'file', allowedRoots, 'wxr')
if (input.sitemapPath !== undefined)
sitemapReal = await validateInputPath(input.sitemapPath, 'file', allowedRoots, 'sitemap')
if (input.mediaDir !== undefined)
mediaReal = await validateInputPath(input.mediaDir, 'directory', allowedRoots, 'media')
if (wxrReal === undefined && sitemapReal === undefined && mediaReal === undefined) {
throw new SourceRegisterError('NO_INPUTS', '未提供任何输入(wxrPath / sitemapPath / mediaDir 至少其一)')
if (input.mediaMapPath !== undefined)
mediaMapReal = await validateInputPath(input.mediaMapPath, 'file', allowedRoots, 'media-map')
if (
wxrReal === undefined &&
sitemapReal === undefined &&
mediaReal === undefined &&
mediaMapReal === undefined
) {
throw new SourceRegisterError(
'NO_INPUTS',
'未提供任何输入(wxrPath / sitemapPath / mediaDir / mediaMapPath 至少其一)',
)
}
// 阶段 1:摄取(hash + 副本落位)——顺序固定 wxr → sitemap → media(报告输入清单稳定序)
// 阶段 1:摄取(hash + 副本落位)——顺序固定 wxr → sitemap → media → media-map(报告输入清单稳定序)
const ingested: Ingested[] = []
if (wxrReal !== undefined) ingested.push(await ingestFile('wxr', wxrReal, storageRoot))
if (sitemapReal !== undefined) ingested.push(await ingestFile('sitemap', sitemapReal, storageRoot))
if (mediaReal !== undefined) ingested.push(await ingestMediaDir(mediaReal, storageRoot))
if (mediaMapReal !== undefined)
ingested.push(await ingestFile('manual_url_list', mediaMapReal, storageRoot))
// 阶段 2:DB 写入——单事务(site→connection→snapshots 全成或全回滚;"任何失败零落库"
// 由此成立)。connection 的 select-then-insert 在事务内靠 advisory xact lock 串行化。