export_meeting_transcripts.py

import argparse
import asyncio
import base64
import hashlib
import json
import os
import re
import secrets
import sys
import time
from dataclasses import dataclass
from datetime import date, datetime, timedelta
from typing import Any, Iterable
from urllib.parse import parse_qs, urlparse, urlsplit, urlunsplit

import httpx
from mcp import ClientSession, types
from mcp.client.streamable_http import streamable_http_client
from mcp.shared.auth_utils import calculate_token_expiry


MCP_SERVER_URL = "https://mcp.notion.com/mcp"
DEFAULT_TOKEN_ENV = "NOTION_MCP_TOKEN"
MANIFEST_FILENAME = ".manifest.json"
DEFAULT_TOKEN_CACHE = os.path.join(os.path.expanduser("~"), ".notion-mcp-token.json")


@dataclass(frozen=True)
class MeetingNote:
    title: str
    url: str
    created_time: datetime
    notion_id: str


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Export Notion meeting notes (summary + transcript) via MCP",
    )
    parser.add_argument(
        "--output-dir",
        default="./meeting_transcripts",
        help="Directory to store exported markdown files",
    )
    parser.add_argument(
        "--since",
        help="Start date (YYYY-MM-DD). Defaults to last exported date + 1 day, or 365 days ago.",
    )
    parser.add_argument(
        "--until",
        help="End date (YYYY-MM-DD). Defaults to today.",
    )
    parser.add_argument(
        "--window-days",
        type=int,
        default=7,
        help="Query window size in days for paging meeting notes",
    )
    parser.add_argument(
        "--limit",
        type=int,
        default=0,
        help="Maximum number of meetings to export (0 = no limit)",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Print what would be exported without writing files",
    )
    parser.add_argument(
        "--debug",
        action="store_true",
        help="Print timestamped debug logs",
    )
    parser.add_argument(
        "--token-env",
        help="Environment variable containing the MCP auth token (requires --no-oauth or --env-token)",
    )
    parser.add_argument(
        "--no-oauth",
        action="store_true",
        help="Disable OAuth and use environment variable token instead",
    )
    parser.add_argument(
        "--env-token",
        action="store_true",
        help="Use environment variable token (NOTION_MCP_TOKEN by default) instead of OAuth",
    )
    parser.add_argument(
        "--token-cache",
        default=DEFAULT_TOKEN_CACHE,
        help="Path to cache OAuth tokens",
    )
    parser.add_argument(
        "--oauth-redirect-uri",
        default="http://localhost:3000/callback",
        help="OAuth redirect URI used for manual code paste",
    )
    parser.add_argument(
        "--server-url",
        default=MCP_SERVER_URL,
        help="MCP server URL",
    )
    return parser.parse_args()


def parse_iso_datetime(value: str) -> datetime:
    if value.endswith("Z"):
        value = value[:-1] + "+00:00"
    return datetime.fromisoformat(value)


def parse_date(value: str) -> date:
    return datetime.strptime(value, "%Y-%m-%d").date()


def debug_log(enabled: bool, message: str) -> None:
    if not enabled:
        return
    timestamp = datetime.now().isoformat(timespec="seconds")
    print(f"[{timestamp}] {message}")


def sanitize_title(title: str) -> str:
    cleaned = title.strip()
    cleaned = cleaned.replace(" \u2023", "").strip()
    cleaned = re.sub(r"\s+", " ", cleaned)
    cleaned = re.sub(r"[^A-Za-z0-9\- _]", "", cleaned)
    cleaned = cleaned.strip("-_ ")
    if not cleaned:
        return "meeting"
    return cleaned[:80]


def extract_notion_id(url: str) -> str:
    match = re.search(r"([0-9a-fA-F]{32})", url)
    if not match:
        return ""
    return match.group(1)


def manifest_path(output_dir: str) -> str:
    return os.path.join(output_dir, MANIFEST_FILENAME)


def load_token_cache(path: str) -> dict[str, Any] | None:
    if not os.path.exists(path):
        return None
    try:
        with open(path, "r", encoding="utf-8") as handle:
            data = json.load(handle)
            if isinstance(data, dict):
                return data
    except Exception:
        return None
    return None


def save_token_cache(path: str, data: dict[str, Any]) -> None:
    os.makedirs(os.path.dirname(path), exist_ok=True)
    existing = load_token_cache(path) or {}
    merged = {**existing, **data}
    with open(path, "w", encoding="utf-8") as handle:
        json.dump(merged, handle, indent=2, sort_keys=True)


def token_is_valid(token: dict[str, Any]) -> bool:
    expires_at = token.get("expires_at")
    if not expires_at:
        return True
    try:
        return time.time() < float(expires_at) - 60
    except TypeError, ValueError:
        return False


def build_pkce_pair() -> tuple[str, str]:
    verifier = (
        base64.urlsafe_b64encode(secrets.token_bytes(32)).decode("ascii").rstrip("=")
    )
    digest = hashlib.sha256(verifier.encode("ascii")).digest()
    challenge = base64.urlsafe_b64encode(digest).decode("ascii").rstrip("=")
    return verifier, challenge


def discovery_base_url(mcp_server_url: str) -> str:
    parsed = urlsplit(mcp_server_url)
    path = parsed.path or ""
    if path.endswith("/mcp"):
        path = path[: -len("/mcp")]
    elif path.endswith("/mcp/"):
        path = path[: -len("/mcp/")]
    if not path:
        path = "/"
    return urlunsplit((parsed.scheme, parsed.netloc, path, "", ""))


async def discover_oauth_endpoints(
    client: httpx.AsyncClient,
    mcp_server_url: str,
) -> dict[str, Any]:
    base_url = discovery_base_url(mcp_server_url)
    protected_resource_url = (
        f"{base_url.rstrip('/')}/.well-known/oauth-protected-resource"
    )
    response = await client.get(protected_resource_url)
    response.raise_for_status()
    data = response.json()
    servers = data.get("authorization_servers", [])
    if not servers:
        raise RuntimeError(
            "No authorization servers found in protected resource metadata"
        )

    oauth_server_url = servers[0]
    metadata_url = (
        f"{oauth_server_url.rstrip('/')}/.well-known/oauth-authorization-server"
    )
    response = await client.get(metadata_url)
    response.raise_for_status()
    metadata = response.json()
    auth_endpoint = metadata.get("authorization_endpoint")
    token_endpoint = metadata.get("token_endpoint")
    registration_endpoint = metadata.get("registration_endpoint")
    token_auth_methods = metadata.get("token_endpoint_auth_methods_supported")
    if not auth_endpoint or not token_endpoint:
        raise RuntimeError("OAuth metadata missing authorization or token endpoint")
    return {
        "authorization_endpoint": auth_endpoint,
        "token_endpoint": token_endpoint,
        "registration_endpoint": registration_endpoint,
        "token_endpoint_auth_methods_supported": token_auth_methods,
    }


def select_token_auth_method(supported: list[str] | None) -> str:
    if not supported:
        return "client_secret_post"
    if "none" in supported:
        return "none"
    if "client_secret_basic" in supported:
        return "client_secret_basic"
    if "client_secret_post" in supported:
        return "client_secret_post"
    return supported[0]


def token_auth_settings(
    client_id: str,
    client_secret: str | None,
    method: str | None,
    payload: dict[str, Any],
) -> httpx.Auth | None:
    if method == "client_secret_basic":
        if not client_secret:
            raise SystemExit("OAuth requires client secret for client_secret_basic")
        return httpx.BasicAuth(client_id, client_secret)
    if method == "client_secret_post":
        if not client_secret:
            raise SystemExit("OAuth requires client secret for client_secret_post")
        payload["client_secret"] = client_secret
        return None
    return None


async def register_oauth_client(
    client: httpx.AsyncClient,
    args: argparse.Namespace,
    metadata: dict[str, Any],
) -> dict[str, Any]:
    registration_endpoint = metadata.get("registration_endpoint")
    if not registration_endpoint:
        raise SystemExit("OAuth server does not expose registration_endpoint")

    token_auth_methods = metadata.get("token_endpoint_auth_methods_supported")
    token_auth_method = select_token_auth_method(token_auth_methods)

    registration_payload = {
        "redirect_uris": [args.oauth_redirect_uri],
        "grant_types": ["authorization_code", "refresh_token"],
        "response_types": ["code"],
        "client_name": "Notion MCP Exporter",
        "token_endpoint_auth_method": token_auth_method,
    }

    response = await client.post(registration_endpoint, json=registration_payload)
    response.raise_for_status()
    client_data = response.json()

    client_id = client_data.get("client_id")
    if not client_id:
        raise SystemExit("Dynamic client registration did not return client_id")

    return {
        "client_id": client_id,
        "client_secret": client_data.get("client_secret"),
        "token_endpoint_auth_method": token_auth_method,
    }


async def oauth_authorization_code_flow(
    args: argparse.Namespace, cached: dict[str, Any] | None
) -> str:
    async with httpx.AsyncClient() as client:
        debug_log(args.debug, "Discovering OAuth endpoints")
        metadata = await discover_oauth_endpoints(
            client,
            args.server_url,
        )

        token_auth_method = "client_secret_post"

        if cached and cached.get("client_id"):
            client_id = cached.get("client_id")
            client_secret = cached.get("client_secret")
            token_auth_method = (
                cached.get("token_endpoint_auth_method") or token_auth_method
            )
        else:
            debug_log(args.debug, "Registering OAuth client")
            registration = await register_oauth_client(client, args, metadata)
            client_id = registration["client_id"]
            client_secret = registration.get("client_secret")
            token_auth_method = registration.get(
                "token_endpoint_auth_method", token_auth_method
            )
            save_token_cache(args.token_cache, registration)

        if not client_id:
            raise SystemExit("OAuth client registration failed")
        client_id = str(client_id)

        auth_endpoint = metadata.get("authorization_endpoint")
        token_endpoint = metadata.get("token_endpoint")
        if not auth_endpoint or not token_endpoint:
            raise SystemExit("OAuth metadata missing authorization or token endpoint")

        verifier, challenge = build_pkce_pair()
        state = secrets.token_urlsafe(16)

        auth_params = {
            "response_type": "code",
            "client_id": client_id,
            "redirect_uri": args.oauth_redirect_uri,
            "state": state,
            "code_challenge": challenge,
            "code_challenge_method": "S256",
        }

        request = client.build_request("GET", auth_endpoint, params=auth_params)
        auth_url = str(request.url)
        print("Open this URL in your browser and authorize:")
        print(auth_url)
        callback_url = input("Paste the full callback URL: ").strip()
        parsed = urlparse(callback_url)
        params = parse_qs(parsed.query)
        code_values = params.get("code")
        state_values = params.get("state")
        if not code_values:
            raise SystemExit("Callback URL missing 'code' parameter")
        if state_values and state_values[0] != state:
            raise SystemExit("State mismatch in callback URL")
        code = code_values[0]

        token_payload = {
            "grant_type": "authorization_code",
            "client_id": client_id,
            "code": code,
            "redirect_uri": args.oauth_redirect_uri,
            "code_verifier": verifier,
        }
        auth = token_auth_settings(
            client_id, client_secret, token_auth_method, token_payload
        )
        auth_param = auth if auth is not None else httpx.USE_CLIENT_DEFAULT

        debug_log(args.debug, "Exchanging authorization code for token")
        response = await client.post(
            token_endpoint, data=token_payload, auth=auth_param
        )
        response.raise_for_status()
        token_data = response.json()
        token_data["expires_at"] = calculate_token_expiry(token_data.get("expires_in"))
        save_token_cache(args.token_cache, token_data)
        return str(token_data["access_token"])


async def refresh_oauth_token(
    args: argparse.Namespace, refresh_token: str, cached: dict[str, Any] | None
) -> str | None:
    async with httpx.AsyncClient() as client:
        debug_log(args.debug, "Refreshing OAuth token")
        metadata = await discover_oauth_endpoints(
            client,
            args.server_url,
        )
        client_id = cached.get("client_id") if cached else None
        client_secret = cached.get("client_secret") if cached else None
        token_auth_method = "client_secret_post"
        if cached and cached.get("token_endpoint_auth_method"):
            token_auth_method = (
                cached.get("token_endpoint_auth_method") or token_auth_method
            )

        if not client_id:
            return None

        client_id = str(client_id)
        token_payload = {
            "grant_type": "refresh_token",
            "client_id": client_id,
            "refresh_token": refresh_token,
        }

        auth = token_auth_settings(
            client_id, client_secret, token_auth_method, token_payload
        )
        auth_param = auth if auth is not None else httpx.USE_CLIENT_DEFAULT
        token_endpoint = metadata.get("token_endpoint")
        if not token_endpoint:
            return None

        response = await client.post(
            token_endpoint, data=token_payload, auth=auth_param
        )
        if response.status_code >= 400:
            return None
        token_data = response.json()
        token_data["expires_at"] = calculate_token_expiry(token_data.get("expires_in"))
        if "refresh_token" not in token_data:
            token_data["refresh_token"] = refresh_token
        save_token_cache(args.token_cache, token_data)
        return str(token_data["access_token"])


def load_manifest(output_dir: str) -> dict[str, str]:
    path = manifest_path(output_dir)
    if not os.path.exists(path):
        return {}
    try:
        with open(path, "r", encoding="utf-8") as handle:
            data = json.load(handle)
            if isinstance(data, dict):
                return {str(k): str(v) for k, v in data.items()}
    except Exception:
        return {}
    return {}


def save_manifest(output_dir: str, manifest: dict[str, str]) -> None:
    path = manifest_path(output_dir)
    with open(path, "w", encoding="utf-8") as handle:
        json.dump(manifest, handle, indent=2, sort_keys=True)


def latest_exported_date(output_dir: str) -> date | None:
    if not os.path.isdir(output_dir):
        return None
    latest: date | None = None
    for entry in os.listdir(output_dir):
        if not entry.endswith(".md"):
            continue
        match = re.match(r"(\d{4}-\d{2}-\d{2})[ _]", entry)
        if not match:
            continue
        try:
            entry_date = parse_date(match.group(1))
        except ValueError:
            continue
        if latest is None or entry_date > latest:
            latest = entry_date
    return latest


def iter_date_windows(
    start: date, end: date, window_days: int
) -> Iterable[tuple[date, date]]:
    current = start
    step = timedelta(days=window_days)
    while current <= end:
        window_end = min(end, current + step - timedelta(days=1))
        yield current, window_end
        current = window_end + timedelta(days=1)


def normalize_block(text: str) -> str:
    cleaned = text.replace("<empty-block/>", "")
    cleaned = re.sub(r"\[\^\{\{.*?\}\}\]", "", cleaned)
    cleaned = re.sub(r"\[\^https://www\.notion\.so/[^\]]+\]", "", cleaned)
    cleaned = re.sub(r"<mention-date[^>]*/>", "", cleaned)
    lines = cleaned.splitlines()
    lines = [line.replace("\t", "").rstrip() for line in lines]
    while lines and not lines[0].strip():
        lines.pop(0)
    while lines and not lines[-1].strip():
        lines.pop()
    return "\n".join(lines)


def extract_tag(text: str, tag: str) -> str:
    match = re.search(rf"<{tag}>\s*(.*?)\s*</{tag}>", text, re.DOTALL)
    if not match:
        return ""
    return match.group(1)


def extract_sections(page_text: str) -> tuple[str, str]:
    content_block = extract_tag(page_text, "content")
    if content_block:
        marker = "<empty-block/>"
        last_index = content_block.rfind(marker)
        if last_index != -1:
            summary_raw = content_block[:last_index]
            transcript_raw = content_block[last_index + len(marker) :]
        else:
            summary_raw = content_block
            transcript_raw = ""
        summary = normalize_block(summary_raw)
        transcript = normalize_block(transcript_raw)
        return summary, transcript

    summary = normalize_block(extract_tag(page_text, "summary"))
    transcript = normalize_block(extract_tag(page_text, "transcript"))
    return summary, transcript


def render_markdown(note: MeetingNote, summary: str, transcript: str) -> str:
    title = note.title.strip()
    created_date = note.created_time.date().isoformat()
    lines = [
        "---",
        f"title: {title}",
        f"date: {created_date}",
        f"notion_url: {note.url}",
        f"notion_id: {note.notion_id}",
        "---",
        f"# {title}",
        "",
    ]

    if summary:
        lines.extend(["## Summary", "", summary, ""])

    if transcript:
        lines.extend(["> [!Transcript]-", ">", transcript, ""])

    if not summary and not transcript:
        lines.extend(["## Notes", "", "(No summary or transcript content found.)", ""])

    return "\n".join(lines)


def get_text_content(result: Any) -> str:
    parts: list[str] = []
    for content in result.content:
        if isinstance(content, types.TextContent):
            parts.append(content.text)
    return "".join(parts).strip()


def ensure_tool_success(result: Any, tool_name: str) -> None:
    if getattr(result, "is_error", False):
        message = get_text_content(result) or "Unknown tool error"
        raise RuntimeError(f"Tool {tool_name} failed: {message}")


def get_structured_content(result: Any) -> Any:
    structured = getattr(result, "structured_content", None)
    if structured:
        return structured
    text = get_text_content(result)
    if not text:
        return None
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        return None


async def query_meeting_notes(
    session: ClientSession,
    start: date,
    end: date,
) -> list[MeetingNote]:
    filter_payload = {
        "operator": "and",
        "filters": [
            {
                "property": "created_time",
                "filter": {
                    "operator": "date_is_within",
                    "value": {
                        "type": "exact",
                        "value": {
                            "type": "daterange",
                            "start_date": start.isoformat(),
                            "end_date": end.isoformat(),
                        },
                    },
                },
            }
        ],
    }

    result = await session.call_tool(
        "notion-query-meeting-notes", {"filter": filter_payload}
    )
    ensure_tool_success(result, "notion-query-meeting-notes")
    structured = get_structured_content(result)
    if not structured or "results" not in structured:
        return []

    notes: list[MeetingNote] = []
    for item in structured.get("results", []):
        title = str(item.get("Title", "")).replace(" \u2023", "").strip()
        url = str(item.get("url", "")).strip()
        created_time_raw = str(item.get("Created time", "")).strip()
        if not url or not created_time_raw:
            continue
        try:
            created_time = parse_iso_datetime(created_time_raw)
        except ValueError:
            continue
        notion_id = extract_notion_id(url)
        if not notion_id:
            continue
        notes.append(
            MeetingNote(
                title=title or "Meeting",
                url=url,
                created_time=created_time,
                notion_id=notion_id,
            )
        )
    return notes


async def fetch_meeting_page(session: ClientSession, url: str) -> str:
    result = await session.call_tool(
        "notion-fetch", {"id": url, "include_transcript": True}
    )
    ensure_tool_success(result, "notion-fetch")
    text = get_text_content(result)
    if text.startswith("{"):
        try:
            parsed = json.loads(text)
        except json.JSONDecodeError:
            return text
        inner_text = parsed.get("text")
        if isinstance(inner_text, str):
            return inner_text
    return text


async def export_meetings(args: argparse.Namespace) -> int:
    token: str | None
    use_env_token = args.no_oauth or args.env_token
    if use_env_token:
        env_var = args.token_env or DEFAULT_TOKEN_ENV
        token = os.getenv(env_var)
        if not token:
            raise SystemExit(f"Missing token: set {env_var} environment variable")
    else:
        cached = load_token_cache(args.token_cache)
        if cached and token_is_valid(cached):
            debug_log(args.debug, "Using cached OAuth token")
            token = str(cached.get("access_token"))
        elif cached and cached.get("refresh_token"):
            token = await refresh_oauth_token(
                args, str(cached.get("refresh_token")), cached
            )
        else:
            token = None

        if not token:
            token = await oauth_authorization_code_flow(args, cached)

    os.makedirs(args.output_dir, exist_ok=True)

    manifest = load_manifest(args.output_dir)

    today = date.today()
    if args.until:
        until = parse_date(args.until)
    else:
        until = today

    if args.since:
        since = parse_date(args.since)
    else:
        latest = latest_exported_date(args.output_dir)
        if latest:
            since = latest + timedelta(days=1)
        else:
            since = today - timedelta(days=365)

    if since > until:
        return 0

    exported = 0
    skipped = 0
    seen_ids: set[str] = set()
    auth_headers = {"Authorization": f"Bearer {token}"}
    async with httpx.AsyncClient(headers=auth_headers) as http_client:
        debug_log(args.debug, f"Connecting to MCP server: {args.server_url}")
        async with streamable_http_client(
            args.server_url,
            http_client=http_client,
        ) as (read_stream, write_stream, _get_session_id):
            async with ClientSession(read_stream, write_stream) as session:
                debug_log(args.debug, "Initializing MCP session")
                await session.initialize()

                for window_start, window_end in iter_date_windows(
                    since, until, args.window_days
                ):
                    debug_log(
                        args.debug,
                        f"Querying meeting notes for {window_start} to {window_end}",
                    )
                    notes = await query_meeting_notes(session, window_start, window_end)
                    for note in notes:
                        if note.notion_id in seen_ids:
                            continue
                        seen_ids.add(note.notion_id)

                        existing = manifest.get(note.notion_id)
                        filename = f"{note.created_time.date().isoformat()} {sanitize_title(note.title)}.md"
                        file_path = os.path.join(args.output_dir, filename)

                        if existing and os.path.exists(existing):
                            skipped += 1
                            continue

                        if os.path.exists(file_path):
                            manifest[note.notion_id] = file_path
                            skipped += 1
                            continue

                        if args.dry_run:
                            print(f"[DRY-RUN] Would export: {file_path}")
                            exported += 1
                        else:
                            debug_log(
                                args.debug, f"Fetching meeting page {note.notion_id}"
                            )
                            page_text = await fetch_meeting_page(session, note.url)
                            debug_log(args.debug, f"Writing markdown {file_path}")
                            summary, transcript = extract_sections(page_text)
                            markdown = render_markdown(note, summary, transcript)
                            with open(file_path, "w", encoding="utf-8") as handle:
                                handle.write(markdown)
                            manifest[note.notion_id] = file_path
                            exported += 1

                        if args.limit and exported >= args.limit:
                            save_manifest(args.output_dir, manifest)
                            print(f"Exported {exported} meeting(s), skipped {skipped}.")
                            return exported

    save_manifest(args.output_dir, manifest)
    print(f"Exported {exported} meeting(s), skipped {skipped}.")
    return exported


def main() -> None:
    stdout_reconfigure = getattr(sys.stdout, "reconfigure", None)
    if callable(stdout_reconfigure):
        stdout_reconfigure(encoding="utf-8", errors="replace")
    stderr_reconfigure = getattr(sys.stderr, "reconfigure", None)
    if callable(stderr_reconfigure):
        stderr_reconfigure(encoding="utf-8", errors="replace")
    args = parse_args()
    asyncio.run(export_meetings(args))


if __name__ == "__main__":
    main()