| from __future__ import annotations |
|
|
| import json |
| from pathlib import Path |
| from typing import Any |
|
|
| from .models import new_id, utc_now |
|
|
|
|
| NOTE_TYPES = ("observation", "hypothesis", "action", "outcome", "question", "guide", "issue") |
|
|
|
|
| class ResearchNotepad: |
| def __init__(self, path: Path): |
| self.path = path |
| self.path.parent.mkdir(parents=True, exist_ok=True) |
|
|
| def _read(self) -> list[dict[str, Any]]: |
| if not self.path.exists(): |
| return [] |
| try: |
| raw = json.loads(self.path.read_text(encoding="utf-8")) |
| except Exception: |
| return [] |
| if isinstance(raw, list): |
| return [item for item in raw if isinstance(item, dict)] |
| if isinstance(raw, dict): |
| for key in ("entries", "notes", "items"): |
| value = raw.get(key) |
| if isinstance(value, list): |
| return [item for item in value if isinstance(item, dict)] |
| return [] |
|
|
| def _write(self, entries: list[dict[str, Any]]) -> None: |
| temp = self.path.with_name(f".{self.path.name}.{new_id('tmp')}") |
| temp.write_text(json.dumps(entries, indent=2, sort_keys=True), encoding="utf-8") |
| temp.replace(self.path) |
|
|
| def list(self, query: str = "", limit: int = 80, note_type: str = "") -> list[dict[str, Any]]: |
| entries = self._read() |
| if note_type: |
| entries = [entry for entry in entries if str(entry.get("type") or "") == note_type] |
| tokens = [token for token in query.lower().replace("/", " ").replace("_", " ").split() if token] |
| if tokens: |
| filtered = [] |
| for entry in entries: |
| hay = json.dumps(entry, sort_keys=True, default=str).lower() |
| if all(token in hay for token in tokens): |
| filtered.append(entry) |
| entries = filtered |
| return entries[-max(1, min(int(limit or 80), 500)) :] |
|
|
| def add(self, payload: dict[str, Any], source: str = "operator") -> dict[str, Any]: |
| note_type = str(payload.get("type") or "observation").strip().lower() |
| if note_type not in NOTE_TYPES: |
| note_type = "observation" |
| entry = { |
| "id": str(payload.get("id") or new_id("note")), |
| "type": note_type, |
| "title": str(payload.get("title") or note_type), |
| "content": str(payload.get("content") or payload.get("text") or payload.get("message") or ""), |
| "tags": list(payload.get("tags") or []), |
| "links": list(payload.get("links") or []), |
| "confidence": payload.get("confidence"), |
| "source": source, |
| "created_at": str(payload.get("created_at") or utc_now()), |
| "updated_at": utc_now(), |
| } |
| entries = self._read() |
| entries.append(entry) |
| self._write(entries) |
| return entry |
|
|
| def clear(self) -> dict[str, Any]: |
| count = len(self._read()) |
| self._write([]) |
| return {"ok": True, "cleared_notes": count} |
|
|
| def summary(self) -> dict[str, Any]: |
| entries = self._read() |
| by_type: dict[str, int] = {} |
| for entry in entries: |
| by_type[str(entry.get("type") or "observation")] = by_type.get(str(entry.get("type") or "observation"), 0) + 1 |
| latest = entries[-8:] |
| open_questions = [entry for entry in entries if str(entry.get("type") or "") == "question"][-8:] |
| hypotheses = [entry for entry in entries if str(entry.get("type") or "") == "hypothesis"][-8:] |
| outcomes = [entry for entry in entries if str(entry.get("type") or "") == "outcome"][-8:] |
| return { |
| "schema": "key_os.research_notepad.summary/v1", |
| "total": len(entries), |
| "by_type": by_type, |
| "latest": latest, |
| "open_questions": open_questions, |
| "hypotheses": hypotheses, |
| "outcomes": outcomes, |
| } |
|
|
