mirror of
https://github.com/safishamsi/graphify.git
synced 2026-07-13 19:07:10 +00:00
Safi
317d73f8f7
- confidence_score required on every edge (INFERRED: 0.4-0.9, EXTRACTED: 1.0, AMBIGUOUS: 0.1-0.3) - semantically_similar_to edges for non-obvious cross-file conceptual links - hyperedges for 3+ node group relationships - fixed cache and merge pipeline that was silently dropping them - check_semantic_cache returns 4-tuple including cached_hyperedges - extract.py: mine the "why" - module/class/function docstrings and rationale comments (# NOTE: # IMPORTANT: # HACK: # WHY: # RATIONALE: # TODO: # FIXME:) as rationale_for nodes - skill.md: rationale_for in relation schema, doc files extract design rationale - obsidian output opt-in (--obsidian flag) - default output is graph.html + graph.json + GRAPH_REPORT.md only - hooks.py: post-checkout hook added alongside post-commit - graph rebuilds on branch switch - claude install: writes .claude/settings.json PreToolUse hook on Glob/Grep - Claude checks graph before searching raw files - README updated with all v2 features
125 lines
3.8 KiB
Python
125 lines
3.8 KiB
Python
# per-file extraction cache - skip unchanged files on re-run
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
|
|
|
|
def file_hash(path: Path) -> str:
|
|
"""SHA256 of file contents, hex digest."""
|
|
return hashlib.sha256(Path(path).read_bytes()).hexdigest()
|
|
|
|
|
|
def cache_dir(root: Path = Path(".")) -> Path:
|
|
"""Returns graphify-out/cache/ - creates it if needed."""
|
|
d = Path(root) / "graphify-out" / "cache"
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
return d
|
|
|
|
|
|
def load_cached(path: Path, root: Path = Path(".")) -> dict | None:
|
|
"""Return cached extraction for this file if hash matches, else None.
|
|
|
|
Cache key: SHA256 of file contents.
|
|
Cache value: stored as graphify-out/cache/{hash}.json
|
|
Returns None if no cache entry or file has changed.
|
|
"""
|
|
try:
|
|
h = file_hash(path)
|
|
except OSError:
|
|
return None
|
|
entry = cache_dir(root) / f"{h}.json"
|
|
if not entry.exists():
|
|
return None
|
|
try:
|
|
return json.loads(entry.read_text())
|
|
except (json.JSONDecodeError, OSError):
|
|
return None
|
|
|
|
|
|
def save_cached(path: Path, result: dict, root: Path = Path(".")) -> None:
|
|
"""Save extraction result for this file.
|
|
|
|
Stores as graphify-out/cache/{hash}.json where hash = SHA256 of current file contents.
|
|
result should be a dict with 'nodes' and 'edges' lists.
|
|
"""
|
|
h = file_hash(path)
|
|
entry = cache_dir(root) / f"{h}.json"
|
|
entry.write_text(json.dumps(result))
|
|
|
|
|
|
def cached_files(root: Path = Path(".")) -> set[str]:
|
|
"""Return set of file paths that have a valid cache entry (hash still matches)."""
|
|
d = cache_dir(root)
|
|
return {p.stem for p in d.glob("*.json")}
|
|
|
|
|
|
def clear_cache(root: Path = Path(".")) -> None:
|
|
"""Delete all graphify-out/cache/*.json files."""
|
|
d = cache_dir(root)
|
|
for f in d.glob("*.json"):
|
|
f.unlink()
|
|
|
|
|
|
def check_semantic_cache(
|
|
files: list[str],
|
|
root: Path = Path("."),
|
|
) -> tuple[list[dict], list[dict], list[dict], list[str]]:
|
|
"""Check semantic extraction cache for a list of absolute file paths.
|
|
|
|
Returns (cached_nodes, cached_edges, cached_hyperedges, uncached_files).
|
|
Uncached files need Claude extraction; cached files are merged directly.
|
|
"""
|
|
cached_nodes: list[dict] = []
|
|
cached_edges: list[dict] = []
|
|
cached_hyperedges: list[dict] = []
|
|
uncached: list[str] = []
|
|
|
|
for fpath in files:
|
|
result = load_cached(Path(fpath), root)
|
|
if result is not None:
|
|
cached_nodes.extend(result.get("nodes", []))
|
|
cached_edges.extend(result.get("edges", []))
|
|
cached_hyperedges.extend(result.get("hyperedges", []))
|
|
else:
|
|
uncached.append(fpath)
|
|
|
|
return cached_nodes, cached_edges, cached_hyperedges, uncached
|
|
|
|
|
|
def save_semantic_cache(
|
|
nodes: list[dict],
|
|
edges: list[dict],
|
|
hyperedges: list[dict] | None = None,
|
|
root: Path = Path("."),
|
|
) -> int:
|
|
"""Save semantic extraction results to cache, keyed by source_file.
|
|
|
|
Groups nodes and edges by source_file, then saves one cache entry per file.
|
|
Returns the number of files cached.
|
|
"""
|
|
from collections import defaultdict
|
|
|
|
by_file: dict[str, dict] = defaultdict(lambda: {"nodes": [], "edges": [], "hyperedges": []})
|
|
for n in nodes:
|
|
src = n.get("source_file", "")
|
|
if src:
|
|
by_file[src]["nodes"].append(n)
|
|
for e in edges:
|
|
src = e.get("source_file", "")
|
|
if src:
|
|
by_file[src]["edges"].append(e)
|
|
for h in (hyperedges or []):
|
|
src = h.get("source_file", "")
|
|
if src:
|
|
by_file[src]["hyperedges"].append(h)
|
|
|
|
saved = 0
|
|
for fpath, result in by_file.items():
|
|
p = Path(fpath)
|
|
if p.exists():
|
|
save_cached(p, result, root)
|
|
saved += 1
|
|
return saved
|