claude_code_cli.py

"""
Claude Code CLI integration - spawns claude-code as subprocess for proper repo exploration.
Uses GitHub API for PR creation.
"""
import os
import subprocess
import tempfile
import logging
from typing import Optional, Dict, Any

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


def run_claude_code_on_repo(
    repo_url: str,
    feature_description: str,
    branch_name: str,
    github_token: str,
    anthropic_key: str
) -> Dict[str, Any]:
    """
    Clone repo, run Claude Code CLI to implement feature, return changes.

    Args:
        repo_url: GitHub repo URL
        feature_description: What to implement
        branch_name: Branch name for changes
        github_token: GitHub access token
        anthropic_key: Anthropic API key for Claude Code

    Returns:
        Dict with success status and branch info
    """
    try:
        # Create temp directory for repo
        with tempfile.TemporaryDirectory() as tmpdir:
            logger.info(f"Cloning {repo_url} to {tmpdir}")

            # Clone repo with auth
            auth_url = repo_url.replace('https://', f'https://{github_token}@')
            clone_result = subprocess.run(
                ['git', 'clone', auth_url, tmpdir],
                capture_output=True,
                text=True,
                timeout=60
            )

            if clone_result.returncode != 0:
                return {
                    'success': False,
                    'message': f'Failed to clone repo: {clone_result.stderr}'
                }

            logger.info(f"Cloned successfully, creating branch {branch_name}")

            # Create new branch
            subprocess.run(
                ['git', 'checkout', '-b', branch_name],
                cwd=tmpdir,
                check=True
            )

            # Run Claude Code CLI
            logger.info(f"Running Claude Code: {feature_description}")

            # Set environment for Claude Code
            env = os.environ.copy()
            env['ANTHROPIC_API_KEY'] = anthropic_key

            # Run claude-code with the feature request
            claude_cmd = [
                'claude-code',
                '-c',
                f'Implement this feature: {feature_description}\n\n'
                f'Guidelines:\n'
                f'- Explore the codebase first to understand existing architecture\n'
                f'- Modify existing files rather than creating new ones\n'
                f'- Follow existing patterns and conventions\n'
                f'- Test your changes work before finishing\n'
                f'- When done, stage and commit all changes with message: "feat: {feature_description}"\n'
                f'- Use "exit" when complete'
            ]

            try:
                result = subprocess.run(
                    claude_cmd,
                    cwd=tmpdir,
                    env=env,
                    capture_output=True,
                    text=True,
                    timeout=300  # 5 minute timeout
                )

                logger.info(f"Claude Code output:\n{result.stdout}")

                if result.returncode != 0:
                    logger.error(f"Claude Code error:\n{result.stderr}")

            except subprocess.TimeoutExpired:
                logger.warning("Claude Code timed out after 5 minutes")

            # Check if there are changes to commit
            status_result = subprocess.run(
                ['git', 'status', '--porcelain'],
                cwd=tmpdir,
                capture_output=True,
                text=True
            )

            has_changes = bool(status_result.stdout.strip())

            if has_changes:
                # Stage all changes
                subprocess.run(['git', 'add', '-A'], cwd=tmpdir, check=True)

                # Commit if not already committed
                commit_check = subprocess.run(
                    ['git', 'diff', '--cached', '--quiet'],
                    cwd=tmpdir
                )

                if commit_check.returncode != 0:  # Has staged changes
                    logger.info("Committing changes...")
                    subprocess.run(
                        ['git', 'commit', '-m', f'feat: {feature_description}\n\nGenerated by Claude Code via Omi'],
                        cwd=tmpdir,
                        check=True
                    )

            # Get default branch for PR
            default_branch_result = subprocess.run(
                ['git', 'remote', 'show', 'origin'],
                cwd=tmpdir,
                capture_output=True,
                text=True
            )

            default_branch = 'main'
            for line in default_branch_result.stdout.split('\n'):
                if 'HEAD branch:' in line:
                    default_branch = line.split(':')[1].strip()
                    break

            logger.info(f"Default branch: {default_branch}")

            # Push to remote
            logger.info(f"Pushing branch {branch_name} to remote...")
            push_result = subprocess.run(
                ['git', 'push', 'origin', branch_name],
                cwd=tmpdir,
                capture_output=True,
                text=True
            )

            if push_result.returncode != 0:
                return {
                    'success': False,
                    'message': f'Failed to push: {push_result.stderr}'
                }

            logger.info(f"Successfully pushed to branch {branch_name}")

            return {
                'success': True,
                'branch': branch_name,
                'default_branch': default_branch,
                'message': f'Claude Code implemented feature and pushed to {branch_name}'
            }

    except Exception as e:
        logger.error(f"Error running Claude Code: {e}", exc_info=True)
        return {
            'success': False,
            'message': f'Failed to run Claude Code: {str(e)}'
        }


def get_default_branch(owner: str, repo: str, github_token: str) -> str:
    """
    Get the default branch of a GitHub repository.

    Args:
        owner: Repository owner
        repo: Repository name
        github_token: GitHub access token

    Returns:
        Default branch name (e.g., 'main', 'master', 'flutterflow')
    """
    import requests

    url = f'https://api.github.com/repos/{owner}/{repo}'
    headers = {
        'Authorization': f'token {github_token}',
        'Accept': 'application/vnd.github.v3+json'
    }

    try:
        response = requests.get(url, headers=headers)
        if response.status_code == 200:
            default_branch = response.json().get('default_branch', 'main')
            logger.info(f"Default branch for {owner}/{repo}: {default_branch}")
            return default_branch
    except Exception as e:
        logger.error(f"Failed to get default branch: {e}")

    # Fallback to 'main'
    return 'main'


def create_pr_with_github_api(
    owner: str,
    repo: str,
    branch: str,
    title: str,
    body: str,
    github_token: str,
    base_branch: str = 'main'
) -> Optional[Dict[str, Any]]:
    """
    Create a pull request using GitHub API.

    Args:
        owner: Repository owner
        repo: Repository name
        branch: Branch with changes
        title: PR title
        body: PR description
        github_token: GitHub access token
        base_branch: Base branch to merge into (default: 'main')

    Returns:
        Dict with pr_url and pr_number if successful, None otherwise
    """
    import requests

    url = f'https://api.github.com/repos/{owner}/{repo}/pulls'
    headers = {
        'Authorization': f'token {github_token}',
        'Accept': 'application/vnd.github.v3+json'
    }
    data = {
        'title': title,
        'body': body,
        'head': branch,
        'base': base_branch
    }

    logger.info(f"Creating PR: {branch} -> {base_branch}")

    try:
        response = requests.post(url, headers=headers, json=data)

        if response.status_code == 201:
            pr_data = response.json()
            pr_url = pr_data.get('html_url')
            pr_number = pr_data.get('number')
            logger.info(f"PR created successfully: {pr_url}")
            return {
                'pr_url': pr_url,
                'pr_number': pr_number
            }
        else:
            error_msg = f"Failed to create PR: {response.status_code} - {response.text}"
            logger.error(error_msg)
            return None
    except Exception as e:
        logger.error(f"Exception creating PR: {e}", exc_info=True)
        return None


def merge_pr_with_github_api(
    owner: str,
    repo: str,
    pr_number: int,
    github_token: str,
    merge_method: str = 'squash'
) -> bool:
    """
    Merge a pull request using GitHub API.

    Args:
        owner: Repository owner
        repo: Repository name
        pr_number: PR number to merge
        github_token: GitHub access token
        merge_method: Merge method ('merge', 'squash', or 'rebase')

    Returns:
        True if merged successfully, False otherwise
    """
    import requests

    url = f'https://api.github.com/repos/{owner}/{repo}/pulls/{pr_number}/merge'
    headers = {
        'Authorization': f'token {github_token}',
        'Accept': 'application/vnd.github.v3+json'
    }
    data = {
        'merge_method': merge_method
    }

    logger.info(f"Merging PR #{pr_number} using {merge_method} method")

    try:
        response = requests.put(url, headers=headers, json=data)

        if response.status_code == 200:
            logger.info(f"PR #{pr_number} merged successfully")
            return True
        else:
            error_msg = f"Failed to merge PR: {response.status_code} - {response.text}"
            logger.error(error_msg)
            return False
    except Exception as e:
        logger.error(f"Exception merging PR: {e}", exc_info=True)
        return False