Flask CSV上传应用文件路径问题：提交后提示文件未找到求助

开发了一个CSV文件上传站点，后台用Flask调用6sense的v1/enrichment/people接口，以上传文件数据作为输入。但无法正确配置文件路径，代码中with open(**HELP HERE**, "r") as file:的位置不知道填什么路径，点击提交按钮后出现“文件未找到”错误。

附完整Flask代码（授权令牌已省略）：

from flask import Flask, render_template, request, send_file
import requests
import csv
import os

app = Flask(__name__)

def call_api_and_generate_files(csv_path):
    url = "https://api.6sense.com/v1/enrichment/people"

    with open(**HELP HERE**, "r") as file:
        data = list(csv.reader(file, delimiter=","))
        data.pop(0)
        output = []


        for i in range(len(data)):
            payload = str(data[i])
            payload = payload[2:]
            payload = payload[:-2]
            payload = 'email=' + payload
            headers = {
                'Content-Type': 'application/x-www-form-urlencoded',
                'Cache-Control': 'no-cache',
                'Authorization': 'Token '
            }

            response = requests.request("POST", url, headers=headers, data=payload)
            output.append(response.text)

        count_missing_emails = 0
        count_no_return = 0
        count_missing_domain = 0
        count_missing_phone = 0
        count_missing_title = 0
        count_missing_level = 0
        count_missing_function = 0
        count_error_calls = 0

        output_csv_path = 'output.csv'
        counts_csv_path = 'counts.csv'

        with open(output_csv_path, 'w', newline='', encoding='utf-8') as myfile:
            writeout = csv.writer(myfile, quoting=csv.QUOTE_MINIMAL, lineterminator='\n')
            writeout.writerow(["JSON Return", "Email", "Domain", "Phone", "Job Title", "Job Level", "Job Function"])
            for li in output:
                partition_email = li.partition('"email":"')[2]
                partition_email = partition_email.partition('"')[0]
                partition_domain = li.partition('"domain":"')[2]
                partition_domain = partition_domain.partition('"')[0]
                partition_phone = li.partition('"phone":"')[2]
                partition_phone = partition_phone.partition('"')[0]
                partition_title = li.partition('"title":"')[2]
                partition_title = partition_title.partition('"')[0]
                partition_level = li.partition('"level":"')[2]
                partition_level = partition_level.partition('"')[0]
                partition_function = li.partition('"function":"')[2]
                partition_function = partition_function.partition('"')[0]

                if li != "" and "\"error\"" not in li:
                    writeout.writerow([li, partition_email, partition_domain, partition_phone, partition_title, partition_level, partition_function])
                if "\"email\":\"\"" in li:
                    count_missing_emails += 1
                if li == "":
                    count_no_return += 1
                if "\"domain\":\"\"" in li:
                    count_missing_domain += 1
                if "\"phone\":\"\"" in li:
                    count_missing_phone += 1
                if "\"title\":\"\"" in li:
                    count_missing_title += 1
                if "\"level\":\"\"" in li:
                    count_missing_level += 1
                if "\"function\":\"\"" in li:
                    count_missing_function += 1
                if "\"error\"" in li:
                    count_error_calls += 1

        with open(counts_csv_path, 'w', newline='', encoding='utf-8') as myfile2:
            writeout2 = csv.writer(myfile2, quoting=csv.QUOTE_MINIMAL, lineterminator='\n')
            writeout2.writerow(["# of Records From File: {}".format(len(data))])
            writeout2.writerow(["# of Calls Made: {}".format(len(output))])
            writeout2.writerow(["# of Calls Returned With Firmographics: {}".format(len(output) - count_no_return - count_error_calls)])
            writeout2.writerow(["# of Calls Not Returned/Matched: {}".format(count_no_return)])
            writeout2.writerow(["# of Calls With Errors: {}".format(count_error_calls)])
            writeout2.writerow(["# of Records Returned With Email: {}".format(len(output) - count_no_return - count_missing_emails - count_error_calls)])
            writeout2.writerow(["# of Records Returned With Domain: {}".format(len(output) - count_no_return - count_missing_domain - count_error_calls)])
            writeout2.writerow(["# of Records Returned With Phone: {}".format(len(output) - count_no_return - count_missing_phone - count_error_calls)])
            writeout2.writerow(["# of Records Returned With Title: {}".format(len(output) - count_no_return - count_missing_title - count_error_calls)])
            writeout2.writerow(["# of Records Returned With Level: {}".format(len(output) - count_no_return - count_missing_level - count_error_calls)])
            writeout2.writerow(["# of Records Returned With Function: {}".format(len(output) - count_no_return - count_missing_function - count_error_calls)])

    return counts_csv_path

@app.route('/')
def index():
    return render_template('index.html')

@app.route('/upload', methods=['POST'])
def upload():
    if 'csv-file' not in request.files:
        return "No file part"

    csv_file = request.files['csv-file']

    if csv_file.filename == '':
        return "No selected file"

    # Save the uploaded CSV file
    uploaded_csv_path = os.path.join('/Users/patrickcunning/PycharmProjects/frontendTEST', 'uploaded.csv')
    csv_file.save(uploaded_csv_path)


    # Call API and generate files
    counts_csv_path = call_api_and_generate_files(uploaded_csv_path)

    # Read the counts.csv file and pass its contents to the template
    with open(counts_csv_path, 'r') as counts_file:
        counts_data = counts_file.read()

    return render_template('index.html', counts_data=counts_data)

@app.route('/download_output')
def download_output():
    # Logic to send the output.csv file to the user
    return send_file('output.csv', as_attachment=True)

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

HTML页面代码：

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>CSV Upload and API Call</title>
    <link rel="stylesheet" href="{{ url_for('static', filename='styles.css') }}">
</head>
<body>
    <h1>Upload CSV and Perform API Call</h1>
    <form id="upload-form" action="/upload" method="POST" enctype="multipart/form-data">
        <input type="file" name="csv-file" accept=".csv">
        <button type="submit">Upload and Call API</button>
    </form>

    <div id="results" class="hidden">
        <h2>Results from counts.csv:</h2>
        <pre id="counts-results"></pre>
        <a href="{{ url_for('download_output') }}" download>Download output.csv</a>
    </div>

    <script>
        // JavaScript to show/hide results div after successful API call
        const resultsDiv = document.getElementById('results');
        const countsResults = document.getElementById('counts-results');

        function showResults(resultsText) {
            countsResults.textContent = resultsText;
            resultsDiv.classList.remove('hidden');
        }
    </script>
</body>
</html>

1. 修复函数内的文件路径参数 #

call_api_and_generate_files函数已经接收csv_path参数，直接将**HELP HERE**替换为csv_path即可，这样就能正确读取上传后保存的文件。

修改后该部分代码：

def call_api_and_generate_files(csv_path):
    url = "https://api.6sense.com/v1/enrichment/people"

    with open(csv_path, "r") as file:
        data = list(csv.reader(file, delimiter=","))
        # 后续代码不变

2. 替换硬编码绝对路径，使用动态路径 #

当前代码中uploaded_csv_path用了固定绝对路径，换环境后会失效。改用Flask提供的app.root_path获取项目根目录，生成动态路径，确保文件保存和读取的一致性。同时，生成的output.csv和counts.csv也建议用动态路径：

修改后的核心代码片段：

@app.route('/upload', methods=['POST'])
def upload():
    if 'csv-file' not in request.files:
        return "No file part"

    csv_file = request.files['csv-file']

    if csv_file.filename == '':
        return "No selected file"

    # 基于项目根目录生成上传文件路径
    uploaded_csv_path = os.path.join(app.root_path, 'uploaded.csv')
    csv_file.save(uploaded_csv_path)

    # Call API and generate files
    counts_csv_path = call_api_and_generate_files(uploaded_csv_path)

    # 读取统计文件
    with open(counts_csv_path, 'r') as counts_file:
        counts_data = counts_file.read()

    return render_template('index.html', counts_data=counts_data)

def call_api_and_generate_files(csv_path):
    url = "https://api.6sense.com/v1/enrichment/people"

    with open(csv_path, "r") as file:
        data = list(csv.reader(file, delimiter=","))
        data.pop(0)
        output = []

        # 接口调用逻辑不变

        # 动态生成输出文件路径
        output_csv_path = os.path.join(app.root_path, 'output.csv')
        counts_csv_path = os.path.join(app.root_path, 'counts.csv')

        # 后续文件写入逻辑不变

3. 修复前端结果展示逻辑 #

当前HTML中的JavaScript没有处理模板传递的counts_data，导致上传后结果不会自动显示。修改脚本部分，在页面加载时检查是否有counts_data并展示：

修改后的HTML脚本：

<script>
    // JavaScript to show/hide results div after successful API call
    const resultsDiv = document.getElementById('results');
    const countsResults = document.getElementById('counts-results');

    function showResults(resultsText) {
        countsResults.textContent = resultsText;
        resultsDiv.classList.remove('hidden');
    }

    // 页面加载时检查是否有结果数据
    {% if counts_data %}
        showResults("{{ counts_data | escapejs }}");
    {% endif %}
</script>

额外优化建议 #

• 为上传文件创建专门的uploads目录，避免根目录文件混乱：

  upload_dir = os.path.join(app.root_path, 'uploads')
  os.makedirs(upload_dir, exist_ok=True)
  uploaded_csv_path = os.path.join(upload_dir, 'uploaded.csv')
  

• 处理文件上传时的重名问题，比如用原文件名或者生成唯一文件名；
• 接口调用部分加入异常处理，避免单个请求失败导致整个流程中断。

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