orchestrata_ultimate.integrations.fireflies_sync

Reference › orchestrata_ultimate › integrations › fireflies_sync

fireflies_sync is the write side of specification 018: it copies records from the Fireflies service into the knowledge graph as :Document nodes. FirefliesKnowledgeSync takes a FirefliesClient (fireflies_client), an injected AsyncGraphRepository from orchestrata_ultimate.core.graph (it never builds its own) and optional extra tags. The gateway's Fireflies handler is its importer.

sync_recent fetches the most recent records with their summaries (default 20) and injects each one. inject_transcript upserts a single record: it derives a deterministic node id from the source id, builds the node properties (title, date, duration, participants, topics, action items, questions and the summary as JSON), merges the configured tags, and writes with a MERGE statement through the repository, so repeated syncs update the same node and never duplicate it. It returns the node id. _str_list coerces values to lists of strings because the graph stores only primitive arrays.

At a glance

Class diagram

Dependency diagram

API

Classes

FirefliesKnowledgeSync

Line 43

Sync Fireflies transcripts into the knowledge graph as :Document nodes.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/integrations/fireflies_sync.py

"""Write-side Fireflies → knowledge-graph sync (SPEC-018).

The read-side (``FirefliesHandler`` / ``FirefliesClient``) fetches meeting
transcripts on demand. This module adds the write-side: pull recent transcripts
(+ AI summaries) and inject them into the graph as idempotent ``:Document`` nodes so
meeting knowledge is queryable alongside the rest of the graph.

Triggering (a gateway ``fireflies.sync`` RPC, a cron poller, or a CLI command)
is a thin wrapper over :class:`FirefliesKnowledgeSync`.
"""

from __future__ import annotations

import json
import logging
from typing import TYPE_CHECKING, Any

from orchestrata_ultimate.integrations.fireflies_client import FirefliesClient

if TYPE_CHECKING:
    from orchestrata_ultimate.core.graph import AsyncGraphRepository

logger = logging.getLogger(__name__)

# Idempotent upsert keyed on the deterministic node id — re-syncing a transcript
# updates the node rather than creating a duplicate.
_MERGE_DOCUMENT = (
    "MERGE (d:Document {id: $id}) "
    "ON CREATE SET d.created_at = timestamp() "
    "SET d += $props, d.updated_at = timestamp() "
    "RETURN d.id AS id"
)


def _str_list(value: Any) -> list[str]:
    """Coerce a value to a list of strings (Neo4j stores only primitive arrays;
    non-string items are JSON-encoded)."""
    if not isinstance(value, list):
        return []
    return [x if isinstance(x, str) else json.dumps(x, default=str) for x in value]


class FirefliesKnowledgeSync:
    """Sync Fireflies transcripts into the knowledge graph as ``:Document`` nodes."""

    def __init__(
        self,
        client: FirefliesClient | None,
        repo: AsyncGraphRepository,
        *,
        tags: dict[str, Any] | None = None,
    ) -> None:
        """
        Args:
            client: A ``FirefliesClient`` used to fetch transcripts (only needed
                for :meth:`sync_recent`; :meth:`inject_transcript` does not use it).
            repo: An ``AsyncGraphRepository`` (backlog #163 phase 5b) — always
                injected/shared; this class never builds its own.
            tags: Extra properties merged into every node (e.g. a sync batch id).
        """
        self._client = client
        self._repo = repo
        self._tags = dict(tags or {})

    async def sync_recent(self, limit: int = 20) -> dict[str, Any]:
        """Pull the most recent transcripts + summaries and inject each as a
        ``:Document`` node. Idempotent: re-syncing updates rather than duplicates.

        Returns:
            ``{"synced": n, "node_ids": [...], "errors": [...]}``.
        """
        if self._client is None:
            raise ValueError("sync_recent requires a FirefliesClient")
        listing = await self._client.list_transcripts(limit=limit)
        node_ids: list[str] = []
        errors: list[str] = []
        for entry in listing.get("transcripts", []):
            tid = entry.get("id")
            if not tid:
                continue
            try:
                transcript = await self._client.get_transcript(tid)
                summary = await self._client.summarize_transcript(tid)
                node_ids.append(await self.inject_transcript(tid, transcript, summary))
            except Exception as exc:  # one bad transcript must not abort the sync
                logger.warning("Fireflies sync failed for %s: %s", tid, exc)
                errors.append(str(tid))
        return {"synced": len(node_ids), "node_ids": node_ids, "errors": errors}

    async def inject_transcript(
        self,
        transcript_id: str,
        transcript: dict[str, Any],
        summary: dict[str, Any],
    ) -> str:
        """Upsert a single transcript (+ summary) as a ``:Document`` node.

        Returns the node id. Idempotent via MERGE on the deterministic id.
        """
        node_id = f"ff_transcript_{transcript_id}"
        props: dict[str, Any] = {
            "id": node_id,
            "name": transcript.get("title", ""),
            "type": "document",
            "source": "fireflies",
            "transcript_id": str(transcript_id),
            "title": transcript.get("title", ""),
            "date": str(transcript.get("date", "")),
            "duration": transcript.get("duration", 0) or 0,
            "participants": _str_list(transcript.get("participants")),
            "topics": _str_list(summary.get("topics")),
            "action_items": _str_list(summary.get("action_items")),
            "questions": _str_list(summary.get("questions")),
            "summary_json": json.dumps(summary, default=str),
        }
        props.update(self._tags)
        rows = await self._repo.execute_write(
            _MERGE_DOCUMENT, {"id": node_id, "props": props}
        )
        return rows[0]["id"] if rows else node_id