Repository navigation
Expand file tree
/
Copy pathcache.py
More file actions
136 lines (111 loc) · 4.05 KB
/
Copy pathcache.py
File metadata and controls
136 lines (111 loc) · 4.05 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
# per-file extraction cache - skip unchanged files on re-run
from __future__ import annotations
import hashlib
import json
import os
from pathlib import Path
def file_hash(path: Path) -> str:
"""SHA256 of file contents + resolved path. Prevents cache collisions on identical content."""
p = Path(path)
h = hashlib.sha256()
h.update(p.read_bytes())
h.update(b"\x00")
h.update(str(p.resolve()).encode())
return h.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"
tmp = entry.with_suffix(".tmp")
try:
tmp.write_text(json.dumps(result))
os.replace(tmp, entry)
except Exception:
tmp.unlink(missing_ok=True)
raise
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