This commit is contained in:
jinli.yl 2026-05-26 16:23:42 +08:00
parent 9f90c448f1
commit 71e8ec1b46
4 changed files with 97 additions and 155 deletions

View file

@ -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*."""

View file

@ -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

View file

@ -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",
)

View file

@ -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