export_channel.py

#!/usr/bin/env python3
"""
Slack Private Channel Exporter
Exports a single Slack private channel to JSON or text format.
"""

import argparse
import getpass
import json
import os
import sys
from datetime import datetime
from typing import List, Dict, Optional
import requests
import time
from dotenv import load_dotenv

# Load environment variables from .env file
load_dotenv()


class SlackChannelExporter:
    def __init__(self, token: str):
        self.token = token
        self.base_url = "https://slack.com/api"
        self.headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json"
        }

    def _make_request(self, endpoint: str, params: Dict) -> Dict:
        """Make a request to Slack API with rate limit handling."""
        url = f"{self.base_url}/{endpoint}"
        response = requests.get(url, headers=self.headers, params=params)
        data = response.json()

        if not data.get("ok"):
            error = data.get("error", "Unknown error")
            raise Exception(f"Slack API error: {error}")

        # Handle rate limiting
        if response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", 1))
            time.sleep(retry_after)
            return self._make_request(endpoint, params)

        return data

    def fetch_messages(self, channel_id: str, oldest: Optional[str] = None,
                      latest: Optional[str] = None) -> List[Dict]:
        """Fetch all messages from a channel with pagination."""
        messages = []
        cursor = None

        while True:
            params = {
                "channel": channel_id,
                "limit": 1000
            }

            if oldest:
                params["oldest"] = oldest
            if latest:
                params["latest"] = latest
            if cursor:
                params["cursor"] = cursor

            print(f"Fetching messages... (total so far: {len(messages)})", file=sys.stderr)
            data = self._make_request("conversations.history", params)

            messages.extend(data.get("messages", []))

            if not data.get("has_more"):
                break

            cursor = data.get("response_metadata", {}).get("next_cursor")
            if not cursor:
                break

        # Sort messages chronologically (oldest first)
        messages.sort(key=lambda m: float(m.get("ts", 0)))
        return messages

    def fetch_thread_replies(self, channel_id: str, thread_ts: str) -> List[Dict]:
        """Fetch all replies in a thread."""
        params = {
            "channel": channel_id,
            "ts": thread_ts
        }

        data = self._make_request("conversations.replies", params)
        replies = data.get("messages", [])

        # Remove the parent message (first in list)
        if replies and replies[0].get("ts") == thread_ts:
            replies = replies[1:]

        return replies

    def export_channel(self, channel_id: str, oldest: Optional[str] = None,
                      latest: Optional[str] = None) -> List[Dict]:
        """Export channel with all messages and threads."""
        print(f"Fetching messages from channel {channel_id}...", file=sys.stderr)
        messages = self.fetch_messages(channel_id, oldest, latest)

        # Fetch thread replies for messages that have them
        for i, message in enumerate(messages):
            if message.get("thread_ts") and message.get("reply_count", 0) > 0:
                # Only fetch threads for parent messages
                if message.get("ts") == message.get("thread_ts"):
                    print(f"Fetching thread {i+1}/{len(messages)}...", file=sys.stderr)
                    replies = self.fetch_thread_replies(channel_id, message["thread_ts"])
                    message["thread_replies"] = replies

        print(f"Export complete: {len(messages)} messages", file=sys.stderr)
        return messages

    def save_as_json(self, messages: List[Dict], output_file: str):
        """Save messages as JSON file."""
        with open(output_file, 'w', encoding='utf-8') as f:
            json.dump(messages, f, indent=2, ensure_ascii=False)
        print(f"Saved to {output_file}", file=sys.stderr)

    def save_as_text(self, messages: List[Dict], output_file: str):
        """Save messages as formatted text file preserving thread structure."""
        with open(output_file, 'w', encoding='utf-8') as f:
            for message in messages:
                self._write_message(f, message)

                # Write thread replies if any
                if "thread_replies" in message:
                    for reply in message["thread_replies"]:
                        self._write_message(f, reply, is_reply=True)

                f.write("\n" + "="*80 + "\n\n")

        print(f"Saved to {output_file}", file=sys.stderr)

    def _write_message(self, f, message: Dict, is_reply: bool = False):
        """Write a single message to text file."""
        timestamp = float(message.get("ts", 0))
        dt = datetime.fromtimestamp(timestamp)
        user = message.get("user", "Unknown")
        text = message.get("text", "")

        prefix = "    ↳ " if is_reply else ""

        f.write(f"{prefix}[{dt.strftime('%Y-%m-%d %H:%M:%S')}] {user}\n")
        f.write(f"{prefix}{text}\n")
        if not is_reply:
            f.write("\n")


def parse_date(date_str: str) -> str:
    """Parse date string to Unix timestamp."""
    try:
        dt = datetime.strptime(date_str, "%Y-%m-%d")
        return str(int(dt.timestamp()))
    except ValueError:
        raise argparse.ArgumentTypeError(f"Invalid date format: {date_str}. Use YYYY-MM-DD")


def main():
    parser = argparse.ArgumentParser(
        description="Export a Slack private channel to JSON or text format"
    )
    parser.add_argument("channel_id", nargs="?",
                       help="Slack channel ID (e.g., C1234567890)")
    parser.add_argument("--output", "-o", required=True, help="Output file path")
    parser.add_argument("--format", choices=["json", "text"], default="json",
                       help="Output format (default: json)")
    parser.add_argument("--start-date", type=parse_date,
                       help="Start date for export (YYYY-MM-DD)")
    parser.add_argument("--end-date", type=parse_date,
                       help="End date for export (YYYY-MM-DD)")

    args = parser.parse_args()

    # Get token from environment variable or prompt interactively
    token = os.getenv("SLACK_BOT_TOKEN")
    if not token:
        token = getpass.getpass("Enter Slack Bot Token: ")
        if not token:
            print("Error: Slack token is required", file=sys.stderr)
            sys.exit(1)

    # Get channel ID from CLI arg or environment variable
    channel_id = args.channel_id or os.getenv("SLACK_CHANNEL_ID")
    if not channel_id:
        print("Error: Channel ID required. Provide as argument or set SLACK_CHANNEL_ID environment variable",
              file=sys.stderr)
        sys.exit(1)

    try:
        exporter = SlackChannelExporter(token)
        messages = exporter.export_channel(channel_id, args.start_date, args.end_date)

        if args.format == "json":
            exporter.save_as_json(messages, args.output)
        else:
            exporter.save_as_text(messages, args.output)

        print(f"Successfully exported {len(messages)} messages", file=sys.stderr)

    except Exception as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()