Flask执行Python函数并在/logging页实时展示终端输出实现方案

你之前/logging流式输出不生效，核心原因基本是这几个：响应被web服务器缓冲、没用标准流传输协议、ansible/terraform这类子进程输出没逐行捕获、业务逻辑阻塞了响应生成器。下面是可直接落地的完整实现：

1. 先统一项目结构 #

提前创建好对应目录，避免路径报错：

your_project/
├── app.py              # Flask主应用
├── core_logic.py       # 你已经写好的、包含infra_create的核心业务模块
├── templates/
│   ├── index.html      # 参数输入表单页
│   └── log_view.html   # 实时日志展示页
├── static/
│   └── style.css       # 样式文件
└── logs/               # 持久化日志存储目录，手动提前创建

2. 核心逻辑改造：统一捕获所有输出 #

不管是Python原生print内容，还是ansible、terraform这类子进程的终端输出，都通过统一方法同时推流+写日志文件，避免多用户访问时日志串流。

2.1 先在app.py里初始化基础配置 #

from flask import Flask, render_template, request, Response
import queue
import threading
import time
import uuid
import os
import subprocess
from datetime import datetime
# 导入你已经写好的核心业务函数
from core_logic import infra_create

app = Flask(__name__)
# 每个任务分配独立队列存日志，避免多请求串内容
task_queues = {}
LOG_DIR = os.path.join(os.path.dirname(__file__), 'logs')
os.makedirs(LOG_DIR, exist_ok=True)

# 统一日志推送方法：同时写队列+写本地日志文件
def log_push(queue_obj, content, log_file_path=None):
    log_line = f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {content.strip()}\n"
    queue_obj.put(log_line)
    if log_file_path:
        with open(log_file_path, 'a', encoding='utf-8') as f:
            f.write(log_line)

2.2 改造infra_create的输出逻辑 #

注意：之前如果用subprocess.run()/subprocess.check_output()调用ansible、terraform，会等命令全量执行完才返回所有输出，根本没法流式传输，必须改成Popen逐行读输出的模式。

# 改造后的infra_create示例，按你自己的业务逻辑调整即可
def infra_create(params, log_queue, log_file_path):
    # 原有目录创建逻辑
    os.makedirs(params['work_dir'], exist_ok=True)
    log_push(log_queue, "工作目录创建完成", log_file_path)

    # ansible-playbook调用改造：逐行读输出
    ansible_cmd = ["ansible-playbook", "your_playbook.yml", "-e", f"env={params['env']}"]
    proc = subprocess.Popen(
        ansible_cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        cwd=params['work_dir']
    )
    for line in iter(proc.stdout.readline, ''):
        if line:
            log_push(log_queue, line, log_file_path)
    proc.wait()
    if proc.returncode != 0:
        log_push(log_queue, f"ansible执行失败，返回码：{proc.returncode}", log_file_path)
        return False

    # terraform调用改造：和上面逻辑一致
    tf_init_cmd = ["terraform", "init"]
    proc = subprocess.Popen(
        tf_init_cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        cwd=params['work_dir']
    )
    for line in iter(proc.stdout.readline, ''):
        if line:
            log_push(log_queue, line, log_file_path)
    proc.wait()
    # 后续terraform plan、apply逻辑同上逐行捕获即可

    log_push(log_queue, "基础设施创建流程执行完成", log_file_path)
    return True

如果函数里有原生print输出，直接替换成log_push调用即可，不要用全局重定向sys.stdout的方案，多用户访问时必然串日志。

3. Flask路由实现 #

# 表单首页路由
@app.route('/', methods=['GET'])
def index():
    return render_template('index.html')

# 表单提交路由：启动后台线程跑任务，不阻塞请求
@app.route('/submit', methods=['POST'])
def submit_form():
    # 按你自己的表单字段收集参数即可
    params = {
        "env": request.form.get("env"),
        "cluster_name": request.form.get("cluster_name"),
        "work_dir": request.form.get("work_dir", f"/tmp/infra_{int(time.time())}")
    }
    # 生成唯一任务ID，做日志隔离
    task_id = str(uuid.uuid4())
    task_queues[task_id] = queue.Queue()
    log_file_path = os.path.join(LOG_DIR, f"task_{task_id}.log")

    # 后台线程执行业务逻辑
    def run_task():
        try:
            infra_create(params, task_queues[task_id], log_file_path)
        except Exception as e:
            log_push(task_queues[task_id], f"任务执行异常：{str(e)}", log_file_path)
        finally:
            # 任务结束放终止标记
            task_queues[task_id].put(None)
    threading.Thread(target=run_task, daemon=True).start()

    # 跳转到对应任务的日志页
    return f'<script>window.location.href="/logging?task_id={task_id}"</script>'

# 日志页+流式输出路由
@app.route('/logging')
def logging_page():
    task_id = request.args.get("task_id")
    if not task_id or task_id not in task_queues:
        return "无效任务ID", 404

    # SSE协议流式响应，这是之前最容易写错的部分
    if request.headers.get('Accept') == 'text/event-stream':
        def generate_stream():
            q = task_queues[task_id]
            while True:
                line = q.get()
                if line is None:
                    yield "data: [流程执行结束]\n\n"
                    break
                # SSE格式必须严格遵守data:内容\n\n，少一个换行前端都解析失败
                yield f"data: {line.replace('\n', '<br>')}\n\n"
        return Response(
            generate_stream(),
            mimetype='text/event-stream',
            headers={
                'Cache-Control': 'no-cache',
                'X-Accel-Buffering': 'no', # 自动关nginx反向代理缓冲，不用额外改nginx配置
                'Connection': 'keep-alive'
            }
        )
    # 普通访问返回静态日志页
    return render_template('log_view.html', task_id=task_id)

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

4. 前端页面代码 #

4.1 表单页 templates/index.html #

<!DOCTYPE html>
<html>
<head>
    <title>基础设施创建参数提交</title>
    <link rel="stylesheet" href="/static/style.css">
</head>
<body>
    <div class="form-container">
        <h2>提交创建参数</h2>
        <form method="post" action="/submit">
            <div class="form-item">
                <label>部署环境</label>
                <input type="text" name="env" required placeholder="例如prod/test">
            </div>
            <div class="form-item">
                <label>集群名称</label>
                <input type="text" name="cluster_name" required>
            </div>
            <div class="form-item">
                <label>工作目录（可选）</label>
                <input type="text" name="work_dir" placeholder="留空自动生成">
            </div>
            <button type="submit">提交执行</button>
        </form>
    </div>
</body>
</html>

4.2 日志展示页 templates/log_view.html #

<!DOCTYPE html>
<html>
<head>
    <title>执行日志</title>
    <link rel="stylesheet" href="/static/style.css">
</head>
<body>
    <div class="log-container">
        <h2>实时执行日志</h2>
        <div id="log-output" class="log-box"></div>
    </div>
    <script>
        const taskId = "{{ task_id }}";
        const logBox = document.getElementById('log-output');
        // 原生JS接SSE流，不需要引第三方库
        const eventSource = new EventSource(`/logging?task_id=${taskId}`);
        eventSource.onmessage = function(event) {
            const line = document.createElement('div');
            line.innerHTML = event.data;
            logBox.appendChild(line);
            // 自动滚动到日志底部
            logBox.scrollTop = logBox.scrollHeight;
            if (event.data.includes('[流程执行结束]')) {
                eventSource.close();
            }
        };
        eventSource.onerror = function() {
            const line = document.createElement('div');
            line.innerText = '[连接中断，可刷新重试，或查看logs目录下的本地日志文件]';
            logBox.appendChild(line);
            eventSource.close();
        };
    </script>
</body>
</html>

4.3 样式文件 static/style.css #

.form-container, .log-container {
    width: 80%;
    max-width: 1000px;
    margin: 30px auto;
}
.form-item {
    margin: 15px 0;
}
.form-item label {
    display: block;
    margin-bottom: 5px;
}
.form-item input {
    width: 100%;
    padding: 8px;
    border: 1px solid #eee;
    border-radius: 4px;
}
button {
    padding: 8px 20px;
    background: #409eff;
    color: #fff;
    border: none;
    border-radius: 4px;
    cursor: pointer;
}
.log-box {
    background: #1e1e1e;
    color: #00ff00;
    padding: 20px;
    border-radius: 4px;
    height: 600px;
    overflow-y: auto;
    font-family: monospace;
    line-height: 1.5;
}

部署注意事项 #

• 如果用gunicorn部署，启动参数要加--worker-class gthread --timeout 0，用线程模式、关闭长连接超时切断
• 所有日志按任务ID存在项目的logs目录下，可直接下载归档
• 不要开Flask的debug多进程模式（use_reloader=True），会导致队列跨进程不共享，日志收不到

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