Streamlit应用如何等待外部Python程序生成的CSV文件？

针对你遇到的问题，这里提供几种实用的解决思路，从简单易实现到资源友好型都有：

1. 轮询检测文件（最直接方案） #

核心逻辑是定期检查目标CSV文件是否存在且写入完成，同时用Streamlit的动态组件更新状态，避免页面卡顿。

代码示例（带状态提示） #

import streamlit as st
import pandas as pd
import os
import time

def wait_for_csv(file_path, check_interval=1):
    # 初始化占位符用于动态更新状态
    status_placeholder = st.empty()
    
    # 处理文件已存在的情况
    if os.path.exists(file_path):
        status_placeholder.success("CSV文件已存在，直接读取")
        return
    
    start_time = time.time()
    while True:
        # 检查文件是否存在且写入完成（通过文件大小稳定判断）
        if os.path.exists(file_path):
            size1 = os.path.getsize(file_path)
            time.sleep(0.5)
            size2 = os.path.getsize(file_path)
            if size1 == size2:
                break
        
        elapsed = int(time.time() - start_time)
        status_placeholder.info(f"等待CSV生成中... 已耗时 {elapsed} 秒")
        time.sleep(check_interval)
    
    status_placeholder.success("CSV文件生成完成！")

# 实际调用
target_csv = "external_output.csv"
wait_for_csv(target_csv)

# 读取并可视化
df = pd.read_csv(target_csv)
st.line_chart(df.set_index(df.columns[0]))

扩展：添加进度条 #

如果能预估文件生成的大致时长（比如最多100秒），可以把状态提示换成进度条：

def wait_for_csv_with_progress(file_path, max_wait=100):
    status_placeholder = st.empty()
    progress_bar = st.progress(0)
    
    if os.path.exists(file_path):
        status_placeholder.success("CSV已存在")
        progress_bar.progress(100)
        return
    
    start_time = time.time()
    while time.time() - start_time < max_wait:
        if os.path.exists(file_path):
            size1 = os.path.getsize(file_path)
            time.sleep(0.5)
            if size1 == os.path.getsize(file_path):
                progress_bar.progress(100)
                status_placeholder.success("CSV生成完成！")
                return
        
        progress = min(int((time.time() - start_time)/max_wait * 100), 99)
        progress_bar.progress(progress)
        status_placeholder.info(f"进度：{progress}%")
        time.sleep(1)
    
    status_placeholder.error("等待超时，请检查外部程序状态")

2. 文件系统事件监听（资源友好方案） #

如果不想用轮询浪费CPU，可以用watchdog库监听目标目录的文件创建事件，一旦CSV生成就触发后续逻辑。

代码示例 #

import streamlit as st
import pandas as pd
import os
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class CSVMonitor(FileSystemEventHandler):
    def __init__(self, target_file, on_ready):
        self.target_file = os.path.abspath(target_file)
        self.on_ready = on_ready
        self.ready = False

    def on_created(self, event):
        # 排除目录，只处理目标文件
        if event.is_directory:
            return
        if os.path.abspath(event.src_path) == self.target_file:
            # 等待文件写入完成
            while True:
                size1 = os.path.getsize(self.target_file)
                time.sleep(0.5)
                size2 = os.path.getsize(self.target_file)
                if size1 == size2:
                    self.ready = True
                    self.on_ready()
                    break

def wait_for_csv_via_watch(file_path):
    status_placeholder = st.empty()
    status_placeholder.info("正在监听CSV文件生成...")

    def on_csv_ready():
        status_placeholder.success("CSV生成完成！")

    event_handler = CSVMonitor(file_path, on_csv_ready)
    observer = Observer()
    # 监听CSV所在目录
    observer.schedule(event_handler, path=os.path.dirname(file_path), recursive=False)
    observer.start()

    # 等待事件触发
    while not event_handler.ready:
        time.sleep(0.5)
    
    observer.stop()
    observer.join()

# 调用示例
target_csv = "external_output.csv"
wait_for_csv_via_watch(target_csv)

df = pd.read_csv(target_csv)
st.bar_chart(df)

注意：需要先安装依赖：pip install watchdog

3. 外部程序主动通知（最优解） #

如果能修改生成CSV的外部程序，让它在完成后发送一个"完成信号"，比如：

• 创建一个标记文件（如csv_done.flag）
• 通过本地HTTP请求通知Streamlit
• 用本地消息队列传递状态

这种方式最可靠，比如外部程序最后添加一行：

# 外部程序的收尾代码
open("csv_done.flag", "w").write("completed")

然后Streamlit的等待逻辑改成检测这个标记文件，读取完CSV后记得删除标记，避免下次误判。

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