快捷導(dǎo)航

node項(xiàng)目中調(diào)用執(zhí)行python腳本的方法詳解

更新時(shí)間：2026年03月05日 08:35:32 作者：允許部分打工人先富起來(lái)

這篇文章主要為大家詳細(xì)介紹了node項(xiàng)目中調(diào)用執(zhí)行python腳本的兩個(gè)常用方法,文中的示例代碼講解詳細(xì),感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下

一、環(huán)境配置

Windows + PowerShell 命令：

在node項(xiàng)目中新建python文件夾，以及子文件夾scripts用于放置python腳本
創(chuàng)建虛擬環(huán)境 cd python -> python -m venv venv
進(jìn)入虛擬環(huán)境并install腳本中的依賴(lài)：venv\Scripts\Activate.ps1 -> 單個(gè)安裝 pip install pandas
不要把 venv 提交到 git

// .gitignore
python/venv/
__pycache__/
*.pyc

只提交依賴(lài)清單

執(zhí)行：pip freeze > requirements.txt 提交這個(gè)文件即可

服務(wù)器部署時(shí)重新創(chuàng)建環(huán)境

python -m venv venv
pip install -r requirements.txt

二、運(yùn)行python腳本

方法一：用spawn 子進(jìn)程通信（不建議）

缺點(diǎn)：頻繁 spawn 會(huì)很慢，stdout 容易污染

注意：

絕對(duì)路徑的拼接，以及腳本里如果有讀取文件代碼時(shí)的文件路徑
Python腳本里不能隨便 print，因?yàn)閚ode必須自己保證輸出是 JSON

const { spawn } = require("child_process");
const path = require("path");
const rootPath = path.resolve(__dirname, "..");
// 1?? 定位 Python 解釋器路徑：E:\node\xxx\python\venv\Scripts\python.exe
const pythonPath = path.join(
  rootPath,
  "python",
  "venv",
  "Scripts",
  "python.exe",
);
// 2?? 定位 Python 腳本路徑：E:\node\xxx\python\scripts\test.py
const scriptPath = path.join(
  rootPath,
  "python",
  "scripts",
  "test.py",
);
// ?? 全局執(zhí)行鎖
let isRunning = false;

// 注意腳本文件里訪(fǎng)問(wèn)文件時(shí)的路徑，要用絕對(duì)路徑
exports.get_info = (req, res) => {
  if (isRunning) {
    return res.send({
      status: 429,
      message: "任務(wù)正在執(zhí)行中，請(qǐng)稍后再試",
    });
  }

  isRunning = true;

  // 第二個(gè)參數(shù)是數(shù)組，表示執(zhí)行哪個(gè)腳本，和傳給 Python 的參數(shù)
  const py = spawn(pythonPath, [scriptPath]);

  let error = "";

  py.stderr.on("data", (data) => {
    error += data.toString();
  });

  py.on("error", (err) => {
    console.error("啟動(dòng)失敗:", err);
    isRunning = false;
    return res.send({
      status: 500,
      message: "Python 啟動(dòng)失敗",
      error: err.message,
    });
  });
  // 超時(shí)殺進(jìn)程
  const timeout = setTimeout(() => {
    py.kill("SIGTERM");
  }, 300000); // 300秒超時(shí)

  py.on("close", (code) => {
    clearTimeout(timeout);
    isRunning = false;
    if (code === 0) {
      return res.send({
        status: 200,
        message: "Python腳本執(zhí)行成功",
      });
    }

    console.error("Python執(zhí)行錯(cuò)誤:", error);

    return res.send({
      status: 500,
      message: "Python執(zhí)行失敗",
      error: error || "未知錯(cuò)誤",
    });
  });
};

每個(gè)請(qǐng)求都 spawn 一個(gè) Python 進(jìn)程，如果接口被并發(fā)打（比如前端連點(diǎn)多次請(qǐng)求、或者有壓測(cè)），就會(huì)同時(shí)起多個(gè) python.exe 導(dǎo)致內(nèi)存瞬間爆掉，所以這里限制了單進(jìn)程：

①在腳本test.py里強(qiáng)制限制線(xiàn)程，放在所有import之前

import os
os.environ["OPENBLAS_NUM_THREADS"] = "1"
os.environ["OMP_NUM_THREADS"] = "1"
os.environ["MKL_NUM_THREADS"] = "1"

②加單進(jìn)程鎖（不允許并發(fā)執(zhí)行），以及加超時(shí)殺進(jìn)程（防卡死）

同一時(shí)間只執(zhí)行一個(gè) Python 腳本，后面的請(qǐng)求直接拒絕，在“單進(jìn)程 + spawn 模式”下，只要 Python 進(jìn)程正常退出，內(nèi)存會(huì)被系統(tǒng)回收。

方法二：使用FastAPI（推薦）

用 Python 起一個(gè) API 服務(wù)，Node 通過(guò) HTTP 調(diào)用，穩(wěn)定性最高，避免：spawn 死鎖、buffer 溢出、JSON污染

Step1：進(jìn)入虛擬環(huán)境安裝 FastAPI 和 uvicorn

cd python 
venv\Scripts\Activate.ps1
pip install fastapi "uvicorn[standard]"

Step2：Python 起 API 服務(wù)，新建入口文件 api_server.py

# api_server.py 
import os
from fastapi import FastAPI
import sys

# 關(guān)鍵：讓 python 能識(shí)別 scripts 目錄
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
sys.path.append(BASE_DIR)
# 從 scripts 引入
from scripts.test import get_test_info   

app = FastAPI()
# 注冊(cè)路由：
@app.get("/aaa/bbb")
async def ccc():
    result = await get_test_info()
    return result

Step3：把腳本文件改成接口驅(qū)動(dòng)模式的函數(shù)寫(xiě)法

比如異步任務(wù)時(shí) if __name__ == "__main__": 改成async 函數(shù) async def get_test_info(): 并return json：

return {
    "status": 200,
    "file": result_file
}

注意：FastAPI 本身已經(jīng)運(yùn)行在事件循環(huán)里，不能在腳本文件里 asyncio.run()

Step4：在不同端口啟動(dòng)這個(gè) FastAPI 應(yīng)用實(shí)例：uvicorn api_server:app --host 0.0.0.0 --port 8000 --reload

Step5：不再 spawn，改成 axios 請(qǐng)求：

const axios = require("axios");
// 全局執(zhí)行鎖
let isRunning = false;

// 注意腳本文件里訪(fǎng)問(wèn)文件時(shí)的路徑，要用絕對(duì)路徑
exports.get_info = async (req, res) => {
  if (isRunning) {
    return res.send({
      status: 429,
      message: "任務(wù)正在執(zhí)行中，請(qǐng)稍后再試",
    });
  }

  isRunning = true;

  try {
    const response = await axios.get(
      "http://127.0.0.1:8000/aaa/bbb",
      {
        timeout: 300000, // 5分鐘超時(shí)
      }
    );

    isRunning = false;
    
    return res.send({
      status: 200,
      message: "Python腳本執(zhí)行成功",
      data: response.data,
    });
  } catch (error) {
    isRunning = false;

    return res.send({
      status: 500,
      message: "FastAPI調(diào)用失敗",
      error: error.message,
    });
  }
};

到此這篇關(guān)于node項(xiàng)目中調(diào)用執(zhí)行python腳本的方法詳解的文章就介紹到這了,更多相關(guān)node調(diào)用python腳本內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: