基于Flask的Python工具调用Ruby API处理CSV的技术问询

嘿，这个需求其实挺常见的，你猜想的传递文件名作为命令行参数完全可行，而且还有几个其他方案可以根据你的场景灵活选择。我来给你详细拆解每个方法的实现：

这个思路就是Flask先接收用户上传的CSV，保存到服务器本地，然后调用Ruby脚本时把保存后的文件路径作为参数传过去，Ruby脚本读取该文件处理后，把结果输出到标准输出（STDOUT），Flask捕获这个输出再返回给用户。

Flask端代码示例 #

from flask import Flask, request, jsonify
import subprocess
import os
from werkzeug.utils import secure_filename

app = Flask(__name__)
# 配置上传文件保存目录，确保目录存在
UPLOAD_FOLDER = './uploads'
os.makedirs(UPLOAD_FOLDER, exist_ok=True)
app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER

@app.route('/upload-csv', methods=['POST'])
def upload_csv():
    if 'csv_file' not in request.files:
        return jsonify({'error': '未上传文件'}), 400
    file = request.files['csv_file']
    if file.filename == '':
        return jsonify({'error': '文件名不能为空'}), 400
    if file and file.filename.endswith('.csv'):
        # 用secure_filename避免路径注入风险
        filename = secure_filename(file.filename)
        file_path = os.path.join(app.config['UPLOAD_FOLDER'], filename)
        file.save(file_path)
        
        try:
            # 调用Ruby脚本，传递文件路径作为参数
            result = subprocess.run(
                ['ruby', 'your_ruby_script.rb', file_path],
                capture_output=True,
                text=True,
                check=True  # 如果Ruby脚本返回非0状态码，会抛出异常
            )
            # 返回Ruby脚本的输出结果
            return jsonify({
                'message': '处理完成',
                'result': result.stdout.strip()
            })
        except subprocess.CalledProcessError as e:
            # 捕获Ruby脚本执行错误
            return jsonify({
                'error': '脚本执行失败',
                'details': e.stderr.strip()
            }), 500
        finally:
            # 可选：处理完后删除上传的文件，避免磁盘占用
            if os.path.exists(file_path):
                os.remove(file_path)
    else:
        return jsonify({'error': '仅支持CSV文件'}), 400

if __name__ == '__main__':
    app.run(debug=True)

Ruby端代码示例（your_ruby_script.rb） #

# 检查是否传入了文件路径参数
if ARGV.empty?
  puts "错误：请传入CSV文件路径"
  exit 1
end

file_path = ARGV[0]
# 检查文件是否存在
unless File.exist?(file_path)
  puts "错误：文件不存在 #{file_path}"
  exit 1
end

# 读取并处理CSV文件
require 'csv'
processed_data = []
CSV.foreach(file_path, headers: true) do |row|
  # 这里写你的处理逻辑，比如修改字段、计算统计值等
  processed_row = row.to_h.transform_values { |v| v.upcase } # 示例：所有值转大写
  processed_data << processed_row
end

# 将处理结果输出到标准输出，Flask会捕获这个内容
puts "处理完成，共处理#{processed_data.length}行数据"
puts JSON.generate(processed_data) # 如果需要返回结构化数据，用JSON格式更方便

注意事项 #

• 一定要用secure_filename处理文件名，防止路径注入攻击；
• 确保Ruby脚本有读取上传目录文件的权限；
• 大文件上传时要考虑磁盘空间限制，或者设置上传文件大小限制；
• subprocess.run的check=True会在脚本返回非0时抛出异常，方便捕获错误。

如果不想把CSV保存到磁盘，可以直接把上传的文件内容通过标准输入传给Ruby脚本，适合小体积的CSV文件，减少磁盘IO开销。

Flask端代码修改部分 #

# 替换方案1中保存文件和调用脚本的部分
file_content = file.read().decode('utf-8') # 读取文件内容
try:
    result = subprocess.run(
        ['ruby', 'your_ruby_script.rb'],
        input=file_content,
        capture_output=True,
        text=True,
        check=True
    )
    # 后续逻辑和方案1一致

Ruby端代码修改部分 #

# 直接从STDIN读取CSV内容
require 'csv'
require 'json'

processed_data = []
CSV.parse(STDIN.read, headers: true) do |row|
  # 处理逻辑和之前一样
  processed_row = row.to_h.transform_values { |v| v.upcase }
  processed_data << processed_row
end

puts "处理完成，共处理#{processed_data.length}行数据"
puts JSON.generate(processed_data)

如果你的Ruby脚本需要被多个服务调用，或者处理逻辑比较复杂，可以把它封装成一个独立的HTTP API（比如用Sinatra框架快速搭建），然后Flask作为客户端发送请求给这个API，这种方式耦合度更低，扩展性更好。

Ruby API端代码（用Sinatra） #

require 'sinatra'
require 'csv'
require 'json'

post '/process-csv' do
  content_type :json
  # 获取上传的CSV文件
  csv_file = params['csv_file']
  unless csv_file
    return { error: '未上传CSV文件' }.to_json, 400
  end

  begin
    processed_data = []
    CSV.parse(csv_file.read, headers: true) do |row|
      processed_row = row.to_h.transform_values { |v| v.upcase }
      processed_data << processed_row
    end
    {
      message: '处理完成',
      row_count: processed_data.length,
      data: processed_data
    }.to_json
  rescue => e
    { error: "处理失败：#{e.message}" }.to_json, 500
  end
end

# 启动服务：ruby ruby_api.rb -p 4567

Flask端调用代码 #

import requests

@app.route('/upload-csv', methods=['POST'])
def upload_csv():
    if 'csv_file' not in request.files:
        return jsonify({'error': '未上传文件'}), 400
    file = request.files['csv_file']
    if file.filename.endswith('.csv'):
        try:
            # 发送请求给Ruby API
            response = requests.post(
                'http://localhost:4567/process-csv',
                files={'csv_file': file}
            )
            response.raise_for_status()
            return jsonify(response.json())
        except requests.exceptions.RequestException as e:
            return jsonify({'error': '调用Ruby API失败', 'details': str(e)}), 500
    else:
        return jsonify({'error': '仅支持CSV文件'}), 400

总结一下：

• 快速实现选方案1，逻辑直观，适合小项目；
• 追求无磁盘IO选方案2，适合小文件；
• 分布式或多服务场景选方案3，耦合度低，扩展性强。

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