Fathom

{
  "name": "fathom",
  "description": "Fetch meetings, transcripts, summaries, and action items from Fathom API. Use when user asks to get Fathom recordings, s",
  "author": {
    "name": "Gleb Kalinin"
  },
  "repository": "https://github.com/glebis/claude-skills",
  "license": "MIT"
}

FATHOM_API_KEY=your-api-key-here

# download_from_fathom.py

import sys
import subprocess
from urllib.parse import urlparse
import os
import requests
import re
import argparse
from dotenv import load_dotenv


def get_m3u8_content(url):
    try:
        response = requests.get(url)
        response.raise_for_status()  # Raise an exception for bad status codes
        return response.text
    except requests.exceptions.RequestException as e:
        print(f"Error fetching m3u8 file: {e}")
        sys.exit(1)


def parse_m3u8_chunks(m3u8_content, base_url):
    chunk_urls = []
    # Extract directory path from base_url
    base_path = "/".join(base_url.split("/")[:-1]) + "/"
    for line in m3u8_content.splitlines():
        # Simple check for lines that are not comments or directives
        if not line.startswith("#"):
            # Construct the full URL for the chunk
            chunk_url = base_path + line.strip()
            chunk_urls.append(chunk_url)
    return chunk_urls


def download_fathom_video(url, output_path):
    output_path = os.path.abspath(output_path)
    print(f"Downloading video from: {url}")
    print(f"Output: {output_path}")

    # Get m3u8 content and parse chunks
    m3u8_content = get_m3u8_content(url)
    chunk_list = parse_m3u8_chunks(m3u8_content, url)
    total_chunks = len(chunk_list)
    print(f"Found {total_chunks} video chunks.")

    # Step 1: Download HLS to a raw .ts container.
    # Fathom's HLS server can drop connections on long recordings, so we
    # retry up to 3 times. ffmpeg's resumable HLS download resumes from
    # where it left off when given the same output file.
    raw_output = output_path + ".raw.ts"
    command = [
        "ffmpeg", "-y",
        "-reconnect", "1",
        "-reconnect_streamed", "1",
        "-reconnect_delay_max", "30",
        "-http_persistent", "false",
        "-i", url,
        "-c", "copy",
        raw_output,
    ]

    max_retries = 3
    for attempt in range(1, max_retries + 1):
        try:
            process = subprocess.Popen(
                command, stderr=subprocess.PIPE, universal_newlines=True
            )

            downloaded_chunks = 0
            chunk_regex = re.compile(r"Opening '(.*?)' for reading")

            while True:
                output = process.stderr.readline()
                if output == "" and process.poll() is not None:
                    break
                if output:
                    match = chunk_regex.search(output)
                    if match:
                        downloaded_chunks += 1
                        print(
                            f"Downloaded chunk {downloaded_chunks}/{total_chunks}", end="\r"
                        )

            print(f"Downloaded chunk {downloaded_chunks}/{total_chunks}")

            if process.returncode == 0:
                print(f"Download complete: {raw_output}")
                break
            elif downloaded_chunks >= total_chunks * 0.85:
                # Got most chunks — treat partial download as usable
                print(f"ffmpeg exited {process.returncode} but got {downloaded_chunks}/{total_chunks} chunks — using partial download")
                break
            else:
                print(f"ffmpeg exited {process.returncode} at chunk {downloaded_chunks}/{total_chunks} (attempt {attempt}/{max_retries})")
                if attempt < max_retries:
                    wait = attempt * 10
                    print(f"Retrying in {wait}s...")
                    import time
                    time.sleep(wait)
                else:
                    print(f"All {max_retries} attempts failed")
                    if os.path.exists(raw_output):
                        os.remove(raw_output)
                    sys.exit(1)

        except subprocess.CalledProcessError as e:
            print(f"ffmpeg failed on attempt {attempt}: {e}")
            if attempt >= max_retries:
                if os.path.exists(raw_output):
                    os.remove(raw_output)
                sys.exit(1)

    if not os.path.exists(raw_output):
        print(f"ERROR: Raw download file not created: {raw_output}")
        sys.exit(1)

    raw_size = os.path.getsize(raw_output) / (1024 * 1024)
    print(f"Raw .ts file: {raw_size:.0f} MB")

    # Step 2: Remux .ts → .mp4 with faststart moov atom
    print("Remuxing to MP4 with faststart...")
    remux_command = [
        "ffmpeg", "-y",
        "-i", raw_output,
        "-c", "copy",
        "-movflags", "faststart",
        output_path,
    ]

    try:
        subprocess.run(remux_command, check=True, capture_output=True)
        os.remove(raw_output)
        print(f"Remux complete: {output_path}")
    except subprocess.CalledProcessError as e:
        print(f"Remux failed: {e}")
        if os.path.exists(raw_output):
            os.remove(raw_output)
        if os.path.exists(output_path):
            os.remove(output_path)
        sys.exit(1)

    # Validate output file
    _validate_video_output(output_path)


def _validate_video_output(path):
    """Verify the downloaded video is a valid, playable MP4."""
    if not os.path.exists(path):
        print(f"ERROR: Output file not created: {path}")
        sys.exit(1)

    size_mb = os.path.getsize(path) / (1024 * 1024)
    if size_mb < 1:
        print(f"ERROR: Output file too small ({size_mb:.1f} MB): {path}")
        sys.exit(1)

    result = subprocess.run(
        ['ffprobe', '-v', 'error', '-print_format', 'json',
         '-show_format', '-show_streams', path],
        capture_output=True, text=True, timeout=30
    )

    if result.returncode != 0:
        print(f"ERROR: Video file is corrupt: {result.stderr.strip()}")
        print(f"  File: {path} ({size_mb:.1f} MB)")
        print("  The HLS download may have failed silently. Try again or use a different source.")
        sys.exit(1)

    import json as _json
    try:
        probe = _json.loads(result.stdout)
        duration = float(probe.get('format', {}).get('duration', 0))
        streams = probe.get('streams', [])
        has_video = any(s.get('codec_type') == 'video' for s in streams)
        if not has_video:
            print(f"ERROR: No video stream in output file: {path}")
            sys.exit(1)
        print(f"✓ Video validated: {duration/60:.0f} min, {size_mb:.0f} MB")
    except Exception as e:
        print(f"WARNING: Could not parse ffprobe output: {e}")


if __name__ == "__main__":
    load_dotenv()  # Load environment variables from .env file

    parser = argparse.ArgumentParser(description="Download a Fathom video.")
    parser.add_argument("fathom_url", help="The base URL of the Fathom video.")
    parser.add_argument(
        "--output-name",
        help="Optional name for the output video file (without extension).",
    )
    args = parser.parse_args()

    fathom_url = args.fathom_url
    output_name = args.output_name

    # Clean the URL - remove any query parameters like ?tab=summary
    if "?" in fathom_url:
        fathom_url = fathom_url.split("?")[0]
    
    # Append "/video.m3u8" to the URL
    fathom_url_with_m3u8 = fathom_url.rstrip("/") + "/video.m3u8"

    # Extract the ID from the URL path for default output name
    parsed_url = urlparse(fathom_url_with_m3u8)
    path_segments = parsed_url.path.split("/")
    if len(path_segments) >= 2 and path_segments[-1] == "video.m3u8":
        video_id = path_segments[-2]
    else:
        video_id = "output"  # Fallback if the URL format is unexpected

    # Determine the final output filename
    if output_name:
        final_output_name = output_name
    else:
        final_output_name = video_id

    output_filename = f"{final_output_name}.mp4"

    # Get output directory from environment variable, default to current directory
    output_dir = os.getenv("OUTPUT_DIR", ".")
    os.makedirs(output_dir, exist_ok=True)  # Create directory if it doesn't exist

    full_output_path = os.path.join(output_dir, output_filename)

    download_fathom_video(fathom_url_with_m3u8, full_output_path)

#!/usr/bin/env python3
"""
Fathom meeting fetcher CLI.

Fetches meetings, transcripts, summaries, and action items from Fathom API.
"""

import argparse
import sys
import subprocess
from pathlib import Path
from datetime import date

from utils import FathomClient, format_meeting_markdown, meeting_filename

# Default output directory (Obsidian vault)
DEFAULT_OUTPUT = Path.home() / 'Brains' / 'brain'
TRANSCRIPT_ANALYZER = Path.home() / '.claude' / 'skills' / 'transcript-analyzer' / 'scripts'
VIDEO_DOWNLOADER = Path(__file__).parent / 'download_video.py'


def list_meetings(client: FathomClient, limit: int = 10):
    """List recent meetings."""
    meetings = client.list_meetings(limit=limit, include_transcript=False)

    if not meetings:
        print("No meetings found.")
        return

    print(f"\n{'ID':<40} {'Date':<12} {'Title'}")
    print("-" * 80)

    for m in meetings:
        mid = str(m.get('recording_id', ''))
        created = m.get('created_at', '')[:10]
        title = (m.get('meeting_title') or m.get('title', 'Untitled'))[:40]
        print(f"{mid:<40} {created:<12} {title}")


def fetch_meeting(client: FathomClient, recording_id: str, output_dir: Path, analyze: bool = False, download_vid: bool = False):
    """Fetch a specific meeting and save to file."""
    print(f"Fetching meeting {recording_id}...")

    # Get meeting with transcript
    meetings = client.list_meetings(include_transcript=True, limit=100)
    meeting = None
    for m in meetings:
        if str(m.get('recording_id')) == recording_id or recording_id in m.get('url', ''):
            meeting = m
            break

    if not meeting:
        print(f"Meeting {recording_id} not found")
        return None

    # Try to get additional summary if available
    try:
        summary = client.get_summary(recording_id)
    except:
        summary = None

    # Format and save
    markdown = format_meeting_markdown(meeting, summary=summary)
    filename = meeting_filename(meeting)
    output_path = output_dir / filename

    output_path.write_text(markdown)
    print(f"Saved: {output_path}")

    # Optionally download video
    if download_vid:
        share_url = meeting.get('share_url', '')
        if share_url:
            download_video(share_url, output_path)
        else:
            print("No share_url found for video download")

    # Optionally run transcript analyzer
    if analyze:
        run_analyzer(output_path, output_dir)

    return output_path


def fetch_today(client: FathomClient, output_dir: Path, analyze: bool = False, download_vid: bool = False):
    """Fetch all meetings from today."""
    today = date.today().isoformat()
    print(f"Fetching meetings from {today}...")

    meetings = client.list_meetings(created_after=today, include_transcript=True)

    if not meetings:
        print("No meetings found for today.")
        return []

    saved = []
    for meeting in meetings:
        markdown = format_meeting_markdown(meeting)
        filename = meeting_filename(meeting)
        output_path = output_dir / filename

        output_path.write_text(markdown)
        print(f"Saved: {output_path}")
        saved.append(output_path)

        if download_vid:
            share_url = meeting.get('share_url', '')
            if share_url:
                download_video(share_url, output_path)

        if analyze:
            run_analyzer(output_path, output_dir)

    return saved


def fetch_since(client: FathomClient, since_date: str, output_dir: Path, analyze: bool = False, download_vid: bool = False):
    """Fetch all meetings since a date."""
    print(f"Fetching meetings since {since_date}...")

    meetings = client.list_meetings(created_after=since_date, include_transcript=True)

    if not meetings:
        print(f"No meetings found since {since_date}.")
        return []

    saved = []
    for meeting in meetings:
        markdown = format_meeting_markdown(meeting)
        filename = meeting_filename(meeting)
        output_path = output_dir / filename

        output_path.write_text(markdown)
        print(f"Saved: {output_path}")
        saved.append(output_path)

        if download_vid:
            share_url = meeting.get('share_url', '')
            if share_url:
                download_video(share_url, output_path)

        if analyze:
            run_analyzer(output_path, output_dir)

    return saved


def run_analyzer(transcript_path: Path, output_dir: Path):
    """Run transcript-analyzer on a transcript file."""
    if not TRANSCRIPT_ANALYZER.exists():
        print("transcript-analyzer skill not found, skipping analysis")
        return

    analysis_name = transcript_path.stem + '-analysis.md'
    analysis_path = output_dir / 'Projects' / analysis_name

    print(f"Running transcript analysis...")
    try:
        subprocess.run(
            ['npm', 'run', 'cli', '--', str(transcript_path), '-o', str(analysis_path)],
            cwd=str(TRANSCRIPT_ANALYZER),
            check=True,
            capture_output=True
        )
        print(f"Analysis saved: {analysis_path}")
    except subprocess.CalledProcessError as e:
        print(f"Analysis failed: {e}")
    except FileNotFoundError:
        print("npm not found, skipping analysis")


def verify_video(video_path: Path) -> bool:
    """Verify video file is valid using ffprobe."""
    try:
        result = subprocess.run(
            ['ffprobe', '-v', 'error', '-show_format', '-show_streams', str(video_path)],
            capture_output=True,
            timeout=10
        )
        return result.returncode == 0
    except (subprocess.CalledProcessError, subprocess.TimeoutExpired, FileNotFoundError):
        return False


def download_video(share_url: str, output_path: Path, max_retries: int = 3):
    """Download video using fathom video downloader with validation and retry."""
    if not VIDEO_DOWNLOADER.exists():
        print("Video downloader not found, skipping video download")
        return

    # Generate output filename based on meeting markdown filename
    video_filename = output_path.stem + '.mp4'
    video_path = output_path.parent / video_filename

    for attempt in range(1, max_retries + 1):
        print(f"Downloading video from {share_url}... (attempt {attempt}/{max_retries})")

        # Remove corrupted file if exists
        if video_path.exists():
            video_path.unlink()

        try:
            subprocess.run(
                ['python3', str(VIDEO_DOWNLOADER), share_url, '--output-name', str(video_path.stem)],
                cwd=str(output_path.parent),
                check=True,
                timeout=1800  # 30 minute timeout
            )

            # Verify the downloaded video
            if video_path.exists() and verify_video(video_path):
                print(f"Video saved and verified: {video_path}")
                return
            else:
                print(f"Video verification failed (attempt {attempt}/{max_retries})")

        except subprocess.CalledProcessError as e:
            print(f"Video download failed: {e}")
        except subprocess.TimeoutExpired:
            print(f"Video download timed out (attempt {attempt}/{max_retries})")
        except FileNotFoundError:
            print("python3 not found, skipping video download")
            return

    print(f"Failed to download valid video after {max_retries} attempts")


def main():
    parser = argparse.ArgumentParser(
        description='Fetch meetings from Fathom API',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  python fetch.py --list                    # List recent meetings
  python fetch.py --id abc123               # Fetch specific meeting
  python fetch.py --today                   # Fetch all today's meetings
  python fetch.py --since 2025-01-01        # Fetch since date
  python fetch.py --today --analyze         # Fetch and analyze
  python fetch.py --id abc123 --download-video  # Fetch meeting and download video
        """
    )

    parser.add_argument('--list', action='store_true', help='List recent meetings')
    parser.add_argument('--id', type=str, help='Fetch specific meeting by recording ID')
    parser.add_argument('--today', action='store_true', help='Fetch all meetings from today')
    parser.add_argument('--since', type=str, help='Fetch meetings since date (YYYY-MM-DD)')
    parser.add_argument('--analyze', action='store_true', help='Run transcript-analyzer on fetched meetings')
    parser.add_argument('--download-video', action='store_true', help='Download video recording (requires ffmpeg)')
    parser.add_argument('--output', '-o', type=str, default=str(DEFAULT_OUTPUT),
                        help=f'Output directory (default: {DEFAULT_OUTPUT})')
    parser.add_argument('--limit', type=int, default=10, help='Max meetings to list (default: 10)')

    args = parser.parse_args()
    output_dir = Path(args.output)

    if not output_dir.exists():
        print(f"Output directory does not exist: {output_dir}")
        sys.exit(1)

    try:
        client = FathomClient()
    except ValueError as e:
        print(f"Error: {e}")
        sys.exit(1)

    if args.list:
        list_meetings(client, limit=args.limit)
    elif args.id:
        fetch_meeting(client, args.id, output_dir, analyze=args.analyze, download_vid=args.download_video)
    elif args.today:
        fetch_today(client, output_dir, analyze=args.analyze, download_vid=args.download_video)
    elif args.since:
        fetch_since(client, args.since, output_dir, analyze=args.analyze, download_vid=args.download_video)
    else:
        # Default: list meetings
        list_meetings(client, limit=args.limit)


if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""Fathom API client utilities."""

import os
import time
import requests
from pathlib import Path
from datetime import datetime, date
from typing import Optional, List, Dict, Any
from dotenv import load_dotenv

# Load environment variables
env_path = Path(__file__).parent / '.env'
load_dotenv(env_path)

API_KEY = os.getenv('FATHOM_API_KEY')
BASE_URL = 'https://api.fathom.ai/external/v1'
RATE_LIMIT_DELAY = 1.0  # seconds between requests (60/min limit)

class FathomClient:
    """Client for Fathom API."""

    def __init__(self, api_key: str = None):
        self.api_key = api_key or API_KEY
        if not self.api_key:
            raise ValueError("FATHOM_API_KEY not set")
        self.headers = {'X-Api-Key': self.api_key}
        self._last_request_time = 0

    def _rate_limit(self):
        """Ensure we don't exceed rate limits."""
        elapsed = time.time() - self._last_request_time
        if elapsed < RATE_LIMIT_DELAY:
            time.sleep(RATE_LIMIT_DELAY - elapsed)
        self._last_request_time = time.time()

    def _get(self, endpoint: str, params: dict = None) -> dict:
        """Make GET request to API."""
        self._rate_limit()
        url = f"{BASE_URL}/{endpoint}"
        response = requests.get(url, headers=self.headers, params=params)
        response.raise_for_status()
        return response.json()

    def list_meetings(
        self,
        created_after: str = None,
        created_before: str = None,
        recorded_by: List[str] = None,
        include_transcript: bool = False,
        limit: int = 50
    ) -> List[dict]:
        """
        List meetings with optional filters.

        Args:
            created_after: ISO date string (YYYY-MM-DD)
            created_before: ISO date string (YYYY-MM-DD)
            recorded_by: List of email addresses
            include_transcript: Include full transcript in response
            limit: Max number of meetings to return

        Returns:
            List of meeting objects
        """
        params = {'include_transcript': str(include_transcript).lower()}
        if created_after:
            params['created_after'] = created_after
        if created_before:
            params['created_before'] = created_before
        if recorded_by:
            params['recorded_by[]'] = recorded_by

        meetings = []
        cursor = None

        while len(meetings) < limit:
            if cursor:
                params['cursor'] = cursor

            data = self._get('meetings', params)
            items = data.get('items', [])
            meetings.extend(items)

            cursor = data.get('next_cursor')
            if not cursor or not items:
                break

        return meetings[:limit]

    def get_meeting(self, recording_id: str, include_transcript: bool = True) -> dict:
        """Get a specific meeting by recording ID."""
        params = {'include_transcript': str(include_transcript).lower()}
        # Meeting details are in the list endpoint, filtered
        meetings = self._get('meetings', params)
        for meeting in meetings.get('items', []):
            if str(meeting.get('recording_id')) == recording_id or recording_id in meeting.get('url', ''):
                return meeting
        return None

    def get_summary(self, recording_id: str) -> str:
        """Get AI summary for a recording."""
        data = self._get(f'recordings/{recording_id}/summary')
        summary = data.get('summary', {})
        if isinstance(summary, dict):
            return summary.get('markdown_formatted', '')
        return str(summary) if summary else ''

    def get_transcript(self, recording_id: str) -> str:
        """Get full transcript for a recording."""
        data = self._get(f'recordings/{recording_id}/transcript')
        return data.get('markdown', data.get('transcript', ''))

    def get_today_meetings(self) -> List[dict]:
        """Get all meetings from today."""
        today = date.today().isoformat()
        return self.list_meetings(created_after=today, include_transcript=True)

    def get_meetings_since(self, since_date: str) -> List[dict]:
        """Get all meetings since a specific date."""
        return self.list_meetings(created_after=since_date, include_transcript=True)


def format_meeting_markdown(meeting: dict, summary: str = None, transcript: str = None) -> str:
    """
    Format a meeting as markdown for Obsidian.

    Args:
        meeting: Meeting object from API
        summary: Optional pre-fetched summary
        transcript: Optional pre-fetched transcript

    Returns:
        Markdown string
    """
    # Extract metadata
    title = meeting.get('meeting_title') or meeting.get('title', 'Untitled Meeting')
    recording_id = str(meeting.get('recording_id', ''))
    created = meeting.get('created_at', '')[:10]  # YYYY-MM-DD

    # Parse participants
    invitees = meeting.get('calendar_invitees', [])
    participants = [inv.get('name') or inv.get('email', '') for inv in invitees]

    # Calculate duration
    start = meeting.get('recording_start_time', '')
    end = meeting.get('recording_end_time', '')
    duration = ''
    if start and end:
        try:
            start_dt = datetime.fromisoformat(start.replace('Z', '+00:00'))
            end_dt = datetime.fromisoformat(end.replace('Z', '+00:00'))
            delta = end_dt - start_dt
            hours, remainder = divmod(int(delta.total_seconds()), 3600)
            minutes = remainder // 60
            duration = f"{hours:02d}:{minutes:02d}"
        except:
            pass

    # Get summary from meeting object if not provided
    if not summary:
        default_summary = meeting.get('default_summary', {})
        summary = default_summary.get('markdown', '')

    # Get transcript from meeting object if not provided
    if not transcript:
        transcript_data = meeting.get('transcript', [])
        if isinstance(transcript_data, list):
            transcript_lines = []
            for entry in transcript_data:
                speaker = entry.get('speaker', {}).get('name', 'Unknown')
                text = entry.get('text', '')
                transcript_lines.append(f"**{speaker}**: {text}")
            transcript = '\n\n'.join(transcript_lines)
        else:
            transcript = str(transcript_data)

    # Build action items
    action_items = meeting.get('action_items', [])
    action_items_md = ''
    if action_items:
        items = []
        for item in action_items:
            desc = item.get('description', '')
            assignee = item.get('assignee', {}).get('name', '')
            completed = item.get('completed', False)
            checkbox = '[x]' if completed else '[ ]'
            assignee_str = f" (@{assignee})" if assignee else ''
            items.append(f"- {checkbox} {desc}{assignee_str}")
        action_items_md = '\n'.join(items)

    # Build frontmatter
    frontmatter = f"""---
fathom_id: {recording_id}
title: "{title}"
date: {created}
participants: {participants}
duration: {duration}
fathom_url: {meeting.get('url', '')}
share_url: {meeting.get('share_url', '')}
---"""

    # Build document
    sections = [frontmatter, f"# {title}"]

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

    if action_items_md:
        sections.append("## Action Items")
        sections.append(action_items_md)

    if transcript:
        sections.append("## Transcript")
        sections.append(transcript)

    return '\n\n'.join(sections)


def slugify(text: str) -> str:
    """Convert text to URL-friendly slug."""
    import re
    text = text.lower()
    text = re.sub(r'[^\w\s-]', '', text)
    text = re.sub(r'[-\s]+', '-', text)
    return text.strip('-')


def meeting_filename(meeting: dict) -> str:
    """Generate filename for a meeting."""
    title = meeting.get('meeting_title') or meeting.get('title', 'meeting')
    created = meeting.get('created_at', '')[:10].replace('-', '')
    slug = slugify(title)[:50]  # Limit length
    return f"{created}-{slug}.md"