subprocess 调用 ffmpeg 和 logging 的配置
在很久之后我尝试再使用以前写的那个 subprocess 代码来运行 ffmpeg，我碰到了不少问题。
被测试的 Command
比如这次参与测试的两条 command 是这样的:
# ffmpeg -version
command = [
str(FFMPEG_PATH),
"-version"
]
# ffmpeg -i example.wav -vn -af aresample=async=1 -acodec pcm_s16le -ar 44100 -ac 2 example.wav
command = [
str(FFMPEG_PATH),
'-i', str(input_mp4_path.absolute()),
'-vn', # 禁用视频流
'-af', 'aresample=async=1', # 音频滤波器，异步重采样
'-acodec', 'pcm_s16le', # 音频编码器，PCM s16le (16-bit signed little-endian)
'-ar', '44100', # 音频采样率，44100 Hz
'-ac', '2', # 音频通道数，2 (立体声)
str(output_wav_path.absolute())
]使用的代码
参见:使用 subprocess 调用 python 脚本并且打印执行脚本的 output 一般写法。
关于 shell=True 参数
指定它，你需要把命令以str的形式传入而不是list[str]。
指定它，你可以直接使用cd,mkdir,>,&& 这样的指令，而如果不指定，则需要用/user/bin/cd这样的。
大多时候，不建议开启shell=True,除非真的必要。
关于 subprocess.run(check=True)
如果 returncode=False 直接中断程序。相反如果设置为 False , 那么子进程会一直运行。直到结束或者异常。通常来说这没有必要，但多执行一点可能可以看到更多的 BUG 信息。所以默认我设为 False。
碰到的 Bug
- 如果没有删除 wav ,再次运行会卡死在那一步。
ffmpeg -version的输出不可见。
关于再次运行卡死的解决:
加入 -y 参数,强制覆盖输出文件:
# ffmpeg -i example.wav -vn -af aresample=async=1 -acodec pcm_s16le -ar 44100 -ac 2 -y example.wav
command = [
str(FFMPEG_PATH),
'-i', str(input_mp4_path.absolute()),
'-vn', # 禁用视频流
'-af', 'aresample=async=1', # 音频滤波器，异步重采样
'-acodec', 'pcm_s16le', # 音频编码器，PCM s16le (16-bit signed little-endian)
'-ar', '44100', # 音频采样率，44100 Hz
'-ac', '2', # 音频通道数，2 (立体声)
'-y',
str(output_wav_path.absolute())
]部分输出可见，部分输出不可见
Logger 优先级和日志可见性
似乎问题出在我把 logger的level设置为info，但是我的日志等级分别是DEBUG和WARNING，最终只输出了WARNING。
一般在主程序入口需要设置 logger_level:
import logging
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(name)s - %(message)s')
if __name__ == "__main__":
# ... 调用 run_shell_command ...
logging.info("This is a info message from main program") # info 等级日志。
logging.debug("This is a debug message from main program") # DEBUG 等级日志。CRITICAL - 50
ERROR - 40
WARNING - 30
INFO - 20
DEBUG - 10
NOTSET - 0 (表示未设置级别，会继承父 logger 的级别，或者默认为 WARNING)对于上面那个程序。如果我设置 level=DEBUG,我将能看到两条日志。
如果我设置 level=info ，那么我将只能看到第一条info日志。如果有更高级warning,ERROR也会看到。
另外，
logger = logging.getLogger(__name__)意思是，以函数名作为 logger名称，但是实际上可以拼接，把函数再划分为模块。
最终定下来的代码：
可以通过log_level 来设定WARNING和ERROR,哪些情况应该被警告。
以及，即使不输出日志，用户也可以通过result.returncode:bool来判断脚本是否执行成功，然后写下处理信息。
loggger_name 用于整理归类 logs 。
import subprocess
import logging
def run_shell_command(command: list[str],log_level_stdout:int=logging.DEBUG, log_level_stderr:int=logging.WARNING,logger_name:str="" , check_returncode:bool=False):
"""
运行 shell 命令并记录输出，可以控制日志级别和是否检查返回码。
参数:
command (list[str]): 要运行的命令。
log_level_stdout (int): 标准输出的日志级别，默认为 DEBUG。
log_level_stderr (int): 标准错误的日志级别，默认为 WARNING。
check_returncode (bool): 是否检查返回码，如果为 True 且返回码非零，则抛出异常。 默认为 False。
返回:
subprocess.CompletedProcess: 包含命令执行结果的对象。
抛出:
subprocess.CalledProcessError: 如果 check_returncode=True 且命令返回非零返回码。
"""
command_str = ' '.join(command)
if logger_name:
logger = logging.getLogger(logger_name+" - "+__name__)
else:
logger = logging.getLogger(__name__)
logger.debug(f"执行命令: {command_str}")
try:
result = subprocess.run(
command,
capture_output=True,
text=True,
encoding='utf-8',
check=check_returncode # 根据参数决定是否检查返回码
)
if result.stdout:
logger.log(log_level_stdout, result.stdout) # 使用可配置的日志级别
if result.stderr:
logger.log(log_level_stderr, result.stderr) # 使用可配置的日志级别
return result
except FileNotFoundError as e:
logger.error(f"命令未找到错误: {e}")
return subprocess.CompletedProcess(args=command, returncode=-1, stdout="", stderr=str(e))
except subprocess.CalledProcessError as e: # 只在 check_returncode=True 时可能抛出
logger.error(f"子进程调用错误: {e}")
logger.error(f"错误输出:\n{e.stderr}")
raise e # 重新抛出异常，让调用者处理
except Exception as e:
logger.exception(f"发生未知错误: {e}")
return subprocess.CompletedProcess(args=command, returncode=-1, stdout="", stderr=str(e))