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()