feat(import): term 物化与缺失自建
Plan 3 Task 2：createTermWriter 管道写入器——双源物化分类域。 - wp:term 事件：taxonomies（key→object_type 映射 + public_base_path 前置事实 9 常量）+ terms（source_id=term_id 字符串） - row 的 taxonomyRefs miss 时自建（source_id=auto:<nicename>、slug=nicename； phase2-smoke 实证真实 WXR 的 wp:term 表缺 category/post_tag 整域）；同 nicename 多次引用只建一条（内存 (taxonomy,slug) 索引 + UNIQUE 兜底） - 单遍流式 + 收编：WXR 文档序 items 先于 wp:term——row 先到自建、wp:term 后到 就地 UPDATE 扶正 source_id/name，两种顺序终态一致（测试钉死顺序无关性） - content_terms：is_primary 每 (entry, taxonomy) 首个 true 余 false（WP 单主分类 近似）；row 无 content entry → term 仍物化、关联跳过 + issue（禁静默） - ctx 加法扩展 events: AsyncIterable<AdapterEvent>（测试手构注入，Task 6 接真流） - Task 1 遗留两行：ImportErrorCode 补 REPORT_WRITE_FAILED、ctx.db 事务边界 JSDoc Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
浅念
@@ -34,6 +34,7 @@ import {
|
||||
MAPPING_VERSION,
|
||||
PARSER_VERSION,
|
||||
SANITIZER_VERSION,
|
||||
type AdapterEvent,
|
||||
} from '@sf/adapter-wp'
|
||||
import { importJobs, sites, sourceConnections, sourceSnapshots } from '../../db/schema.js'
|
||||
import type * as schema from '../../db/schema.js'
|
||||
@@ -51,6 +52,7 @@ export type ImportErrorCode =
|
||||
| 'SNAPSHOT_NOT_FOUND'
|
||||
| 'NO_WXR_SNAPSHOT'
|
||||
| 'PIPELINE_FAILED'
|
||||
| 'REPORT_WRITE_FAILED'
|
||||
|
||||
/** 导入失败（可预期错误）：CLI 以 code 区分呈现；不裹程序 bug（那些原样上抛） */
|
||||
export class ImportError extends Error {
|
||||
@@ -83,6 +85,11 @@ export interface SnapshotRef {
|
||||
|
||||
/** 管道执行上下文：Task 6 真实管道的全部输入（站点 locale 接缝见前置事实 8） */
|
||||
export interface ImportJobContext {
|
||||
/**
|
||||
* 写入器自行管理事务：ctx.db 为直连（非事务代理），需要原子性的多语句写入用
|
||||
* db.transaction 自行包裹；单语句 upsert（on conflict do nothing/update）本身原子，
|
||||
* 无需显式事务。
|
||||
*/
|
||||
db: ImportDb
|
||||
jobId: string
|
||||
siteId: string
|
||||
@@ -100,6 +107,13 @@ export interface ImportJobContext {
|
||||
* ctx.report.issues.push(...)）。合并规则见 {@link ImportPipeline}——返回值优先。
|
||||
*/
|
||||
report: ImportPipelineOutcome
|
||||
/**
|
||||
* 适配器事件流（runWxrAdapter 产物：row/unsupported/error/term 按文档序交错）。
|
||||
* Task 2 加法扩展：测试经 {@link CreateImportJobInput.events} 手构注入；Task 6 编排
|
||||
* 时改为从 wxr 快照构造真流（runWxrAdapter(storagePath 流)）——多个写入器共享同流
|
||||
* 时的重迭代策略（每写入器一新生成器等）由 Task 6 决定，本字段只承诺"可迭代一次"。
|
||||
*/
|
||||
events: AsyncIterable<AdapterEvent>
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -130,6 +144,12 @@ export interface CreateImportJobInput {
|
||||
connectionName?: string
|
||||
/** 报告根（落 <reportRoot>/<jobId>/reports/import.json）；缺省 repoRoot/artifacts/import */
|
||||
reportRoot?: string
|
||||
/**
|
||||
* 适配器事件流（ctx.events 的注入位）：Task 2 测试手构注入；缺省空流（空管道下无消费
|
||||
* 方）。Task 6 编排时由此改为从 wxr 快照构造 runWxrAdapter 真流（前置事实 8：mapConfig
|
||||
* .locale 必须传 site.default_locale）。
|
||||
*/
|
||||
events?: AsyncIterable<AdapterEvent>
|
||||
/** 管道钩子（测试/Task 6 注入）；缺省空管道 */
|
||||
pipeline?: ImportPipeline
|
||||
}
|
||||
@@ -344,7 +364,7 @@ export async function createImportJob(
|
||||
|
||||
// 阶段 5：管道（本任务缺省空管道；Task 6 接真实写入器）——抛错 → failed 终态。
|
||||
// 产出汇合规则（ImportPipeline JSDoc 钉死）：返回值 counts 覆盖累加器同名键、issues 追加在后。
|
||||
let failed: { code: string; summary: string } | null = null
|
||||
let failed: { code: ImportErrorCode; summary: string } | null = null
|
||||
const ctx: ImportJobContext = {
|
||||
db,
|
||||
jobId,
|
||||
@@ -357,6 +377,7 @@ export async function createImportJob(
|
||||
manualUrlListSnapshot,
|
||||
versions: VERSIONS,
|
||||
report: { counts: {}, issues: [] },
|
||||
events: input.events ?? emptyEvents(),
|
||||
}
|
||||
let outcome: ImportPipelineOutcome | null = null
|
||||
try {
|
||||
@@ -497,3 +518,7 @@ function slimRef(ref: SnapshotRef): {
|
||||
async function emptyPipeline(_ctx: ImportJobContext): Promise<ImportPipelineOutcome> {
|
||||
return { counts: {}, issues: [] }
|
||||
}
|
||||
|
||||
/** 缺省空事件流（input.events 缺席时的 ctx.events 占位；每次调用产新迭代器） */
|
||||
async function* emptyEvents(): AsyncGenerator<AdapterEvent, void, void> {}
|
||||
|
||||
|
||||
@@ -0,0 +1,311 @@
|
||||
// term 物化写入器（Plan 3 Task 2，spec §5.4/§7.5、前置事实 1/9、phase2-smoke 移交项①）：
|
||||
// 消费 ctx.events（AdapterEvent 流）把分类域落库——**双源物化**：
|
||||
// 1. wp:term 事件（表数据旁路直通）：taxonomies 行（key→object_type 映射 +
|
||||
// public_base_path 前置事实 9 常量）+ terms 行（source_id = term_id 字符串）；
|
||||
// 2. row 的 taxonomyRefs（(domain, nicename) 锚点——phase2-smoke 实证：真实 WXR 的
|
||||
// wp:term 表可整域缺席 category/post_tag，148+659 条引用只能靠引用侧自建）：miss 时
|
||||
// 自动建 term（source_id=`auto:<nicename>`、slug=nicename）。
|
||||
//
|
||||
// 单遍流式（Task 决策）：term 事件即建、row 到达时 ensure-term + 关联；内存只留已见
|
||||
// (taxonomy, slug) 索引（不囤 row）。WXR 文档序 items 先于 wp:term（mini 与真实导出同
|
||||
// 序）——row 先到时按 nicename 先自建，对应 wp:term 事件后到时**收编**：就地 UPDATE
|
||||
// source_id/name 扶正为权威值（content_terms 挂的是 term id，不受收编影响）——两种事件
|
||||
// 顺序终态一致（测试钉死顺序无关性）。
|
||||
//
|
||||
// name 的自建回退（如实声明）：UniversalRow.taxonomyRefs 只携带 (taxonomy,
|
||||
// sourceTermId=nicename)，不携带 <category> 文本 value（契约冻结，升版须评审）——自建
|
||||
// term 的 name 暂用 nicename；显示名回填留待后续契约决策。
|
||||
//
|
||||
// is_primary（WP 单主分类近似，钉死）：WXR 不导出 Yoast 主分类标记，按 refs 文档序
|
||||
// **每 (entry, taxonomy) 首个引用**记 true 余 false（"每 entry 每 taxonomy 至多一个
|
||||
// primary"由本写入器保证——库层刻意无部分唯一索引）。
|
||||
//
|
||||
// 事务边界：所有 upsert 均为单语句原子操作（insert ... on conflict do nothing +
|
||||
// returning / 单行 update），无跨语句不变量需要显式事务（约定见 ImportJobContext.db）。
|
||||
import { and, desc, eq, isNotNull } from 'drizzle-orm'
|
||||
import type { WxrTerm } from '@sf/adapter-wp'
|
||||
import type { UniversalRow } from '@sf/contracts'
|
||||
import { contentTerms, sourceRecords, taxonomies, terms } from '../../db/schema.js'
|
||||
import { isUniqueViolation } from '../source/register.service.js'
|
||||
import type { ImportPipeline } from './import.service.js'
|
||||
|
||||
/** 自建 term 的 source_id 前缀（与 wp:term 的 term_id 字符串源身份空间隔离） */
|
||||
const AUTO_SOURCE_ID_PREFIX = 'auto:'
|
||||
|
||||
/** taxonomy key → (object_type, public_base_path) 钉死常量（前置事实 9：WP 默认路径） */
|
||||
export interface TermTaxonomyDefaults {
|
||||
objectType: string
|
||||
publicBasePath: string
|
||||
}
|
||||
|
||||
export const TERM_TAXONOMY_DEFAULTS: Readonly<Record<string, TermTaxonomyDefaults>> = Object.freeze({
|
||||
category: Object.freeze({ objectType: 'post', publicBasePath: '/category/' }),
|
||||
post_tag: Object.freeze({ objectType: 'post', publicBasePath: '/tag/' }),
|
||||
product_cat: Object.freeze({ objectType: 'product', publicBasePath: '/product-category/' }),
|
||||
product_tag: Object.freeze({ objectType: 'product', publicBasePath: '/product-tag/' }),
|
||||
})
|
||||
|
||||
/**
|
||||
* 表外 taxonomy 的兜底映射（真实数据存在：nav_menu/elementor_library_type 有 wp:term
|
||||
* 事件但不在常量表）：object_type 'post'、public_base_path `/<key>/`——仅记录不消费
|
||||
* （路由生成不读表外 taxonomy 的路径），并逐 key 记 issue（不静默）。
|
||||
*/
|
||||
function taxonomyDefaultsFor(key: string): TermTaxonomyDefaults {
|
||||
if (Object.hasOwn(TERM_TAXONOMY_DEFAULTS, key)) return TERM_TAXONOMY_DEFAULTS[key]!
|
||||
return { objectType: 'post', publicBasePath: `/${key}/` }
|
||||
}
|
||||
|
||||
/** 已见 term 的内存索引项（countedAuto：是否已计入 autoCreatedTerms——收编时回退计数） */
|
||||
interface SeenTerm {
|
||||
id: string
|
||||
sourceId: string
|
||||
countedAuto: boolean
|
||||
}
|
||||
|
||||
/**
|
||||
* term 物化管道写入器。产出 counts：
|
||||
* - termsProcessed：处理的 wp:term 事件数；
|
||||
* - autoCreatedTerms：**终态**自建数（引用侧自建且未被 wp:term 收编的行数——瞬态自建
|
||||
* 被收编即回退，不虚报"wp:term 表缺失"）；
|
||||
* - contentTermsCreated：新建的 (entry, term) 关联数（幂等重跑为 0）。
|
||||
* issues：无 entry 可关联的 row 分类引用、表外 taxonomy 兜底、缺 term_id 的 wp:term
|
||||
* 事件、收编撞唯一约束——全部显式报告（§7.6 禁静默）。
|
||||
*/
|
||||
export function createTermWriter(): ImportPipeline {
|
||||
return async (ctx) => {
|
||||
const db = ctx.db
|
||||
const counts = { termsProcessed: 0, autoCreatedTerms: 0, contentTermsCreated: 0 }
|
||||
const issues: string[] = []
|
||||
const taxonomyIds = new Map<string, string>()
|
||||
const slugIndex = new Map<string, SeenTerm>()
|
||||
const flaggedUnknownTaxonomies = new Set<string>()
|
||||
|
||||
const idxKey = (taxonomyKey: string, slug: string): string => `${taxonomyKey}\u0000${slug}`
|
||||
const isAuto = (sourceId: string): boolean => sourceId.startsWith(AUTO_SOURCE_ID_PREFIX)
|
||||
|
||||
/** taxonomy 行 ensure（UNIQUE(site,key,object_type) 幂等；常量表外 key 记 issue） */
|
||||
async function ensureTaxonomy(key: string): Promise<string> {
|
||||
const cached = taxonomyIds.get(key)
|
||||
if (cached !== undefined) return cached
|
||||
const defaults = taxonomyDefaultsFor(key)
|
||||
if (!flaggedUnknownTaxonomies.has(key) && !Object.hasOwn(TERM_TAXONOMY_DEFAULTS, key)) {
|
||||
flaggedUnknownTaxonomies.add(key)
|
||||
issues.push(
|
||||
`未知 taxonomy key：${JSON.stringify(key)}（不在前置事实 9 常量表）——object_type 兜底 '${defaults.objectType}'、public_base_path 兜底 '${defaults.publicBasePath}'（仅记录，路由不消费）`,
|
||||
)
|
||||
}
|
||||
const inserted = await db
|
||||
.insert(taxonomies)
|
||||
.values({ siteId: ctx.siteId, key, objectType: defaults.objectType, publicBasePath: defaults.publicBasePath })
|
||||
.onConflictDoNothing()
|
||||
.returning({ id: taxonomies.id })
|
||||
let id: string
|
||||
if (inserted.length > 0) {
|
||||
id = inserted[0]!.id
|
||||
} else {
|
||||
// 撞唯一三元组（重跑/既有行）：重查复用——插入冲突即行存在，必命中
|
||||
const found = await db
|
||||
.select({ id: taxonomies.id })
|
||||
.from(taxonomies)
|
||||
.where(
|
||||
and(
|
||||
eq(taxonomies.siteId, ctx.siteId),
|
||||
eq(taxonomies.key, key),
|
||||
eq(taxonomies.objectType, defaults.objectType),
|
||||
),
|
||||
)
|
||||
.limit(1)
|
||||
id = found[0]!.id
|
||||
}
|
||||
taxonomyIds.set(key, id)
|
||||
return id
|
||||
}
|
||||
|
||||
/** 按 (taxonomy, slug) 查既有 term（重跑/跨管道持久化行；slug 无唯一索引，取文档序首行） */
|
||||
async function findTermBySlug(taxonomyId: string, slug: string): Promise<SeenTerm | null> {
|
||||
const rows = await db
|
||||
.select({ id: terms.id, sourceId: terms.sourceId })
|
||||
.from(terms)
|
||||
.where(and(eq(terms.siteId, ctx.siteId), eq(terms.taxonomyId, taxonomyId), eq(terms.slug, slug)))
|
||||
.orderBy(terms.sourceId)
|
||||
.limit(1)
|
||||
if (rows.length === 0) return null
|
||||
return { id: rows[0]!.id, sourceId: rows[0]!.sourceId, countedAuto: false }
|
||||
}
|
||||
|
||||
/** term 行插入（UNIQUE(site,taxonomy,source_id) 兜底：冲突即重查复用——同键必命中） */
|
||||
async function insertTerm(args: {
|
||||
taxonomyId: string
|
||||
sourceId: string
|
||||
slug: string
|
||||
name: string
|
||||
data: Record<string, unknown>
|
||||
}): Promise<string> {
|
||||
const inserted = await db
|
||||
.insert(terms)
|
||||
.values({
|
||||
siteId: ctx.siteId,
|
||||
taxonomyId: args.taxonomyId,
|
||||
sourceId: args.sourceId,
|
||||
slug: args.slug,
|
||||
name: args.name,
|
||||
data: args.data,
|
||||
})
|
||||
.onConflictDoNothing()
|
||||
.returning({ id: terms.id })
|
||||
if (inserted.length > 0) return inserted[0]!.id
|
||||
const found = await db
|
||||
.select({ id: terms.id })
|
||||
.from(terms)
|
||||
.where(
|
||||
and(
|
||||
eq(terms.siteId, ctx.siteId),
|
||||
eq(terms.taxonomyId, args.taxonomyId),
|
||||
eq(terms.sourceId, args.sourceId),
|
||||
),
|
||||
)
|
||||
.limit(1)
|
||||
return found[0]!.id
|
||||
}
|
||||
|
||||
/** 收编：自建行就地 UPDATE 为 wp:term 权威身份（source_id/name）；撞唯一约束则保留自建行并记 issue */
|
||||
async function adoptTerm(seen: SeenTerm, realSourceId: string, name: string, taxonomyKey: string): Promise<void> {
|
||||
try {
|
||||
await db.update(terms).set({ sourceId: realSourceId, name }).where(eq(terms.id, seen.id))
|
||||
} catch (err) {
|
||||
if (isUniqueViolation(err)) {
|
||||
issues.push(
|
||||
`term 收编撞唯一约束：taxonomy=${taxonomyKey} source_id=${realSourceId} 已被他行占用——自建行保留（source_id=${seen.sourceId}）`,
|
||||
)
|
||||
return
|
||||
}
|
||||
throw err
|
||||
}
|
||||
if (seen.countedAuto) counts.autoCreatedTerms--
|
||||
seen.sourceId = realSourceId
|
||||
seen.countedAuto = false
|
||||
}
|
||||
|
||||
/** wp:term 事件：建 taxonomy/term；同 slug 已有自建行（row 先到）→ 收编 */
|
||||
async function handleTermEvent(term: WxrTerm): Promise<void> {
|
||||
counts.termsProcessed++
|
||||
const taxonomyId = await ensureTaxonomy(term.taxonomy)
|
||||
const realSourceId = term.termId === null ? null : String(term.termId)
|
||||
const key = idxKey(term.taxonomy, term.slug)
|
||||
let seen = slugIndex.get(key) ?? null
|
||||
if (seen === null) {
|
||||
const dbTerm = await findTermBySlug(taxonomyId, term.slug)
|
||||
if (dbTerm !== null) {
|
||||
slugIndex.set(key, dbTerm)
|
||||
seen = dbTerm
|
||||
}
|
||||
}
|
||||
if (seen !== null) {
|
||||
if (isAuto(seen.sourceId) && realSourceId !== null && seen.sourceId !== realSourceId) {
|
||||
await adoptTerm(seen, realSourceId, term.name, term.taxonomy)
|
||||
}
|
||||
return // 已物化（同流重复事件/重跑既有行）：幂等跳过
|
||||
}
|
||||
if (realSourceId === null) {
|
||||
issues.push(
|
||||
`wp:term 事件缺 term_id（${term.rawLocation}，taxonomy=${term.taxonomy} slug=${term.slug}）——source_id 以 ${AUTO_SOURCE_ID_PREFIX}<slug> 落位`,
|
||||
)
|
||||
}
|
||||
const sourceId = realSourceId ?? `${AUTO_SOURCE_ID_PREFIX}${term.slug}`
|
||||
const id = await insertTerm({
|
||||
taxonomyId,
|
||||
sourceId,
|
||||
slug: term.slug,
|
||||
name: term.name,
|
||||
data:
|
||||
realSourceId === null
|
||||
? { origin: 'wp:term-no-id' }
|
||||
: { origin: 'wp:term', rawLocation: term.rawLocation },
|
||||
})
|
||||
slugIndex.set(key, { id, sourceId, countedAuto: false })
|
||||
}
|
||||
|
||||
/** row 的稳定身份 → content entry（source_records.content_entry_id 是内容写入器的接缝列） */
|
||||
async function resolveContentEntryId(row: UniversalRow): Promise<string | null> {
|
||||
const rows = await db
|
||||
.select({ contentEntryId: sourceRecords.contentEntryId })
|
||||
.from(sourceRecords)
|
||||
.where(
|
||||
and(
|
||||
eq(sourceRecords.siteId, ctx.siteId),
|
||||
eq(sourceRecords.sourceConnectionId, ctx.connectionId),
|
||||
eq(sourceRecords.sourceType, row.sourceType),
|
||||
eq(sourceRecords.sourceId, row.sourceId),
|
||||
eq(sourceRecords.locale, row.locale),
|
||||
isNotNull(sourceRecords.contentEntryId),
|
||||
),
|
||||
)
|
||||
.orderBy(desc(sourceRecords.createdAt))
|
||||
.limit(1)
|
||||
return rows.length > 0 ? rows[0]!.contentEntryId : null
|
||||
}
|
||||
|
||||
/** row 事件：先收集 refs 建齐 term（含自建缺失），再关联 content_terms（顺序无关性） */
|
||||
async function handleRow(row: UniversalRow): Promise<void> {
|
||||
if (row.taxonomyRefs.length === 0) return
|
||||
const resolvedTermIds: string[] = []
|
||||
for (const ref of row.taxonomyRefs) {
|
||||
const taxonomyId = await ensureTaxonomy(ref.taxonomy)
|
||||
const key = idxKey(ref.taxonomy, ref.sourceTermId)
|
||||
let seen = slugIndex.get(key) ?? null
|
||||
if (seen === null) {
|
||||
const dbTerm = await findTermBySlug(taxonomyId, ref.sourceTermId)
|
||||
if (dbTerm !== null) {
|
||||
slugIndex.set(key, dbTerm)
|
||||
seen = dbTerm
|
||||
}
|
||||
}
|
||||
if (seen === null) {
|
||||
// 自建缺失 term（移交项①）：name 用 nicename——UniversalRow 不携带 <category>
|
||||
// 文本 value（契约冻结），显示名回填留待后续契约决策
|
||||
const sourceId = `${AUTO_SOURCE_ID_PREFIX}${ref.sourceTermId}`
|
||||
const id = await insertTerm({
|
||||
taxonomyId,
|
||||
sourceId,
|
||||
slug: ref.sourceTermId,
|
||||
name: ref.sourceTermId,
|
||||
data: { origin: 'auto', firstSeenRawLocation: row.rawLocation },
|
||||
})
|
||||
seen = { id, sourceId, countedAuto: true }
|
||||
slugIndex.set(key, seen)
|
||||
counts.autoCreatedTerms++
|
||||
}
|
||||
resolvedTermIds.push(seen.id)
|
||||
}
|
||||
const entryId = await resolveContentEntryId(row)
|
||||
if (entryId === null) {
|
||||
issues.push(
|
||||
`row 的分类引用无对应 content entry（${row.sourceType}/${row.sourceId}/${row.locale}，refs=${row.taxonomyRefs.length}）——term 已物化，content_terms 跳过`,
|
||||
)
|
||||
return
|
||||
}
|
||||
const primarySeen = new Set<string>()
|
||||
for (let i = 0; i < row.taxonomyRefs.length; i++) {
|
||||
const ref = row.taxonomyRefs[i]!
|
||||
const isPrimary = !primarySeen.has(ref.taxonomy)
|
||||
if (isPrimary) primarySeen.add(ref.taxonomy)
|
||||
const inserted = await db
|
||||
.insert(contentTerms)
|
||||
.values({ siteId: ctx.siteId, contentEntryId: entryId, termId: resolvedTermIds[i]!, isPrimary })
|
||||
.onConflictDoNothing()
|
||||
.returning({ contentEntryId: contentTerms.contentEntryId })
|
||||
if (inserted.length > 0) counts.contentTermsCreated++
|
||||
}
|
||||
}
|
||||
|
||||
for await (const ev of ctx.events) {
|
||||
if (ev.kind === 'term') {
|
||||
await handleTermEvent(ev.term)
|
||||
} else if (ev.kind === 'row') {
|
||||
await handleRow(ev.row)
|
||||
}
|
||||
// unsupported/error 事件：无分类信息，本写入器不消费（Task 6 各写入器各取所需）
|
||||
}
|
||||
return { counts, issues }
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user