如何在TypeScript服务端文件中调用Python脚本实现XML更新？

方案1：将Python功能封装为HTTP接口（最常用） #

把Python的XML处理逻辑做成HTTP服务，TypeScript通过HTTP请求直接调用，适合跨环境、易扩展的场景。

Python端（FastAPI示例） #

from fastapi import FastAPI, Body
import xml.etree.ElementTree as ET

app = FastAPI()

@app.post("/update-xml")
async def update_xml(xml_content: str = Body(...), updates: dict = Body(...)):
    # 解析XML内容
    root = ET.fromstring(xml_content)
    # 执行自定义更新逻辑（示例：根据路径更新节点文本）
    for node_path, new_text in updates.items():
        target_elements = root.findall(node_path)
        for elem in target_elements:
            elem.text = new_text
    # 返回更新后的XML字符串
    return {"updated_xml": ET.tostring(root, encoding='unicode')}

启动服务：uvicorn main:app --host 0.0.0.0 --port 8000

TypeScript端（Node.js环境，使用fetch） #

async function invokeXmlUpdate(xmlContent: string, updates: Record<string, string>) {
    const response = await fetch('http://localhost:8000/update-xml', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ xml_content: xmlContent, updates })
    });
    const result = await response.json();
    return result.updated_xml;
}

// 调用示例
const rawXml = '<root><title>Old Title</title><content>Old Content</content></root>';
const updateMap = { './title': 'New Title', './content': 'Updated Content' };
invokeXmlUpdate(rawXml, updateMap).then(updatedXml => console.log(updatedXml));

方案2：通过子进程直接调用Python脚本 #

在TypeScript服务端用Node.js的child_process模块执行本地Python脚本，适合轻量、无网络依赖的场景。

Python脚本（xml_processor.py） #

import sys
import json
import xml.etree.ElementTree as ET

def process_xml():
    # 从命令行参数获取输入数据
    input_payload = json.loads(sys.argv[1])
    xml_content = input_payload['xml']
    update_rules = input_payload['rules']
    
    root = ET.fromstring(xml_content)
    # 执行更新逻辑
    for path, value in update_rules.items():
        elements = root.findall(path)
        for elem in elements:
            elem.text = value
    
    # 输出结果到标准输出
    print(json.dumps({'result': ET.tostring(root, encoding='unicode')}))

if __name__ == '__main__':
    process_xml()

TypeScript端 #

import { exec } from 'child_process';
import { promisify } from 'util';

const execAsync = promisify(exec);

async function runPythonXmlProcessor(xmlContent: string, updateRules: Record<string, string>) {
    const inputData = JSON.stringify({ xml: xmlContent, rules: updateRules });
    const { stdout } = await execAsync(`python3 xml_processor.py '${inputData}'`);
    const result = JSON.parse(stdout);
    return result.result;
}

// 调用示例
const rawXml = '<root><version>1.0</version></root>';
const updateRules = { './version': '2.0' };
runPythonXmlProcessor(rawXml, updateRules).then(updated => console.log(updated));

注意：需确保服务器环境已安装Python，且脚本路径正确；若输入含特殊字符，需做转义处理避免命令注入风险。

方案3：使用RPC框架（适合复杂交互） #

用gRPC定义标准化服务接口，生成Python服务端和TypeScript客户端代码，适合高性能、多语言协作的复杂场景。

1. 定义proto文件（xml_service.proto） #

syntax = "proto3";

service XmlProcessor {
  rpc UpdateXml (XmlUpdateRequest) returns (XmlUpdateResponse);
}

message XmlUpdateRequest {
  string xml_content = 1;
  map<string, string> update_rules = 2;
}

message XmlUpdateResponse {
  string updated_xml = 1;
}

2. 生成代码 #

• Python端：用grpcio-tools生成服务端代码

  python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. xml_service.proto
  

• TypeScript端：用grpc-tools生成客户端代码

  grpc_tools_node_protoc --js_out=import_style=commonjs,binary:. --grpc_out=. --plugin=protoc-gen-grpc=`which grpc_tools_node_protoc_plugin` xml_service.proto
  

3. Python服务端实现 #

import grpc
from xml_service_pb2 import XmlUpdateResponse
from xml_service_pb2_grpc import XmlProcessorServicer, add_XmlProcessorServicer_to_server
import xml.etree.ElementTree as ET
from concurrent import futures

class XmlProcessorImpl(XmlProcessorServicer):
    def UpdateXml(self, request, context):
        root = ET.fromstring(request.xml_content)
        for path, value in request.update_rules.items():
            elements = root.findall(path)
            for elem in elements:
                elem.text = value
        return XmlUpdateResponse(updated_xml=ET.tostring(root, encoding='unicode'))

def start_server():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    add_XmlProcessorServicer_to_server(XmlProcessorImpl(), server)
    server.add_insecure_port('[::]:50051')
    server.start()
    server.wait_for_termination()

if __name__ == '__main__':
    start_server()

4. TypeScript客户端实现 #

import * as grpc from '@grpc/grpc-js';
import * as protoLoader from '@grpc/proto-loader';

const packageDef = protoLoader.loadSync('xml_service.proto', {
    keepCase: true,
    longs: String,
    enums: String,
    defaults: true,
    oneofs: true
});
const xmlService = grpc.loadPackageDefinition(packageDef).XmlProcessor;

const client = new xmlService('localhost:50051', grpc.credentials.createInsecure());

function callXmlUpdate(xmlContent: string, updateRules: Record<string, string>) {
    return new Promise<string>((resolve, reject) => {
        client.UpdateXml({ xml_content: xmlContent, update_rules: updateRules }, (err, res) => {
            if (err) reject(err);
            else resolve(res.updated_xml);
        });
    });
}

// 调用示例
const rawXml = '<root><status>inactive</status></root>';
const updateRules = { './status': 'active' };
callXmlUpdate(rawXml, updateRules).then(updated => console.log(updated)).catch(err => console.error(err));

内容的提问来源于stack exchange，提问作者vinayak