From 71e8ec1b46eb467b7ae7c2643fe35b27abaeb228 Mon Sep 17 00:00:00 2001 From: "jinli.yl" Date: Tue, 26 May 2026 16:23:42 +0800 Subject: [PATCH] up --- .../components/file_graph/base_file_graph.py | 60 ++++-------- .../components/file_graph/local_file_graph.py | 94 ++++++++---------- .../components/file_graph/neo4j_file_graph.py | 2 +- reme4/components/file_graph/nx_file_graph.py | 96 ++++++++----------- 4 files changed, 97 insertions(+), 155 deletions(-) diff --git a/reme4/components/file_graph/base_file_graph.py b/reme4/components/file_graph/base_file_graph.py index f68488fd..c34d53e0 100644 --- a/reme4/components/file_graph/base_file_graph.py +++ b/reme4/components/file_graph/base_file_graph.py @@ -1,7 +1,6 @@ """Abstract base for file-graph backends.""" from abc import abstractmethod -from pathlib import Path from ..base_component import BaseComponent from ...enumeration import ComponentEnum, LinkScopeEnum @@ -9,17 +8,17 @@ from ...schema import FileLink, FileNode class BaseFileGraph(BaseComponent): - """Abstract base for file-graph backends.""" + """Abstract base for file-graph backends. + + Link scope (``get_outlinks`` / ``get_inlinks``): + REAL edges touching an indexed node + VIRTUAL edges touching a dangling placeholder + (referenced but never upserted, or already deleted) + ALL both + """ component_type = ComponentEnum.FILE_GRAPH - def __init__(self, graph_name: str = "default", graph_version: str = "v1", **kwargs): - super().__init__(**kwargs) - self.graph_name: str = graph_name or self.name - self.graph_version: str = graph_version - self.graph_path: Path = self.vault_metadata_path / self.component_type.value - self.graph_path.mkdir(parents=True, exist_ok=True) - # -- Lifecycle --------------------------------------------------------- async def _start(self) -> None: @@ -31,13 +30,7 @@ class BaseFileGraph(BaseComponent): await super()._close() async def load(self) -> None: - """Load persisted state. No-op for backends without local files. - - Called at the end of ``_start()`` after base resources are ready - but before subclass-specific resources are initialised. Backends - that need their own resources for loading should override - ``_start()`` instead of this hook. - """ + """Restore persisted state. No-op for backends without local files.""" async def dump(self) -> None: """Persist state. No-op for backends without local files.""" @@ -46,15 +39,15 @@ class BaseFileGraph(BaseComponent): @abstractmethod async def upsert_nodes(self, nodes: list[FileNode]) -> None: - """Insert or update nodes in the graph.""" + """Insert or update nodes.""" @abstractmethod async def delete_nodes(self, paths: list[str]) -> None: - """Delete nodes by path.""" + """Remove nodes by path.""" @abstractmethod async def get_nodes(self, paths: list[str] | None = None) -> list[FileNode]: - """Return nodes by paths; None = all real nodes; [] = [].""" + """Return nodes by paths; ``None`` = all real nodes.""" @abstractmethod async def rebuild_links(self) -> None: @@ -67,30 +60,9 @@ class BaseFileGraph(BaseComponent): # -- Link access ------------------------------------------------------- @abstractmethod - async def get_outlinks( - self, - path: str, - scope: LinkScopeEnum = LinkScopeEnum.REAL, - ) -> list[FileLink]: - """Return outgoing links for *path*. - - ``scope=REAL`` (default) → edges whose target is an indexed - (real) node. ``scope=VIRTUAL`` → only dangling edges (target - was referenced but never upserted, or was deleted). ``ALL`` - → both. - """ + async def get_outlinks(self, path: str, scope: LinkScopeEnum = LinkScopeEnum.REAL) -> list[FileLink]: + """Outgoing links from *path*.""" @abstractmethod - async def get_inlinks( - self, - path: str, - scope: LinkScopeEnum = LinkScopeEnum.REAL, - ) -> list[FileLink]: - """Return incoming links for *path*. - - ``scope=REAL`` (default) → returns the inbound edges only when - *path* itself is a real node. ``scope=VIRTUAL`` → returns - inbound edges only when *path* is virtual (useful for retarget - / lint that must see references to non-existent targets). - ``ALL`` → both. - """ + async def get_inlinks(self, path: str, scope: LinkScopeEnum = LinkScopeEnum.REAL) -> list[FileLink]: + """Inbound links to *path*.""" diff --git a/reme4/components/file_graph/local_file_graph.py b/reme4/components/file_graph/local_file_graph.py index f34293de..6ec573d1 100644 --- a/reme4/components/file_graph/local_file_graph.py +++ b/reme4/components/file_graph/local_file_graph.py @@ -15,9 +15,10 @@ class LocalFileGraph(BaseFileGraph): def __init__(self, **kwargs): super().__init__(**kwargs) self._nodes: dict[str, FileNode] = {} - self._inverse: dict[str, set[str]] = {} # target → {sources} - self._pending: dict[str, set[str]] = {} # virtual target → {sources} - self._graph_file: Path = self.graph_path / f"{self.graph_name}_{self.graph_version}.jsonl" + self._inverse: dict[str, set[str]] = {} # real target → sources + self._pending: dict[str, set[str]] = {} # virtual target → sources + self.component_metadata_path.mkdir(parents=True, exist_ok=True) + self._graph_file: Path = self.component_metadata_path / f"{self.name}.jsonl" # -- Lifecycle --------------------------------------------------------- @@ -26,20 +27,19 @@ class LocalFileGraph(BaseFileGraph): await self.rebuild_links() async def load(self) -> None: - """Load nodes from JSONL file into memory; keep current state on failure.""" if not self._graph_file.exists(): return try: with open(self._graph_file, "r", encoding="utf-8") as f: - self._nodes.update( - (n.path, n) for line in f if line.strip() for n in [FileNode.model_validate_json(line)] - ) + for line in f: + if line.strip(): + node = FileNode.model_validate_json(line) + self._nodes[node.path] = node self.logger.info(f"Loaded {len(self._nodes)} nodes from {self._graph_file}") except Exception as e: self.logger.exception(f"Failed to load {self._graph_file}: {e}") async def dump(self) -> None: - """Persist all nodes to JSONL via atomic rename.""" try: tmp = self._graph_file.with_suffix(".tmp") with open(tmp, "w", encoding="utf-8") as f: @@ -49,22 +49,29 @@ class LocalFileGraph(BaseFileGraph): except Exception as e: self.logger.exception(f"Failed to write {self._graph_file}: {e}") - # -- Edge bookkeeping -------------------------------------------------- + # -- Internals --------------------------------------------------------- + + @staticmethod + def _targets(node: FileNode) -> list[str]: + return [lnk.target_path for lnk in node.links if lnk.target_path] def _add_edge(self, src: str, target: str) -> None: - """Register src→target; route to pending if target is virtual.""" bucket = self._inverse if target in self._nodes else self._pending bucket.setdefault(target, set()).add(src) def _remove_edge(self, src: str, target: str) -> None: - """Remove src→target from both inverse and pending buckets.""" for bucket in (self._inverse, self._pending): srcs = bucket.get(target) - if srcs is None or src not in srcs: - continue - srcs.discard(src) - if not srcs: - del bucket[target] + if srcs and src in srcs: + srcs.discard(src) + if not srcs: + del bucket[target] + + def _scope_match(self, target: str, scope: LinkScopeEnum) -> bool: + if scope is LinkScopeEnum.ALL: + return True + is_real = target in self._nodes + return is_real if scope is LinkScopeEnum.REAL else not is_real # -- Node CRUD --------------------------------------------------------- @@ -73,14 +80,11 @@ class LocalFileGraph(BaseFileGraph): path = node.path old = self._nodes.get(path) if old is not None: - for link in old.links: - if link.target_path: - self._remove_edge(path, link.target_path) + for target in self._targets(old): + self._remove_edge(path, target) self._nodes[path] = node - for link in node.links: - if link.target_path: - self._add_edge(path, link.target_path) - # Promote pending edges that now target a real node. + for target in self._targets(node): + self._add_edge(path, target) promoted = self._pending.pop(path, None) if promoted: self._inverse.setdefault(path, set()).update(promoted) @@ -90,10 +94,8 @@ class LocalFileGraph(BaseFileGraph): node = self._nodes.pop(path, None) if node is None: continue - for link in node.links: - if link.target_path: - self._remove_edge(path, link.target_path) - # Demote inbound edges to pending (sources still reference this path). + for target in self._targets(node): + self._remove_edge(path, target) demoted = self._inverse.pop(path, None) if demoted: self._pending.setdefault(path, set()).update(demoted) @@ -104,13 +106,11 @@ class LocalFileGraph(BaseFileGraph): return [self._nodes[p] for p in paths if p in self._nodes] async def rebuild_links(self) -> None: - """Rebuild inverse/pending indexes from all node link payloads.""" self._inverse.clear() self._pending.clear() for src, node in self._nodes.items(): - for link in node.links: - if link.target_path: - self._add_edge(src, link.target_path) + for target in self._targets(node): + self._add_edge(src, target) async def clear(self): self._nodes.clear() @@ -120,39 +120,21 @@ class LocalFileGraph(BaseFileGraph): # -- Link access ------------------------------------------------------- - async def get_outlinks( - self, - path: str, - scope: LinkScopeEnum = LinkScopeEnum.REAL, - ) -> list[FileLink]: - # Source must be real (only real nodes carry a ``links`` payload). - # Targets may be real or virtual; ``scope`` selects which to surface. + async def get_outlinks(self, path: str, scope: LinkScopeEnum = LinkScopeEnum.REAL) -> list[FileLink]: node = self._nodes.get(path) if node is None: return [] - return [lnk for lnk in node.links if lnk.target_path and _match_target(lnk.target_path, self._nodes, scope)] + return [lnk for lnk in node.links if lnk.target_path and self._scope_match(lnk.target_path, scope)] - async def get_inlinks( - self, - path: str, - scope: LinkScopeEnum = LinkScopeEnum.REAL, - ) -> list[FileLink]: - # ``_inverse`` keys real targets; ``_pending`` keys virtual ones. - # The queried ``path`` lives in at most one bucket, so ``scope`` - # is satisfied by selecting which bucket to read. + async def get_inlinks(self, path: str, scope: LinkScopeEnum = LinkScopeEnum.REAL) -> list[FileLink]: sources: set[str] = set() if scope in (LinkScopeEnum.REAL, LinkScopeEnum.ALL): sources |= self._inverse.get(path, set()) if scope in (LinkScopeEnum.VIRTUAL, LinkScopeEnum.ALL): sources |= self._pending.get(path, set()) return [ - link for src in sources if src in self._nodes for link in self._nodes[src].links if link.target_path == path + link + for src in sources if src in self._nodes + for link in self._nodes[src].links + if link.target_path == path ] - - -def _match_target(target_path: str, nodes: dict, scope: LinkScopeEnum) -> bool: - """Whether an edge into ``target_path`` should be surfaced under ``scope``.""" - if scope is LinkScopeEnum.ALL: - return True - is_real = target_path in nodes - return is_real if scope is LinkScopeEnum.REAL else not is_real diff --git a/reme4/components/file_graph/neo4j_file_graph.py b/reme4/components/file_graph/neo4j_file_graph.py index 828d2776..ec22a7ef 100644 --- a/reme4/components/file_graph/neo4j_file_graph.py +++ b/reme4/components/file_graph/neo4j_file_graph.py @@ -103,7 +103,7 @@ class Neo4jFileGraph(BaseFileGraph): ) real, virtual, edges = await self._counts(session) self.logger.info( - f"Neo4jFileGraph '{self.graph_name}' connected at " + f"Neo4jFileGraph '{self.name}' connected at " f"{self._uri}/{self._database}: " f"{real} nodes, {edges} edges, {virtual} virtual", ) diff --git a/reme4/components/file_graph/nx_file_graph.py b/reme4/components/file_graph/nx_file_graph.py index e41cafe7..5157d5d0 100644 --- a/reme4/components/file_graph/nx_file_graph.py +++ b/reme4/components/file_graph/nx_file_graph.py @@ -16,121 +16,109 @@ from ...schema import FileLink, FileNode @R.register("nx") class NxFileGraph(BaseFileGraph): - """Networkx-backed file graph; uses FileLink.target_path for adjacency.""" + """Networkx-backed file graph; uses FileLink.target_path for adjacency. + + Real node carries ``node`` attr; virtual (dangling target) does not. + """ def __init__(self, **kwargs): super().__init__(**kwargs) if nx is None: raise ImportError("NxFileGraph requires networkx — pip install networkx") self._graph: nx.MultiDiGraph = nx.MultiDiGraph() - self._graph_file: Path = self.graph_path / f"{self.graph_name}_{self.graph_version}.pkl" + self.component_metadata_path.mkdir(parents=True, exist_ok=True) + self._graph_file: Path = self.component_metadata_path / f"{self.name}.pkl" # -- Lifecycle --------------------------------------------------------- async def load(self) -> None: - """Load graph from pickle file; keep current graph on failure.""" if not self._graph_file.exists(): return try: with open(self._graph_file, "rb") as f: self._graph = pickle.load(f) - n_real = sum(1 for _, d in self._graph.nodes(data=True) if "node" in d) - self.logger.info(f"Loaded {n_real} nodes from {self._graph_file}") + self.logger.info(f"Loaded {self._real_count()} nodes from {self._graph_file}") except Exception as e: self.logger.exception(f"Failed to load {self._graph_file}: {e}") async def dump(self) -> None: - """Persist graph to pickle via atomic rename.""" try: tmp = self._graph_file.with_suffix(".tmp") with open(tmp, "wb") as f: pickle.dump(self._graph, f, protocol=pickle.HIGHEST_PROTOCOL) tmp.replace(self._graph_file) - n_real = sum(1 for _, d in self._graph.nodes(data=True) if "node" in d) - self.logger.info(f"Saved {n_real} nodes to {self._graph_file}") + self.logger.info(f"Saved {self._real_count()} nodes to {self._graph_file}") except Exception as e: self.logger.exception(f"Failed to write {self._graph_file}: {e}") + # -- Internals --------------------------------------------------------- + + def _real_count(self) -> int: + return sum(1 for _, d in self._graph.nodes(data=True) if "node" in d) + + def _is_real(self, key: str) -> bool: + return "node" in self._graph.nodes[key] + + @staticmethod + def _edges_from(src: str, node: FileNode): + return ((src, lnk.target_path, {"link": lnk}) for lnk in node.links if lnk.target_path) + + def _scope_match(self, key: str, scope: LinkScopeEnum) -> bool: + if scope is LinkScopeEnum.ALL: + return True + is_real = self._is_real(key) + return is_real if scope is LinkScopeEnum.REAL else not is_real + # -- Node CRUD --------------------------------------------------------- async def upsert_nodes(self, nodes: list[FileNode]) -> None: for node in nodes: path = node.path if self._graph.has_node(path): - # Drop outgoing edges; inbound stay intact. self._graph.remove_edges_from(list(self._graph.out_edges(path, keys=True))) - self._graph.add_node(path, node=node) # promotes virtual node if present - # Missing targets become attr-less virtual nodes. - self._graph.add_edges_from((path, lnk.target_path, {"link": lnk}) for lnk in node.links if lnk.target_path) + self._graph.add_node(path, node=node) # promotes virtual placeholder + self._graph.add_edges_from(self._edges_from(path, node)) async def delete_nodes(self, paths: list[str]) -> None: for path in paths: if not self._graph.has_node(path): continue self._graph.remove_edges_from(list(self._graph.out_edges(path, keys=True))) - # Demote to virtual: keep inbound edges, drop node payload. - self._graph.nodes[path].pop("node", None) + self._graph.nodes[path].pop("node", None) # demote to virtual if self._graph.in_degree(path) == 0: - self._graph.remove_node(path) # remove orphan virtual node + self._graph.remove_node(path) async def get_nodes(self, paths: list[str] | None = None) -> list[FileNode]: - nodes_view = self._graph.nodes + view = self._graph.nodes if paths is None: - return [d["node"] for _, d in nodes_view(data=True) if "node" in d] - return [nodes_view[path]["node"] for path in paths if path in nodes_view and "node" in nodes_view[path]] + return [d["node"] for _, d in view(data=True) if "node" in d] + return [view[p]["node"] for p in paths if p in view and "node" in view[p]] async def rebuild_links(self) -> None: - """Rebuild all edges from real node payloads; drop virtual nodes.""" self._graph.remove_edges_from(list(self._graph.edges(keys=True))) virtual = [n for n, d in self._graph.nodes(data=True) if "node" not in d] self._graph.remove_nodes_from(virtual) - self._graph.add_edges_from( - (path, lnk.target_path, {"link": lnk}) - for path, data in self._graph.nodes(data=True) - for lnk in data["node"].links - if lnk.target_path - ) + for path, data in list(self._graph.nodes(data=True)): + self._graph.add_edges_from(self._edges_from(path, data["node"])) async def clear(self): - """Remove all nodes and edges, and remove persisted file.""" self._graph.clear() self._graph_file.unlink(missing_ok=True) # -- Link access ------------------------------------------------------- - async def get_outlinks( - self, - path: str, - scope: LinkScopeEnum = LinkScopeEnum.REAL, - ) -> list[FileLink]: - # Source must be real; targets may be virtual placeholders. - # ``scope`` picks real / virtual / both targets. - nodes_view = self._graph.nodes - if path not in nodes_view or "node" not in nodes_view[path]: + async def get_outlinks(self, path: str, scope: LinkScopeEnum = LinkScopeEnum.REAL) -> list[FileLink]: + view = self._graph.nodes + if path not in view or "node" not in view[path]: return [] return [ d["link"] for _, tgt, d in self._graph.out_edges(path, data=True) - if "link" in d and _match_node(nodes_view, tgt, scope) + if "link" in d and self._scope_match(tgt, scope) ] - async def get_inlinks( - self, - path: str, - scope: LinkScopeEnum = LinkScopeEnum.REAL, - ) -> list[FileLink]: - # ``path`` is a single node — its realness selects which scope - # produces a non-empty result (REAL ↔ real node, VIRTUAL ↔ - # virtual placeholder; ALL is always allowed). - nodes_view = self._graph.nodes - if path not in nodes_view or not _match_node(nodes_view, path, scope): + async def get_inlinks(self, path: str, scope: LinkScopeEnum = LinkScopeEnum.REAL) -> list[FileLink]: + view = self._graph.nodes + if path not in view or not self._scope_match(path, scope): return [] return [d["link"] for _, _, d in self._graph.in_edges(path, data=True) if "link" in d] - - -def _match_node(nodes_view, key: str, scope: LinkScopeEnum) -> bool: - """Whether ``key`` satisfies ``scope`` under the nx node-realness convention.""" - if scope is LinkScopeEnum.ALL: - return True - is_real = "node" in nodes_view[key] - return is_real if scope is LinkScopeEnum.REAL else not is_real