Topic 3: The running project

ls notes/
cat notes/lab-sync-2026-09-02.md

"""Business logic for the research notes assistant.

This module knows nothing about MCP. It reads and writes a folder of Markdown
notes, so it can be unit tested and reused by any server, host, or script.
"""
from __future__ import annotations

import re
from dataclasses import dataclass, field
from datetime import date
from pathlib import Path

NOTE_ID_PATTERN = re.compile(r"^[a-z0-9][a-z0-9-]{0,63}$")
WORD_PATTERN = re.compile(r"[a-z0-9]+")
MAX_TITLE_LENGTH = 120
MAX_BODY_LENGTH = 20_000


class NoteError(Exception):
    """Base class for every error the store raises on purpose."""


class NoteNotFound(NoteError):
    """Raised when a note id does not match any file."""


class NoteExists(NoteError):
    """Raised when creating a note whose id is already taken."""


class InvalidNote(NoteError):
    """Raised when a note id, title, body, or tag is not acceptable."""


@dataclass(frozen=True)
class Note:
    note_id: str
    title: str
    tags: tuple[str, ...]
    created: str
    body: str

    def snippet(self, length: int = 120) -> str:
        """Return the first line of the body, cut to `length` characters."""
        first_line = self.body.strip().splitlines()[0] if self.body.strip() else ""
        return first_line if len(first_line) <= length else first_line[: length - 3] + "..."


@dataclass(frozen=True)
class SearchHit:
    note_id: str
    title: str
    score: int
    snippet: str
    tags: tuple[str, ...] = field(default_factory=tuple)


def slugify(title: str) -> str:
    """Turn a title into a note id: lowercase words joined by hyphens."""
    words = WORD_PATTERN.findall(title.lower())
    return "-".join(words)[:64].strip("-")


def validate_note_id(note_id: str) -> str:
    """Reject ids that could escape the folder or are not simple slugs."""
    if not NOTE_ID_PATTERN.fullmatch(note_id):
        raise InvalidNote(
            f"Invalid note id {note_id!r}. Use lowercase letters, digits, and hyphens, "
            "for example 'sleep-and-memory'."
        )
    return note_id


def _parse(note_id: str, text: str) -> Note:
    """Split a Markdown file into frontmatter fields and body."""
    meta: dict[str, str] = {}
    body = text
    if text.startswith("---\n"):
        header, _, body = text[4:].partition("\n---\n")
        for line in header.splitlines():
            key, _, value = line.partition(":")
            meta[key.strip()] = value.strip()
    raw_tags = meta.get("tags", "").strip("[]")
    tags = tuple(t.strip() for t in raw_tags.split(",") if t.strip())
    return Note(
        note_id=note_id,
        title=meta.get("title", note_id),
        tags=tags,
        created=meta.get("created", ""),
        body=body.strip(),
    )


class NoteStore:
    """A folder of Markdown notes, one note per file, named <note_id>.md."""

    def __init__(self, root: Path | str) -> None:
        self.root = Path(root)
        if not self.root.is_dir():
            raise NoteError(f"Notes folder not found: {self.root}")

    def _path(self, note_id: str) -> Path:
        return self.root / f"{validate_note_id(note_id)}.md"

    def list_notes(self) -> list[Note]:
        """Every note, sorted by id so the order is deterministic."""
        return [self.get(p.stem) for p in sorted(self.root.glob("*.md"))]

    def get(self, note_id: str) -> Note:
        path = self._path(note_id)
        if not path.is_file():
            raise NoteNotFound(f"No note with id {note_id!r}.")
        return _parse(note_id, path.read_text(encoding="utf-8"))

    def search(self, query: str, limit: int = 5, tag: str | None = None) -> list[SearchHit]:
        """Rank notes by how often the query words appear (title words count 3 times)."""
        terms = WORD_PATTERN.findall(query.lower())
        if not terms:
            raise InvalidNote("Query must contain at least one letter or digit.")
        hits: list[SearchHit] = []
        for note in self.list_notes():
            if tag and tag not in note.tags:
                continue
            body_words = WORD_PATTERN.findall(note.body.lower())
            title_words = WORD_PATTERN.findall(note.title.lower())
            tag_words = [t.lower() for t in note.tags]
            score = sum(
                body_words.count(t) + 3 * title_words.count(t) + 2 * tag_words.count(t)
                for t in terms
            )
            if score > 0:
                hits.append(SearchHit(note.note_id, note.title, score, note.snippet(), note.tags))
        hits.sort(key=lambda h: (-h.score, h.note_id))
        return hits[:limit]

    def create(self, title: str, body: str, tags: list[str] | tuple[str, ...] = ()) -> Note:
        """Write a new note file. Never overwrites an existing note."""
        title = title.strip()
        if not title or len(title) > MAX_TITLE_LENGTH:
            raise InvalidNote(f"Title must be 1 to {MAX_TITLE_LENGTH} characters.")
        if not body.strip() or len(body) > MAX_BODY_LENGTH:
            raise InvalidNote(f"Body must be 1 to {MAX_BODY_LENGTH} characters.")
        clean_tags = tuple(slugify(t) for t in tags if slugify(t))
        note_id = slugify(title)
        if not note_id:
            raise InvalidNote("Title must contain at least one letter or digit.")
        path = self._path(note_id)
        if path.exists():
            raise NoteExists(f"A note with id {note_id!r} already exists.")
        header = (
            f"---\ntitle: {title}\ntags: [{', '.join(clean_tags)}]\n"
            f"created: {date.today().isoformat()}\n---\n"
        )
        path.write_text(header + body.strip() + "\n", encoding="utf-8")
        return self.get(note_id)

"""Module 1: a tour of NoteStore, the business logic every later module builds on.

Run from the repository root:  PYTHONPATH=. python examples/m01_store_tour.py
"""
from notes_assistant.store import InvalidNote, NoteNotFound, NoteStore

store = NoteStore("notes")

print("All notes:")
for note in store.list_notes():
    print(f"  {note.note_id:<24} {note.created}  tags={list(note.tags)}")

print("\nsearch('sleep memory', limit=3):")
for hit in store.search("sleep memory", limit=3):
    print(f"  score={hit.score:>2}  {hit.note_id:<24} {hit.snippet[:60]}")

print("\nsearch('sleep', tag='meeting'):")
for hit in store.search("sleep", tag="meeting"):
    print(f"  score={hit.score:>2}  {hit.note_id}")

note = store.get("lab-sync-2026-09-02")
print(f"\nget('lab-sync-2026-09-02').title = {note.title!r}")
print(f"body has {len(note.body.splitlines())} lines")

for bad in ["../secrets", "no-such-note"]:
    try:
        store.get(bad)
    except (InvalidNote, NoteNotFound) as exc:
        print(f"\nget({bad!r}) raised {type(exc).__name__}: {exc}")

print(f"\nsearch('quantum') returned {store.search('quantum')}")

# create() writes a file, so try it in a throwaway folder, never in notes/.
import tempfile  # noqa: E402

from notes_assistant.store import NoteExists  # noqa: E402

with tempfile.TemporaryDirectory() as scratch:
    scratch_store = NoteStore(scratch)
    created = scratch_store.create("Nap study consent form", "Draft due 15 September.", tags=["Meeting", "sleep"])
    print(f"\ncreated {created.note_id!r} with tags {list(created.tags)}")
    try:
        scratch_store.create("Nap study: consent form!", "Second try.")
    except NoteExists as exc:
        print(f"second create raised NoteExists: {exc}")

"""One helper for three LLM providers: Groq, Gemini, and Ollama.

All three expose an OpenAI-compatible Chat Completions endpoint, so a single
client library (`openai`) talks to each of them. Pick a provider with the
LLM_PROVIDER environment variable and override the model with LLM_MODEL.
"""
from __future__ import annotations

import json
import os
from dataclasses import dataclass, field
from typing import Any

from openai import OpenAI

PROVIDERS: dict[str, dict[str, str]] = {
    "groq": {
        "base_url": "https://api.groq.com/openai/v1",
        "key_env": "GROQ_API_KEY",
        "default_model": "llama-3.3-70b-versatile",
    },
    "gemini": {
        "base_url": "https://generativelanguage.googleapis.com/v1beta/openai/",
        "key_env": "GEMINI_API_KEY",
        "default_model": "gemini-2.5-flash",
    },
    "ollama": {
        "base_url": "http://localhost:11434/v1",
        "key_env": "",
        "default_model": "qwen3:8b",
    },
}


@dataclass
class ToolCall:
    id: str
    name: str
    arguments: dict[str, Any]


@dataclass
class ChatReply:
    content: str | None
    tool_calls: list[ToolCall] = field(default_factory=list)

    def as_message(self) -> dict[str, Any]:
        """The assistant message to append to the conversation history."""
        message: dict[str, Any] = {"role": "assistant", "content": self.content or ""}
        if self.tool_calls:
            message["tool_calls"] = [
                {
                    "id": call.id,
                    "type": "function",
                    "function": {"name": call.name, "arguments": json.dumps(call.arguments)},
                }
                for call in self.tool_calls
            ]
        return message


def make_client(provider: str | None = None) -> tuple[OpenAI, str]:
    """Build an OpenAI-compatible client for the chosen provider and return it with the model name."""
    name = (provider or os.environ.get("LLM_PROVIDER", "groq")).lower()
    if name not in PROVIDERS:
        raise ValueError(f"Unknown LLM_PROVIDER {name!r}. Choose one of {sorted(PROVIDERS)}.")
    settings = PROVIDERS[name]
    api_key = os.environ.get(settings["key_env"], "") if settings["key_env"] else "ollama"
    if not api_key:
        raise RuntimeError(f"Set {settings['key_env']} in your environment to use {name}.")
    base_url = os.environ.get("OLLAMA_BASE_URL", settings["base_url"]) if name == "ollama" else settings["base_url"]
    model = os.environ.get("LLM_MODEL", settings["default_model"])
    return OpenAI(api_key=api_key, base_url=base_url), model


def chat(
    messages: list[dict[str, Any]],
    tools: list[dict[str, Any]] | None = None,
    provider: str | None = None,
    temperature: float = 0.0,
) -> ChatReply:
    """Send a conversation (and optional tool definitions) and return text or tool calls."""
    client, model = make_client(provider)
    kwargs: dict[str, Any] = {"model": model, "messages": messages, "temperature": temperature}
    if tools:
        kwargs["tools"] = tools
    response = client.chat.completions.create(**kwargs)
    message = response.choices[0].message
    calls = [
        ToolCall(id=c.id, name=c.function.name, arguments=json.loads(c.function.arguments or "{}"))
        for c in (message.tool_calls or [])
        if c.type == "function"
    ]
    return ChatReply(content=message.content, tool_calls=calls)