mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 11:13:43 +08:00
fix(wiki): dedup directory-scanned files by source path, not just content hash (#272)
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
This commit is contained in:
倪程伟
GitHub
parent
46f3d425e0
commit
d3a432d8e3
117
docs/fix-duplicate-raws.sql
Normal file
117
docs/fix-duplicate-raws.sql
Normal file
@ -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;
|
||||
@ -83,28 +83,45 @@ public class WikiRawMaterialService {
|
||||
return rawMapper.selectOne(
|
||||
new LambdaQueryWrapper<WikiRawMaterialEntity>()
|
||||
.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
|
||||
* <p>Dedup strategy (in priority order):
|
||||
* <ol>
|
||||
* <li><b>Source path</b>: 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.</li>
|
||||
* <li><b>Content hash</b>: if a different file with identical content already exists
|
||||
* in the KB, the existing raw is reused (copy/duplicate scenario).</li>
|
||||
* </ol>
|
||||
*
|
||||
* @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<WikiRawMaterialEntity>()
|
||||
.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
|
||||
* <p>Dedup strategy (in priority order):
|
||||
* <ol>
|
||||
* <li><b>Source path</b>: if a raw for {@code (kbId, absolutePath)} already exists,
|
||||
* compare hashes. Unchanged → skip; changed → update in-place.</li>
|
||||
* <li><b>Content hash</b>: if a different file with identical bytes already exists,
|
||||
* the existing raw is reused.</li>
|
||||
* </ol>
|
||||
*
|
||||
* @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<WikiRawMaterialEntity>()
|
||||
@ -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
|
||||
|
||||
Loading…
Reference in New Issue
Block a user