影刀RPA 命令行执行调用外部程序与脚本作者林焱什么情况用你的影刀流程需要调用一个Python脚本处理数据需要执行一个CMD命令来操作文件需要启动另一个程序并等待它完成需要用PowerShell执行系统管理操作命令行执行是影刀RPA与其他程序交互的桥梁。虽然影刀内置了很多功能但有时直接调用外部程序或命令行工具更方便。比如调用FFmpeg处理视频、用ImageMagick转换图片、执行数据库备份命令。本文讲清楚怎么在影刀RPA中执行命令行命令、怎么获取执行结果、怎么处理超时和错误。怎么做一、影刀RPA中的命令行执行影刀提供了【执行CMD命令】或【执行系统命令】指令【执行CMD命令】 命令python script.py --input data.csv --output result.xlsx 工作目录D:/scripts 等待方式等待执行完成 超时时间60秒 → 输出变量: output_text二、用subprocess执行命令importsubprocessimportos# 基本执行 # 方式1subprocess.run()推荐resultsubprocess.run([python,script.py],capture_outputTrue,# 捕获输出textTrue,# 输出为文本而非bytesencodingutf-8# 指定编码)print(f返回码:{result.returncode})print(f标准输出:{result.stdout})print(f标准错误:{result.stderr})# 方式2用字符串命令简单但不安全resultsubprocess.run(python script.py --input data.csv,capture_outputTrue,textTrue,shellTrue# 通过shell执行)# 带工作目录的执行 resultsubprocess.run([python,script.py],capture_outputTrue,textTrue,cwdD:/scripts# 工作目录)# 带环境变量的执行 envos.environ.copy()env[PYTHONPATH]D:/libsresultsubprocess.run([python,script.py],capture_outputTrue,textTrue,envenv)三、执行结果处理importsubprocessdefrun_command(command,cwdNone,timeout60,encodingutf-8): 执行命令并返回结构化结果 try:ifisinstance(command,str):# 字符串命令用shell执行resultsubprocess.run(command,capture_outputTrue,textTrue,encodingencoding,cwdcwd,timeouttimeout,shellTrue)else:# 列表命令直接执行更安全resultsubprocess.run(command,capture_outputTrue,textTrue,encodingencoding,cwdcwd,timeouttimeout)return{success:result.returncode0,returncode:result.returncode,stdout:result.stdout.strip()ifresult.stdoutelse,stderr:result.stderr.strip()ifresult.stderrelse,}exceptsubprocess.TimeoutExpired:return{success:False,error:f命令执行超时{timeout}秒}exceptFileNotFoundError:return{success:False,error:命令或程序不存在}exceptExceptionase:return{success:False,error:str(e)}# 使用示例# 1. 执行Python脚本resultrun_command([python,data_processor.py,--input,data.csv,--output,result.xlsx],cwdD:/scripts,timeout120)ifresult[success]:print(f脚本执行成功:{result[stdout]})else:print(f脚本执行失败:{result.get(stderr,result.get(error))})# 2. 执行CMD命令resultrun_command(dir D:\\data /B)ifresult[success]:filesresult[stdout].split(\n)print(f找到{len(files)}个文件)# 3. 执行PowerShell命令resultrun_command([powershell,-Command,Get-Process | Where-Object {$_.CPU -gt 10} | Select-Object Name,CPU])ifresult[success]:print(result[stdout])四、常见命令行场景场景1调用Python脚本处理数据defrun_python_script(script_path,argsNone,cwdNone,timeout300):调用Python脚本cmd[python,script_path]ifargs:cmd.extend(args)resultrun_command(cmd,cwdcwd,timeouttimeout)ifnotresult[success]:print(f脚本执行失败:)print(f 命令:{ .join(cmd)})print(f 错误:{result.get(stderr,result.get(error))})returnresult# 使用resultrun_python_script(D:/scripts/clean_data.py,args[--input,raw.csv,--output,clean.csv,--mode,strict],cwdD:/scripts,timeout300)场景2文件操作命令# 复制文件run_command([copy,source.txt,target.txt],shellFalse)# 或用xcopy复制文件夹run_command([xcopy,D:/source/*,D:/target/,/E,/I,/Y])# 删除文件run_command([del,D:/temp/old_files.txt])# 创建目录run_command([mkdir,D:/new_folder])# 移动文件run_command([move,D:/temp/file.txt,D:/archive/])场景3调用FFmpeg处理视频defconvert_video(input_file,output_file,codeclibx264):用FFmpeg转换视频格式cmd[ffmpeg,-i,input_file,# 输入文件-c:v,codec,# 视频编码-c:a,aac,# 音频编码-y,# 覆盖输出output_file]resultrun_command(cmd,timeout600)ifresult[success]:print(f视频转换成功:{output_file})else:print(f视频转换失败:{result.get(stderr)})returnresult场景4数据库操作defbackup_mysql(host,username,password,database,output_file):用mysqldump备份MySQL数据库cmd[mysqldump,-h,host,-u,username,f-p{password},# 密码直接跟在-p后面注意安全database,--result-file,output_file]resultrun_command(cmd,timeout600)returnresult五、后台执行与进程管理importsubprocessimporttimeimportosdefstart_background_process(command,cwdNone):启动后台进程不等待完成ifisinstance(command,str):processsubprocess.Popen(command,shellTrue,cwdcwd,stdoutsubprocess.PIPE,stderrsubprocess.PIPE,textTrue,encodingutf-8)else:processsubprocess.Popen(command,cwdcwd,stdoutsubprocess.PIPE,stderrsubprocess.PIPE,textTrue,encodingutf-8)print(f已启动后台进程PID:{process.pid})returnprocessdefcheck_process_status(process):检查进程状态statusprocess.poll()ifstatusisNone:returnrunningelifstatus0:returncompletedelse:returnffailed (code{status})defwait_for_process(process,timeout300):等待进程完成try:stdout,stderrprocess.communicate(timeouttimeout)return{success:process.returncode0,stdout:stdout,stderr:stderr,returncode:process.returncode}exceptsubprocess.TimeoutExpired:process.kill()return{success:False,error:进程超时被强制终止}# 使用启动一个长时间运行的进程processstart_background_process([python,long_task.py],cwdD:/scripts)# 做其他事情time.sleep(5)# 检查进程状态statuscheck_process_status(process)print(f进程状态:{status})# 等待完成ifstatusrunning:resultwait_for_process(process,timeout300)print(f结果:{result})六、管道与链式命令# 用shell执行管道命令resultsubprocess.run(dir D:\\data /B | findstr .csv,capture_outputTrue,textTrue,encodingutf-8,shellTrue)# 列出data目录下所有CSV文件# 多个命令串联resultsubprocess.run(cd D:\\scripts python build.py python deploy.py,capture_outputTrue,textTrue,shellTrue)有什么坑坑1shellTrue的安全风险现象如果命令字符串中包含用户输入shellTrue可能导致命令注入攻击。解决尽量用列表形式传命令[python, script.py]不用shellTrue。如果必须用shell确保命令字符串中不包含不受信任的输入。对用户输入做转义或用shlex.quote()。坑2中文输出乱码现象CMD命令输出的中文在Python中显示为乱码。原因Windows CMD默认使用GBK编码但Python的subprocess可能用UTF-8解码。解决明确指定encoding参数resultsubprocess.run(dir,capture_outputTrue,textTrue,encodinggbk# Windows CMD用GBK编码)或者读取bytes后手动解码resultsubprocess.run(dir,capture_outputTrue)textresult.stdout.decode(gbk,errorsreplace)坑3subprocess.run阻塞导致流程卡死现象调用的程序一直运行不退出subprocess.run()永远等不到结果。解决始终设置timeout参数。如果程序可能长时间运行用subprocess.Popen()启动后台进程定期检查状态。超时后用process.kill()强制终止。坑4路径中的空格和特殊字符现象subprocess.run([python, C:/Program Files/script.py])报错找不到文件。原因路径中有空格在shell模式下会被拆分成多个参数。解决用列表形式传参不用shellTrue列表中的每个元素会被当作一个完整的参数不受空格影响。如果必须用shell用引号包裹路径C:/Program Files/script.py。坑5影刀的【执行CMD命令】和Python的subprocess行为不一致现象同一个命令在影刀的【执行CMD命令】中能执行但在Python代码块的subprocess中报错。原因影刀的【执行CMD命令】可能默认使用shell模式会加载用户环境变量和PATH。Python的subprocess默认不通过shell可能找不到某些命令。解决在Python中用shellTrue让命令通过shell执行。或者用完整路径调用程序。检查环境变量PATH是否包含目标程序的目录。在影刀中统一用一种方式——要么都用【执行CMD命令】指令要么都用Python代码块。