"""Snippet service — reusable functions/components recorded for recall. A *snippet* is a Note with ``note_type='snippet'``: a named, reusable function or component recorded once so a later session can recall it before writing a one-off. It carries a name, language, signature, canonical location (repo · path · symbol), a one-line "when to reach for it", and the code itself. Structured fields are stored as a **body-convention** — no dedicated column, so a snippet inherits everything a note has (embeddings, ACL, project/System association, dedup) and, crucially, becomes eligible for semantic recall the moment it's embedded: - ``title`` = ``"{name} — {when_to_use}"``. The title is exactly what the title-first auto-inject surfaces, so this one line self-describes the snippet in a recall menu. - ``tags`` = ``[language, "snippet", *caller_tags]``. - ``body`` = templated markdown (When to use / Signature / Location, then a fenced code block). The public field API (name/language/signature/location/when_to_use/code) lives here, so the underlying storage could later move to a structured column without changing any caller (MCP tool, REST route, UI). """ from __future__ import annotations import re from scribe.services import knowledge as knowledge_svc from scribe.services import notes as notes_svc SNIPPET_NOTE_TYPE = "snippet" SNIPPET_TAG = "snippet" # --- serialize: structured fields -> note (title/body/tags) ------------------ def compose_title(name: str, when_to_use: str = "") -> str: """`name — when to use` (or just `name` when no usage note is given).""" name = (name or "").strip() when = (when_to_use or "").strip() return f"{name} — {when}" if when else name def compose_tags(language: str = "", tags: list[str] | None = None) -> list[str]: """Language (lowercased) first, then the `snippet` marker, then caller tags — de-duplicated, order preserved.""" out: list[str] = [] lang = (language or "").strip().lower() if lang: out.append(lang) out.append(SNIPPET_TAG) for t in tags or []: t = (t or "").strip() if t and t not in out: out.append(t) return out def _location_line(repo: str = "", path: str = "", symbol: str = "") -> str: parts = [p.strip() for p in (repo, path, symbol) if (p or "").strip()] return " · ".join(f"`{p}`" for p in parts) def compose_body( *, code: str, language: str = "", signature: str = "", when_to_use: str = "", repo: str = "", path: str = "", symbol: str = "", ) -> str: """Render structured fields into the snippet body markdown. Empty fields are omitted so the body stays clean.""" header: list[str] = [] if (when_to_use or "").strip(): header.append(f"**When to use:** {when_to_use.strip()}") if (signature or "").strip(): header.append(f"**Signature:** `{signature.strip()}`") loc = _location_line(repo, path, symbol) if loc: header.append(f"**Location:** {loc}") fence_lang = (language or "").strip().lower() code_block = f"```{fence_lang}\n{(code or '').rstrip()}\n```" if header: return "\n\n".join(header) + "\n\n" + code_block + "\n" return code_block + "\n" # --- parse: note -> structured fields (best-effort, never raises) ------------ _WHEN_RE = re.compile(r"^\*\*When to use:\*\*\s*(.+?)\s*$", re.MULTILINE) _SIG_RE = re.compile(r"^\*\*Signature:\*\*\s*(.+?)\s*$", re.MULTILINE) _LOC_RE = re.compile(r"^\*\*Location:\*\*\s*(.+?)\s*$", re.MULTILINE) _CODE_RE = re.compile(r"```([\w+.#-]*)\n(.*?)\n```", re.DOTALL) def parse_snippet_fields( title: str, body: str, tags: list[str] | None = None ) -> dict: """Recover structured fields from a snippet note. Tolerant by design: a field that isn't present comes back empty and this never raises, so a hand-edited body can't break the edit form.""" title = title or "" body = body or "" name, _, when_from_title = title.partition(" — ") fields = { "name": name.strip(), "when_to_use": when_from_title.strip(), "signature": "", "language": "", "repo": "", "path": "", "symbol": "", "code": "", } m = _WHEN_RE.search(body) if m: fields["when_to_use"] = m.group(1).strip() m = _SIG_RE.search(body) if m: fields["signature"] = m.group(1).strip().strip("`").strip() m = _LOC_RE.search(body) if m: loc_parts = [p.strip().strip("`").strip() for p in m.group(1).split("·")] for key, val in zip(("repo", "path", "symbol"), loc_parts): fields[key] = val m = _CODE_RE.search(body) if m: fields["language"] = m.group(1).strip() fields["code"] = m.group(2) if not fields["language"]: for t in tags or []: if t and t != SNIPPET_TAG: fields["language"] = t break return fields def snippet_to_dict(note) -> dict: """Note serialization plus a parsed ``snippet`` sub-object of structured fields, so callers get both the raw record and the typed view.""" data = note.to_dict() data["snippet"] = parse_snippet_fields(note.title, note.body, note.tags) return data # --- service wrappers over notes_svc ---------------------------------------- async def create_snippet( user_id: int, *, name: str, code: str, language: str = "", signature: str = "", when_to_use: str = "", repo: str = "", path: str = "", symbol: str = "", tags: list[str] | None = None, project_id: int | None = None, ): """Create a snippet note. Returns the created Note.""" return await notes_svc.create_note( user_id, title=compose_title(name, when_to_use), body=compose_body( code=code, language=language, signature=signature, when_to_use=when_to_use, repo=repo, path=path, symbol=symbol, ), note_type=SNIPPET_NOTE_TYPE, tags=compose_tags(language, tags), project_id=project_id, ) async def get_snippet(user_id: int, snippet_id: int): """Fetch a snippet by id, or None if it doesn't exist / isn't a snippet.""" note = await notes_svc.get_note(user_id, snippet_id) if note is None or note.note_type != SNIPPET_NOTE_TYPE: return None return note async def list_snippets( user_id: int, *, q: str | None = None, tag: str = "", limit: int = 50, offset: int = 0, ) -> tuple[list[dict], int]: """List snippets (id/title/tags/preview dicts), most-recently-updated first.""" return await knowledge_svc.query_knowledge( user_id=user_id, note_type=SNIPPET_NOTE_TYPE, tags=[tag] if tag else [], sort="modified", q=q, limit=max(1, min(limit, 100)), offset=max(0, offset), ) async def update_snippet( user_id: int, snippet_id: int, *, name: str | None = None, code: str | None = None, language: str | None = None, signature: str | None = None, when_to_use: str | None = None, repo: str | None = None, path: str | None = None, symbol: str | None = None, tags: list[str] | None = None, project_id: int | None = None, ): """Partial update: only fields passed (not None) change. Re-serializes the merged field set back into title/body/tags. Returns the Note, or None if the id isn't a snippet.""" note = await notes_svc.get_note(user_id, snippet_id) if note is None or note.note_type != SNIPPET_NOTE_TYPE: return None cur = parse_snippet_fields(note.title, note.body, note.tags) overlay = { "name": name, "code": code, "language": language, "signature": signature, "when_to_use": when_to_use, "repo": repo, "path": path, "symbol": symbol, } merged = {**cur, **{k: v for k, v in overlay.items() if v is not None}} fields: dict = { "title": compose_title(merged["name"], merged["when_to_use"]), "body": compose_body( code=merged["code"], language=merged["language"], signature=merged["signature"], when_to_use=merged["when_to_use"], repo=merged["repo"], path=merged["path"], symbol=merged["symbol"], ), } # Recompute tags: keep any non-language, non-marker tags the note already had # (or the caller's replacement set), then re-derive language + marker. existing_extra = [ t for t in (note.tags or []) if t not in (SNIPPET_TAG, cur.get("language", "")) ] fields["tags"] = compose_tags( merged["language"], tags if tags is not None else existing_extra ) if project_id is not None: fields["project_id"] = project_id return await notes_svc.update_note(user_id, snippet_id, **fields)