快捷導(dǎo)航

Python?Flask中文件與異常處理的實(shí)踐指南

更新時(shí)間：2025年05月20日 08:24:00 作者：碼農(nóng)阿豪@新空間

在開發(fā)Web應(yīng)用時(shí),文件處理和異常處理是常見的需求,本文將通過一個(gè)實(shí)際案例,分析如何優(yōu)化Python?Flask應(yīng)用中的文件處理邏輯,感興趣的小伙伴可以了解下

引言

在開發(fā)Web應(yīng)用時(shí)，文件處理和異常處理是常見的需求。特別是在使用Flask框架進(jìn)行文件上傳、處理和下載時(shí)，正確處理返回值類型和異常情況至關(guān)重要。本文將通過一個(gè)實(shí)際案例，分析如何優(yōu)化Python Flask應(yīng)用中的文件處理邏輯，并解決常見的異常問題。

問題背景

在一個(gè)基于Flask的文件處理工具中，用戶上傳Excel文件，系統(tǒng)處理后返回結(jié)果文件。但在實(shí)際運(yùn)行中，出現(xiàn)了以下錯(cuò)誤：

AttributeError: 'list' object has no attribute 'read'

該錯(cuò)誤表明，F(xiàn)lask的 send_file 方法期望接收一個(gè)文件路徑（字符串），但實(shí)際傳入的是一個(gè)列表（list），導(dǎo)致無法讀取文件內(nèi)容。

此外，還出現(xiàn)了以下日志：

2025-05-20 01:05:37,500 - ERROR - process_and_export_results 返回了無效類型

這說明后端處理函數(shù)的返回值類型不符合預(yù)期，導(dǎo)致后續(xù)操作失敗。

問題分析

1. 錯(cuò)誤原因

process_single_thread 函數(shù)返回的是 [output_file]（包含單個(gè)文件路徑的列表），但 send_file 需要的是 output_file（字符串）。

調(diào)用方未對返回值進(jìn)行類型檢查，直接傳給 send_file，導(dǎo)致 AttributeError。

2. 深層原因

接口設(shè)計(jì)不一致：處理函數(shù)返回列表，但調(diào)用方期望字符串。

異常處理不足：未對返回值做校驗(yàn)，導(dǎo)致錯(cuò)誤傳播到Flask中間件。

日志信息不完整：錯(cuò)誤日志未能清晰指出問題所在。

解決方案

1. 優(yōu)化 process_single_thread 返回值

原代碼：

def process_single_thread(raw_results, cookie, timestamp, base_filename, secretKey, receiver_email):
    # ...處理邏輯...
    return [output_file]  # 返回列表

優(yōu)化后：

def process_single_thread(raw_results, cookie, timestamp, base_filename, secretKey, receiver_email):
    """單線程處理數(shù)據(jù)

    Args:
        raw_results: 待處理的原始數(shù)據(jù)列表
        cookie: 用于處理的cookie
        timestamp: 時(shí)間戳，用于生成文件名
        base_filename: 基礎(chǔ)文件名
        receiver_email: 接收結(jié)果的郵箱地址

    Returns:
        str: 輸出文件路徑（直接返回字符串，而非列表）
    """
    # ...處理邏輯...
    return output_file  # 直接返回字符串

優(yōu)化點(diǎn)：

修改返回值為字符串，符合 send_file 的預(yù)期。
更新函數(shù)文檔，明確返回值類型。

2. 調(diào)用方增強(qiáng)校驗(yàn)

在 app.py 中，增加對返回值的檢查：

try:
    output_file = process_and_export_results(filepath, cookie, nationwide, receiver_email)
    
    # 檢查返回值是否為有效路徑
    if not isinstance(output_file, str):
        logger.error(f"無效返回值類型: {type(output_file)}")
        return "處理錯(cuò)誤：內(nèi)部服務(wù)異常", 500
    
    if not os.path.exists(output_file):
        logger.error(f"文件不存在: {output_file}")
        return "處理錯(cuò)誤：結(jié)果文件未生成", 500
    
    return send_file(output_file, as_attachment=True, download_name='result.xlsx')

except Exception as e:
    logger.error(f"文件處理異常: {str(e)}", exc_info=True)
    return f"處理錯(cuò)誤：{str(e)}", 500

優(yōu)化點(diǎn)：

檢查返回值是否為字符串。
確保文件存在，避免 FileNotFoundError。
捕獲并記錄異常，返回友好的錯(cuò)誤信息。

3. 日志優(yōu)化

在關(guān)鍵步驟添加詳細(xì)日志，便于排查問題：

logger.info(f"開始處理文件: {filepath}")
logger.info(f"全國匹配模式: {'開啟' if nationwide else '關(guān)閉'}")
logger.info(f"接收郵箱: {receiver_email}")

output_file = process_and_export_results(filepath, cookie, nationwide, receiver_email)
logger.info(f"處理完成，輸出文件: {output_file}")

完整優(yōu)化后的代碼

deal_excel_file.py（優(yōu)化后）

import os
import logging
from datetime import datetime

logger = logging.getLogger(__name__)

def process_single_thread(raw_results, cookie, timestamp, base_filename, secretKey, receiver_email):
    """單線程處理數(shù)據(jù)，返回文件路徑（字符串）"""
    final_results = []
    total_count = len(raw_results)
    success_count = 0

    for idx, item in enumerate(raw_results, 1):
        record = process_single_item(item, idx, cookie, secretKey, False)
        final_results.append(record)
        if record["匹配狀態(tài)"] == "成功":
            success_count += 1

    success_rate = (success_count / total_count) * 100 if total_count > 0 else 0
    output_file = f"result_{timestamp}_{base_filename}.xlsx"

    logger.info(
        f"[{base_filename}] 處理完成 - 總數(shù): {total_count}, "
        f"成功: {success_count}, 失敗: {total_count - success_count}, "
        f"成功率: {success_rate:.2f}%"
    )

    export_to_excel(final_results, output_file)

    if receiver_email:
        try:
            send_email_with_attachment(output_file, receiver_email)
            logger.info(f"[{base_filename}] 結(jié)果已發(fā)送至郵箱: {receiver_email}")
        except Exception as e:
            logger.error(f"[{base_filename}] 郵件發(fā)送失敗: {str(e)}")

    return output_file  # 直接返回字符串

app.py（優(yōu)化后）

from flask import Flask, request, send_file
import os
import logging

app = Flask(__name__)
app.config['UPLOAD_FOLDER'] = 'uploads'
os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True)

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[logging.FileHandler('app.log'), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

@app.route('/', methods=['POST'])
def upload_file():
    try:
        cookie = request.form.get('cookie', '').strip()
        nationwide = request.form.get('nationwide') == '1'
        receiver_email = request.form.get('email', '').strip()

        logger.info(f"開始處理請求，Cookie: {cookie[:10]}...")  # 避免日志泄露完整Cookie

        if not cookie:
            return "請?zhí)峁┯行У腃ookie", 400

        # 檢查文件上傳
        if 'file' not in request.files:
            return "未上傳文件", 400

        file = request.files['file']
        if not file.filename.endswith('.xlsx'):
            return "僅支持.xlsx文件", 400

        # 保存上傳文件
        timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
        filepath = os.path.join(app.config['UPLOAD_FOLDER'], f'upload_{timestamp}.xlsx')
        file.save(filepath)

        # 處理文件
        output_file = process_and_export_results(filepath, cookie, nationwide, receiver_email)

        # 校驗(yàn)返回值
        if not isinstance(output_file, str):
            logger.error(f"無效的返回值類型: {type(output_file)}")
            return "內(nèi)部服務(wù)錯(cuò)誤", 500

        if not os.path.exists(output_file):
            logger.error(f"文件不存在: {output_file}")
            return "結(jié)果文件生成失敗", 500

        return send_file(output_file, as_attachment=True, download_name='result.xlsx')

    except Exception as e:
        logger.error(f"處理請求時(shí)出錯(cuò): {str(e)}", exc_info=True)
        return f"服務(wù)器錯(cuò)誤: {str(e)}", 500

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)