pig.py/repo.py

"""
repo.py - VCS Repository Detection, Cloning, and Indexing

Detects git/hg/svn/fossil repositories and clones them for indexing.
Files are stored in content-addressed vault with symlinks preserving tree structure.

Like Hydra mode for RSS feeds, VCS detection is a "smart source" that
bypasses slow HTTP crawling for high-priority ingestion.
"""

import asyncio
import os
import re
import subprocess
from pathlib import Path
from typing import Optional, Tuple, Iterator, AsyncIterator
from urllib.parse import urlparse

# VCS type detection by domain
VCS_HOSTS = {
    # Git hosts
    'github.com': 'git',
    'gitlab.com': 'git',
    'git.unturf.com': 'git',
    'bitbucket.org': 'git',
    'codeberg.org': 'git',
    'sr.ht': 'git',
    'git.sr.ht': 'git',
    'gitea.com': 'git',
    'gitee.com': 'git',
    'salsa.debian.org': 'git',
    'git.savannah.gnu.org': 'git',
    'git.kernel.org': 'git',
    'git.zx2c4.com': 'git',
    'git.openwrt.org': 'git',

    # Mercurial hosts
    'hg.mozilla.org': 'hg',
    'hg.python.org': 'hg',
    'hg.sr.ht': 'hg',
    'foss.heptapod.net': 'hg',

    # SVN hosts (legacy)
    'svn.apache.org': 'svn',
    'svn.code.sf.net': 'svn',
}

# URL patterns that indicate VCS type
VCS_URL_PATTERNS = [
    (r'\.git/?$', 'git'),
    (r'git@[^:]+:', 'git'),  # git@github.com:user/repo
    (r'/trunk/?$', 'svn'),
    (r'/branches/?', 'svn'),
    (r'/tags/?$', 'svn'),
    (r'\.fossil$', 'fossil'),
]

# Directories to skip when walking files
VCS_DIRS = {'.git', '.hg', '.svn', '_FOSSIL_', '.fossil-settings'}

# Binary file extensions to skip for text indexing
BINARY_EXTENSIONS = {
    '.png', '.jpg', '.jpeg', '.gif', '.bmp', '.ico', '.webp', '.avif',
    '.mp4', '.webm', '.mov', '.avi', '.mkv',
    '.mp3', '.wav', '.ogg', '.flac', '.m4a',
    '.zip', '.tar', '.gz', '.bz2', '.xz', '.7z', '.rar',
    '.exe', '.dll', '.so', '.dylib', '.a', '.o',
    '.pyc', '.pyo', '.class', '.jar',
    '.pdf', '.doc', '.docx', '.xls', '.xlsx',
    '.ttf', '.otf', '.woff', '.woff2', '.eot',
    '.sqlite', '.db', '.sqlite3',
}


def detect_vcs(uri: str) -> Tuple[Optional[str], Optional[str]]:
    """
    Detect VCS type from URI.

    Returns:
        (vcs_type, clone_url) or (None, None) if not a repo

    Examples:
        >>> detect_vcs('https://github.com/user/repo')
        ('git', 'https://github.com/user/repo.git')
        >>> detect_vcs('git@github.com:user/repo.git')
        ('git', 'git@github.com:user/repo.git')
        >>> detect_vcs('https://hg.mozilla.org/mozilla-central')
        ('hg', 'https://hg.mozilla.org/mozilla-central')
    """
    # Check SSH-style URLs first (git@host:path)
    ssh_match = re.match(r'^(git|hg)@([^:]+):(.+)$', uri)
    if ssh_match:
        vcs_type = ssh_match.group(1)
        return vcs_type, uri

    # Parse URL
    parsed = urlparse(uri)
    host = parsed.netloc.lower()

    # Remove port if present
    if ':' in host:
        host = host.split(':')[0]

    # Check URL patterns
    for pattern, vcs_type in VCS_URL_PATTERNS:
        if re.search(pattern, uri):
            return vcs_type, uri

    # Check known hosts
    if host in VCS_HOSTS:
        vcs_type = VCS_HOSTS[host]
        clone_url = uri

        # Ensure .git suffix for git repos if not present
        if vcs_type == 'git' and not uri.endswith('.git'):
            # Strip trailing slashes and add .git
            clone_url = uri.rstrip('/') + '.git'

        return vcs_type, clone_url

    return None, None


def get_repo_path(uri: str, base_path: Path) -> Path:
    """
    Generate local path for a repository.

    Structure: base_path / domain / org / repo

    Examples:
        >>> get_repo_path('https://github.com/user/repo', Path('repo_vault'))
        Path('repo_vault/github.com/user/repo')
    """
    # Handle SSH URLs
    ssh_match = re.match(r'^(?:git|hg)@([^:]+):(.+?)(?:\.git)?$', uri)
    if ssh_match:
        host = ssh_match.group(1)
        path = ssh_match.group(2)
    else:
        parsed = urlparse(uri)
        host = parsed.netloc.lower()
        if ':' in host:
            host = host.split(':')[0]
        path = parsed.path.strip('/')

        # Remove .git suffix
        if path.endswith('.git'):
            path = path[:-4]

    return base_path / host / path


def run_cmd(cmd: list, cwd: Optional[Path] = None, timeout: int = 300) -> Tuple[int, str, str]:
    """Run a command and return (returncode, stdout, stderr)."""
    try:
        result = subprocess.run(
            cmd,
            cwd=cwd,
            capture_output=True,
            text=True,
            timeout=timeout
        )
        return result.returncode, result.stdout, result.stderr
    except subprocess.TimeoutExpired:
        return -1, '', f'Command timed out after {timeout}s'
    except Exception as e:
        return -1, '', str(e)


async def run_cmd_async(cmd: list, cwd: Optional[Path] = None, timeout: int = 300) -> Tuple[int, str, str]:
    """Run a command asynchronously."""
    return await asyncio.to_thread(run_cmd, cmd, cwd, timeout)


def clone_repo(uri: str, dest: Path, vcs_type: str, shallow: bool = True) -> Tuple[bool, str]:
    """
    Clone a repository.

    Args:
        uri: Clone URL
        dest: Destination path
        vcs_type: 'git', 'hg', 'svn', or 'fossil'
        shallow: Use shallow clone if supported (git only)

    Returns:
        (success, message)
    """
    dest.parent.mkdir(parents=True, exist_ok=True)

    if vcs_type == 'git':
        cmd = ['git', 'clone']
        if shallow:
            cmd.extend(['--depth', '1'])
        cmd.extend([uri, str(dest)])

    elif vcs_type == 'hg':
        cmd = ['hg', 'clone', uri, str(dest)]

    elif vcs_type == 'svn':
        cmd = ['svn', 'checkout', uri, str(dest)]

    elif vcs_type == 'fossil':
        # Fossil needs a two-step process
        fossil_file = dest.parent / f'{dest.name}.fossil'
        ret, out, err = run_cmd(['fossil', 'clone', uri, str(fossil_file)])
        if ret != 0:
            return False, f'fossil clone failed: {err}'
        dest.mkdir(parents=True, exist_ok=True)
        ret, out, err = run_cmd(['fossil', 'open', str(fossil_file)], cwd=dest)
        if ret != 0:
            return False, f'fossil open failed: {err}'
        return True, 'Cloned successfully'

    else:
        return False, f'Unknown VCS type: {vcs_type}'

    ret, out, err = run_cmd(cmd, timeout=600)
    if ret != 0:
        return False, f'{vcs_type} clone failed: {err}'

    return True, 'Cloned successfully'


async def clone_repo_async(uri: str, dest: Path, vcs_type: str, shallow: bool = True) -> Tuple[bool, str]:
    """Clone a repository asynchronously."""
    return await asyncio.to_thread(clone_repo, uri, dest, vcs_type, shallow)


def pull_repo(path: Path) -> Tuple[bool, str]:
    """
    Update an existing repository.

    Auto-detects VCS type from directory contents.

    Returns:
        (success, message)
    """
    if (path / '.git').exists():
        ret, out, err = run_cmd(['git', 'pull'], cwd=path)
        if ret != 0:
            return False, f'git pull failed: {err}'
        return True, out.strip() or 'Already up to date'

    elif (path / '.hg').exists():
        ret, out, err = run_cmd(['hg', 'pull', '-u'], cwd=path)
        if ret != 0:
            return False, f'hg pull failed: {err}'
        return True, out.strip()

    elif (path / '.svn').exists():
        ret, out, err = run_cmd(['svn', 'update'], cwd=path)
        if ret != 0:
            return False, f'svn update failed: {err}'
        return True, out.strip()

    elif (path / '_FOSSIL_').exists() or (path / '.fslckout').exists():
        ret, out, err = run_cmd(['fossil', 'update'], cwd=path)
        if ret != 0:
            return False, f'fossil update failed: {err}'
        return True, out.strip()

    else:
        return False, 'No VCS directory found'


async def pull_repo_async(path: Path) -> Tuple[bool, str]:
    """Update a repository asynchronously."""
    return await asyncio.to_thread(pull_repo, path)


def get_commit_hash(path: Path) -> Optional[str]:
    """Get current commit/revision hash."""
    if (path / '.git').exists():
        ret, out, err = run_cmd(['git', 'rev-parse', 'HEAD'], cwd=path)
        if ret == 0:
            return out.strip()

    elif (path / '.hg').exists():
        ret, out, err = run_cmd(['hg', 'id', '-i'], cwd=path)
        if ret == 0:
            return out.strip()

    elif (path / '.svn').exists():
        ret, out, err = run_cmd(['svn', 'info', '--show-item', 'revision'], cwd=path)
        if ret == 0:
            return out.strip()

    elif (path / '_FOSSIL_').exists() or (path / '.fslckout').exists():
        ret, out, err = run_cmd(['fossil', 'info'], cwd=path)
        if ret == 0:
            for line in out.split('\n'):
                if line.startswith('checkout:'):
                    return line.split()[1]

    return None


def walk_files(repo_path: Path, include_binary: bool = False) -> Iterator[Path]:
    """
    Walk files in a repository, excluding VCS directories.

    Args:
        repo_path: Path to repository root
        include_binary: Include binary files (default: skip them)

    Yields:
        Path objects for each file
    """
    for root, dirs, files in os.walk(repo_path):
        # Skip VCS directories
        dirs[:] = [d for d in dirs if d not in VCS_DIRS]

        for name in files:
            file_path = Path(root) / name

            # Skip binary files unless requested
            if not include_binary:
                if file_path.suffix.lower() in BINARY_EXTENSIONS:
                    continue

            # Skip empty files
            try:
                if file_path.stat().st_size == 0:
                    continue
            except OSError:
                continue

            yield file_path


async def symlink_to_vault(
    repo_path: Path,
    vault,  # AsyncVault
    on_progress: Optional[callable] = None
) -> AsyncIterator[Tuple[str, Path, Path]]:
    """
    Replace repo files with symlinks to content-addressed vault.

    Args:
        repo_path: Path to repository root
        vault: AsyncVault instance for content storage
        on_progress: Optional callback(hash, file_path, vault_path)

    Yields:
        (md5_hash, original_path, vault_path) for each file
    """
    from filevault import content_hash

    for file_path in walk_files(repo_path, include_binary=True):
        try:
            content = file_path.read_bytes()
            h = content_hash(content)
            ext = file_path.suffix or '.txt'

            # Store in vault
            vault_path = await vault.store(h, content, ext)

            # Replace file with symlink
            file_path.unlink()
            rel_path = os.path.relpath(vault_path, file_path.parent)
            file_path.symlink_to(rel_path)

            if on_progress:
                on_progress(h, file_path, vault_path)

            yield h, file_path, vault_path

        except Exception as e:
            # Log error but continue with other files
            print(f'Error processing {file_path}: {e}')
            continue


def is_binary_file(path: Path) -> bool:
    """Check if a file is binary by extension or content."""
    if path.suffix.lower() in BINARY_EXTENSIONS:
        return True

    # Check first bytes for null characters
    try:
        with open(path, 'rb') as f:
            chunk = f.read(8192)
            if b'\x00' in chunk:
                return True
    except Exception:
        pass

    return False


def get_file_language(path: Path) -> Optional[str]:
    """Guess programming language from file extension."""
    ext_to_lang = {
        '.py': 'python',
        '.js': 'javascript',
        '.ts': 'typescript',
        '.jsx': 'javascript',
        '.tsx': 'typescript',
        '.rb': 'ruby',
        '.go': 'go',
        '.rs': 'rust',
        '.c': 'c',
        '.h': 'c',
        '.cpp': 'cpp',
        '.hpp': 'cpp',
        '.cc': 'cpp',
        '.java': 'java',
        '.kt': 'kotlin',
        '.scala': 'scala',
        '.swift': 'swift',
        '.m': 'objective-c',
        '.php': 'php',
        '.pl': 'perl',
        '.pm': 'perl',
        '.sh': 'shell',
        '.bash': 'shell',
        '.zsh': 'shell',
        '.fish': 'shell',
        '.lua': 'lua',
        '.r': 'r',
        '.R': 'r',
        '.jl': 'julia',
        '.ex': 'elixir',
        '.exs': 'elixir',
        '.erl': 'erlang',
        '.hs': 'haskell',
        '.ml': 'ocaml',
        '.fs': 'fsharp',
        '.clj': 'clojure',
        '.lisp': 'lisp',
        '.el': 'elisp',
        '.vim': 'vim',
        '.sql': 'sql',
        '.html': 'html',
        '.htm': 'html',
        '.css': 'css',
        '.scss': 'scss',
        '.sass': 'sass',
        '.less': 'less',
        '.json': 'json',
        '.yaml': 'yaml',
        '.yml': 'yaml',
        '.toml': 'toml',
        '.xml': 'xml',
        '.md': 'markdown',
        '.rst': 'rst',
        '.txt': 'text',
        '.ini': 'ini',
        '.cfg': 'ini',
        '.conf': 'conf',
        '.dockerfile': 'dockerfile',
        '.makefile': 'makefile',
        '.cmake': 'cmake',
        '.gradle': 'gradle',
        '.groovy': 'groovy',
        '.tf': 'terraform',
        '.nix': 'nix',
        '.zig': 'zig',
        '.v': 'v',
        '.nim': 'nim',
        '.d': 'd',
        '.pas': 'pascal',
        '.asm': 'assembly',
        '.s': 'assembly',
        '.wasm': 'wasm',
        '.wat': 'wat',
    }

    ext = path.suffix.lower()
    if ext in ext_to_lang:
        return ext_to_lang[ext]

    # Check filename for special cases
    name = path.name.lower()
    if name == 'makefile':
        return 'makefile'
    elif name == 'dockerfile':
        return 'dockerfile'
    elif name == 'jenkinsfile':
        return 'groovy'
    elif name == 'gemfile':
        return 'ruby'
    elif name == 'rakefile':
        return 'ruby'
    elif name == 'vagrantfile':
        return 'ruby'
    elif name == 'procfile':
        return 'text'
    elif name.startswith('.') and name.endswith('rc'):
        return 'shell'

    return None