Win10环境下Flask服务如何检测Python脚本运行状态避免重复调用

推荐使用锁文件方案实现运行状态检测，该方案不受Flask多worker部署、服务重启等场景影响，稳定性更高，修改后的完整代码如下：

from flask import Flask
from flask import request
import os
import json
import datetime
import shlex # 用于安全转义命令参数

app = Flask(__name__)
# 定义锁文件路径，只要文件存在就代表检测脚本正在运行
LOCK_FILE_PATH = "./detect_running.lock"
#url test-> http://192.168.1.186:8081/detectForCar?source=D:/test.mp4

@app.route('/detectForCar', methods=['GET','POST'])
def detectForCar():
    source = request.args.get("source")
    ip = request.remote_addr 
    # 先判断锁文件是否存在，存在则直接返回正在运行提示
    if os.path.exists(LOCK_FILE_PATH):
        str_response = {'message': 'detect task is already running, please try again later'}
        return json.dumps(str_response, ensure_ascii=False), 429
    
    if source:
        if os.path.exists(source):
            try:
                # 创建锁文件
                with open(LOCK_FILE_PATH, 'w') as f:
                    f.write(f"running, start time: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
                
                # 用shlex.quote转义参数，避免命令注入风险和特殊路径字符导致的执行失败
                cmd = f'python detect_for_all.py --source {shlex.quote(source)}'
                os.system(cmd)
                
                str_response = {'message': 'success'}
                json_response = json.dumps(str_response,ensure_ascii=False)
                return json_response
            except Exception as ex:
                str_response = {'message': f'error: {str(ex)}'}
                json_response = json.dumps(str_response,ensure_ascii=False)
                return json_response
            finally:
                # 无论脚本执行成功还是报错，都删除锁文件，避免死锁
                if os.path.exists(LOCK_FILE_PATH):
                    os.remove(LOCK_FILE_PATH)
    # 参数缺失的返回补充
    str_response = {'message': 'invalid source parameter'}
    return json.dumps(str_response, ensure_ascii=False), 400
            
if __name__ == '__main__':
    # 服务启动时先删除残留的锁文件，避免服务异常退出导致锁文件未删除
    if os.path.exists(LOCK_FILE_PATH):
        os.remove(LOCK_FILE_PATH)
    app.run(host='192.168.1.186' ,port=8081, debug=False)

核心修改点说明 #

• 新增锁文件作为运行状态标记，请求进入时先判断锁文件是否存在，存在直接返回运行中提示
• 新增finally逻辑，无论脚本执行成功还是报错，都会删除锁文件，避免永久死锁
• 服务启动时自动清理残留锁文件，避免服务异常崩溃后锁文件未删除导致的无法执行问题
• 新增shlex.quote转义参数，避免命令注入漏洞

可选优化（异步返回） #

如果不需要等待脚本执行完成再返回结果，可以将os.system(cmd)替换为subprocess.Popen实现异步执行，对应修改代码段即可：

import subprocess
# 替换os.system(cmd)为以下代码
subprocess.Popen(cmd, shell=True)

注意异步场景下需要把删除锁文件的逻辑移到detect_for_all.py的末尾，无论脚本执行成功还是失败都删除锁文件，不要在Flask接口的finally代码块中删除。

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