Spaces:
Sleeping
Sleeping
| 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} | |