diff --git a/examples/granola/honcho_granola.py b/examples/granola/honcho_granola.py
index 4babb0cd..b3881eba 100644
--- a/examples/granola/honcho_granola.py
+++ b/examples/granola/honcho_granola.py
@@ -26,29 +26,34 @@ import sys
import threading
import traceback
import webbrowser
+from dataclasses import dataclass, field
from datetime import datetime, timezone
from http.server import HTTPServer, BaseHTTPRequestHandler
-from typing import Any, TypedDict
+from typing import Any
from urllib.parse import parse_qs, urlencode, urlparse
import httpx
-class Participant(TypedDict):
+@dataclass
+class Participant:
name: str
- email: str | None
- org: str | None
+ email: str | None = None
+ org: str | None = None
-class ParsedParticipants(TypedDict):
- note_creator: Participant | None
- others: list[Participant]
+@dataclass
+class ParsedParticipants:
+ note_creator: Participant | None = None
+ others: list[Participant] = field(default_factory=list)
-class TranscriptTurn(TypedDict):
+@dataclass
+class TranscriptTurn:
speaker: str
text: str
+
# Granola MCP + OAuth endpoints
GRANOLA_MCP_URL = "https://mcp.granola.ai/mcp"
AUTH_BASE = "https://mcp-auth.granola.ai"
@@ -58,32 +63,28 @@ OAUTH_REDIRECT_URI = f"http://localhost:{OAUTH_REDIRECT_PORT}/callback"
# Honcho message size limit (25000 max, leave headroom)
MAX_MESSAGE_LEN = 24000
-# Module-level state
-http_client = httpx.AsyncClient(timeout=60.0)
-access_token: str | None = None
-auth_result: dict[str, str | None] = {"code": None, "error": None}
-
# ---------------------------------------------------------------------------
# OAuth callback handler (must be a class for BaseHTTPRequestHandler)
# ---------------------------------------------------------------------------
class _OAuthCallback(BaseHTTPRequestHandler):
+ auth_result: dict[str, str | None] = {"code": None, "error": None}
+
def do_GET(self):
- global auth_result
params = parse_qs(urlparse(self.path).query)
if "code" in params:
- auth_result["code"] = params["code"][0]
+ _OAuthCallback.auth_result["code"] = params["code"][0]
self.send_response(200)
self.send_header("Content-Type", "text/html")
self.end_headers()
self.wfile.write(b"Authenticated! You can close this window.
")
elif "error" in params:
- auth_result["error"] = params.get("error_description", params["error"])[0]
+ _OAuthCallback.auth_result["error"] = params.get("error_description", params["error"])[0]
self.send_response(400)
self.send_header("Content-Type", "text/html")
self.end_headers()
- self.wfile.write(f"Error: {auth_result['error']}
".encode())
+ self.wfile.write(f"Error: {_OAuthCallback.auth_result['error']}
".encode())
else:
self.send_response(404)
self.end_headers()
@@ -96,10 +97,9 @@ class _OAuthCallback(BaseHTTPRequestHandler):
# Granola OAuth + MCP
# ---------------------------------------------------------------------------
-async def authenticate() -> bool:
- """Perform OAuth (DCR + PKCE) with Granola. Sets module-level access_token."""
- global access_token, auth_result
- auth_result = {"code": None, "error": None}
+async def authenticate(http_client: httpx.AsyncClient) -> str:
+ """Perform OAuth (DCR + PKCE) with Granola. Returns access token."""
+ _OAuthCallback.auth_result = {"code": None, "error": None}
print("\nAuthenticating with Granola...")
@@ -115,8 +115,7 @@ async def authenticate() -> bool:
},
)
if resp.status_code not in (200, 201):
- print(f" Client registration failed: {resp.status_code}")
- return False
+ raise RuntimeError(f"Client registration failed: {resp.status_code}")
client_id = resp.json().get("client_id")
# PKCE
@@ -142,12 +141,11 @@ async def authenticate() -> bool:
thread.join(timeout=120)
server.server_close()
+ auth_result = _OAuthCallback.auth_result
if auth_result["error"]:
- print(f" Authentication failed: {auth_result['error']}")
- return False
+ raise RuntimeError(f"Authentication failed: {auth_result['error']}")
if not auth_result["code"]:
- print(" Authentication timed out")
- return False
+ raise RuntimeError("Authentication timed out")
# Exchange code for token
resp = await http_client.post(
@@ -162,15 +160,18 @@ async def authenticate() -> bool:
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
if resp.status_code != 200:
- print(f" Token exchange failed: {resp.status_code}")
- return False
+ raise RuntimeError(f"Token exchange failed: {resp.status_code}")
- access_token = resp.json().get("access_token")
print(" Authenticated successfully!")
- return True
+ return resp.json()["access_token"]
-async def call_mcp_tool(tool_name: str, arguments: dict[str, Any] | None = None) -> dict[str, Any]:
+async def call_mcp_tool(
+ http_client: httpx.AsyncClient,
+ access_token: str,
+ tool_name: str,
+ arguments: dict[str, Any] | None = None,
+) -> dict[str, Any]:
"""Call a Granola MCP tool, handling both JSON and SSE responses."""
resp = await http_client.post(
GRANOLA_MCP_URL,
@@ -187,7 +188,7 @@ async def call_mcp_tool(tool_name: str, arguments: dict[str, Any] | None = None)
},
)
if resp.status_code != 200:
- raise Exception(f"MCP call failed: {resp.status_code} - {resp.text}")
+ raise RuntimeError(f"MCP call failed: {resp.status_code} - {resp.text}")
# SSE response
if "text/event-stream" in resp.headers.get("content-type", ""):
@@ -199,38 +200,44 @@ async def call_mcp_tool(tool_name: str, arguments: dict[str, Any] | None = None)
if "result" in parsed:
result = parsed
elif "error" in parsed:
- raise Exception(f"MCP error: {parsed['error']}")
+ raise RuntimeError(f"MCP error: {parsed['error']}")
except json.JSONDecodeError:
continue
if result:
final = result.get("result", {})
return final if isinstance(final, dict) else {"result": final}
- raise Exception("No result in SSE response")
+ raise RuntimeError("No result in SSE response")
# JSON response
result = resp.json()
if "error" in result:
- raise Exception(f"MCP error: {result['error']}")
+ raise RuntimeError(f"MCP error: {result['error']}")
return result.get("result", {})
def extract_mcp_text(result: dict[str, Any]) -> str:
- """Extract text from the first content block of an MCP result."""
+ """Extract text from the first content block of an MCP result.
+
+ Raises ValueError if the response structure is unexpected.
+ """
content = result.get("content", [])
- if isinstance(content, list) and content:
- first = content[0]
- if isinstance(first, dict) and "text" in first:
- return str(first["text"])
- return ""
+ if not isinstance(content, list) or not content:
+ raise ValueError(f"MCP response missing content array: {list(result.keys())}")
+ first = content[0]
+ if not isinstance(first, dict) or "text" not in first:
+ raise ValueError(f"MCP content block missing 'text' field: {first}")
+ return str(first["text"])
# ---------------------------------------------------------------------------
# Granola data fetching
# ---------------------------------------------------------------------------
-async def list_meetings(limit: int = 100) -> list[dict[str, Any]]:
+async def list_meetings(
+ http_client: httpx.AsyncClient, access_token: str, limit: int = 100,
+) -> list[dict[str, Any]]:
"""List meetings from Granola MCP. Parses Granola's XML-like response format."""
- result = await call_mcp_tool("list_meetings", {"limit": limit})
+ result = await call_mcp_tool(http_client, access_token, "list_meetings", {"limit": limit})
text = extract_mcp_text(result)
meetings: list[dict[str, Any]] = []
@@ -249,37 +256,91 @@ async def list_meetings(limit: int = 100) -> list[dict[str, Any]]:
return meetings
-async def get_meeting_details(meeting_id: str) -> dict[str, Any]:
- """Get full meeting details including notes. Returns raw XML-like content."""
- result = await call_mcp_tool("get_meetings", {"meeting_ids": [meeting_id]})
+async def get_meeting_details(
+ http_client: httpx.AsyncClient, access_token: str, meeting_id: str,
+) -> dict[str, Any]:
+ """Get full meeting details including notes."""
+ result = await call_mcp_tool(http_client, access_token, "get_meetings", {"meeting_ids": [meeting_id]})
text = extract_mcp_text(result)
return {"id": meeting_id, "raw_content": text}
-async def get_meeting_transcript(meeting_id: str) -> str | None:
- """Get transcript for a meeting (paid tiers only)."""
- try:
- result = await call_mcp_tool("get_meeting_transcript", {"meeting_id": meeting_id})
- except Exception as e:
- print(f" Transcript unavailable: {e}")
- return None
+async def get_meeting_transcript(
+ http_client: httpx.AsyncClient, access_token: str, meeting_id: str,
+ max_retries: int = 3,
+) -> str | None:
+ """Get transcript for a meeting (paid tiers only).
+
+ Retries on rate limit responses with exponential backoff.
+ """
+ for attempt in range(max_retries):
+ try:
+ result = await call_mcp_tool(http_client, access_token, "get_meeting_transcript", {"meeting_id": meeting_id})
+ text = extract_mcp_text(result)
+ except Exception as e:
+ print(f" Transcript unavailable: {e}")
+ return None
+
+ if not text or "no transcript" in text.lower():
+ return None
+
+ # Granola returns rate limit errors as content text, not HTTP errors
+ if "rate limit" in text.lower():
+ wait = 2 ** attempt * 3 # 3s, 6s, 12s
+ print(f" ⚠ Granola rate limit hit (attempt {attempt + 1}/{max_retries}), waiting {wait}s...")
+ await asyncio.sleep(wait)
+ continue
- text = extract_mcp_text(result)
- if text and "no transcript" not in text.lower():
return text
- if not text:
- t = result.get("transcript")
- return str(t) if t else None
+
+ print(f" ⚠ Transcript skipped after {max_retries} rate limit retries")
return None
+async def fetch_all_meetings(
+ http_client: httpx.AsyncClient, access_token: str,
+) -> list[dict[str, Any]]:
+ """Fetch meeting list and enrich each with transcript and details."""
+ print("\nFetching meetings from Granola...")
+ meetings = await list_meetings(http_client, access_token, limit=500)
+ if not meetings:
+ print("No meetings found.")
+ return []
+ print(f" Found {len(meetings)} meetings. Fetching content...\n")
+
+ for i, m in enumerate(meetings, 1):
+ mid = m.get("id")
+ if not mid:
+ continue
+
+ transcript = await get_meeting_transcript(http_client, access_token, mid)
+ if transcript:
+ m["transcript"] = transcript
+
+ try:
+ m.update(await get_meeting_details(http_client, access_token, mid))
+ except Exception as exc:
+ print(f" Failed to fetch details for {mid}: {exc}")
+
+ has_t = "transcript" in m
+ has_s = bool(extract_summary(m))
+ label = "transcript+summary" if has_t and has_s else "transcript only" if has_t else "summary only" if has_s else "basic only"
+ print(f" [{i}/{len(meetings)}] {label}: {m.get('title', 'Untitled')[:45]}")
+ await asyncio.sleep(1.5) # rate limit
+
+ return meetings
+
+
# ---------------------------------------------------------------------------
# Parsing helpers
# ---------------------------------------------------------------------------
def parse_participants(participants_str: str) -> ParsedParticipants:
- """Parse Granola's participant string into {note_creator, others}."""
- result: ParsedParticipants = {"note_creator": None, "others": []}
+ """Parse Granola's participant string into structured participants.
+
+ Warns on unparseable entries instead of silently dropping them.
+ """
+ result = ParsedParticipants()
if not participants_str:
return result
@@ -310,22 +371,26 @@ def parse_participants(participants_str: str) -> ParsedParticipants:
email = email_match.group(1) if email_match else None
name = re.sub(r"\s*<[^>]+>", "", clean).strip()
+ if not name:
+ print(f" Warning: could not parse participant entry: {entry!r}")
+ continue
+
org = None
org_match = re.match(r"(.+?)\s+from\s+(.+)", name)
if org_match:
name, org = org_match.group(1).strip(), org_match.group(2).strip()
- person: Participant = {"name": name, "email": email, "org": org}
+ person = Participant(name=name, email=email, org=org)
if is_creator:
- result["note_creator"] = person
+ result.note_creator = person
else:
- result["others"].append(person)
+ result.others.append(person)
return result
def parse_transcript_turns(raw: str) -> list[TranscriptTurn]:
- """Split a Granola transcript into speaker turns [{speaker, text}, ...]."""
+ """Split a Granola transcript into speaker turns."""
# Unwrap JSON wrapper if present
try:
parsed = json.loads(raw)
@@ -340,7 +405,7 @@ def parse_transcript_turns(raw: str) -> list[TranscriptTurn]:
while i < len(parts) - 1:
text = parts[i + 1].strip()
if text:
- turns.append({"speaker": parts[i], "text": text})
+ turns.append(TranscriptTurn(speaker=parts[i], text=text))
i += 2
return turns
@@ -379,13 +444,199 @@ def sanitize(text: str) -> str:
def parse_date(date_str: str) -> datetime:
- """Parse Granola's date format into a timezone-aware datetime."""
+ """Parse Granola's date format into a timezone-aware datetime.
+
+ Raises ValueError if the date string doesn't match any known format.
+ """
for fmt in ["%b %d, %Y %I:%M %p", "%b %d, %Y %I:%M:%S %p", "%B %d, %Y %I:%M %p"]:
try:
return datetime.strptime(date_str, fmt).replace(tzinfo=timezone.utc)
except ValueError:
continue
- return datetime.now(timezone.utc)
+ raise ValueError(f"Unrecognized date format: {date_str!r}")
+
+
+# ---------------------------------------------------------------------------
+# Honcho import helpers
+# ---------------------------------------------------------------------------
+
+def build_messages(
+ peer: Any,
+ content: str,
+ metadata: dict[str, object] | None,
+ created_at: datetime,
+) -> list[Any]:
+ """Build chunked messages for a single peer, attaching metadata to the first chunk."""
+ messages = []
+ content = sanitize(content)
+ for start in range(0, len(content), MAX_MESSAGE_LEN):
+ chunk = content[start:start + MAX_MESSAGE_LEN]
+ msg_meta = metadata if start == 0 else None
+ messages.append(peer.message(chunk, metadata=msg_meta, created_at=created_at))
+ return messages
+
+
+def send_messages(session: Any, messages: list[Any]) -> None:
+ """Send messages to a session in batches of 100."""
+ for batch_start in range(0, len(messages), 100):
+ session.add_messages(messages[batch_start:batch_start + 100])
+
+
+def import_two_person(
+ honcho: Any,
+ session: Any,
+ me_peer_id: str,
+ them_peer_id: str,
+ turns: list[TranscriptTurn],
+ metadata: dict[str, object],
+ created_at: datetime,
+) -> None:
+ """Import a two-person meeting with speaker attribution."""
+ me_peer = honcho.peer(me_peer_id)
+ them_peer = honcho.peer(them_peer_id)
+
+ # Merge consecutive same-speaker turns
+ merged: list[TranscriptTurn] = []
+ for t in turns:
+ if merged and merged[-1].speaker == t.speaker:
+ merged[-1].text += " " + t.text
+ else:
+ merged.append(TranscriptTurn(speaker=t.speaker, text=t.text))
+
+ messages: list[Any] = []
+ for i, t in enumerate(merged):
+ peer = me_peer if t.speaker == "Me" else them_peer
+ msg_meta = metadata if i == 0 else None
+ messages.extend(build_messages(peer, t.text, msg_meta, created_at))
+
+ send_messages(session, messages)
+ print(f" -> Imported as 2-person ({me_peer_id} + {them_peer_id})")
+
+
+def import_summary(
+ honcho: Any,
+ session: Any,
+ me_peer_id: str,
+ meeting: dict[str, Any],
+ metadata: dict[str, object],
+ created_at: datetime,
+) -> None:
+ """Import a meeting as a summary message."""
+ me_peer = honcho.peer(me_peer_id)
+ summary = extract_summary(meeting)
+ if not summary:
+ raw_t = meeting.get("transcript", "")
+ try:
+ parsed = json.loads(raw_t)
+ summary = str(parsed.get("transcript", "")) if isinstance(parsed, dict) else raw_t
+ except (json.JSONDecodeError, TypeError):
+ summary = raw_t
+ summary = summary or "No content available"
+
+ title = meeting.get("title", "Untitled")
+ date = meeting.get("date", "")
+ header = f"Meeting: {title}\nDate: {date}\nParticipants: {meeting.get('participants', '')}\n\n"
+
+ messages = build_messages(me_peer, header + summary, metadata, created_at)
+ send_messages(session, messages)
+ print(" -> Imported as summary")
+
+
+def resolve_them_participant(others: list[Participant]) -> Participant | None:
+ """Ask user to pick which participant is 'Them' from a multi-person meeting."""
+ for j, p in enumerate(others, 1):
+ email_str = f" <{p.email}>" if p.email else ""
+ print(f" {j}. {p.name}{email_str}")
+ idx_str = input(f" Who is 'Them'? [1-{len(others)}]: ").strip()
+ try:
+ return others[int(idx_str) - 1]
+ except (ValueError, IndexError):
+ print(" Invalid selection.")
+ return None
+
+
+def review_meeting(
+ index: int,
+ total: int,
+ meeting: dict[str, Any],
+ participants: ParsedParticipants,
+ turns: list[TranscriptTurn],
+) -> tuple[str, Participant | None]:
+ """Display meeting info and get user's import choice.
+
+ Returns (mode, them_participant) where mode is one of:
+ - "two_person": import with speaker attribution using them_participant
+ - "summary": import as a single summary message
+ - "skip": skip this meeting
+ """
+ title = meeting.get("title", "Untitled")
+ date = meeting.get("date", "")
+ creator = participants.note_creator
+ others = participants.others
+
+ me_turns = sum(1 for t in turns if t.speaker == "Me")
+ them_turns = len(turns) - me_turns
+ total_words = sum(len(t.text.split()) for t in turns)
+
+ print(f"\n{'─' * 60}")
+ print(f" [{index}/{total}] {title}")
+ print(f" Date: {date}")
+ if creator:
+ print(f" You: {creator.name} <{creator.email}>")
+ for j, p in enumerate(others, 1):
+ email_str = f" <{p.email}>" if p.email else ""
+ org_str = f" ({p.org})" if p.org else ""
+ print(f" {j}. {p.name}{email_str}{org_str}")
+
+ has_transcript = bool(meeting.get("transcript"))
+ if turns:
+ print(f" Transcript: {me_turns} Me, {them_turns} Them, ~{total_words} words")
+ if them_turns == 0:
+ print(" ** No 'Them' turns — nobody else spoke **")
+ if total_words < 30:
+ print(" ** Very short — might be empty **")
+ elif has_transcript:
+ raw = meeting["transcript"]
+ print(f" Transcript: present ({len(raw)} chars) but could not parse speaker turns")
+ print(f" Preview: {raw[:200]!r}")
+ else:
+ print(f" Content: {'summary available' if extract_summary(meeting) else 'metadata only'}")
+
+ # Two-person default: exactly one other participant with transcript
+ if len(others) == 1 and them_turns > 0:
+ them_label = others[0].name + (f" <{others[0].email}>" if others[0].email else "")
+ print(f"\n Detected: 2-person call (you + {them_label})")
+ choice = input(" [Enter] 2-person / [s]ummary / [k] skip: ").strip().lower()
+ while choice not in ("", "s", "k"):
+ choice = input(" [Enter] 2-person / [s]ummary / [k] skip: ").strip().lower()
+ if choice == "k":
+ return ("skip", None)
+ if choice == "s":
+ return ("summary", None)
+ return ("two_person", others[0])
+
+ # Multi-person with transcript
+ if len(others) > 1 and them_turns > 0:
+ print(f"\n {len(others)} participants")
+ choice = input(" [Enter] summary / [2] 2-person / [k] skip: ").strip().lower()
+ while choice not in ("", "2", "k"):
+ choice = input(" [Enter] summary / [2] 2-person / [k] skip: ").strip().lower()
+ if choice == "k":
+ return ("skip", None)
+ if choice == "2":
+ them = resolve_them_participant(others)
+ if them is None:
+ return ("summary", None)
+ return ("two_person", them)
+ return ("summary", None)
+
+ # No transcript or no other speakers
+ choice = input(" [Enter] summary / [k] skip: ").strip().lower()
+ while choice not in ("", "k"):
+ choice = input(" [Enter] summary / [k] skip: ").strip().lower()
+ if choice == "k":
+ return ("skip", None)
+ return ("summary", None)
# ---------------------------------------------------------------------------
@@ -402,233 +653,98 @@ async def main():
print(" Get your key at: https://app.honcho.dev/api-keys")
sys.exit(1)
- try:
- if not await authenticate():
- sys.exit(1)
+ async with httpx.AsyncClient(timeout=60.0) as http_client:
+ try:
+ access_token = await authenticate(http_client)
+ meetings = await fetch_all_meetings(http_client, access_token)
+ if not meetings:
+ sys.exit(0)
- # Fetch meeting list
- print("\nFetching meetings from Granola...")
- meetings = await list_meetings(limit=500)
- if not meetings:
- print("No meetings found.")
- sys.exit(0)
- print(f" Found {len(meetings)} meetings. Fetching content...\n")
+ from honcho import Honcho
- # Fetch content for each meeting
- for i, m in enumerate(meetings, 1):
- mid = m.get("id")
- if not mid:
- continue
+ honcho = Honcho(workspace_id="granola_test")
+ seen_peers: set[str] = set()
+ results = {"imported": 0, "skipped": 0, "failed": 0}
- transcript = await get_meeting_transcript(mid)
- if transcript:
- m["transcript"] = transcript
+ print("\n" + "=" * 60)
+ print(" Review each meeting")
+ print("=" * 60)
- try:
- m.update(await get_meeting_details(mid))
- except Exception as exc:
- print(f" Failed to fetch details for {mid}: {exc}")
+ for i, m in enumerate(meetings, 1):
+ mid = m.get("id")
+ if not mid:
+ continue
- has_t = "transcript" in m
- has_s = bool(extract_summary(m))
- label = "transcript+summary" if has_t and has_s else "transcript only" if has_t else "summary only" if has_s else "basic only"
- print(f" [{i}/{len(meetings)}] {label}: {m.get('title', 'Untitled')[:45]}")
- await asyncio.sleep(1.5) # rate limit
+ participants = parse_participants(m.get("participants", ""))
+ turns = parse_transcript_turns(m["transcript"]) if m.get("transcript") else []
- # Initialize Honcho
- from honcho import Honcho
+ mode, them = review_meeting(i, len(meetings), m, participants, turns)
- honcho = Honcho(workspace_id="granola")
- seen_peers: set[str] = set()
- results = {"imported": 0, "skipped": 0, "failed": 0}
+ if mode == "skip":
+ print(" -> Skipped")
+ results["skipped"] += 1
+ continue
- # Interactive review
- print("\n" + "=" * 60)
- print(" Review each meeting")
- print("=" * 60)
+ # Resolve creator peer
+ creator = participants.note_creator
+ me_source = (creator.email or creator.name) if creator else None
+ if not me_source:
+ print(" -> Skipped (no creator identifier)")
+ results["skipped"] += 1
+ continue
- for i, m in enumerate(meetings, 1):
- mid = m.get("id")
- if not mid:
- continue
+ me_peer_id = peer_id_from(me_source)
+ if me_peer_id not in seen_peers:
+ print(f" New peer: {me_source} ({me_peer_id})")
+ seen_peers.add(me_peer_id)
- title = m.get("title", "Untitled")
- date = m.get("date", "")
- participants = parse_participants(m.get("participants", ""))
- creator = participants["note_creator"]
- others = participants["others"]
+ try:
+ created_at = parse_date(m.get("date", ""))
+ session = honcho.session(f"meeting-{mid}")
+ metadata: dict[str, object] = {
+ "title": m.get("title", "Untitled"),
+ "date": m.get("date", ""),
+ "granola_meeting_id": mid,
+ "mode": mode,
+ }
- # Parse transcript
- turns = parse_transcript_turns(m["transcript"]) if m.get("transcript") else []
- me_turns = sum(1 for t in turns if t["speaker"] == "Me")
- them_turns = len(turns) - me_turns
- total_words = sum(len(t["text"].split()) for t in turns)
-
- # Display meeting info
- print(f"\n{'─' * 60}")
- print(f" [{i}/{len(meetings)}] {title}")
- print(f" Date: {date}")
- if creator:
- print(f" You: {creator['name']} <{creator['email']}>")
- for j, p in enumerate(others, 1):
- email_str = f" <{p['email']}>" if p["email"] else ""
- org_str = f" ({p['org']})" if p["org"] else ""
- print(f" {j}. {p['name']}{email_str}{org_str}")
-
- if turns:
- print(f" Transcript: {me_turns} Me, {them_turns} Them, ~{total_words} words")
- if them_turns == 0:
- print(" ** No 'Them' turns — nobody else spoke **")
- if total_words < 30:
- print(" ** Very short — might be empty **")
- else:
- print(f" Content: {'summary available' if extract_summary(m) else 'metadata only'}")
-
- # Ask what to do
- if len(others) == 1 and them_turns > 0:
- them_label = others[0]["name"] + (f" <{others[0]['email']}>" if others[0]["email"] else "")
- print(f"\n Detected: 2-person call (you + {them_label})")
- choice = input(" [Enter] 2-person / [s]ummary / [k] skip: ").strip().lower()
- while choice not in ("", "s", "k"):
- choice = input(" [Enter] 2-person / [s]ummary / [k] skip: ").strip().lower()
- elif len(others) > 1 and them_turns > 0:
- print(f"\n {len(others)} participants")
- choice = input(" [Enter] summary / [2] 2-person / [k] skip: ").strip().lower()
- while choice not in ("", "2", "k"):
- choice = input(" [Enter] summary / [2] 2-person / [k] skip: ").strip().lower()
- else:
- choice = input(" [Enter] summary / [k] skip: ").strip().lower()
- while choice not in ("", "k"):
- choice = input(" [Enter] summary / [k] skip: ").strip().lower()
-
- if choice == "k":
- print(" -> Skipped")
- results["skipped"] += 1
- continue
-
- # Resolve creator peer
- creator_email = creator["email"] if creator else None
- me_source = creator_email or (creator["name"] if creator else None)
- if not me_source:
- print(" -> Skipped (no creator identifier)")
- results["skipped"] += 1
- continue
-
- me_peer_id = peer_id_from(me_source)
- if me_peer_id not in seen_peers:
- print(f" New peer: {me_source} ({me_peer_id})")
- seen_peers.add(me_peer_id)
-
- try:
- created_at = parse_date(date)
- session = honcho.session(f"meeting-{mid}")
-
- # Two-person mode
- if (choice == "" and len(others) == 1 and them_turns > 0) or choice == "2":
- them: Participant | None = None
- if choice == "2":
- for j, p in enumerate(others, 1):
- print(f" {j}. {p['name']}" + (f" <{p['email']}>" if p["email"] else ""))
- idx_str = input(f" Who is 'Them'? [1-{len(others)}]: ").strip()
- try:
- them = others[int(idx_str) - 1]
- except (ValueError, IndexError):
- print(" Invalid — importing as summary instead.")
- choice = "s" # fall through to summary below
- else:
- them = others[0]
-
- if choice != "s" and them is not None:
- them_source: str = them["email"] or them["name"]
+ if mode == "two_person" and them is not None:
+ them_source = them.email or them.name
them_peer_id = peer_id_from(them_source)
if them_peer_id not in seen_peers:
print(f" New peer: {them_source} ({them_peer_id})")
seen_peers.add(them_peer_id)
+ import_two_person(honcho, session, me_peer_id, them_peer_id, turns, metadata, created_at)
+ else:
+ import_summary(honcho, session, me_peer_id, m, metadata, created_at)
- me_peer = honcho.peer(me_peer_id)
- them_peer = honcho.peer(them_peer_id)
+ results["imported"] += 1
- metadata: dict[str, object] = {
- "title": title, "date": date,
- "granola_meeting_id": mid, "mode": "two_person",
- }
+ except ValueError as e:
+ print(f" -> FAILED: {e}")
+ results["failed"] += 1
+ except Exception as e:
+ print(f" -> FAILED: {e}")
+ traceback.print_exc()
+ results["failed"] += 1
- # Merge consecutive same-speaker turns
- merged: list[TranscriptTurn] = []
- for t in turns:
- if merged and merged[-1]["speaker"] == t["speaker"]:
- merged[-1]["text"] += " " + t["text"]
- else:
- merged.append({"speaker": t["speaker"], "text": t["text"]})
+ # Done
+ print("\n" + "=" * 60)
+ print(" Transfer Complete!")
+ print("=" * 60)
+ print(f"\n Imported: {results['imported']}")
+ print(f" Skipped: {results['skipped']}")
+ print(f" Failed: {results['failed']}")
+ print(" Workspace: granola")
+ print(f" Peers: {sorted(seen_peers)}")
- messages: list[Any] = []
- for j, t in enumerate(merged):
- peer = me_peer if t["speaker"] == "Me" else them_peer
- content = sanitize(t["text"])
- for start in range(0, len(content), MAX_MESSAGE_LEN):
- chunk = content[start:start + MAX_MESSAGE_LEN]
- msg_meta = metadata if j == 0 and start == 0 else None
- messages.append(peer.message(chunk, metadata=msg_meta, created_at=created_at))
-
- for batch_start in range(0, len(messages), 100):
- session.add_messages(messages[batch_start:batch_start + 100])
-
- print(f" -> Imported as 2-person ({me_peer_id} + {them_peer_id})")
- results["imported"] += 1
- continue
-
- # Summary mode
- me_peer = honcho.peer(me_peer_id)
- summary = extract_summary(m)
- if not summary:
- raw_t = m.get("transcript", "")
- try:
- parsed = json.loads(raw_t)
- summary = str(parsed.get("transcript", "")) if isinstance(parsed, dict) else raw_t
- except (json.JSONDecodeError, TypeError):
- summary = raw_t
- summary = summary or "No content available"
-
- header = f"Meeting: {title}\nDate: {date}\nParticipants: {m.get('participants', '')}\n\n"
- full = sanitize(header + summary)
-
- messages: list[Any] = []
- metadata: dict[str, object] = {"title": title, "date": date, "granola_meeting_id": mid, "mode": "summary"}
- for start in range(0, len(full), MAX_MESSAGE_LEN):
- chunk = full[start:start + MAX_MESSAGE_LEN]
- msg_meta = metadata if start == 0 else None
- messages.append(me_peer.message(chunk, metadata=msg_meta, created_at=created_at))
-
- for batch_start in range(0, len(messages), 100):
- session.add_messages(messages[batch_start:batch_start + 100])
-
- print(" -> Imported as summary")
- results["imported"] += 1
-
- except Exception as e:
- print(f" -> FAILED: {e}")
- traceback.print_exc()
- results["failed"] += 1
-
- # Done
- print("\n" + "=" * 60)
- print(" Transfer Complete!")
- print("=" * 60)
- print(f"\n Imported: {results['imported']}")
- print(f" Skipped: {results['skipped']}")
- print(f" Failed: {results['failed']}")
- print(" Workspace: granola")
- print(f" Peers: {sorted(seen_peers)}")
-
- except KeyboardInterrupt:
- print("\n\nAborted.")
- sys.exit(0)
- except Exception as e:
- print(f"\nTransfer failed: {e}")
- traceback.print_exc()
- sys.exit(1)
- finally:
- await http_client.aclose()
+ except KeyboardInterrupt:
+ print("\n\nAborted.")
+ sys.exit(0)
+ except Exception as e:
+ print(f"\nTransfer failed: {e}")
+ traceback.print_exc()
+ sys.exit(1)
if __name__ == "__main__":