From d3a432d8e3bc16b36327b6e9f14d3895b28dcc8e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=80=AA=E7=A8=8B=E4=BC=9F?= Date: Sun, 7 Jun 2026 19:50:21 +0800 Subject: [PATCH] fix(wiki): dedup directory-scanned files by source path, not just content hash (#272) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Directory-scan ingestion deduped only by content hash, so when a file at a known path changed, the new hash missed the existing raw and a second row was inserted for the same source_path — both rows then generated wiki pages, accumulating duplicates. Make source path the primary dedup key: same path + same hash skips, same path + changed hash updates the existing raw in place (reset to pending, re-process), falling back to the content-hash check only for genuine copies at new paths. Reprocessing reuses the same rawId, so deleteExclusiveBySourceRawId cleans the old pages before regeneration — no duplicate rows and no duplicate pages. findBySourcePath gains LIMIT 1 to tolerate pre-existing duplicates; docs/fix-duplicate-raws.sql remediates existing data. Closes #271 --- docs/fix-duplicate-raws.sql | 117 ++++++++++++++++++ .../wiki/service/WikiRawMaterialService.java | 115 ++++++++++++++--- 2 files changed, 217 insertions(+), 15 deletions(-) create mode 100644 docs/fix-duplicate-raws.sql diff --git a/docs/fix-duplicate-raws.sql b/docs/fix-duplicate-raws.sql new file mode 100644 index 00000000..9e420007 --- /dev/null +++ b/docs/fix-duplicate-raws.sql @@ -0,0 +1,117 @@ +-- ============================================================ +-- 修复：知识库原始材料重复入库 +-- 适用：MySQL 8.0+（使用 JSON 函数处理 source_raw_ids） +-- 说明：同一 (kb_id, source_path) 可能因文件内容变更 +-- 被多次 INSERT 而形成多行。本脚本保留最新行， +-- 并级联清理其关联的 chunk、citation、page。 +-- ============================================================ + +-- ────────────────────────────────────────────────────────── +-- STEP 0：预览（只读，不改数据，先跑这一步确认影响范围） +-- ────────────────────────────────────────────────────────── + +-- 0-A：查看所有重复组（按 kb_id + source_path 分组，count > 1） +SELECT + kb_id, + source_path, + COUNT(*) AS duplicate_count, + MAX(id) AS keep_id, + GROUP_CONCAT(id ORDER BY id DESC) AS all_ids +FROM mate_wiki_raw_material +WHERE source_path IS NOT NULL +GROUP BY kb_id, source_path +HAVING COUNT(*) > 1; + +-- 0-B：查看待删除的具体行（排除每组最新的那一行） +SELECT + r.id, r.kb_id, r.source_path, + r.content_hash, r.processing_status, r.create_time +FROM mate_wiki_raw_material r +WHERE r.source_path IS NOT NULL + AND r.id NOT IN ( + SELECT MAX(id) + FROM mate_wiki_raw_material + WHERE source_path IS NOT NULL + GROUP BY kb_id, source_path + ) +ORDER BY r.kb_id, r.source_path, r.id; + + +-- ────────────────────────────────────────────────────────── +-- STEP 1：开事务，执行清理（确认 STEP 0 结果后再运行） +-- ────────────────────────────────────────────────────────── + +START TRANSACTION; + +-- 1-A：把待删除的 raw id 暂存到临时表，后续步骤复用 +CREATE TEMPORARY TABLE IF NOT EXISTS _stale_raw_ids AS +SELECT id AS raw_id, kb_id +FROM mate_wiki_raw_material +WHERE source_path IS NOT NULL + AND id NOT IN ( + SELECT MAX(id) + FROM mate_wiki_raw_material + WHERE source_path IS NOT NULL + GROUP BY kb_id, source_path + ); + +-- 1-B：删除这些 raw 产生的 citation（通过 chunk_id 关联） +DELETE c +FROM mate_wiki_page_citation c + INNER JOIN mate_wiki_chunk ch ON c.chunk_id = ch.id + INNER JOIN _stale_raw_ids s ON ch.raw_id = s.raw_id; + +-- 1-C：删除 chunk +DELETE ch +FROM mate_wiki_chunk ch + INNER JOIN _stale_raw_ids s ON ch.raw_id = s.raw_id; + +-- 1-D：删除仅由该 raw 派生的 page（source_raw_ids 数组长度为 1） +-- 使用 JSON_CONTAINS 判断 page 是否引用了待删 raw +DELETE p +FROM mate_wiki_page p +WHERE JSON_LENGTH(p.source_raw_ids) = 1 + AND EXISTS ( + SELECT 1 + FROM _stale_raw_ids s + WHERE JSON_CONTAINS(p.source_raw_ids, CAST(s.raw_id AS CHAR)) + ); + +-- 1-E：对多来源 page，将待删 raw 从 source_raw_ids 中移除 +-- 通过 JSON_TABLE 把数组展开再重组，排除掉 stale raw id +UPDATE mate_wiki_page p +SET p.source_raw_ids = ( + SELECT JSON_ARRAYAGG(jt.v) + FROM JSON_TABLE(p.source_raw_ids, '$[*]' COLUMNS (v BIGINT PATH '$')) jt + WHERE jt.v NOT IN (SELECT raw_id FROM _stale_raw_ids) +) +WHERE JSON_LENGTH(p.source_raw_ids) > 1 + AND EXISTS ( + SELECT 1 + FROM _stale_raw_ids s + WHERE JSON_CONTAINS(p.source_raw_ids, CAST(s.raw_id AS CHAR)) + ); + +-- 1-F：删除 stale raw 行 +DELETE r +FROM mate_wiki_raw_material r + INNER JOIN _stale_raw_ids s ON r.id = s.raw_id; + +-- 1-G：确认结果 +SELECT + 'stale raws deleted' AS action, + ROW_COUNT() AS affected_rows; + +SELECT + 'remaining duplicates' AS check_item, + COUNT(*) AS count +FROM mate_wiki_raw_material +WHERE source_path IS NOT NULL +GROUP BY kb_id, source_path +HAVING COUNT(*) > 1; + +-- 确认无误后提交；如有问题改为 ROLLBACK +COMMIT; +-- ROLLBACK; + +DROP TEMPORARY TABLE IF EXISTS _stale_raw_ids; diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiRawMaterialService.java b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiRawMaterialService.java index b9a6d697..d3b9dc4c 100644 --- a/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiRawMaterialService.java +++ b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiRawMaterialService.java @@ -83,28 +83,45 @@ public class WikiRawMaterialService { return rawMapper.selectOne( new LambdaQueryWrapper() .eq(WikiRawMaterialEntity::getKbId, kbId) - .eq(WikiRawMaterialEntity::getSourcePath, sourcePath)); + .eq(WikiRawMaterialEntity::getSourcePath, sourcePath) + .last("LIMIT 1")); } /** - * Import a text file discovered by a directory scan, detecting content - * changes by hash: unchanged content (a raw with the same hash already - * exists) is a no-op, while changed content creates a new raw and triggers - * processing — so a modified file is re-ingested rather than silently - * skipped. The originating path is recorded for diagnostics. + * Import a text file discovered by a directory scan. * - * @return {@code true} when the file was newly ingested (new or changed + *

Dedup strategy (in priority order): + *

    + *
  1. Source path: if a raw for {@code (kbId, absolutePath)} already exists, + * compare hashes. Unchanged → skip; changed → update in-place so the same raw + * row is reused rather than creating a duplicate row for the same file path.
    2. + *
  2. Content hash: if a different file with identical content already exists + * in the KB, the existing raw is reused (copy/duplicate scenario).
    3. + *
+ * + * @return {@code true} when the file was newly ingested or updated (new or changed * content), {@code false} when skipped as unchanged */ public boolean ingestTextFileFromScan(Long kbId, String fileName, String absolutePath, String content) { String hash = computeHash(content); + + // Primary dedup: same source path already in this KB + WikiRawMaterialEntity byPath = findBySourcePath(kbId, absolutePath); + if (byPath != null) { + if (hash != null && hash.equals(byPath.getContentHash())) { + return false; // unchanged + } + // File changed: update existing raw in-place to avoid a duplicate row + updateTextContentFromScan(byPath.getId(), fileName, content, absolutePath); + return true; + } + + // Secondary dedup: different path but identical content (copy of an existing file) WikiRawMaterialEntity sameContent = rawMapper.selectOne( new LambdaQueryWrapper() .eq(WikiRawMaterialEntity::getKbId, kbId) .eq(WikiRawMaterialEntity::getContentHash, hash) .last("LIMIT 1")); - // addText dedups internally by hash, so this reuses sameContent when - // unchanged and inserts + triggers processing when the content differs. WikiRawMaterialEntity raw = addText(kbId, fileName, content); // Only stamp the path on a genuinely new raw. When the content matched // an existing raw (possibly a different file with identical content), @@ -116,13 +133,45 @@ public class WikiRawMaterialService { } /** - * Import a binary file discovered by a directory scan, detecting content - * changes by hashing the bytes: unchanged content (a raw with the same hash - * exists) is skipped, while changed content is re-ingested via - * {@link #addFile}. The unchanged case reads the file once; only a - * new/changed file is read again by addFile. + * Update an existing text raw in-place when a directory-scanned file has changed content. + * Resets processing state and triggers re-processing so new pages are generated from the + * updated content without leaving a stale duplicate row alongside the new one. + */ + @Transactional + public void updateTextContentFromScan(Long rawId, String title, String content, String sourcePath) { + WikiRawMaterialEntity entity = rawMapper.selectById(rawId); + if (entity == null) return; + String hash = computeHash(content); + entity.setTitle(title); + entity.setOriginalContent(content); + entity.setContentHash(hash); + entity.setFileSize((long) content.getBytes(StandardCharsets.UTF_8).length); + entity.setSourcePath(sourcePath); + entity.setProcessingStatus("pending"); + entity.setErrorMessage(null); + entity.setExtractedText(null); + entity.setProgressPhase(null); + entity.setProgressDone(0); + entity.setProgressTotal(0); + rawMapper.updateById(entity); + log.info("[Wiki] Text raw updated in-place from scan: id={}, kbId={}, newHash={}", rawId, entity.getKbId(), hash); + if (properties.isAutoProcessOnUpload()) { + eventPublisher.publishEvent(new WikiProcessingEvent(this, rawId, entity.getKbId())); + } + } + + /** + * Import a binary file discovered by a directory scan. * - * @return {@code true} when newly ingested, {@code false} when unchanged + *

Dedup strategy (in priority order): + *

    + *
  1. Source path: if a raw for {@code (kbId, absolutePath)} already exists, + * compare hashes. Unchanged → skip; changed → update in-place.
    2. + *
  2. Content hash: if a different file with identical bytes already exists, + * the existing raw is reused.
    3. + *
+ * + * @return {@code true} when newly ingested or updated, {@code false} when unchanged */ public boolean ingestBinaryFileFromScan(Long kbId, String title, String sourceType, String absolutePath, long fileSize) { @@ -132,6 +181,19 @@ public class WikiRawMaterialService { } catch (Exception e) { log.warn("[Wiki] Could not hash file for change detection: {}", e.getMessage()); } + + // Primary dedup: same source path already in this KB + WikiRawMaterialEntity byPath = findBySourcePath(kbId, absolutePath); + if (byPath != null) { + if (hash != null && hash.equals(byPath.getContentHash())) { + return false; // unchanged + } + // File changed: update existing raw in-place + updateBinaryFileFromScan(byPath.getId(), absolutePath, fileSize, hash); + return true; + } + + // Secondary dedup: same content hash → copy at a different path if (hash != null) { WikiRawMaterialEntity sameContent = rawMapper.selectOne( new LambdaQueryWrapper() @@ -147,6 +209,29 @@ public class WikiRawMaterialService { return true; } + /** + * Update an existing binary raw in-place when a directory-scanned file has changed content. + */ + @Transactional + public void updateBinaryFileFromScan(Long rawId, String sourcePath, long fileSize, String hash) { + WikiRawMaterialEntity entity = rawMapper.selectById(rawId); + if (entity == null) return; + entity.setContentHash(hash); + entity.setFileSize(fileSize); + entity.setSourcePath(sourcePath); + entity.setProcessingStatus("pending"); + entity.setErrorMessage(null); + entity.setExtractedText(null); + entity.setProgressPhase(null); + entity.setProgressDone(0); + entity.setProgressTotal(0); + rawMapper.updateById(entity); + log.info("[Wiki] Binary raw updated in-place from scan: id={}, kbId={}, newHash={}", rawId, entity.getKbId(), hash); + if (properties.isAutoProcessOnUpload()) { + eventPublisher.publishEvent(new WikiProcessingEvent(this, rawId, entity.getKbId())); + } + } + /** * Record the originating file path on a raw material via a partial update, * so a later directory re-scan can dedup it by source path. Used for