Python执行Linux命令的多种方法

目录

• 一、使用os和subprocess模块（推荐）
• 1. 基础方法
• 2. 使用 shell 的 cd 命令（不推荐）
• 二、完整的最佳实践解决方案
• 1. 带错误处理和路径扩展的版本
• 2. 命令行工具版本
• 三、高级功能：完整的Git操作封装
• 四、最简实现（一行代码）
• 五、总结与建议
• 推荐方案：
• 注意事项：

Python 实现shell脚本各种命令有多种方法，下面我将以 cd 和 git 命令为例子详细介绍几种常用的方式：

import os
import subprocess

# 方法1：分别执行
def git_branches_separate():
    """分别执行cd和git命令"""
    try:
        # 1. 改变工作目录
        os.chdir("~/GoogleTV5/gitfromMTK/device/changhong/hakata")
        print(f"当前目录: {os.getcwd()}")
        
        # 2. 执行git命令
        result = subprocess.run(
            ["git", "branch", "-a"],
            capture_output=True,
            text=True
        )
        
        print("Git分支列表:")
        print(result.stdout)
        
        if result.stderr:
            print(f"错误信息: {result.stderr}")
            
    except FileNotFoundError as e:
        print(f"目录不存在: {e}")
    except Exception as e:
        print(f"执行错误: {e}")

# 使用
git_branches_separate()

import subprocess

def git_branches_with_shell():
    """使用shell执行复合命令"""
    try:
        # 使用shell=True执行复合命令
        result = subprocess.run(
            "cd ~/GoogleTV5/gitfromMTK/device/changhong/hakata && git branch -a",
            shell=True,
            capture_output=True,
            text=True
        )
        
        print("Git分支列表:")
        print(result.stdout)
        
        if result.returncode != 0:
            print(f"命令执行失败: {result.stderr}")
            
    except Exception as e:
        print(f"执行错误: {e}")

# 使用
git_branches_with_shell()

import os
import subprocess
from pathlib import Path

class GitRepository:
    """Git仓库操作类"""
    
    def __init__(self, repo_path=None):
        """
        初始化Git仓库对象
        
        Args:
            repo_path: 仓库路径，支持相对路径、绝对路径和~扩展
        """
        self.original_dir = os.getcwd()
        self.repo_path = self._expand_path(repo_path) if repo_path else None
    
    def _expand_path(self, path):
        """展开路径，支持~和变量"""
        # 展开用户目录 ~
        path = os.path.expanduser(path)
        # 展开环境变量
        path = os.path.expandvars(path)
        # 获取绝对路径
        return os.path.abspath(path)
    
    def change_directory(self, path):
        """改变工作目录"""
        try:
            expanded_path = self._expand_path(path)
            
            if not os.path.exists(expanded_path):
                raise FileNotFoundError(f"目录不存在: {expanded_path}")
            
            if not os.path.isdir(expanded_path):
                raise NotADirectoryError(f"不是目录: {expanded_path}")
            
            os.chdir(expanded_path)
            print(f"✅ 切换到目录: {os.getcwd()}")
            self.repo_path = expanded_path
            return True
            
        except Exception as e:
            print(f"❌ 切换目录失败: {e}")
            return False
    
    def run_git_command(self, command, capture_output=True):
        """
        执行git命令
        
        Args:
            command: git命令，如 ["branch", "-a"] 或 "branch -a"
            capture_output: 是否捕获输出
        """
        try:
            # 确保在正确的目录
            if self.repo_path and os.getcwd() != self.repo_path:
                self.change_directory(self.repo_path)
            
            # 处理命令格式
            if isinstance(command, str):
                git_cmd = ["git"] + command.split()
            else:
                git_cmd = ["git"] + list(command)
            
            # 执行命令
            result = subprocess.run(
                git_cmd,
                capture_output=capture_output,
                text=True,
                encoding='utf-8'
            )
            
            return {
                'success': result.returncode == 0,
                'returncode': result.returncode,
                'stdout': result.stdout if capture_output else None,
                'stderr': result.stderr if capture_output else None,
                'command': ' '.join(git_cmd)
            }
            
        except FileNotFoundError:
            return {
                'success': False,
                'error': 'git命令未找到，请确保已安装Git',
                'command': ' '.join(git_cmd) if 'git_cmd' in locals() else 'git'
            }
        except Exception as e:
            return {
                'success': False,
                'error': str(e),
                'command': ' '.join(git_cmd) if 'git_cmd' in locals() else 'git'
            }
    
    def list_branches(self, show_all=True):
        """列出分支"""
        command = ["branch", "-a"] if show_all else ["branch"]
        return self.run_git_command(command)
    
    def get_current_branch(self):
        """获取当前分支"""
        result = self.run_git_command(["branch", "--show-current"])
        if result['success'] and result['stdout']:
            return result['stdout'].strip()
        return None
    
    def get_status(self):
        """获取仓库状态"""
        return self.run_git_command(["status"])
    
    def restore_directory(self):
        """恢复原始目录"""
        os.chdir(self.original_dir)
        print(f"🔙 恢复到原始目录: {self.original_dir}")
    
    def __enter__(self):
        """上下文管理器入口"""
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        """上下文管理器退出"""
        self.restore_directory()

# 使用示例
def main():
    # 方法1：直接使用
    repo = GitRepository()
    
    # 切换目录
    if repo.change_directory("~/GoogleTV5/gitfromMTK/device/changhong/hakata"):
        # 列出所有分支
        result = repo.list_branches(show_all=True)
        
        if result['success']:
            print("✅ Git分支列表:")
            print(result['stdout'])
            
            # 获取当前分支
            current_branch = repo.get_current_branch()
            if current_branch:
                print(f"📌 当前分支: {current_branch}")
        else:
            print(f"❌ 执行失败: {result.get('error', '未知错误')}")
            if result.get('stderr'):
                print(f"错误信息: {result['stderr']}")
    
    # 恢复原始目录
    repo.restore_directory()
    
    print("n" + "="*50 + "n")
    
    # 方法2：使用上下文管理器（推荐）
    repo_path = "~/GoogleTV5/gitfromMTK/device/changhong/hakata"
    
    with GitRepository(repo_path) as repo:
        # 在这个代码块中，会自动切换到指定目录
        # 退出代码块时，会自动恢复原始目录
        
        print(f"📁 当前在仓库: {os.getcwd()}")
        
        # 执行多个git命令
        result1 = repo.list_branches()
        if result1['success']:
            print("✅ 所有分支:")
            print(result1['stdout'])
        
        result2 = repo.get_status()
        if result2['success']:
            print("✅ 仓库状态:")
            print(result2['stdout'])
        
        # 也可以执行自定义git命令
        result3 = repo.run_git_command("log --oneline -5")
        if result3['success']:
            print("✅ 最近5次提交:")
            print(result3['stdout'])

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Git仓库操作工具
用法: python git_tool.py [命令] [参数]
"""

import os
import subprocess
import argparse
import sys

def execute_in_directory(directory, command, use_shell=False):
    """
    在指定目录执行命令
    
    Args:
        directory: 目标目录
        command: 要执行的命令
        use_shell: 是否使用shell执行
    """
    original_dir = os.getcwd()
    
    try:
        # 展开路径
        directory = os.path.expanduser(directory)
        directory = os.path.expandvars(directory)
        
        # 检查目录是否存在
        if not os.path.exists(directory):
            print(f"错误: 目录不存在 - {directory}")
            return False
        
        # 切换目录
        os.chdir(directory)
        print(f"切换到: {os.getcwd()}")
        
        # 执行命令
        if use_shell:
            # 使用shell执行
            result = subprocess.run(
                command,
                shell=True,
                capture_output=True,
                text=True,
                encoding='utf-8'
            )
        else:
            # 不使用shell
            result = subprocess.run(
                command,
                capture_output=True,
                text=True,
                encoding='utf-8'
            )
        
        # 输出结果
        if result.stdout:
            print(result.stdout)
        
        if result.stderr:
            print(f"错误: {result.stderr}", file=sys.stderr)
        
        # 返回是否成功
        return result.returncode == 0
        
    except Exception as e:
        print(f"执行出错: {e}", file=sys.stderr)
        return False
        
    finally:
        # 恢复原始目录
        os.chdir(original_dir)

def main():
    parser = argparse.ArgumentParser(
        description='在指定目录执行Git命令',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
示例:
  %(prog)s ~/project git branch -a
  %(prog)s ~/project "git status"
  %(prog)s --shell ~/project "cd src && git log --oneline"
        """
    )
    
    parser.add_argument(
        'directory',
        help='要切换的目录'
    )
    
    parser.add_argument(
        'command',
        nargs='+',
        help='要执行的命令'
    )
    
    parser.add_argument(
        '--shell',
        action='store_true',
        help='使用shell执行命令（支持&&等操作符）'
    )
    
    parser.add_argument(
        '--verbose',
        '-v',
        action='store_true',
        help='显示详细信息'
    )
    
    args = parser.parse_args()
    
    if args.verbose:
        print(f"目录: {args.directory}")
        print(f"命令: {' '.join(args.command)}")
        print(f"使用shell: {args.shell}")
    
    # 如果是git命令且没有指定shell，自动添加git前缀
    if not args.shell and args.command[0] != 'git':
        args.command = ['git'] + args.command
    
    command_str = ' '.join(args.command) if args.shell else args.command
    
    success = execute_in_directory(
        args.directory,
        command_str,
        use_shell=args.shell
    )
    
    sys.exit(0 if success else 1)

if __name__ == '__main__':
    main()

import os
import subprocess
import json
from datetime import datetime
from typing import List, Dict, Optional, Any

class AdvancedGitManager:
    """高级Git仓库管理器"""
    
    def __init__(self, repo_path=None):
        self.repo_path = os.path.expanduser(repo_path) if repo_path else None
        self.original_dir = os.getcwd()
        
        if self.repo_path and not os.path.exists(self.repo_path):
            raise FileNotFoundError(f"仓库目录不存在: {self.repo_path}")
    
    def _ensure_in_repo(self):
        """确保在正确的仓库目录"""
        if not self.repo_path:
            raise ValueError("未指定仓库路径")
        
        if os.getcwd() != self.repo_path:
            os.chdir(self.repo_path)
    
    def _run_git(self, args, **kwargs):
        """执行git命令的通用方法"""
        self._ensure_in_repo()
        
        try:
            result = subprocess.run(
                ['git'] + args,
                capture_output=True,
                text=True,
                encoding='utf-8',
                **kwargs
            )
            
            return {
                'success': result.returncode == 0,
                'returncode': result.returncode,
                'stdout': result.stdout.strip(),
                'stderr': result.stderr.strip(),
                'command': 'git ' + ' '.join(args)
            }
        except Exception as e:
            return {
                'success': False,
                'error': str(e),
                'command': 'git ' + ' '.join(args)
            }
    
    def get_all_branches(self, verbose=False):
        """获取所有分支（包括远程分支）"""
        result = self._run_git(['branch', '-a'])
        
        if not result['success']:
            return result
        
        branches = []
        for line in result['stdout'].split('n'):
            line = line.strip()
            if not line:
                continue
            
            # 解析分支信息
            is_current = line.startswith('*')
            is_remote = 'remotes/' in line
            
            if is_current:
                line = line[2:]  # 移除 "* "
            
            branch_info = {
                'name': line.replace('remotes/', '') if is_remote else line,
                'is_current': is_current,
                'is_remote': is_remote,
                'remote': None
            }
            
            # 解析远程分支
            if is_remote:
                parts = line.split('/')
                if len(parts) >= 2:
                    branch_info['remote'] = parts[0]
                    branch_info['name'] = '/'.join(parts[1:])
            
            branches.append(branch_info)
        
        if verbose:
            print("🌿 分支列表:")
            for branch in branches:
                prefix = "📍 " if branch['is_current'] else "  "
                remote = f" ({branch['remote']})" if branch['is_remote'] else ""
                print(f"{prefix}{branch['name']}{remote}")
        
        return {
            'success': True,
            'branches': branches,
            'current': next((b for b in branches if b['is_current']), None),
            'local': [b for b in branches if not b['is_remote']],
            'remote': [b for b in branches if b['is_remote']]
        }
    
    def get_branch_info(self, branch_name=None):
        """获取分支详细信息"""
        if not branch_name:
            # 获取当前分支
            result = self._run_git(['branch', '--show-current'])
            if not result['success']:
                return result
            branch_name = result['stdout']
        
        # 获取最后提交信息
        log_result = self._run_git([
            'log', branch_name, '-1',
            '--format=%H|%an|%ae|%ad|%s',
            '--date=iso'
        ])
        
        if not log_result['success']:
            return log_result
        
        if log_result['stdout']:
            commit_hash, author, email, date, subject = log_result['stdout'].split('|', 4)
        else:
            commit_hash = author = email = date = subject = ''
        
        return {
            'success': True,
            'branch': branch_name,
            'last_commit': {
                'hash': commit_hash,
                'author': author,
                'email': email,
                'date': date,
                'subject': subject
            }
        }
    
    def switch_branch(self, branch_name, create_new=False):
        """切换分支"""
        if create_new:
            result = self._run_git(['checkout', '-b', branch_name])
        else:
            result = self._run_git(['checkout', branch_name])
        
        if result['success']:
            print(f"✅ 切换到分支: {branch_name}")
        
        return result
    
    def fetch_all(self):
        """获取所有远程更新"""
        result = self._run_git(['fetch', '--all'])
        
        if result['success']:
            print("✅ 已获取所有远程更新")
        
        return result
    
    def pull_current(self):
        """拉取当前分支更新"""
        result = self._run_git(['pull'])
        
        if result['success']:
            print("✅ 已拉取当前分支更新")
        
        return result
    
    def create_report(self, output_file=None):
        """生成仓库报告"""
        report = {
            'timestamp': datetime.now().isoformat(),
            'repository': os.getcwd(),
            'branches': [],
            'status': {}
        }
        
        # 获取分支信息
        branches_result = self.get_all_branches()
        if branches_result['success']:
            report['branches'] = branches_result
        
        # 获取仓库状态
        status_result = self._run_git(['status', '--porcelain'])
        if status_result['success']:
            status_lines = status_result['stdout'].split('n')
            report['status'] = {
                'modified': [],
                'added': [],
                'deleted': [],
                'untracked': []
            }
            
            for line in status_lines:
                if not line:
                    continue
                
                status = line[:2]
                filename = line[3:]
                
                if status == ' M':
                    report['status']['modified'].append(filename)
                elif status == 'A ' or status == 'M ':
                    report['status']['added'].append(filename)
                elif status == ' D':
                    report['status']['deleted'].append(filename)
                elif status == '??':
                    report['status']['untracked'].append(filename)
        
        # 写入文件或返回
        if output_file:
            with open(output_file, 'w', encoding='utf-8') as f:
                if output_file.endswith('.json'):
                    json.dump(report, f, indent=2, ensure_ascii=False)
                else:
                    # 文本格式
                    f.write(f"Git仓库报告n")
                    f.write(f"生成时间: {report['timestamp']}n")
                    f.write(f"仓库路径: {report['repository']}nn")
                    
                    f.write("分支信息:n")
                    for branch in report['branches'].get('branches', []):
                        current = "*" if branch['is_current'] else " "
                        remote = f" [{branch.get('remote', 'local')}]" if branch['is_remote'] else ""
                        f.write(f"  {current} {branch['name']}{remote}n")
                    
                    f.write("n文件状态:n")
                    for status_type, files in report['status'].items():
                        if files:
                            f.write(f"  {status_type}:n")
                            for file in files:
                                f.write(f"    - {file}n")
            
            print(f"✅ 报告已生成: {output_file}")
        
        return report
    
    def __enter__(self):
        """上下文管理器入口"""
        if self.repo_path:
            os.chdir(self.repo_path)
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        """上下文管理器退出"""
        os.chdir(self.original_dir)

# 使用示例
def advanced_example():
    repo_path = "~/GoogleTV5/gitfromMTK/device/changhong/hakata"
    
    try:
        # 使用上下文管理器
        with AdvancedGitManager(repo_path) as git:
            print(f"📁 当前仓库: {os.getcwd()}")
            
            # 1. 获取所有分支
            print("n1. 获取所有分支:")
            branches = git.get_all_branches(verbose=True)
            
            # 2. 获取当前分支信息
            print("n2. 当前分支信息:")
            current_branch = git.get_branch_info()
            if current_branch['success']:
                commit = current_branch['last_commit']
                print(f"   分支: {current_branch['branch']}")
                print(f"   最后提交: {commit['subject']}")
                print(f"   作者: {commit['author']}")
                print(f"   时间: {commit['date']}")
            
            # 3. 获取远程更新
            print("n3. 获取远程更新:")
            git.fetch_all()
            
            # 4. 生成报告
            print("n4. 生成仓库报告:")
            git.create_report("git_report.json")
            
            # 5. 切换分支示例
            print("n5. 切换到特定分支:")
            # 假设我们想切换到master分支
            # git.switch_branch("master")
            
    except FileNotFoundError as e:
        print(f"❌ 错误: {e}")
    except Exception as e:
        print(f"❌ 未知错误: {e}")

if __name__ == "__main__":
    advanced_example()

如果你只需要最简单的功能：

import os, subprocess
os.chdir(os.path.expanduser("~/GoogleTV5/gitfromMTK/device/changhong/hakata"))
subprocess.run(["git", "branch", "-a"])

对于简单需求：

import os
import subprocess

# 切换目录
os.chdir(os.path.expanduser("~/GoogleTV5/gitfromMTK/device/changhong/hakata"))

# 执行git命令
result = subprocess.run(["git", "branch", "-a"], capture_output=True, text=True)
print(result.stdout)

对于复杂需求：
使用 GitRepository 或 AdvancedGitManager 类，它们提供了：

• 错误处理
• 路径扩展（支持 ~）
• 上下文管理器（自动恢复目录）
• 更丰富的功能

1. 路径处理：使用 os.path.expanduser() 处理 ~
2. 错误处理：总是检查目录是否存在
3. 安全性：避免使用 shell=True，除非必要
4. 资源管理：使用上下文管理器或确保恢复原始目录

通过以上方法，你可以在 Python 中安全、有效地执行 cd 和 git 命令。

以上就是Python执行Linux命令的多种方法的详细内容，更多关于Python执行Linux命令的资料请关注风君子博客其它相关文章！

您可能感兴趣的文章:

• 在Python代码中执行Linux命令的详细用法教程
• python执行linux系统命令的三种方式小结
• Python通过paramiko库实现远程执行linux命令的方法
• python中windows链接linux执行命令并获取执行状态的问题小结
• Python执行dos和Linux命令的方法详解