"""Export/import knowledge plane data (Fáze B4). Poskytuje serializaci/deserializaci knowledge plane komponent: - KnowledgeDocument metadata (SQLiteKnowledgeRepository) - Entity relationship graph edges (KnowledgeNavigator) - Vector embeddings (KnowledgeVectorStore / LanceDB) Vzory z výzkumu: - Mem.nowledge Data Transfer: text-based bundle (JSON manifest + JSONL records + Markdown) https://mem.nowledge.co/docs/data-portability (přístup 2026-11-04) - LanceDB: Apache Arrow / PyArrow interoperability, batch add, load_columns from Parquet https://docs.lancedb.com/ (přístup 2026-11-04) - ByteRover Context Tree: markdown-based knowledge storage, git-like versioning arXiv:2604.01599 (přístup 2026-11-04) - Agent Memory Protocol: markdown-first, graph-native, git-native https://github.com/agentmemoryprotocol/agentmemoryprotocol (přístup 2026-11-04) Formát exports: / manifest.json # verze, čas, counts, included sections, sensitivity_filter documents.jsonl # KnowledgeDocument metadata (jeden záznam na řádek) entities.jsonl # entity-relationship graph edges vectors.jsonl # vektorové záznamy (id, vector, text, metadata) knowledge_docs/ # volitelné původní soubory (kopie) settings.json # konfigurace embeding modelu, vector_dim, atd. Použití: from cyberbrain26.knowledge_plane.export_import import KnowledgeExporter, KnowledgeImporter exporter = KnowledgeExporter(memory_service, navigator, vector_store) export_path = exporter.export(export_dir="/path/to/export", format="zip") importer = KnowledgeImporter(memory_service, navigator, vector_store) summary = importer.import_from(export_path) """ from __future__ import annotations import json import logging import shutil import tempfile import time import zipfile from dataclasses import asdict, dataclass, field from pathlib import Path from typing import Any logger = logging.getLogger(__name__) EXPORT_FORMAT_VERSION = "1.0" EXPORT_MANIFEST_NAME = "manifest.json" EXPORT_SETTINGS_NAME = "settings.json" EXPORT_DOCUMENTS_NAME = "documents.jsonl" EXPORT_ENTITIES_NAME = "entities.jsonl" EXPORT_VECTORS_NAME = "vectors.jsonl" EXPORT_DOCS_DIR_NAME = "knowledge_docs" @dataclass(frozen=True) class ExportManifest: """Manifesť exportu — verzované metadata o exportovaném balíku.""" format_version: str export_timestamp: float export_source: str document_count: int entity_count: int vector_count: int included_sections: list[str] sensitivity_filter: str | None topic_filter: str | None schema_version: str = "cyberbrain26-knowledge-plane-1.0" def to_dict(self) -> dict[str, Any]: return asdict(self) @dataclass class ImportSummary: """Souhrn výsledku importu.""" imported_documents: int = 0 skipped_documents: int = 0 imported_entities: int = 0 imported_vectors: int = 0 errors: list[str] = field(default_factory=list) warnings: list[str] = field(default_factory=list) @property def total_records(self) -> int: return self.imported_documents + self.imported_entities + self.imported_vectors class KnowledgeExporter: """Exportuje knowledge plane data do adresáře nebo ZIP archivu. Formát je text-based bundle (inspirace Mem.nowledge Data Transfer): JSON manifest + JSONL pro strukturovaná data + volitelné originální soubory. """ def __init__( self, memory_service: Any, navigator: Any | None = None, vector_store: Any | None = None, ) -> None: self.memory_service = memory_service self.navigator = navigator self.vector_store = vector_store def export( self, export_dir: str | Path, *, format: str = "zip", sensitivity_filter: str | None = None, topic_filter: str | None = None, include_vectors: bool = True, include_entities: bool = True, include_docs: bool = False, ) -> Path: """Exportuje knowledge data. Args: export_dir: Cílový adresář pro export (nebo soubor pro ZIP). format: "zip" nebo "directory". sensitivity_filter: Pokud nastaveno, exportuje jen dokumenty s citlivostí <= hodnota. topic_filter: Pokud nastaveno, exportuje jen dokumenty s daným topikem. include_vectors: Zahrnout vektorová data z LanceDB. include_entities: Zahrnout entity-relationship graf. include_docs: Zahrnout kopie původních souborů. Returns: Cesta k vytvořenému exportu (ZIP soubor nebo adresář). """ export_path = Path(export_dir) tmp_dir = Path(tempfile.mkdtemp(prefix="cyberbrain26-export-")) try: # 1. Shromáždit dokumenty z MemoryService / Repository documents = self._collect_documents( sensitivity_filter=sensitivity_filter, topic_filter=topic_filter ) # 2. Shromáždit entity graf entities = self._collect_entities() if include_entities else [] # 3. Shromáždit vektory vectors = self._collect_vectors() if include_vectors else [] # 4. Vytvořit soubory tmp_docs_dir = tmp_dir / EXPORT_DOCS_DIR_NAME if include_docs: tmp_docs_dir.mkdir(parents=True, exist_ok=True) self._write_documents_jsonl(tmp_dir / EXPORT_DOCUMENTS_NAME, documents) self._write_entities_jsonl(tmp_dir / EXPORT_ENTITIES_NAME, entities) self._write_vectors_jsonl(tmp_dir / EXPORT_VECTORS_NAME, vectors) if include_docs: self._copy_source_files(tmp_docs_dir, documents) # 5. Nastavení a manifest settings = self._collect_settings() settings_path = tmp_dir / EXPORT_SETTINGS_NAME settings_path.write_text(json.dumps(settings, indent=2), encoding="utf-8") manifest = ExportManifest( format_version=EXPORT_FORMAT_VERSION, export_timestamp=time.time(), export_source=str(export_path), document_count=len(documents), entity_count=len(entities), vector_count=len(vectors), included_sections=[ s for s, inc in [ ("documents", True), ("entities", include_entities), ("vectors", include_vectors), ("docs", include_docs), ] if inc ], sensitivity_filter=sensitivity_filter, topic_filter=topic_filter, ) manifest_path = tmp_dir / EXPORT_MANIFEST_NAME manifest_path.write_text( json.dumps(manifest.to_dict(), indent=2), encoding="utf-8" ) # 6. Zabalit nebo ponechat jako adresář if format == "zip": export_path.parent.mkdir(parents=True, exist_ok=True) with zipfile.ZipFile(export_path, "w", zipfile.ZIP_DEFLATED) as zf: for p in sorted(tmp_dir.rglob("*")): if p.is_file(): arcname = p.relative_to(tmp_dir) zf.write(p, arcname) logger.info( "Export abgeschlossen: %d dokumentů, %d entit, %d vektorů → %s", len(documents), len(entities), len(vectors), export_path, ) return export_path else: shutil.copytree(tmp_dir, export_path, dirs_exist_ok=True) logger.info( "Export abgeschlossen do adresáře: %s", export_path ) return export_path finally: shutil.rmtree(tmp_dir, ignore_errors=True) # ------------------------------------------------------------------ # Collection helpers # ------------------------------------------------------------------ def _collect_documents( self, sensitivity_filter: str | None, topic_filter: str | None ) -> list[dict[str, Any]]: """Shromáždí metadata dokumentů z MemoryService repository.""" documents: list[dict[str, Any]] = [] repository = getattr(self.memory_service, "repository", None) if repository is None: logger.warning("MemoryService nemá repository — nelze exportovat dokumenty") return documents try: # Iterovat přes všechny dokumenty v repository all_doc_ids = self._get_all_document_ids(repository) for doc_id in all_doc_ids: doc = repository.get(doc_id) if doc is None: continue doc_dict = self._document_to_dict(doc) # Sensitivity filter if sensitivity_filter is not None: doc_sensitivity = doc_dict.get("sensitivity", "personal") try: if self._sensitivity_rank(doc_sensitivity) > self._sensitivity_rank( sensitivity_filter ): continue except ValueError: continue # Topic filter if topic_filter is not None: meta_topic = doc_dict.get("metadata", {}).get("topic", "") if meta_topic != topic_filter: continue documents.append(doc_dict) except Exception as exc: # noqa: BLE001 logger.error("Chyba při sběru dokumentů pro export: %s", exc) return documents def _collect_entities(self) -> list[dict[str, Any]]: """Shromáždí entity-relationship edges z navigátoru.""" entities: list[dict[str, Any]] = [] if self.navigator is None: return entities graph = getattr(self.navigator, "_entity_graph", None) if graph is None: return entities edges = getattr(graph, "edges", {}) for _edge_key, edge_list in edges.items(): for edge in edge_list: entities.append( { "source_entity": edge.source_entity, "target_entity": edge.target_entity, "relationship_type": edge.relationship_type, "document_id": edge.document_id, } ) return entities def _collect_vectors(self) -> list[dict[str, Any]]: """Shromáždí vektorová data z LanceDB (batch).""" vectors: list[dict[str, Any]] = [] if self.vector_store is None: return vectors try: table = self.vector_store._table rows = table.to_arrow().to_pylist() for row in rows: vectors.append( { "id": row.get("id", ""), "vector": row.get("vector", []), "text": row.get("text", ""), "source_id": row.get("source_id", ""), "path": row.get("path", ""), "mtime": row.get("mtime", 0.0), "sha256": row.get("sha256", ""), "chunk_index": row.get("chunk_index", 0), "metadata": row.get("metadata", "{}"), } ) except Exception as exc: # noqa: BLE001 logger.error("Chyba při sběru vektorů pro export: %s", exc) return vectors def _collect_settings(self) -> dict[str, Any]: """Shromáždí konfiguraci pro export.""" config = getattr(self.memory_service, "config", None) if config is None: return {} return { "model": config.model, "vector_dim": config.vector_dim, "max_context": config.max_context, "ollama_base_url": config.ollama_base_url, "lancedb_directory": str(config.lancedb_directory), "file_extensions": config.file_extensions, "chunk_size_tokens": config.chunk_size_tokens, "chunk_overlap_tokens": config.chunk_overlap_tokens, } # ------------------------------------------------------------------ # Writing helpers # ------------------------------------------------------------------ def _document_to_dict(self, doc: Any) -> dict[str, Any]: """Převede KnowledgeDocument na serializovatelný dict.""" return { "document_id": doc.document_id, "source_id": doc.source_id, "source_uri": doc.source_uri, "content_sha256": doc.content_sha256, "media_type": doc.media_type, "trust": doc.trust.value if hasattr(doc.trust, "value") else str(doc.trust), "sensitivity": doc.sensitivity.value if hasattr(doc.sensitivity, "value") else str(doc.sensitivity), "status": doc.status.value if hasattr(doc.status, "value") else str(doc.status), "metadata": doc.metadata, "created_at": doc.created_at, "updated_at": doc.updated_at, "valid_from": doc.valid_from, "valid_to": doc.valid_to, "superseded_by": doc.superseded_by, } def _write_jsonl(self, path: Path, records: list[dict[str, Any]]) -> None: with open(path, "w", encoding="utf-8") as f: for record in records: f.write(json.dumps(record, ensure_ascii=False) + "\n") def _write_documents_jsonl(self, path: Path, documents: list[dict[str, Any]]) -> None: self._write_jsonl(path, documents) def _write_entities_jsonl(self, path: Path, entities: list[dict[str, Any]]) -> None: self._write_jsonl(path, entities) def _write_vectors_jsonl(self, path: Path, vectors: list[dict[str, Any]]) -> None: self._write_jsonl(path, vectors) def _copy_source_files(self, docs_dir: Path, documents: list[dict[str, Any]]) -> None: """Kopíruje původní soubory do exports adresáře.""" for doc in documents: source_uri = doc.get("source_uri", "") if not source_uri: continue src = Path(source_uri) if not src.exists() or not src.is_file(): continue # Bezpečná cílová cesta (pouze název souboru, předejít path traversal) dest = docs_dir / src.name try: shutil.copy2(src, dest) except Exception as exc: # noqa: BLE001 logger.warning("Nelze zkopírovat %s do exportu: %s", src, exc) # ------------------------------------------------------------------ # Utility helpers # ------------------------------------------------------------------ def _get_all_document_ids(self, repository: Any) -> list[str]: """Získá všechny dokumentové ID z repository.""" # Repository nemusí mít metodu pro listing všech ID — zkusíme různé přístupy all_ids_method = getattr(repository, "get_all_ids", None) if callable(all_ids_method): try: return list(all_ids_method()) except Exception as exc: # noqa: BLE001 logger.debug("get_all_ids selhal: %s", exc) # Fallback: zkusíme melalui SQLite přímo database = getattr(self.memory_service, "database", None) if database is not None: try: with database.connection() as connection: rows = connection.execute( "SELECT document_id FROM knowledge_documents" ).fetchall() return [row["document_id"] for row in rows] except Exception as exc: # noqa: BLE001 logger.debug("SQLite listing selhal: %s", exc) return [] @staticmethod def _sensitivity_rank(level: str) -> int: rank = {"public": 0, "personal": 1, "sensitive": 2, "restricted": 3} return rank.get(level, 1) class KnowledgeImporter: """Importuje knowledge plane data z adresáře nebo ZIP archivu. Fail-closed: chybějící manifest nebo neznámá verze = chyba importu. """ def __init__( self, memory_service: Any, navigator: Any | None = None, vector_store: Any | None = None, ) -> None: self.memory_service = memory_service self.navigator = navigator self.vector_store = vector_store def import_from( self, export_path: str | Path, *, overwrite: bool = False, sensitivity_override: str | None = None, dedup: bool = True, ) -> ImportSummary: """Importuje knowledge data. Args: export_path: Cesta k ZIP souboru nebo adresáři s exportem. overwrite: Přepsat existující dokumenty se stejným ID. sensitivity_override: Přepsat citlivost všech importovaných dokumentů. dedup: Pokud True, přeskočí duplikáty podle content_sha256. Returns: ImportSummary s počty importovaných/preskočených položek a chybami. """ summary = ImportSummary() export_path = Path(export_path) # Rozbalit ZIP do dočasného adresáře tmp_dir: Path | None = None source_dir: Path if zipfile.is_zipfile(export_path): tmp_dir = Path(tempfile.mkdtemp(prefix="cyberbrain26-import-")) with zipfile.ZipFile(export_path, "r") as zf: zf.extractall(tmp_dir) source_dir = tmp_dir elif export_path.is_dir(): source_dir = export_path else: summary.errors.append(f"Neplatný cesta k exportu: {export_path}") return summary try: # 1. Načíst a ověřit manifest manifest_path = source_dir / EXPORT_MANIFEST_NAME if not manifest_path.exists(): summary.errors.append("Chybí manifest.json v exportu") return summary manifest = json.loads(manifest_path.read_text(encoding="utf-8")) if manifest.get("format_version") != EXPORT_FORMAT_VERSION: summary.errors.append( f"Nepodporovaná verze formátu: {manifest.get('format_version')} " f"(očekáváno {EXPORT_FORMAT_VERSION})" ) return summary logger.info( "Import z exportu verze %s, %d dokumentů, %d entit, %d vektorů", manifest.get("format_version"), manifest.get("document_count", 0), manifest.get("entity_count", 0), manifest.get("vector_count", 0), ) # 2. Importovat dokumenty documents_path = source_dir / EXPORT_DOCUMENTS_NAME if documents_path.exists(): repo = getattr(self.memory_service, "repository", None) if repo is None: summary.errors.append("MemoryService nemá repository pro import dokumentů") else: for line in documents_path.read_text(encoding="utf-8").splitlines(): if not line.strip(): continue try: doc_dict = json.loads(line) doc_id = doc_dict.get("document_id", "") # Duplikace kontrola if dedup and repo.get(doc_id) is not None: if not overwrite: summary.skipped_documents += 1 continue # Sensitive override if sensitivity_override is not None: doc_dict["sensitivity"] = sensitivity_override # Vytvořit KnowledgeDocument doc = self._dict_to_document(doc_dict) repo.register( type( "Registration", (), { "source": doc.source_id, "content": b"", # Obsah není v exportu "media_type": doc.media_type, "metadata": doc.metadata, }, ) ) summary.imported_documents += 1 except Exception as exc: # noqa: BLE001 summary.errors.append(f"Chyba importu dokumentu: {exc}") logger.warning("Import dokumentu selhal: %s", exc) # 3. Importovat entity graf entities_path = source_dir / EXPORT_ENTITIES_NAME if entities_path.exists() and self.navigator is not None: for line in entities_path.read_text(encoding="utf-8").splitlines(): if not line.strip(): continue try: entity = json.loads(line) self.navigator.add_entity_edge( source_entity=entity["source_entity"], target_entity=entity["target_entity"], relationship_type=entity["relationship_type"], document_id=entity["document_id"], ) summary.imported_entities += 1 except Exception as exc: # noqa: BLE001 summary.errors.append(f"Chyba importu entity: {exc}") # 4. Importovat vektory vectors_path = source_dir / EXPORT_VECTORS_NAME if vectors_path.exists() and self.vector_store is not None: from cyberbrain26.knowledge_plane.vector_store import VectorDocument batch: list[VectorDocument] = [] for line in vectors_path.read_text(encoding="utf-8").splitlines(): if not line.strip(): continue try: vec = json.loads(line) batch.append( VectorDocument( id=vec["id"], vector=vec["vector"], text=vec.get("text", ""), source_id=vec.get("source_id", ""), path=vec.get("path", ""), mtime=vec.get("mtime", 0.0), sha256=vec.get("sha256", ""), chunk_index=vec.get("chunk_index", 0), metadata=json.loads(vec.get("metadata", "{}")), ) ) if len(batch) >= 100: self.vector_store.add_documents(batch) summary.imported_vectors += len(batch) batch = [] except Exception as exc: # noqa: BLE001 summary.errors.append(f"Chyba importu vektoru: {exc}") if batch: self.vector_store.add_documents(batch) summary.imported_vectors += len(batch) logger.info( "Import dokončen: %d dokumentů, %d entit, %d vektorů, %d chyb", summary.imported_documents, summary.imported_entities, summary.imported_vectors, len(summary.errors), ) finally: if tmp_dir is not None: shutil.rmtree(tmp_dir, ignore_errors=True) return summary def _dict_to_document(self, doc_dict: dict[str, Any]) -> Any: """Převede dict zpět na KnowledgeDocument.""" from cyberbrain26.knowledge_plane.contracts import ( DocumentStatus, DocumentTrust, KnowledgeDocument, Sensitivity, ) def parse_enum(cls: Any, value: Any, default: Any) -> Any: if isinstance(value, cls): return value try: return cls(value) except (ValueError, TypeError): return default return KnowledgeDocument( document_id=doc_dict["document_id"], source_id=doc_dict.get("source_id", ""), source_uri=doc_dict.get("source_uri", ""), content_sha256=doc_dict.get("content_sha256", ""), media_type=doc_dict.get("media_type", "application/octet-stream"), trust=parse_enum(DocumentTrust, doc_dict.get("trust"), DocumentTrust.REVIEWED), sensitivity=parse_enum( Sensitivity, doc_dict.get("sensitivity"), Sensitivity.PERSONAL ), status=parse_enum( DocumentStatus, doc_dict.get("status"), DocumentStatus.REGISTERED ), metadata=doc_dict.get("metadata", {}), created_at=doc_dict.get("created_at", time.time()), updated_at=doc_dict.get("updated_at", time.time()), valid_from=doc_dict.get("valid_from", 0.0), valid_to=doc_dict.get("valid_to", 0.0), superseded_by=doc_dict.get("superseded_by", ""), ) __all__ = [ "ExportManifest", "ImportSummary", "KnowledgeExporter", "KnowledgeImporter", ]