99 lines
3.2 KiB
Python
99 lines
3.2 KiB
Python
import hashlib
|
|
import json
|
|
|
|
from pathlib import Path
|
|
from typing import Optional, Dict
|
|
|
|
from src.result_types import Result, ErrorInfo, ErrorKind
|
|
|
|
|
|
def get_file_hash(content: str) -> str:
|
|
"""Returns SHA256 hash of the content."""
|
|
return hashlib.sha256(content.encode("utf-8")).hexdigest()
|
|
|
|
class SummaryCache:
|
|
"""
|
|
A hash-based cache for file summaries to avoid redundant processing.
|
|
Invalidates when content hash changes.
|
|
"""
|
|
def __init__(self, cache_file: Optional[str] = None, max_entries: int = 1000):
|
|
if cache_file:
|
|
self.cache_file = Path(cache_file)
|
|
else:
|
|
# Default relative to current working directory
|
|
self.cache_file = Path(".slop_cache/summary_cache.json")
|
|
self.max_entries = max_entries
|
|
self.cache: Dict[str, Dict[str, str]] = {}
|
|
self.load()
|
|
|
|
def load(self) -> Result[bool]:
|
|
"""Loads cache from disk."""
|
|
if not self.cache_file.exists():
|
|
return Result(data=False)
|
|
try:
|
|
with open(self.cache_file, "r", encoding="utf-8") as f:
|
|
self.cache = json.load(f)
|
|
return Result(data=True)
|
|
except (OSError, json.JSONDecodeError) as e:
|
|
self.cache = {}
|
|
return Result(data=False, errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source="summary_cache.load", original=e)])
|
|
|
|
def save(self) -> Result[bool]:
|
|
"""Saves cache to disk."""
|
|
try:
|
|
self.cache_file.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(self.cache_file, "w", encoding="utf-8") as f:
|
|
json.dump(self.cache, f, indent=1)
|
|
return Result(data=True)
|
|
except OSError as e:
|
|
return Result(data=False, errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source="summary_cache.save", original=e)])
|
|
|
|
def get_summary(self, file_path: str, content_hash: str) -> str:
|
|
"""Returns cached summary if hash matches, otherwise ""."""
|
|
entry = self.cache.get(file_path)
|
|
if entry and entry.get("hash") == content_hash:
|
|
# LRU: move to end
|
|
val = self.cache.pop(file_path)
|
|
self.cache[file_path] = val
|
|
return val.get("summary") or ""
|
|
return ""
|
|
|
|
def set_summary(self, file_path: str, content_hash: str, summary: str) -> None:
|
|
"""Stores summary in cache and saves to disk."""
|
|
if file_path in self.cache:
|
|
self.cache.pop(file_path)
|
|
self.cache[file_path] = {
|
|
"hash": content_hash,
|
|
"summary": summary
|
|
}
|
|
# Enforce LRU size limit
|
|
while len(self.cache) > self.max_entries:
|
|
# pop first item (oldest)
|
|
first_key = next(iter(self.cache))
|
|
self.cache.pop(first_key)
|
|
self.save()
|
|
|
|
def clear(self) -> Result[bool]:
|
|
"""Clears the cache both in-memory and on disk."""
|
|
self.cache.clear()
|
|
if not self.cache_file.exists():
|
|
return Result(data=True)
|
|
try:
|
|
self.cache_file.unlink()
|
|
return Result(data=True)
|
|
except OSError as e:
|
|
return Result(data=False, errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source="summary_cache.clear", original=e)])
|
|
|
|
def get_stats(self) -> Result[dict]:
|
|
"""Returns dictionary of cache statistics."""
|
|
size_bytes = 0
|
|
if self.cache_file.exists():
|
|
try:
|
|
size_bytes = self.cache_file.stat().st_size
|
|
except OSError as e:
|
|
return Result(data={"entries": len(self.cache), "size_bytes": 0}, errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source="summary_cache.get_stats", original=e)])
|
|
return Result(data={
|
|
"entries": len(self.cache),
|
|
"size_bytes": size_bytes
|
|
})
|