Codex
2b7688bd75
setup / install-leaves-tree-clean (push) Canceled after 0s
Treat the source registry as authoritative during file-index rebuilds while preserving staged uploads. This prevents deleted archived sources from remaining searchable. Tests: python3 -m unittest tests.test_file_assets tests.test_business_facts
107 lines
4.2 KiB
Python
107 lines
4.2 KiB
Python
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from obsidian_wiki.file_assets import rebuild_from_registry, search_assets
|
|
|
|
|
|
class FileAssetIndexTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tempdir = tempfile.TemporaryDirectory()
|
|
self.vault = Path(self.tempdir.name)
|
|
(self.vault / "_meta").mkdir()
|
|
self.registry_path = self.vault / "_meta" / "source-registry.jsonl"
|
|
self.db_path = self.vault / "_meta" / "file-assets.sqlite"
|
|
|
|
def tearDown(self):
|
|
self.tempdir.cleanup()
|
|
|
|
def test_rebuild_preserves_registry_identifiers_for_part_number_search(self):
|
|
drawing_id = "sha256:" + "a" * 64
|
|
image_id = "sha256:" + "b" * 64
|
|
self._write_registry_record(
|
|
source_id=drawing_id,
|
|
archive_path="_sources/archive/bjd/pdf/a.pdf",
|
|
original_name="北极大-0.5-18客户图.pdf",
|
|
title="北极大 FPC 0.5-18 客户图纸",
|
|
summary="FPC 0.5-18 客户图 PDF 归档。",
|
|
doc_type="drawing",
|
|
source_type="pdf",
|
|
bjd_part_numbers=["101049-X0XXX0", "101049-20XXX0"],
|
|
series=["0.5-18"],
|
|
pitch=["0.5"],
|
|
pin_count=["18"],
|
|
)
|
|
self._write_registry_record(
|
|
source_id=image_id,
|
|
archive_path="_sources/archive/bjd/quality/photo.jpg",
|
|
original_name="微信图片_20250714101049.jpg",
|
|
title="微信图片_20250714101049",
|
|
summary="客诉图片。",
|
|
doc_type="evidence_media",
|
|
source_type="image",
|
|
business_categories=["quality/customer-complaints"],
|
|
)
|
|
|
|
rebuild_from_registry(self.vault, db_path=self.db_path)
|
|
|
|
rows = search_assets(self.vault, "101049 图纸", db_path=self.db_path, limit=5, read_only=True)
|
|
|
|
self.assertGreaterEqual(len(rows), 2)
|
|
self.assertEqual(rows[0]["file_id"], drawing_id)
|
|
self.assertEqual(json.loads(rows[0]["bjd_part_numbers"]), ["101049-X0XXX0", "101049-20XXX0"])
|
|
self.assertEqual(json.loads(rows[0]["series"]), ["0.5-18"])
|
|
self.assertEqual(json.loads(rows[0]["pitch"]), ["0.5"])
|
|
self.assertEqual(json.loads(rows[0]["pin_count"]), ["18"])
|
|
|
|
def test_rebuild_removes_active_records_removed_from_registry(self):
|
|
retained_id = "sha256:" + "c" * 64
|
|
removed_id = "sha256:" + "d" * 64
|
|
self._write_registry_record(
|
|
source_id=retained_id,
|
|
archive_path="_sources/archive/bjd/pdf/retained.pdf",
|
|
original_name="retained.pdf",
|
|
title="Retained source",
|
|
)
|
|
self._write_registry_record(
|
|
source_id=removed_id,
|
|
archive_path="_sources/archive/bjd/xlsx/removed.xlsx",
|
|
original_name="removed.xlsx",
|
|
title="Removed source",
|
|
)
|
|
rebuild_from_registry(self.vault, db_path=self.db_path)
|
|
|
|
retained_record = json.loads(self.registry_path.read_text(encoding="utf-8").splitlines()[0])
|
|
self.registry_path.write_text(json.dumps(retained_record) + "\n", encoding="utf-8")
|
|
rebuild_from_registry(self.vault, db_path=self.db_path)
|
|
|
|
remaining = search_assets(self.vault, "retained", db_path=self.db_path, limit=5, read_only=True)
|
|
removed = search_assets(self.vault, "removed", db_path=self.db_path, limit=5, read_only=True)
|
|
self.assertEqual([row["file_id"] for row in remaining], [retained_id])
|
|
self.assertEqual(removed, [])
|
|
|
|
def _write_registry_record(self, **fields):
|
|
record = {
|
|
"record_type": "source_archive",
|
|
"project": "bjd",
|
|
"visibility": "internal",
|
|
"business_categories": ["products/fpc"],
|
|
"product_lines": ["FPC", "FFC"],
|
|
"customers": [],
|
|
"competitors": [],
|
|
"standards": [],
|
|
"market_tags": [],
|
|
"related_pages": ["projects/bjd/references/example.md"],
|
|
"catalog_paths": ["_sources/catalog/bjd/products/fpc.md"],
|
|
"classification_confidence": 0.9,
|
|
"size_bytes": 1024,
|
|
**fields,
|
|
}
|
|
with self.registry_path.open("a", encoding="utf-8") as fh:
|
|
fh.write(json.dumps(record, ensure_ascii=False) + "\n")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|