"""Fireflies.ai source adapter for meeting summaries and action items."""
from datetime import date, datetime, timedelta
from typing import Any
from zoneinfo import ZoneInfo
import httpx
from daily_briefing.config import settings
from daily_briefing.logging import get_logger
from daily_briefing.models import ActionItem, CalendarEvent, MeetingSummary, Priority
from daily_briefing.sources.base import DataSource
logger = get_logger(__name__)
class FirefliesSource(DataSource):
"""Data source for Fireflies.ai meeting transcripts and summaries."""
def __init__(
self,
api_key: str,
api_url: str = "https://api.fireflies.ai/graphql",
enabled: bool = True,
):
"""Initialize the Fireflies source.
Args:
api_key: Fireflies API key
api_url: GraphQL API URL
enabled: Whether the source is enabled
"""
super().__init__("Fireflies", enabled)
self.api_key = api_key
self.api_url = api_url
self._tz = ZoneInfo(settings.timezone)
async def _execute_query(
self, query: str, variables: dict[str, Any] | None = None
) -> dict[str, Any]:
"""Execute a GraphQL query against the Fireflies API.
Args:
query: GraphQL query string
variables: Query variables
Returns:
Response data from the API
"""
async with httpx.AsyncClient(timeout=60) as client:
response = await client.post(
self.api_url,
json={"query": query, "variables": variables or {}},
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
},
)
response.raise_for_status()
data = response.json()
if "errors" in data:
error_msg = data["errors"][0].get("message", "Unknown error")
raise ValueError(f"GraphQL error: {error_msg}")
return data.get("data", {})
async def get_events(
self, start_date: date, end_date: date
) -> list[CalendarEvent]:
"""Get meeting events from Fireflies (not typically used for calendar).
Args:
start_date: Start of date range
end_date: End of date range
Returns:
Empty list (Fireflies is used for summaries, not events)
"""
return []
async def get_meetings(
self, start_date: date, end_date: date
) -> list[MeetingSummary]:
"""Get meeting summaries from Fireflies.
Args:
start_date: Start of date range
end_date: End of date range
Returns:
List of meeting summaries
"""
if not self.enabled or not self.api_key:
return []
try:
logger.info(
"fetching_fireflies_meetings",
start=start_date.isoformat(),
end=end_date.isoformat(),
)
query = """
query Transcripts($fromDate: DateTime, $toDate: DateTime, $limit: Int) {
transcripts(fromDate: $fromDate, toDate: $toDate, limit: $limit) {
id
title
date
duration
participants
summary {
overview
action_items
topics_discussed
}
}
}
"""
data = await self._execute_query(
query,
{
"fromDate": start_date.isoformat(),
"toDate": end_date.isoformat(),
"limit": 50,
},
)
transcripts = data.get("transcripts", [])
meetings = []
for t in transcripts:
try:
meeting = self._parse_meeting(t)
if meeting:
meetings.append(meeting)
except Exception as e:
logger.warning("meeting_parse_error", error=str(e))
logger.info("fetched_fireflies_meetings", count=len(meetings))
return meetings
except Exception as e:
logger.error("fireflies_fetch_error", error=str(e))
return []
def _parse_meeting(self, transcript: dict[str, Any]) -> MeetingSummary | None:
"""Parse a Fireflies transcript into a MeetingSummary.
Args:
transcript: Fireflies transcript data
Returns:
MeetingSummary or None
"""
if not transcript:
return None
# Parse date
date_str = transcript.get("date")
if date_str:
try:
meeting_date = datetime.fromisoformat(
date_str.replace("Z", "+00:00")
).astimezone(self._tz)
except ValueError:
meeting_date = datetime.now(self._tz)
else:
meeting_date = datetime.now(self._tz)
# Get summary data
summary = transcript.get("summary") or {}
action_items = summary.get("action_items") or []
if isinstance(action_items, str):
action_items = [action_items] if action_items.strip() else []
topics = summary.get("topics_discussed") or []
if isinstance(topics, str):
topics = [topics] if topics.strip() else []
# Get participants
participants = transcript.get("participants") or []
if isinstance(participants, str):
participants = [participants]
return MeetingSummary(
id=transcript.get("id", ""),
title=transcript.get("title", "Untitled Meeting"),
date=meeting_date,
duration_minutes=(transcript.get("duration") or 0) // 60,
participants=participants,
overview=summary.get("overview"),
action_items=action_items,
key_topics=topics,
)
async def get_action_items(self) -> list[ActionItem]:
"""Get action items from recent Fireflies meetings.
Returns:
List of action items
"""
if not self.enabled or not self.api_key:
return []
# Get meetings from the last week
end_date = date.today()
start_date = end_date - timedelta(days=7)
meetings = await self.get_meetings(start_date, end_date)
action_items = []
for meeting in meetings:
for i, item_text in enumerate(meeting.action_items):
action_items.append(
ActionItem(
id=f"{meeting.id}-action-{i}",
text=item_text,
source="Fireflies",
source_id=meeting.id,
meeting_title=meeting.title,
meeting_date=meeting.date.date(),
priority=Priority.MEDIUM,
)
)
return action_items
async def health_check(self) -> dict[str, Any]:
"""Check if the Fireflies API is accessible.
Returns:
Health status dictionary
"""
if not self.enabled:
return {"status": "disabled", "source": self.name}
if not self.api_key:
return {"status": "not_configured", "source": self.name}
try:
query = """
query {
transcripts(limit: 1) {
id
}
}
"""
await self._execute_query(query)
return {"status": "healthy", "source": self.name}
except Exception as e:
return {"status": "error", "source": self.name, "error": str(e)}
