feat(import): 媒体登记与安全校验
Plan 3 Task 4：media-writer（attachment 行三级对拍 + 内容行引用提取双事件消费） + magic-bytes 手写签名表（零新增依赖）。 - 三级对拍（前置事实 7）：①uploads 相对路径清单精确命中 ②侧车 source_url→ file_path→清单实体（精确 miss 按文件名候选，多候选歧义不自动选全列 issue） ③missing 只进报告不插行（sha256 NOT NULL 库层防线） - 魔数校验：命中实体读存储副本头部实际字节——SVG 检出 status=quarantined 不进 产物语义（引用不建 usage）；未知签名 unsupported；扩展名声明 MIME 与检测不符 → issue（按检测登记） - usage 两阶段时序（WXR 文档序 items 先于 attachment）：内容行提取引用 + 修订 锚点（依赖 content-writer 先处理同事件——Task 6 编排顺序契约），finish 统一 对 media_assets 反查（id→source_id / URL→source_url→public_path），顺序无关 - 悬空引用计数 mediaDanglingRefs 不静默（mini 203/204 钉死） - counts 六键 media 前缀独占（勿复用 'skipped' 的键契约注记） - 幂等：(site,source_url) 定向 onConflictDoNothing + usage 复合主键——重跑零增行 308 测试（296+6 单测+6 集成）；typecheck/lint 三绿。 Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
浅念
@@ -0,0 +1,49 @@
|
||||
// 手写魔数签名表（Plan 3 Task 4，spec §11.1、计划"Tech Stack：零新增依赖"）：
|
||||
// 常见六类媒体格式判别 + SVG 文本特征检出。刻意不引 file-type 依赖——导入管道的
|
||||
// 安全决策只须回答两问："是什么格式"与"是否 SVG（隔离）"，一张可审计的小签名表
|
||||
// 足够，且避免为一次头部读取引入整个 MIME 数据库。
|
||||
//
|
||||
// 判别语义（detectMediaType）：
|
||||
// - 命中二进制签名 → {type: MIME, svg: false}（PNG/JPEG/GIF87a/GIF89a/WEBP/PDF）；
|
||||
// - 无二进制命中但前 1KB 文本窗口含 "<svg"（大小写不敏感）→ {type: 'image/svg+xml',
|
||||
// svg: true}——调用方据此 status='quarantined'（spec §11.1：SVG 可携带脚本与外部
|
||||
// 引用，不进产物语义）；
|
||||
// - 均未命中 → null（未知签名——调用方 status='unsupported' + issue，不猜格式）。
|
||||
//
|
||||
// 顺序纪律：二进制签名先于 SVG 文本扫描——PNG/JPEG 正例即便尾部字节碰巧含 "<svg"
|
||||
// 片段也按二进制格式判（文本窗口只对非二进制文件有意义）。窗口 1KB 是钉死的折中：
|
||||
// 覆盖常规 "<?xml …?><svg" 前导（声明/注释/doctype），不给深藏脚本的巨型前缀放行。
|
||||
//
|
||||
// Buffer 读取口径：latin1 单字节解码后小写化做包含检测——不做 UTF-8 多字节解码，
|
||||
// 避免截断的字节序列被解码成意外字符。
|
||||
/** 二进制签名表（前缀匹配；WEBP 的 RIFF+WEBP@8 双段结构在表外单查） */
|
||||
const SIGNATURES: ReadonlyArray<{ mime: string; bytes: Buffer }> = Object.freeze([
|
||||
{ mime: 'image/png', bytes: Buffer.from([0x89, 0x50, 0x4e, 0x47]) },
|
||||
{ mime: 'image/jpeg', bytes: Buffer.from([0xff, 0xd8, 0xff]) },
|
||||
{ mime: 'image/gif', bytes: Buffer.from('GIF87a', 'latin1') },
|
||||
{ mime: 'image/gif', bytes: Buffer.from('GIF89a', 'latin1') },
|
||||
{ mime: 'application/pdf', bytes: Buffer.from('%PDF', 'latin1') },
|
||||
])
|
||||
|
||||
const RIFF = Buffer.from('RIFF', 'latin1')
|
||||
const WEBP = Buffer.from('WEBP', 'latin1')
|
||||
const SVG_WINDOW_BYTES = 1024
|
||||
|
||||
/**
|
||||
* 魔数检测：读文件头部字节判别媒体类型。返回 null = 未知签名（调用方按
|
||||
* unsupported 处置）；svg=true 仅出现在 type='image/svg+xml'（文本特征命中）。
|
||||
*/
|
||||
export function detectMediaType(buf: Buffer): { type: string; svg: boolean } | null {
|
||||
for (const sig of SIGNATURES) {
|
||||
if (buf.length >= sig.bytes.length && buf.subarray(0, sig.bytes.length).equals(sig.bytes)) {
|
||||
return { type: sig.mime, svg: false }
|
||||
}
|
||||
}
|
||||
// WEBP：RIFF 容器头（4B 长度）+ 'WEBP' Magic（@8）——RIFF 家族其他成员（WAV/AVI）不误报
|
||||
if (buf.length >= 12 && buf.subarray(0, 4).equals(RIFF) && buf.subarray(8, 12).equals(WEBP)) {
|
||||
return { type: 'image/webp', svg: false }
|
||||
}
|
||||
const head = buf.subarray(0, Math.min(buf.length, SVG_WINDOW_BYTES)).toString('latin1').toLowerCase()
|
||||
if (head.includes('<svg')) return { type: 'image/svg+xml', svg: true }
|
||||
return null
|
||||
}
|
||||
@@ -0,0 +1,617 @@
|
||||
// 媒体登记写入器（Plan 3 Task 4，spec §5.5/§11.1、前置事实 7/11、评审 Issue 1/3）：
|
||||
// 消费适配器事件流的**双事件**面——attachment 行建资产（media_assets）、内容行提取引用
|
||||
// （media_usages）——媒体域的领域 owner（Task 6 分派规则：content-writer 对 attachment 行
|
||||
// 只留 source_records 足迹，media_assets 的唯一写入方在此）。
|
||||
//
|
||||
// 三级对拍（前置事实 7 钉死，真实数据实证：媒体目录扁平 GUID 命名、与 WXR
|
||||
// attachment_url 零精确命中，唯一映射在旧 cms 库侧车）：
|
||||
// ① uploads 相对路径清单精确命中：attachment_url 掐掉
|
||||
// '…/wp-content/uploads/' 前缀 → manifest[path]（清单 = media_bundle 快照 metadata.
|
||||
// manifest，{path, bytes, sha256}）；
|
||||
// ② 侧车命中：manifest miss → sidecarMap[source_url] → filePath → 清单实体（精确路径
|
||||
// miss 时按**文件名**候选——真实侧车 file_path 是扁平文件名而清单可含子目录；候选
|
||||
// 多条 = 歧义，不自动选、全部列 issue、不插行，spec §5.5）；
|
||||
// ③ missing：两级皆 miss → 只进报告（**不插 media_assets 行**——sha256 NOT NULL 无清单
|
||||
// 实体可取，行插不进是库层防线不是巧合）。
|
||||
// 命中行取值：public_path = 旧 uploads 路径（保旧 URL 语义）、sha256/byte_size 从清单、
|
||||
// storage_path = bundle 副本相对路径 + 清单 path（绝对副本 = metadata.storageRoot + 此列）。
|
||||
//
|
||||
// 魔数安全校验（spec §11.1）：命中实体读存储副本头部实际字节（detectMediaType）——
|
||||
// - SVG 检出（文本特征）→ status='quarantined'：行存在（登记可审计）但不进产物语义，
|
||||
// 其引用不建 media_usages（usage 是发布面消费的引用关系，隔离资产不得经此进产物）；
|
||||
// - 未知签名 → status='unsupported' + issue（不猜格式）；
|
||||
// - 声明 MIME（清单/WXR 均无 MIME——声明侧由 URL 扩展名推断）与检测不符 → issue
|
||||
// （仍登记 ready：内容真实、扩展名说谎是数据质量问题不是安全风险）。
|
||||
//
|
||||
// usage 时序（设计决定，WXR 文档序钉死）：items 先于 attachment——引用先行、资产后到，
|
||||
// 故 usage 建立分两阶段：内容行到达时提取引用并落修订锚点（source_records 六列身份 →
|
||||
// entry_locales.current_revision_id——**依赖 content-writer 先处理同事件**，Task 6 编排
|
||||
// 顺序契约），资产解析统一推迟到 finish() 对 media_assets 反查（id 反查 source_id /
|
||||
// URL 反查 source_url→public_path）——两种文档序终态一致。引用无对应资产且属本站媒体域
|
||||
// （id 引用或 uploads 路径 URL）→ mediaDanglingRefs 计数 + issue（§7.6 禁静默）；外域
|
||||
// 热链 URL 不属本域对拍范围，不计悬空（无清单可比对，静默越过）。
|
||||
//
|
||||
// 幂等：media_assets 唯一 (site, source_url) onConflictDoNothing（重跑零增行）；
|
||||
// media_usages 复合主键 (asset, revision, field_path) onConflictDoNothing——field_path
|
||||
// 由提取序确定（featured/gallery 序号/正文 img 序号），重跑同流同值。事务边界：全部
|
||||
// 单语句原子操作（ctx.db 直连约定，无跨语句不变量——与 term-writer 同纪律）。
|
||||
import { open, readFile } from 'node:fs/promises'
|
||||
import { join } from 'node:path'
|
||||
import { and, eq } from 'drizzle-orm'
|
||||
import type { AdapterEvent } from '@sf/adapter-wp'
|
||||
import type { UniversalRow } from '@sf/contracts'
|
||||
import { entryLocales, mediaAssets, mediaUsages, sourceRecords } from '../../db/schema.js'
|
||||
import { isUniqueViolation, type MediaManifestEntry } from '../source/register.service.js'
|
||||
import { detectMediaType } from './magic-bytes.js'
|
||||
import type { ImportDb, ImportJobContext, ImportPipeline, ImportPipelineOutcome } from './import.service.js'
|
||||
|
||||
const UPLOADS_MARKER = '/wp-content/uploads/'
|
||||
|
||||
/** 未知签名的 detected_mime 落值（列 NOT NULL——不猜格式但要有确定值） */
|
||||
const UNKNOWN_DETECTED_MIME = 'application/octet-stream'
|
||||
|
||||
/** 魔数读取的头部字节（SVG 文本窗口 1KB + 余量） */
|
||||
const HEAD_BYTES = 4096
|
||||
|
||||
/** URL 扩展名 → 声明 MIME（声明侧推断表——与 detectMediaType 的检出表独立） */
|
||||
const EXT_MIME: Readonly<Record<string, string>> = Object.freeze({
|
||||
png: 'image/png',
|
||||
jpg: 'image/jpeg',
|
||||
jpeg: 'image/jpeg',
|
||||
gif: 'image/gif',
|
||||
webp: 'image/webp',
|
||||
pdf: 'application/pdf',
|
||||
svg: 'image/svg+xml',
|
||||
})
|
||||
|
||||
/**
|
||||
* 可分派的媒体写入器句柄（Task 6 编排形状，与 term/content 同构——但 finish 是异步：
|
||||
* usage 解析推迟到流终，见模块头"usage 时序"）。句柄非可重入：一个句柄只服务一遍流。
|
||||
*/
|
||||
export interface MediaEventHandler {
|
||||
/** 单事件分派（term/unsupported/error 为 no-op——无媒体信息） */
|
||||
handleEvent(ev: AdapterEvent): Promise<void>
|
||||
/** 流终：解析悬置引用 → media_usages / 悬空报告，取回累计产出（调用后句柄弃用） */
|
||||
finish(): Promise<ImportPipelineOutcome>
|
||||
}
|
||||
|
||||
/**
|
||||
* 本写入器的 counts 键与归属（**counts 键是跨写入器的扁平全局命名空间**——import.service
|
||||
* 按返回值覆盖累加器同名键合并，各写入器必须各持独占键名，见 ImportPipelineOutcome.counts
|
||||
* 的键契约注记）。本写入器独占六键（media 前缀，**勿复用 content-writer 的 'skipped'**）：
|
||||
* - mediaRegistered：本运行**新插**的 media_assets 行数（含 quarantined/unsupported——
|
||||
* 行已登记；幂等重跑为 0）；
|
||||
* - mediaFromSidecar：其中经级②侧车解析的行数（mediaRegistered 的子集）；
|
||||
* - mediaMissing：级③ missing 的 attachment 数（按事件计——重跑如实再现）；
|
||||
* - mediaQuarantined：本运行检出 SVG 的 attachment 数（按事件计——重跑如实再现，
|
||||
* 与 mediaRegistered 在重跑时不再严格子集：行幂等而检出如实）；
|
||||
* - mediaDanglingRefs：内容引用未解析到资产的条数（按引用计——重跑如实再现）；
|
||||
* - mediaAmbiguous：多候选歧义的 attachment 数（按事件计，不插行）。
|
||||
* issues：missing/歧义/隔离/未知签名/MIME 不符/悬空引用/副本读取失败/public_path 唯一键
|
||||
* 冲突——全部显式报告（§7.6 禁静默）。
|
||||
*/
|
||||
interface MediaWriterCounts {
|
||||
mediaRegistered: number
|
||||
mediaFromSidecar: number
|
||||
mediaMissing: number
|
||||
mediaQuarantined: number
|
||||
mediaDanglingRefs: number
|
||||
mediaAmbiguous: number
|
||||
}
|
||||
|
||||
/** 悬置引用（内容行提取、流终解析）：锚点 + 引用键两半 */
|
||||
interface PendingRef {
|
||||
/** 引用方修订（content-writer 落的 current_revision_id——usage 的不可变锚） */
|
||||
revisionId: string
|
||||
/** 修订数据域内的字段位（同流重跑稳定——复合主键的 field_path 侧） */
|
||||
fieldPath: string
|
||||
kind: 'id' | 'url'
|
||||
/** id 引用 = attachment post id；URL 引用 = 引文原串 */
|
||||
value: string
|
||||
/** URL 引用属本站 uploads 域时的公开路径形态（悬空裁决域）；id 引用恒属本域 */
|
||||
uploadsPublicPath: string | null
|
||||
rawLocation: string
|
||||
}
|
||||
|
||||
/** 侧车 JSON 条目形状（register --media-map 已在登记期校验，此处只做读取防御） */
|
||||
interface SidecarEntry {
|
||||
sourceUrl: string
|
||||
filePath: string
|
||||
}
|
||||
|
||||
export function createMediaEventHandler(identity: {
|
||||
db: ImportDb
|
||||
siteId: string
|
||||
connectionId: string
|
||||
snapshotId: string
|
||||
/** 引用行 locale 校验基准（与 content-writer 同拒收口径——不双报） */
|
||||
defaultLocale: string
|
||||
/** 存储副本绝对根（media_bundle 快照 metadata.storageRoot——登记期事实） */
|
||||
storageRoot: string
|
||||
/** media_bundle 快照的 storagePath（相对 storageRoot；直传清单场景传 ''） */
|
||||
mediaBundleStoragePath: string
|
||||
/** 媒体清单（media_bundle metadata.manifest；缺席 = 空表 → 全部走级③） */
|
||||
mediaManifest: MediaManifestEntry[]
|
||||
/** 侧车映射（manual_url_list：source_url → file_path） */
|
||||
sidecarMap: Map<string, string>
|
||||
}): MediaEventHandler {
|
||||
const { db, siteId } = identity
|
||||
const counts: MediaWriterCounts & Record<string, number> = {
|
||||
mediaRegistered: 0,
|
||||
mediaFromSidecar: 0,
|
||||
mediaMissing: 0,
|
||||
mediaQuarantined: 0,
|
||||
mediaDanglingRefs: 0,
|
||||
mediaAmbiguous: 0,
|
||||
}
|
||||
const issues: string[] = []
|
||||
const pendingRefs: PendingRef[] = []
|
||||
|
||||
// 清单索引：精确路径 + 文件名候选（级②的侧车 file_path 是扁平名，清单可含子目录）
|
||||
const manifestByPath = new Map<string, MediaManifestEntry>()
|
||||
const manifestByBasename = new Map<string, MediaManifestEntry[]>()
|
||||
for (const entry of identity.mediaManifest) {
|
||||
manifestByPath.set(entry.path, entry)
|
||||
const base = basenameOf(entry.path)
|
||||
const bucket = manifestByBasename.get(base)
|
||||
if (bucket === undefined) manifestByBasename.set(base, [entry])
|
||||
else bucket.push(entry)
|
||||
}
|
||||
|
||||
function basenameOf(path: string): string {
|
||||
const idx = path.lastIndexOf('/')
|
||||
return idx === -1 ? path : path.slice(idx + 1)
|
||||
}
|
||||
|
||||
/** 扩展名推断声明 MIME（无扩展名/表外扩展 → null：列可空，无声明则无对拍） */
|
||||
function declaredMimeOf(pathLike: string): string | null {
|
||||
const m = /\.([a-z0-9]+)$/i.exec(pathLike)
|
||||
if (m === null) return null
|
||||
const hit = EXT_MIME[m[1]!.toLowerCase()]
|
||||
return hit ?? null
|
||||
}
|
||||
|
||||
/**
|
||||
* attachment_url → 旧 uploads 相对路径。无 uploads 标记的 URL（外站迁移源）级①无从
|
||||
* 掐前缀：relPath=null（级①跳过），source/public 路径以 URL pathname 兜底。
|
||||
* 注：不做百分号解码——清单 path 是字面文件名，编码差异按 miss 如实进报告。
|
||||
*/
|
||||
function uploadsRelOf(fileUrl: string): { relPath: string | null; pathname: string } {
|
||||
const idx = fileUrl.indexOf(UPLOADS_MARKER)
|
||||
if (idx !== -1) {
|
||||
const rel = fileUrl.slice(idx + UPLOADS_MARKER.length)
|
||||
if (rel !== '') return { relPath: rel, pathname: fileUrl.slice(idx) }
|
||||
}
|
||||
try {
|
||||
return { relPath: null, pathname: new URL(fileUrl).pathname }
|
||||
} catch {
|
||||
// 非 URL 形态的原始串： pathname 兜底保留原文（可审计），级①不可用
|
||||
return { relPath: null, pathname: fileUrl.startsWith('/') ? fileUrl : `/${fileUrl}` }
|
||||
}
|
||||
}
|
||||
|
||||
/** 存储副本头部读取（魔数只需前若干字节——171MB 级文件不做全量读） */
|
||||
async function readHead(abs: string): Promise<Buffer> {
|
||||
const handle = await open(abs, 'r')
|
||||
try {
|
||||
const buf = Buffer.alloc(HEAD_BYTES)
|
||||
const { bytesRead } = await handle.read(buf, 0, HEAD_BYTES, 0)
|
||||
return buf.subarray(0, bytesRead)
|
||||
} finally {
|
||||
await handle.close()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* attachment 行主路径：三级解析 → 魔数 → 插 media_assets。
|
||||
* locale 拒收口径与 content-writer 一致（该行已被 content-writer 拒收并记 issue，
|
||||
* 此处静默跳过不双报——attachment 行的媒体域处理同样以 locale 合法为前提）。
|
||||
*/
|
||||
async function handleAttachmentRow(row: UniversalRow): Promise<void> {
|
||||
if (row.locale !== identity.defaultLocale) return
|
||||
const fileUrl = typeof row.knownFields.attachmentUrl === 'string' ? row.knownFields.attachmentUrl : ''
|
||||
if (fileUrl === '') {
|
||||
counts.mediaMissing++
|
||||
issues.push(
|
||||
`attachment 缺 attachment_url（${row.sourceType}/${row.sourceId}，${row.rawLocation}）——三级对拍无从下手，按 missing 报告`,
|
||||
)
|
||||
return
|
||||
}
|
||||
const { relPath, pathname } = uploadsRelOf(fileUrl)
|
||||
|
||||
// 级①：uploads 相对路径清单精确命中
|
||||
let entry: MediaManifestEntry | null = relPath === null ? null : manifestByPath.get(relPath) ?? null
|
||||
let fromSidecar = false
|
||||
if (entry === null) {
|
||||
// 级②：侧车 source_url → file_path → 清单实体（精确路径 → 文件名候选）
|
||||
const filePath = identity.sidecarMap.get(fileUrl)
|
||||
if (filePath !== undefined) {
|
||||
const exact = manifestByPath.get(filePath)
|
||||
if (exact !== undefined) {
|
||||
entry = exact
|
||||
fromSidecar = true
|
||||
} else {
|
||||
const candidates = manifestByBasename.get(basenameOf(filePath)) ?? []
|
||||
if (candidates.length > 1) {
|
||||
// 歧义不自动选（spec §5.5）：全部候选列报告，不插行
|
||||
counts.mediaAmbiguous++
|
||||
issues.push(
|
||||
`媒体多候选歧义（${row.sourceType}/${row.sourceId}，${fileUrl}）：侧车 file_path=${filePath} 按文件名命中 ${candidates.length} 条清单实体——[${candidates.map((c) => c.path).join('、')}]——不自动选择，未登记`,
|
||||
)
|
||||
return
|
||||
}
|
||||
if (candidates.length === 1) {
|
||||
entry = candidates[0]!
|
||||
fromSidecar = true
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if (entry === null) {
|
||||
// 级③：missing——只进报告不插行（sha256 NOT NULL：无清单实体插不进，库层防线）
|
||||
counts.mediaMissing++
|
||||
issues.push(
|
||||
`媒体 missing（${row.sourceType}/${row.sourceId}，${fileUrl}）：清单与侧车均未命中——不插 media_assets 行（sha256 无来源）`,
|
||||
)
|
||||
return
|
||||
}
|
||||
|
||||
// 魔数：读存储副本实际字节（storageRoot + bundle 相对路径 + 清单 path）
|
||||
const storagePath =
|
||||
identity.mediaBundleStoragePath === ''
|
||||
? entry.path
|
||||
: `${identity.mediaBundleStoragePath}/${entry.path}`
|
||||
let head: Buffer
|
||||
try {
|
||||
head = await readHead(join(identity.storageRoot, storagePath))
|
||||
} catch (err) {
|
||||
issues.push(
|
||||
`媒体副本读取失败（${row.sourceType}/${row.sourceId}，${fileUrl} → ${storagePath}）：${err instanceof Error ? err.message : String(err)}——清单声称存在而副本缺席，未登记`,
|
||||
)
|
||||
return
|
||||
}
|
||||
const detected = detectMediaType(head)
|
||||
const publicPath = relPath === null ? pathname : `${UPLOADS_MARKER}${relPath}`
|
||||
const declared = declaredMimeOf(publicPath)
|
||||
|
||||
let status: 'ready' | 'quarantined' | 'unsupported'
|
||||
let detectedMime: string
|
||||
if (detected === null) {
|
||||
status = 'unsupported'
|
||||
detectedMime = UNKNOWN_DETECTED_MIME
|
||||
issues.push(
|
||||
`媒体未知签名（${row.sourceType}/${row.sourceId}，${publicPath}）：头部字节不匹配签名表——status=unsupported，不猜格式`,
|
||||
)
|
||||
} else if (detected.svg) {
|
||||
status = 'quarantined'
|
||||
detectedMime = detected.type
|
||||
counts.mediaQuarantined++
|
||||
issues.push(
|
||||
`SVG 检出已隔离（${row.sourceType}/${row.sourceId}，${publicPath}）：文本 <svg 特征命中——status=quarantined，不进产物语义（spec §11.1）`,
|
||||
)
|
||||
} else {
|
||||
status = 'ready'
|
||||
detectedMime = detected.type
|
||||
if (declared !== null && declared !== detected.type) {
|
||||
issues.push(
|
||||
`媒体声明与检测不符（${row.sourceType}/${row.sourceId}，${publicPath}）：扩展名推断 ${declared} ≠ 魔数检测 ${detected.type}——按检测登记（ready），数据质量问题已报告`,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
try {
|
||||
const inserted = await db
|
||||
.insert(mediaAssets)
|
||||
.values({
|
||||
siteId,
|
||||
sourceConnectionId: identity.connectionId,
|
||||
sourceId: row.sourceId,
|
||||
sourceUrl: fileUrl,
|
||||
sourcePath: relPath ?? pathname.replace(/^\//, ''),
|
||||
storagePath,
|
||||
publicPath,
|
||||
sha256: entry.sha256,
|
||||
detectedMimeType: detectedMime,
|
||||
declaredMimeType: declared,
|
||||
byteSize: entry.bytes,
|
||||
status,
|
||||
})
|
||||
// 唯一 (site, source_url) 定向幂等：命中零增行；public_path 唯一键冲突**不放行**
|
||||
// （同公开路径双源是真实冲突，须显式报告而非静默吞掉）
|
||||
.onConflictDoNothing({ target: [mediaAssets.siteId, mediaAssets.sourceUrl] })
|
||||
.returning({ id: mediaAssets.id })
|
||||
if (inserted.length > 0) {
|
||||
counts.mediaRegistered++
|
||||
if (fromSidecar) counts.mediaFromSidecar++
|
||||
}
|
||||
} catch (err) {
|
||||
if (isUniqueViolation(err)) {
|
||||
issues.push(
|
||||
`媒体 public_path 冲突（${row.sourceType}/${row.sourceId}，${publicPath}）：同站公开路径已被他行占用——未登记（不静默覆盖）`,
|
||||
)
|
||||
return
|
||||
}
|
||||
throw err
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 内容行引用提取（usage 两阶段的第一阶段）：refs + bodyHtml img src → 悬置引用。
|
||||
* featured/gallery 取 id 锚（WP meta 的 _thumbnail_id/_product_image_gallery）；未知
|
||||
* role 以 refs 原始下标落位（field_path 稳定优先于语义命名）。
|
||||
*/
|
||||
function extractRefs(row: UniversalRow): Array<Omit<PendingRef, 'revisionId'>> {
|
||||
const out: Array<Omit<PendingRef, 'revisionId'>> = []
|
||||
let galleryOrdinal = 0
|
||||
row.attachmentRefs.forEach((ref, i) => {
|
||||
let fieldPath: string
|
||||
if (ref.role === 'featured') fieldPath = 'attachmentRefs.featured'
|
||||
else if (ref.role === 'gallery') fieldPath = `attachmentRefs.gallery[${galleryOrdinal++}]`
|
||||
else fieldPath = `attachmentRefs.raw[${i}]`
|
||||
if (ref.sourceId !== undefined && ref.sourceId !== '') {
|
||||
out.push({ fieldPath, kind: 'id', value: ref.sourceId, uploadsPublicPath: null, rawLocation: row.rawLocation })
|
||||
} else if (ref.sourceUrl !== undefined && ref.sourceUrl !== '') {
|
||||
out.push({
|
||||
fieldPath,
|
||||
kind: 'url',
|
||||
value: ref.sourceUrl,
|
||||
uploadsPublicPath: uploadsPublicPathOf(ref.sourceUrl),
|
||||
rawLocation: row.rawLocation,
|
||||
})
|
||||
}
|
||||
// 双锚皆空：adapter 层已按非法 refs 记 issue（事件 issues 面），此处不重复报
|
||||
})
|
||||
const body = row.bodyHtml ?? ''
|
||||
if (body !== '') {
|
||||
const imgRe = /<img\b[^>]*?\bsrc\s*=\s*["']([^"']+)["']/gi
|
||||
let m: RegExpExecArray | null
|
||||
let imgOrdinal = 0
|
||||
while ((m = imgRe.exec(body)) !== null) {
|
||||
const src = m[1]!
|
||||
if (!src.startsWith('data:')) {
|
||||
// data: URI 是内联数据非资产引用；其余（绝对/相对/协议相对）统一进解析
|
||||
out.push({
|
||||
fieldPath: `data.bodyHtml.img[${imgOrdinal++}]`,
|
||||
kind: 'url',
|
||||
value: src,
|
||||
uploadsPublicPath: uploadsPublicPathOf(src),
|
||||
rawLocation: row.rawLocation,
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
/** URL → 本站 uploads 公开路径形态（无标记 → null：外域热链不属悬空裁决域） */
|
||||
function uploadsPublicPathOf(url: string): string | null {
|
||||
const idx = url.indexOf(UPLOADS_MARKER)
|
||||
return idx === -1 ? null : url.slice(idx)
|
||||
}
|
||||
|
||||
/** 修订锚点：source_records 六列身份 → entry_locales.current_revision_id（content-writer 接缝） */
|
||||
async function resolveRevisionAnchor(row: UniversalRow): Promise<string | null> {
|
||||
const recs = await db
|
||||
.select({ contentEntryId: sourceRecords.contentEntryId })
|
||||
.from(sourceRecords)
|
||||
.where(
|
||||
and(
|
||||
eq(sourceRecords.siteId, siteId),
|
||||
eq(sourceRecords.sourceConnectionId, identity.connectionId),
|
||||
eq(sourceRecords.sourceType, row.sourceType),
|
||||
eq(sourceRecords.sourceId, row.sourceId),
|
||||
eq(sourceRecords.locale, row.locale),
|
||||
eq(sourceRecords.sourceSnapshotId, identity.snapshotId),
|
||||
),
|
||||
)
|
||||
.limit(1)
|
||||
const entryId = recs[0]?.contentEntryId
|
||||
if (entryId === undefined || entryId === null) return null
|
||||
const els = await db
|
||||
.select({ currentRevisionId: entryLocales.currentRevisionId })
|
||||
.from(entryLocales)
|
||||
.where(
|
||||
and(
|
||||
eq(entryLocales.siteId, siteId),
|
||||
eq(entryLocales.contentEntryId, entryId),
|
||||
eq(entryLocales.locale, row.locale),
|
||||
),
|
||||
)
|
||||
.limit(1)
|
||||
return els[0]?.currentRevisionId ?? null
|
||||
}
|
||||
|
||||
/** 内容行（非 attachment）：locale 口径对齐 content-writer（拒收行不双报），提取引用挂锚 */
|
||||
async function handleContentRowRefs(row: UniversalRow): Promise<void> {
|
||||
if (row.locale !== identity.defaultLocale) return
|
||||
const refs = extractRefs(row)
|
||||
if (refs.length === 0) return
|
||||
const anchor = await resolveRevisionAnchor(row)
|
||||
if (anchor === null) {
|
||||
issues.push(
|
||||
`媒体引用行无修订锚点（${row.sourceType}/${row.sourceId}/${row.locale}，refs=${refs.length}）：source_records/entry_locales 无 current 指针（content-writer 未处理该行？）——usage 跳过`,
|
||||
)
|
||||
return
|
||||
}
|
||||
for (const ref of refs) pendingRefs.push({ ...ref, revisionId: anchor })
|
||||
}
|
||||
|
||||
/** 悬置引用 → 资产反查（id → source_id；URL → source_url 精确 → public_path 形态） */
|
||||
async function lookupAsset(
|
||||
ref: PendingRef,
|
||||
): Promise<{ id: string; sourceUrl: string; status: string } | null> {
|
||||
const columns = {
|
||||
id: mediaAssets.id,
|
||||
sourceUrl: mediaAssets.sourceUrl,
|
||||
status: mediaAssets.status,
|
||||
}
|
||||
if (ref.kind === 'id') {
|
||||
const rows = await db
|
||||
.select(columns)
|
||||
.from(mediaAssets)
|
||||
.where(
|
||||
and(
|
||||
eq(mediaAssets.siteId, siteId),
|
||||
eq(mediaAssets.sourceConnectionId, identity.connectionId),
|
||||
eq(mediaAssets.sourceId, ref.value),
|
||||
),
|
||||
)
|
||||
.limit(1)
|
||||
return rows[0] ?? null
|
||||
}
|
||||
const byUrl = await db
|
||||
.select(columns)
|
||||
.from(mediaAssets)
|
||||
.where(and(eq(mediaAssets.siteId, siteId), eq(mediaAssets.sourceUrl, ref.value)))
|
||||
.limit(1)
|
||||
if (byUrl.length > 0) return byUrl[0]!
|
||||
if (ref.uploadsPublicPath !== null) {
|
||||
const byPath = await db
|
||||
.select(columns)
|
||||
.from(mediaAssets)
|
||||
.where(and(eq(mediaAssets.siteId, siteId), eq(mediaAssets.publicPath, ref.uploadsPublicPath)))
|
||||
.limit(1)
|
||||
return byPath[0] ?? null
|
||||
}
|
||||
return null
|
||||
}
|
||||
|
||||
return {
|
||||
async handleEvent(ev) {
|
||||
if (ev.kind !== 'row') return // term/unsupported/error：无媒体信息，本写入器不消费
|
||||
if (ev.row.sourceType === 'wp:attachment') {
|
||||
await handleAttachmentRow(ev.row)
|
||||
return
|
||||
}
|
||||
await handleContentRowRefs(ev.row)
|
||||
},
|
||||
|
||||
/** 流终解析：悬置引用逐条对资产反查——命中且 ready → usage；无资产且属本域 → 悬空报告 */
|
||||
async finish() {
|
||||
for (const ref of pendingRefs) {
|
||||
const asset = await lookupAsset(ref)
|
||||
if (asset === null) {
|
||||
if (ref.kind === 'id' || ref.uploadsPublicPath !== null) {
|
||||
counts.mediaDanglingRefs++
|
||||
const target = ref.kind === 'id' ? `attachment ${ref.value}` : ref.value
|
||||
issues.push(
|
||||
`媒体引用悬空（${ref.rawLocation}，${ref.fieldPath} → ${target}）：无对应 attachment/清单记录——引用保留于报告，不静默丢弃`,
|
||||
)
|
||||
}
|
||||
// 外域 URL（无 uploads 标记）：站外热链不属本域对拍范围，静默越过（JSDoc 模块头）
|
||||
continue
|
||||
}
|
||||
if (asset.status !== 'ready') {
|
||||
// quarantined/unsupported 资产：不进产物语义——引用关系不建 usage（资产侧
|
||||
// issue 已在登记时报告；引用命中非悬空）
|
||||
continue
|
||||
}
|
||||
await db
|
||||
.insert(mediaUsages)
|
||||
.values({
|
||||
siteId,
|
||||
mediaAssetId: asset.id,
|
||||
contentRevisionId: ref.revisionId,
|
||||
fieldPath: ref.fieldPath,
|
||||
// id 引用无引文 URL——落资产登记的 source_url（审计可追）；URL 引用落引文原串
|
||||
sourceUrl: ref.kind === 'url' ? ref.value : asset.sourceUrl,
|
||||
})
|
||||
.onConflictDoNothing()
|
||||
}
|
||||
return { counts, issues }
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 从管道上下文构造媒体写入器句柄（Task 6 编排注入位）：manifest/sidecar 从快照 metadata
|
||||
* 与存储副本解析（media_bundle.metadata.manifest / manual_url_list 副本 JSON）。
|
||||
* bundle 缺席 = 空清单（三级全走 missing——诚实报告而非跳过）；侧车缺席 = 空映射。
|
||||
*/
|
||||
export async function createMediaEventHandlerFromContext(
|
||||
ctx: Pick<
|
||||
ImportJobContext,
|
||||
| 'db'
|
||||
| 'siteId'
|
||||
| 'connectionId'
|
||||
| 'defaultLocale'
|
||||
| 'wxrSnapshot'
|
||||
| 'mediaBundleSnapshot'
|
||||
| 'manualUrlListSnapshot'
|
||||
>,
|
||||
): Promise<MediaEventHandler> {
|
||||
const bundle = ctx.mediaBundleSnapshot
|
||||
const manifestRaw = bundle?.metadata.manifest
|
||||
const manifest: MediaManifestEntry[] = Array.isArray(manifestRaw) ? (manifestRaw as MediaManifestEntry[]) : []
|
||||
let storageRoot = ''
|
||||
if (manifest.length > 0) {
|
||||
const raw = bundle?.metadata.storageRoot
|
||||
if (typeof raw !== 'string' || raw === '') {
|
||||
throw new Error('media_bundle 快照 metadata 缺 storageRoot（登记期事实缺席——无法读取存储副本做魔数校验）')
|
||||
}
|
||||
storageRoot = raw
|
||||
}
|
||||
return createMediaEventHandler({
|
||||
db: ctx.db,
|
||||
siteId: ctx.siteId,
|
||||
connectionId: ctx.connectionId,
|
||||
snapshotId: ctx.wxrSnapshot.id,
|
||||
defaultLocale: ctx.defaultLocale,
|
||||
storageRoot,
|
||||
mediaBundleStoragePath: bundle?.storagePath ?? '',
|
||||
mediaManifest: manifest,
|
||||
sidecarMap: await loadSidecarMap(ctx),
|
||||
})
|
||||
}
|
||||
|
||||
/** 侧车加载：manual_url_list 副本 JSON → Map（source_url → file_path）。形状在登记期已校验 */
|
||||
async function loadSidecarMap(
|
||||
ctx: Pick<ImportJobContext, 'manualUrlListSnapshot'>,
|
||||
): Promise<Map<string, string>> {
|
||||
const snap = ctx.manualUrlListSnapshot
|
||||
if (snap === null) return new Map()
|
||||
const root = snap.metadata.storageRoot
|
||||
if (typeof root !== 'string' || root === '') {
|
||||
throw new Error('manual_url_list 快照 metadata 缺 storageRoot（登记期事实缺席——无法读取侧车副本）')
|
||||
}
|
||||
let parsed: unknown
|
||||
try {
|
||||
parsed = JSON.parse(await readFile(join(root, snap.storagePath), 'utf8'))
|
||||
} catch (err) {
|
||||
throw new Error(
|
||||
`侧车快照副本解析失败（${snap.storagePath}）：${err instanceof Error ? err.message : String(err)}`,
|
||||
)
|
||||
}
|
||||
if (!Array.isArray(parsed)) {
|
||||
throw new Error(`侧车快照副本形状异常（${snap.storagePath}）：顶层不是数组`)
|
||||
}
|
||||
const map = new Map<string, string>()
|
||||
for (const item of parsed) {
|
||||
const entry = item as Partial<SidecarEntry>
|
||||
if (
|
||||
typeof entry.sourceUrl !== 'string' ||
|
||||
entry.sourceUrl === '' ||
|
||||
typeof entry.filePath !== 'string' ||
|
||||
entry.filePath === ''
|
||||
) {
|
||||
throw new Error(`侧车快照副本形状异常（${snap.storagePath}）：存在非 {sourceUrl, filePath} 条目`)
|
||||
}
|
||||
map.set(entry.sourceUrl, entry.filePath)
|
||||
}
|
||||
return map
|
||||
}
|
||||
|
||||
/**
|
||||
* 媒体登记管道写入器（{@link createMediaEventHandler} 的自消费整流便捷壳）：单遍拉完
|
||||
* ctx.events、逐事件分派、流终返回 outcome。与 content-writer 组合消费同流时须用
|
||||
* {@link createMediaEventHandlerFromContext} 句柄形状（单消费方拉流分派，无 tee）。
|
||||
*/
|
||||
export function createMediaWriter(): ImportPipeline {
|
||||
return async (ctx) => {
|
||||
const handler = await createMediaEventHandlerFromContext(ctx)
|
||||
for await (const ev of ctx.events) {
|
||||
await handler.handleEvent(ev)
|
||||
}
|
||||
return handler.finish()
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user