Granola

{
  "name": "granola",
  "description": "This skill should be used when importing, listing, or exporting Granola meeting recordings and transcripts. Queries Gran",
  "author": {
    "name": "Gleb Kalinin"
  },
  "repository": "https://github.com/glebis/claude-skills",
  "license": "MIT"
}

#!/usr/bin/env python3
"""Granola meeting notes CLI — Personal API, export to Obsidian."""

import json
import sys
import os
import argparse
import subprocess
import urllib.request
import urllib.parse
from pathlib import Path
from datetime import datetime

PUBLIC_API_BASE = "https://public-api.granola.ai/v1"
SOPS_ENV_PATH = os.path.expanduser("~/Brains/brain/.env.granola")


def _get_api_key():
    """Decrypt Personal API Key from sops-encrypted .env.granola."""
    try:
        result = subprocess.run(
            ["sops", "-d", SOPS_ENV_PATH],
            capture_output=True, text=True, timeout=10,
        )
        if result.returncode == 0:
            for line in result.stdout.strip().splitlines():
                if line.startswith("GRANOLA_API_KEY="):
                    return line.split("=", 1)[1].strip()
    except (FileNotFoundError, subprocess.TimeoutExpired):
        pass
    print("ERROR: Cannot decrypt Granola API key from", SOPS_ENV_PATH, file=sys.stderr)
    sys.exit(1)


def api_get(path, params=None):
    """GET request to Granola Personal API."""
    key = _get_api_key()
    url = f"{PUBLIC_API_BASE}{path}"
    if params:
        url += "?" + urllib.parse.urlencode(params)
    req = urllib.request.Request(
        url,
        headers={"Authorization": f"Bearer {key}"},
        method="GET",
    )
    try:
        with urllib.request.urlopen(req) as resp:
            return json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as e:
        body = e.read().decode("utf-8", errors="replace")
        print(f"API error {e.code}: {body}", file=sys.stderr)
        sys.exit(1)


def format_time(iso_str):
    if not iso_str:
        return "?"
    try:
        return datetime.fromisoformat(iso_str).strftime("%H:%M")
    except (ValueError, TypeError):
        return "?"


def cmd_list(args):
    """List meetings via Personal API."""
    all_notes = []
    cursor = None

    while True:
        params = {}
        if cursor:
            params["cursor"] = cursor
        if args.after:
            params["created_after"] = args.after

        data = api_get("/notes", params if params else None)
        notes = data.get("notes", [])
        all_notes.extend(notes)

        if not args.all or not data.get("hasMore"):
            break
        cursor = data.get("cursor")
        if not cursor:
            break

    if args.format == "json":
        print(json.dumps({"notes": all_notes, "count": len(all_notes)}, ensure_ascii=False, indent=2))
    else:
        print(f"Found {len(all_notes)} notes:\n")
        for n in all_notes:
            date = n.get("created_at", "")[:10]
            title = n.get("title") or "(Untitled)"
            owner = n.get("owner", {}).get("name", "")
            nid = n.get("id", "")
            attendees = n.get("attendees", [])
            names = ", ".join(a.get("name", a.get("email", "")) for a in attendees[:4])
            att_str = f" with {names}" if names else ""
            print(f"  {date}  {title}{att_str}")
            print(f"    id: {nid}")


def cmd_show(args):
    """Show a single note with summary."""
    params = {"include": "transcript"} if args.transcript else None
    note = api_get(f"/notes/{args.note_id}", params)

    if args.format == "json":
        print(json.dumps(note, ensure_ascii=False, indent=2))
        return

    title = note.get("title") or "(Untitled)"
    date = note.get("created_at", "")[:10]
    owner = note.get("owner", {}).get("name", "")
    attendees = [a.get("name", a.get("email", "")) for a in note.get("attendees", [])]
    cal = note.get("calendar_event", {})

    print(f"# {title}")
    print(f"Date: {date}  Owner: {owner}")
    if cal:
        start = format_time(cal.get("start_time"))
        end = format_time(cal.get("end_time"))
        print(f"Time: {start}–{end}")
    if attendees:
        print(f"Attendees: {', '.join(attendees)}")
    print()

    summary = note.get("summary_markdown") or note.get("summary_text") or ""
    if summary:
        print("## Summary\n")
        print(summary)
        print()

    transcript = note.get("transcript")
    if transcript:
        print("## Transcript\n")
        for u in transcript:
            ts = format_time(u.get("start_time", ""))
            text = u.get("text", "").strip()
            speaker = u.get("speaker", {})
            source = speaker.get("source", "") if isinstance(speaker, dict) else ""
            label = speaker.get("diarization_label", "") if isinstance(speaker, dict) else ""
            tag = label or source or ""
            tag_str = f" [{tag}]" if tag else ""
            print(f"[{ts}]{tag_str} {text}")


def cmd_export(args):
    """Export note to Obsidian markdown (Fathom-compatible format)."""
    note = api_get(f"/notes/{args.note_id}", {"include": "transcript"})

    title = note.get("title") or "(Untitled)"
    note_id = note.get("id", "")
    created_at = note.get("created_at", "")
    date_short = created_at[:10].replace("-", "")
    date_dash = created_at[:10]
    owner = note.get("owner", {}).get("name", "")
    attendees = note.get("attendees", [])
    participant_names = []
    if owner:
        participant_names.append(owner)
    for a in attendees:
        name = a.get("name", a.get("email", ""))
        if name and name not in participant_names:
            participant_names.append(name)

    cal = note.get("calendar_event", {})
    start_time = cal.get("start_time") if cal else None
    end_time = cal.get("end_time") if cal else None

    duration = None
    if start_time and end_time:
        try:
            t0 = datetime.fromisoformat(start_time)
            t1 = datetime.fromisoformat(end_time)
            mins = int((t1 - t0).total_seconds() / 60)
            duration = f"{mins // 60:02d}:{mins % 60:02d}"
        except (ValueError, TypeError):
            pass

    summary = note.get("summary_markdown") or note.get("summary_text") or ""
    transcript = note.get("transcript") or []

    slug = title.lower().strip()
    for ch in ".,!?:;'\"()[]{}":
        slug = slug.replace(ch, "")
    slug = slug.replace(" ", "-").replace("--", "-")[:60].rstrip("-")
    filename = f"{date_short}-{slug}.md"

    lines = ["---"]
    lines.append(f"granola_id: {note_id}")
    lines.append(f'title: "{title}"')
    lines.append(f"date: {date_dash}")
    if participant_names:
        lines.append(f"participants: {json.dumps(participant_names)}")
    if duration:
        lines.append(f"duration: {duration}")
    lines.append("source: granola")
    lines.append("---")
    lines.append("")
    lines.append(f"# {title}")
    lines.append("")

    if summary:
        lines.append("## Summary")
        lines.append("")
        lines.append(summary)
        lines.append("")

    if transcript:
        lines.append("## Transcript")
        lines.append("")
        for u in transcript:
            text = u.get("text", "").strip()
            if not text:
                continue
            speaker = u.get("speaker", {})
            source = speaker.get("source", "") if isinstance(speaker, dict) else ""
            label = speaker.get("diarization_label", "") if isinstance(speaker, dict) else ""
            if label:
                speaker_name = label
            elif source == "microphone":
                speaker_name = participant_names[0] if participant_names else "Speaker"
            elif source == "speaker":
                speaker_name = "Other"
            else:
                speaker_name = "Speaker"
            lines.append(f"**{speaker_name}**: {text}")
            lines.append("")

    content = "\n".join(lines)

    if args.output:
        out_path = Path(args.output)
    else:
        out_path = Path(args.vault) / filename

    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_text(content, encoding="utf-8")
    print(json.dumps({
        "exported": str(out_path),
        "title": title,
        "date": date_short,
        "participants": participant_names,
        "duration": duration,
        "utterances": len(transcript),
    }, ensure_ascii=False, indent=2))


def main():
    parser = argparse.ArgumentParser(description="Granola meeting notes CLI (Personal API)")
    sub = parser.add_subparsers(dest="command", required=True)

    p_list = sub.add_parser("list", help="List notes")
    p_list.add_argument("--format", choices=["text", "json"], default="text")
    p_list.add_argument("--after", help="ISO 8601 date filter (created_after)")
    p_list.add_argument("--all", action="store_true", help="Paginate through all results")
    p_list.set_defaults(func=cmd_list)

    p_show = sub.add_parser("show", help="Show note details")
    p_show.add_argument("note_id", help="Note ID (not_xxxx)")
    p_show.add_argument("--format", choices=["text", "json"], default="text")
    p_show.add_argument("--transcript", action="store_true", help="Include transcript")
    p_show.set_defaults(func=cmd_show)

    p_export = sub.add_parser("export", help="Export note to Obsidian")
    p_export.add_argument("note_id", help="Note ID (not_xxxx)")
    p_export.add_argument("--vault", default=os.path.expanduser("~/Brains/brain"))
    p_export.add_argument("--output", help="Custom output path")
    p_export.set_defaults(func=cmd_export)

    args = parser.parse_args()
    args.func(args)


if __name__ == "__main__":
    main()

#!/bin/bash
# Granola → Obsidian auto-sync
# Checks for new Granola meetings and exports any not yet in the vault.

set -euo pipefail

SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
GRANOLA_PY="$SCRIPT_DIR/granola.py"
VAULT="${GRANOLA_VAULT:-$HOME/Brains/brain}"
LOG="$HOME/Library/Logs/granola-sync.log"

log() {
    echo "$(date '+%Y-%m-%d %H:%M:%S') $1" >> "$LOG"
}

log "=== Sync started ==="

# Collect already-exported granola IDs from vault
KNOWN_IDS=$(grep -rh '^granola_id: ' "$VAULT"/*.md "$VAULT"/**/*.md 2>/dev/null \
    | sed 's/^granola_id: //' | sort -u || true)

# Get recent meetings from API
API_OUTPUT=$(python3 "$GRANOLA_PY" api-list --limit 20 2>>"$LOG") || {
    log "ERROR: api-list failed"
    exit 1
}

# Extract meeting IDs
MEETING_IDS=$(echo "$API_OUTPUT" | python3 -c "
import sys, json
data = json.load(sys.stdin)
for m in data.get('meetings', []):
    print(m['id'])
")

EXPORTED=0
SKIPPED=0

for mid in $MEETING_IDS; do
    if echo "$KNOWN_IDS" | grep -q "$mid"; then
        SKIPPED=$((SKIPPED + 1))
        continue
    fi
    log "Exporting: $mid"
    python3 "$GRANOLA_PY" export "$mid" --vault "$VAULT" 2>>"$LOG" && {
        EXPORTED=$((EXPORTED + 1))
        log "OK: $mid"
    } || {
        log "FAIL: $mid"
    }
done

log "Done: exported=$EXPORTED skipped=$SKIPPED"

{
  "cache": {
    "state": {
      "documents": {
        "def-456": {
          "id": "def-456",
          "title": "Another Meeting",
          "created_at": "2026-02-27T14:00:00Z",
          "people": {
            "creator": {"name": "Gleb Kalinin", "email": "glebis@gmail.com"},
            "attendees": [
              {"email": "alice@example.com", "details": {"person": {"name": {"fullName": "Alice Johnson"}}}}
            ]
          },
          "google_calendar_event": {
            "start": {"dateTime": "2026-02-27T14:00:00+01:00"},
            "end": {"dateTime": "2026-02-27T15:30:00+01:00"}
          },
          "notes_markdown": "",
          "summary": "",
          "type": "meeting"
        }
      },
      "transcripts": {},
      "meetingsMetadata": {}
    },
    "version": 5
  }
}

{
  "cache": {
    "state": "{\"documents\":{\"abc-123\":{\"id\":\"abc-123\",\"title\":\"Test Meeting\",\"created_at\":\"2026-02-28T10:00:00Z\",\"people\":{\"creator\":{\"name\":\"Gleb Kalinin\",\"email\":\"glebis@gmail.com\"},\"attendees\":[{\"email\":\"bob@example.com\",\"details\":{\"person\":{\"name\":{\"fullName\":\"Bob Smith\"}}}}]},\"google_calendar_event\":{\"start\":{\"dateTime\":\"2026-02-28T10:00:00+01:00\"},\"end\":{\"dateTime\":\"2026-02-28T11:00:00+01:00\"}},\"notes_markdown\":\"\",\"summary\":\"Test summary\",\"type\":\"meeting\"}},\"transcripts\":{\"abc-123\":[{\"id\":\"u1\",\"document_id\":\"abc-123\",\"start_timestamp\":\"2026-02-28T10:00:00Z\",\"end_timestamp\":\"2026-02-28T10:05:00Z\",\"text\":\"Hello everyone\",\"source\":\"microphone\",\"is_final\":true},{\"id\":\"u2\",\"document_id\":\"abc-123\",\"start_timestamp\":\"2026-02-28T10:05:00Z\",\"end_timestamp\":\"2026-02-28T10:10:00Z\",\"text\":\"Hi there\",\"source\":\"system\",\"is_final\":true}]},\"meetingsMetadata\":{}}",
    "version": 5
  }
}

import sys
import os
import json
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

import granola

FIXTURES = os.path.join(os.path.dirname(__file__), "fixtures")


def test_cmd_export_generates_fathom_compatible_markdown(monkeypatch, capsys, tmp_path):
    monkeypatch.setattr(granola, "CACHE_PATH",
                        os.path.join(FIXTURES, "cache_string_state.json"))

    output_file = tmp_path / "test-export.md"

    class Args:
        meeting_id = "abc"
        vault = str(tmp_path)
        output = str(output_file)
        local_only = True

    granola.cmd_export(Args())

    # Check JSON output
    result = json.loads(capsys.readouterr().out)
    assert result["title"] == "Test Meeting"
    assert result["utterances"] == 2
    assert "Gleb Kalinin" in result["participants"]

    # Check generated markdown
    content = output_file.read_text()

    # Frontmatter
    assert "granola_id: abc-123" in content
    assert 'title: "Test Meeting"' in content
    assert "source: granola" in content
    assert "date: 2026-02-28" in content

    # Transcript with speaker attribution
    assert "**Gleb Kalinin**: Hello everyone" in content
    assert "**Other**: Hi there" in content

    # Summary
    assert "Test summary" in content

import sys
import os
import json
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

import granola

FIXTURES = os.path.join(os.path.dirname(__file__), "fixtures")


def test_cmd_list_json_output(monkeypatch, capsys):
    monkeypatch.setattr(granola, "CACHE_PATH",
                        os.path.join(FIXTURES, "cache_string_state.json"))

    class Args:
        format = "json"

    granola.cmd_list(Args())
    output = json.loads(capsys.readouterr().out)

    assert len(output["meetings"]) == 1
    m = output["meetings"][0]
    assert m["id"] == "abc-123"
    assert m["title"] == "Test Meeting"
    assert m["has_local_transcript"] is True
    assert m["transcript_utterances"] == 2


def test_cmd_list_text_output(monkeypatch, capsys):
    monkeypatch.setattr(granola, "CACHE_PATH",
                        os.path.join(FIXTURES, "cache_string_state.json"))

    class Args:
        format = "text"

    granola.cmd_list(Args())
    out = capsys.readouterr().out

    assert "Test Meeting" in out
    assert "1 meetings" in out

import sys
import os
import json
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

import pytest
import granola

FIXTURES = os.path.join(os.path.dirname(__file__), "fixtures")


def test_cmd_show_finds_by_id_prefix(monkeypatch, capsys):
    monkeypatch.setattr(granola, "CACHE_PATH",
                        os.path.join(FIXTURES, "cache_string_state.json"))

    class Args:
        meeting_id = "abc"

    granola.cmd_show(Args())
    output = json.loads(capsys.readouterr().out)

    assert output["id"] == "abc-123"
    assert output["title"] == "Test Meeting"
    assert output["summary"] == "Test summary"


def test_cmd_show_finds_by_title_substring(monkeypatch, capsys):
    monkeypatch.setattr(granola, "CACHE_PATH",
                        os.path.join(FIXTURES, "cache_string_state.json"))

    class Args:
        meeting_id = "test meeting"

    granola.cmd_show(Args())
    output = json.loads(capsys.readouterr().out)
    assert output["id"] == "abc-123"


def test_cmd_show_exits_on_not_found(monkeypatch):
    monkeypatch.setattr(granola, "CACHE_PATH",
                        os.path.join(FIXTURES, "cache_string_state.json"))

    class Args:
        meeting_id = "nonexistent"

    with pytest.raises(SystemExit):
        granola.cmd_show(Args())

import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

from granola import compute_duration


def test_compute_duration_from_utterance_timestamps():
    utterances = [
        {"start_timestamp": "2026-02-28T10:00:00Z", "end_timestamp": "2026-02-28T10:05:00Z"},
        {"start_timestamp": "2026-02-28T10:30:00Z", "end_timestamp": "2026-02-28T11:27:00Z"},
    ]
    assert compute_duration(utterances) == "01:27"


def test_compute_duration_returns_none_for_empty_list():
    assert compute_duration([]) is None


def test_compute_duration_returns_none_for_missing_timestamps():
    utterances = [{"start_timestamp": "", "end_timestamp": ""}]
    assert compute_duration(utterances) is None

import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

from granola import extract_calendar_times


def test_extract_calendar_times_returns_start_and_end_from_calendar_event():
    doc = {
        "google_calendar_event": {
            "start": {"dateTime": "2026-02-28T10:00:00+01:00"},
            "end": {"dateTime": "2026-02-28T11:00:00+01:00"}
        }
    }
    start, end = extract_calendar_times(doc)
    assert start == "2026-02-28T10:00:00+01:00"
    assert end == "2026-02-28T11:00:00+01:00"


def test_extract_calendar_times_returns_none_for_missing_event():
    doc = {}
    start, end = extract_calendar_times(doc)
    assert start is None
    assert end is None


def test_extract_calendar_times_returns_none_for_non_dict_event():
    doc = {"google_calendar_event": None}
    start, end = extract_calendar_times(doc)
    assert start is None
    assert end is None

import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

from granola import extract_people


def test_extract_people_returns_names_and_emails_from_nested_attendees():
    # Arrange
    doc = {
        "people": {
            "attendees": [
                {
                    "details": {
                        "person": {
                            "name": {
                                "fullName": "Alice Johnson"
                            }
                        }
                    },
                    "email": "alice@example.com"
                },
                {
                    "details": {
                        "person": {
                            "name": {
                                "fullName": "Bob Smith"
                            }
                        }
                    },
                    "email": "bob@example.com"
                }
            ]
        }
    }

    # Act
    result = extract_people(doc)

    # Assert
    assert result == [
        {"name": "Alice Johnson", "email": "alice@example.com"},
        {"name": "Bob Smith", "email": "bob@example.com"}
    ]

import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

from granola import format_time


def test_format_time_converts_iso_to_hhmm():
    assert format_time("2026-02-28T10:15:00+01:00") == "10:15"


def test_format_time_returns_question_mark_for_none():
    assert format_time(None) == "?"


def test_format_time_returns_question_mark_for_invalid():
    assert format_time("not-a-date") == "?"

import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

import granola

FIXTURES = os.path.join(os.path.dirname(__file__), "fixtures")


def test_load_cache_parses_string_state(monkeypatch):
    monkeypatch.setattr(granola, "CACHE_PATH",
                        os.path.join(FIXTURES, "cache_string_state.json"))
    state = granola.load_cache()
    assert "documents" in state
    assert "abc-123" in state["documents"]
    assert state["documents"]["abc-123"]["title"] == "Test Meeting"


def test_load_cache_parses_object_state(monkeypatch):
    monkeypatch.setattr(granola, "CACHE_PATH",
                        os.path.join(FIXTURES, "cache_object_state.json"))
    state = granola.load_cache()
    assert "documents" in state
    assert "def-456" in state["documents"]
    assert state["documents"]["def-456"]["title"] == "Another Meeting"