node项目中调用执行python脚本的方法详解

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

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",
);
// 🔐 全局执行锁
let isRunning = false;

// 注意脚本文件里访问文件时的路径，要用绝对路径
exports.get_info = (req, res) => {
  if (isRunning) {
    return res.send({
      status: 429,
      message: "任务正在执行中，请稍后再试",
    });
  }

  isRunning = true;

  // 第二个参数是数组，表示执行哪个脚本，和传给 Python 的参数
  const py = spawn(pythonPath, [scriptPath]);

  let error = "";

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

  py.on("error", (err) => {
    console.error("启动失败:", err);
    isRunning = false;
    return res.send({
      status: 500,
      message: "Python 启动失败",
      error: err.message,
    });
  });
  // 超时杀进程
  const timeout = setTimeout(() => {
    py.kill("SIGTERM");
  }, 300000); // 300秒超时

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

    console.error("Python执行错误:", error);

    return res.send({
      status: 500,
      message: "Python执行失败",
      error: error || "未知错误",
    });
  });
};

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

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

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

# 关键：让 python 能识别 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()
# 注册路由：
@app.get("/aaa/bbb")
async def ccc():
    result = await get_test_info()
    return result

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

const axios = require("axios");
// 全局执行锁
let isRunning = false;

// 注意脚本文件里访问文件时的路径，要用绝对路径
exports.get_info = async (req, res) => {
  if (isRunning) {
    return res.send({
      status: 429,
      message: "任务正在执行中，请稍后再试",
    });
  }

  isRunning = true;

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

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

    return res.send({
      status: 500,
      message: "FastAPI调用失败",
      error: error.message,
    });
  }
};