Electron 执行python脚本

1 需求背景

有个需求需要Electron执行在本地执行python脚本。希望通过Electron调用python服务并且实现双向通信。

2 解决思路

使用Electon 的{ exec, spawn, execFile, fork } from "child_process"; 能力来执行python脚本，使用spawn可以实现持续交互，稍后见示例代码。

2.1 在electon打包python和python文件

在开发环境，可以通过直接使用已有python和python文件来测试electron能否执行python。

// 完成代码后后文附上，这里知道在执行python文件就行了
const pythonProcess = spawn(pythonPath, [scriptPath]);

结论：可以执行

2.2 在生产环境测试python执行

在生产环境python包和python文件需要放到resources，打包完成之后，你可以在resources文件夹里面看到python文件夹和你的python文件。
打包配置如下：

"extraResources": [{"from": "python_env.zip","to": "python_env.zip","filter": ["**/*"]},{"from": "electron/main/python","to": "python_scripts","filter": ["**/*.py"]}]

打包后的结果：

2.3 使用python第三方sdk

在实际应用中肯定不能只用python包，也许使用python sdk。继续调研后得到方法，可以直接使用 python 虚拟环境，python虚拟环境是一个包含你所有三方sdk的独立环境，方便移植。
用numpy行测试，输出符合预期。
创建python虚拟环境步骤如下

# 创建虚拟开发环境
`python3 -m venv python_env`# 激活虚拟环境
`source python_env/bin/activate`# 生成 requirement.txt`pip3 freeze > requirements.txt`# 安装依赖`pip3 install -r requirements.txt`

把虚拟环境文件夹python_env打包成zip放到Electron项目里。
注意！！！
需要使用压缩包，在Electron里main直接使用python_env文件夹，可能会打包失败。

2.4 解压缩python虚拟环境，运行python脚本

最近比较忙，到这一步搁置了。以后补上。

示例代码（干货）

ececPy.ts

import { exec, spawn, execFile, fork } from "child_process";
import path from "node:path";
import fs from "fs";
import { fileURLToPath } from "node:url";
const __dirname = path.dirname(fileURLToPath(import.meta.url));
const devPythonPath = path.join(__dirname, "../..");// 查找 Python 可执行文件
function findPython() {console.log("devPythonPath:", devPythonPath);const possibilities1 = [// 在打包的应用中path.join(process.resourcesPath, "python_env", "bin", "python3.9"),];for (const pythonPath of possibilities1) {if (fs.existsSync(pythonPath)) {return pythonPath;}}console.log("Could not find python3 for product, checked", possibilities1);const possibilities2 = [// 在开发环境中path.join(devPythonPath, "python_env", "bin", "python3.9"),];for (const pythonPath of possibilities2) {if (fs.existsSync(pythonPath)) {return pythonPath;}}console.log("Could not find python3 for dev, checked", possibilities2);console.log("测试环境请吧python压缩包解压到项目根目录");const possibilities3 = [// 如果上述路径找不到，尝试系统默认的 python3"python3",];for (const pythonPath of possibilities2) {if (fs.existsSync(pythonPath)) {return pythonPath;}}console.log("Could not find python3 for dev, checked", possibilities3);return null;
}// 启动 Python 进程并进行交互
export async function startPingPong() {console.log("call start pingpong");const pythonPath = findPython();if (!pythonPath) {console.error("Python not found");return;}// 使用 spawn 而不是 execFile 以便进行持续交互// 生产环境路径let scriptPath = path.join(process.resourcesPath,"/python_scripts/pingpong.py");console.log("生产环境 scriptPath:", scriptPath);if (!fs.existsSync(scriptPath)) {scriptPath = "";}// 测试环境路径if (!scriptPath) {scriptPath = path.join(devPythonPath, "/electron/main/python/pingpong.py");console.log("测试环境 scriptPath:", scriptPath);}const pythonProcess = spawn(pythonPath, [scriptPath]);// 处理 Python 输出pythonProcess.stdout.on("data", (data: any) => {try {const response = JSON.parse(data.toString());console.log("Received from Python:", response);// 如果收到 pong，继续发送 pingif (response.action === "pong") {setTimeout(() => {sendPing(pythonProcess, response.count);}, 1000);}} catch (error) {console.error("Error parsing Python response:", error);}});// 处理错误pythonProcess.stderr.on("data", (data: any) => {console.error("Python error:", data.toString());});// 进程退出pythonProcess.on("close", (code: any) => {console.log(`Python process exited with code ${code}`);});// 发送初始 pingsendPing(pythonProcess, 0);
}// 发送 ping 到 Python
function sendPing(process: any, count: any) {const message = {action: "ping",count: count,timestamp: Date.now(),};console.log("Sending to Python:", message);process.stdin.write(JSON.stringify(message) + "\n");
}export const unzipPython = () => {// TODO: 解压python压缩包
};

pingpong.py

import sys
import json
import time
import numpy as np
arr1 = np.array([1, 3, 2, 5, 4])
arr1_str = json.dumps(arr1.tolist())def main():
# 简单的 ping-pong 交互for line in sys.stdin:try:# 解析从 Node.js 发送来的数据data = json.loads(line.strip())if data.get('action') == 'ping':# 收到 ping，回复 pongresponse = {'action': 'pong','timestamp': time.time(),'count': data.get('count', 0) + 1,'arr1_str': arr1_str}print(json.dumps(response))sys.stdout.flush()  # 确保立即发送响应elif data.get('action') == 'exit':# 退出命令breakexcept json.JSONDecodeError:# 处理无效的 JSON 数据error_response = {'error': 'Invalid JSON','received': line.strip()}print(json.dumps(error_response))sys.stdout.flush()if __name__ == '__main__':main()