Spaces:
Sleeping
Sleeping
File size: 6,258 Bytes
2e818da | 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 | from __future__ import annotations
import json
import os
import time
from typing import Any
from app.schemas.project import Project
from app.services.agent_control import AgentControlService
from app.services.annotation_service import get_annotation_service
from app.services.citation_graph import CitationGraphService
from app.services.project_activity import ProjectActivityService
_DEFAULT_ROOT = os.path.expanduser("~/.studybuddy/projects/intelligence")
class ProjectIntelligenceService:
"""Student-facing project status: brief, useful counts, and settings summary."""
def __init__(
self,
root: str | None = None,
citation_service: CitationGraphService | None = None,
agent_control: AgentControlService | None = None,
) -> None:
self.root = root or _DEFAULT_ROOT
self.citation_service = citation_service or CitationGraphService()
self.agent_control = agent_control or AgentControlService()
os.makedirs(self.root, exist_ok=True)
def _path(self, project_id: str) -> str:
safe = project_id.replace("/", "_").replace("\\", "_").replace("..", "_")
return os.path.join(self.root, f"{safe}.json")
def load_brief(self, project_id: str) -> dict[str, Any]:
path = self._path(project_id)
if not os.path.exists(path):
return {"text": "", "updated_at": 0.0, "source": "empty"}
try:
with open(path, encoding="utf-8") as f:
data = json.load(f)
return {
"text": str(data.get("text") or ""),
"updated_at": float(data.get("updated_at") or 0),
"source": str(data.get("source") or "commit"),
}
except Exception:
return {"text": "", "updated_at": 0.0, "source": "error"}
def save_brief(self, project_id: str, text: str, source: str = "commit") -> dict[str, Any]:
brief = {"text": text.strip(), "updated_at": time.time(), "source": source}
with open(self._path(project_id), "w", encoding="utf-8") as f:
json.dump(brief, f, indent=2)
return brief
def update_from_commit(self, project: Project, topic: str, session_summary: str) -> dict[str, Any]:
graph = self.citation_service.load(project.project_id)
paper_titles = [node.title or node.label for node in graph.nodes if node.type == "paper"]
connector_titles = [node.label for node in graph.nodes if node.type != "paper"][:6]
summary = " ".join((session_summary or "").strip().split())[:700]
subject = topic or project.name or "this project"
if paper_titles:
text = f"This project is currently about {subject}, grounded in {', '.join(paper_titles[:3])}."
else:
text = f"This project is currently about {subject}."
if connector_titles:
text += f" The strongest shared threads are {', '.join(connector_titles)}."
if summary:
text += f" Latest Commit signal: {summary}"
else:
text += " Commit again after study activity to deepen this summary."
return self.save_brief(project.project_id, text)
def status(self, project: Project) -> dict[str, Any]:
graph = self.citation_service.load(project.project_id)
annotations = []
annotation_service = get_annotation_service()
for file in project.files:
annotations.extend(annotation_service.get_for_document(file.file_id))
notes = [a for a in annotations if (a.note_text or "").strip()]
pinned_figures = [a for a in annotations if a.image_base64]
highlighted_passages = [
a for a in annotations
if a.target_snippets and any((snippet.text or "").strip() for snippet in a.target_snippets)
]
activity = ProjectActivityService().list(project.project_id, limit=200)
stats = {
"papers": len(project.files),
"notes": len(notes),
"pinned_figures": len(pinned_figures),
"highlighted_passages": len(highlighted_passages),
"commits": sum(1 for event in activity if event.event_type in {"commit", "persona_update", "memory_flush"}),
"recommendation_adds": sum(1 for event in activity if event.event_type == "paper_added" and "recommended" in event.message.lower()),
"verified_citation_links": sum(1 for edge in graph.edges if edge.relation == "cites"),
"unresolved_verified_references": sum(
len(node.metadata.get("verified_unmatched_references", []))
for node in graph.nodes
if node.type == "paper"
),
}
brief = self.load_brief(project.project_id)
if not brief["text"]:
titles = [node.title or node.label for node in graph.nodes if node.type == "paper"]
if titles:
brief = {
"text": (
f"ResearchMate has indexed {len(titles)} paper"
f"{'' if len(titles) == 1 else 's'}: {', '.join(titles[:3])}. "
"Commit this project to write a deeper project brief from your study activity."
),
"updated_at": 0.0,
"source": "metadata",
}
else:
brief = {
"text": "Add papers to let ResearchMate build a project brief. Commit after studying to deepen it.",
"updated_at": 0.0,
"source": "empty",
}
return {
"brief": brief,
"stats": stats,
"settings_summary": self._settings_summary(project.project_id),
}
def _settings_summary(self, project_id: str) -> dict[str, int]:
profile = self.agent_control.get_profile(project_id)
skills = self.agent_control.get_skills(project_id).get("skills", {})
manual_profiles = sum(1 for key in ("student", "project") if profile.get(key, {}).get("manual_text"))
custom_skills = sum(1 for record in skills.values() if record.get("manual_text"))
return {"manual_profiles": manual_profiles, "custom_skills": custom_skills}
|