|
1 | 1 | import subprocess |
2 | 2 | import os |
3 | 3 | import shlex |
| 4 | +import logging |
4 | 5 |
|
| 6 | +logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") |
5 | 7 |
|
6 | | -# 1. Using subprocess.run (Recommended for most use cases) |
7 | | -def run_with_subprocess_run(command): |
| 8 | +def run_with_subprocess_run(command, timeout=10): |
8 | 9 | try: |
9 | 10 | result = subprocess.run( |
10 | | - command, shell=True, check=True, text=True, capture_output=True |
| 11 | + shlex.split(command), check=True, text=True, capture_output=True, timeout=timeout |
11 | 12 | ) |
12 | | - print("Output:", result.stdout) |
| 13 | + logging.info("Output: %s", result.stdout.strip()) |
| 14 | + except subprocess.TimeoutExpired: |
| 15 | + logging.error("Command timed out after %d seconds", timeout) |
13 | 16 | except subprocess.CalledProcessError as e: |
14 | | - print("Error:", e.stderr) |
| 17 | + logging.error("Error: %s", e.stderr.strip()) |
15 | 18 |
|
16 | | - |
17 | | -# 2. Using subprocess.Popen (More control over input/output streams) |
18 | | -def run_with_subprocess_popen(command): |
| 19 | +def run_with_subprocess_popen(command, timeout=10): |
19 | 20 | process = subprocess.Popen( |
20 | 21 | shlex.split(command), stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True |
21 | 22 | ) |
22 | | - stdout, stderr = process.communicate() |
23 | | - if process.returncode == 0: |
24 | | - print("Output:", stdout) |
25 | | - else: |
26 | | - print("Error:", stderr) |
27 | | - |
28 | | - |
29 | | -# 3. Using os.system (Simple but not recommended for new code due to security risks) |
30 | | -def run_with_os_system(command): |
31 | | - exit_code = os.system(command) |
32 | | - print(f"Command exited with code {exit_code}") |
33 | | - |
34 | | - |
35 | | -# 4. Using subprocess.check_output (Useful for capturing command output) |
36 | | -def run_with_check_output(command): |
37 | 23 | try: |
38 | | - output = subprocess.check_output(command, shell=True, text=True) |
39 | | - print("Output:", output) |
| 24 | + stdout, stderr = process.communicate(timeout=timeout) |
| 25 | + if process.returncode == 0: |
| 26 | + logging.info("Output: %s", stdout.strip()) |
| 27 | + else: |
| 28 | + logging.error("Error: %s", stderr.strip()) |
| 29 | + except subprocess.TimeoutExpired: |
| 30 | + process.kill() |
| 31 | + stdout, stderr = process.communicate() |
| 32 | + logging.error("Command timed out and process was killed") |
| 33 | + |
| 34 | +def run_with_check_output(command, timeout=10): |
| 35 | + try: |
| 36 | + output = subprocess.check_output(shlex.split(command), text=True, timeout=timeout) |
| 37 | + logging.info("Output: %s", output.strip()) |
| 38 | + except subprocess.TimeoutExpired: |
| 39 | + logging.error("Command timed out after %d seconds", timeout) |
40 | 40 | except subprocess.CalledProcessError as e: |
41 | | - print("Error:", e.stderr) |
| 41 | + logging.error("Error: %s", e.stderr.strip()) |
42 | 42 |
|
43 | | - |
44 | | -# Example commands for demonstration |
45 | 43 | command1 = "echo Hello, World!" |
46 | 44 | command2 = "ls -l" if os.name != "nt" else "dir" |
47 | 45 |
|
48 | | -# Execute the commands using various methods |
49 | | -print("=== subprocess.run ===") |
| 46 | +logging.info("=== subprocess.run ===") |
50 | 47 | run_with_subprocess_run(command1) |
51 | 48 |
|
52 | | -print("\n=== subprocess.Popen ===") |
| 49 | +logging.info("=== subprocess.Popen ===") |
53 | 50 | run_with_subprocess_popen(command1) |
54 | 51 |
|
55 | | -print("\n=== os.system ===") |
56 | | -run_with_os_system(command1) |
57 | | - |
58 | | -print("\n=== subprocess.check_output ===") |
| 52 | +logging.info("=== subprocess.check_output ===") |
59 | 53 | run_with_check_output(command1) |
0 commit comments