File size: 4,476 Bytes
7e2f74d | 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 | import os
import json
from typing import Dict, Any, Optional
class RepositoryMemoryService:
"""
Foundational data and memory layer for repository intelligence.
Manages in-memory caching with disk persistence for structured
artifacts (profile, graph, summary, and report).
"""
def __init__(self):
self._memory_db: Dict[str, Dict[str, Any]] = {}
self.storage_dir = os.path.join(
os.path.dirname(os.path.dirname(__file__)), "storage", "repos"
)
os.makedirs(self.storage_dir, exist_ok=True)
self._load_persisted()
def _repo_dir(self, repo_id: str) -> str:
return os.path.join(self.storage_dir, repo_id)
def _load_persisted(self) -> None:
if not os.path.isdir(self.storage_dir):
return
for repo_id in os.listdir(self.storage_dir):
repo_dir = self._repo_dir(repo_id)
if not os.path.isdir(repo_dir):
continue
try:
profile_path = os.path.join(repo_dir, "repository_profile.json")
graph_path = os.path.join(repo_dir, "repository_graph.json")
summary_path = os.path.join(repo_dir, "repository_summary.json")
report_path = os.path.join(repo_dir, "repository_report.md")
if not all(os.path.exists(p) for p in (profile_path, graph_path, summary_path, report_path)):
continue
with open(profile_path, "r", encoding="utf-8") as f:
profile = json.load(f)
with open(graph_path, "r", encoding="utf-8") as f:
graph = json.load(f)
with open(summary_path, "r", encoding="utf-8") as f:
summary = json.load(f)
with open(report_path, "r", encoding="utf-8") as f:
report = f.read()
self._memory_db[repo_id] = {
"repo_id": repo_id,
"profile": profile,
"graph": graph,
"summary": summary,
"report": report,
}
except (json.JSONDecodeError, OSError):
continue
def store(
self,
repo_id: str,
profile: Dict[str, Any],
graph: Dict[str, Any],
summary: Dict[str, Any],
report_markdown: str
) -> Dict[str, Any]:
"""
Stores repository intelligence artifacts in memory and on disk.
"""
intelligence_payload = {
"repo_id": repo_id,
"profile": profile,
"graph": graph,
"summary": summary,
"report": report_markdown
}
self._memory_db[repo_id] = intelligence_payload
self.export_as_json_files(repo_id, self._repo_dir(repo_id))
return intelligence_payload
def retrieve(self, repo_id: str) -> Optional[Dict[str, Any]]:
"""
Retrieves repository intelligence artifacts by repo ID.
"""
return self._memory_db.get(repo_id)
def export_as_json_files(self, repo_id: str, output_dir: str) -> Dict[str, str]:
"""
Writes structured intelligence files to a target directory.
"""
data = self.retrieve(repo_id)
if not data:
raise ValueError(f"No intelligence artifacts found for repo: {repo_id}")
os.makedirs(output_dir, exist_ok=True)
profile_path = os.path.join(output_dir, "repository_profile.json")
graph_path = os.path.join(output_dir, "repository_graph.json")
summary_path = os.path.join(output_dir, "repository_summary.json")
report_path = os.path.join(output_dir, "repository_report.md")
with open(profile_path, "w", encoding="utf-8") as f:
json.dump(data["profile"], f, indent=2)
with open(graph_path, "w", encoding="utf-8") as f:
json.dump(data["graph"], f, indent=2)
with open(summary_path, "w", encoding="utf-8") as f:
json.dump(data["summary"], f, indent=2)
with open(report_path, "w", encoding="utf-8") as f:
f.write(data["report"])
return {
"profile": profile_path,
"graph": graph_path,
"summary": summary_path,
"report": report_path
}
# Global singleton service instance for easy access across the controllers
memory_service = RepositoryMemoryService()
|