替代st.file_uploader：Streamlit本地/S3 CSV上传方案问询

一、本地文件上传替代方案 #

如果st.file_uploader无法使用，可采用以下两种替代方式：

1. 粘贴CSV内容解析 #

让用户直接将CSV内容粘贴到文本框，再解析为DataFrame：

import streamlit as st
import pandas as pd

st.subheader("粘贴CSV内容上传")
csv_content = st.text_area("粘贴你的CSV内容", height=200)
if csv_content:
    try:
        df = pd.read_csv(pd.compat.StringIO(csv_content))
        st.dataframe(df)
        # 后续可添加数据插入逻辑
    except Exception as e:
        st.error(f"解析CSV失败: {str(e)}")

2. 本地文件路径读取（仅本地运行场景） #

适合本地开发时，让用户输入CSV文件的绝对路径来读取：

import streamlit as st
import pandas as pd

st.subheader("通过本地文件路径上传")
file_path = st.text_input("输入本地CSV文件绝对路径")
if file_path:
    try:
        df = pd.read_csv(file_path)
        st.dataframe(df)
        # 后续可添加数据插入逻辑
    except Exception as e:
        st.error(f"读取文件失败: {str(e)}")

二、从S3存储桶选择CSV文件 #

通过boto3连接S3，列出桶内CSV文件供用户选择，直接读取文件内容：

核心代码 #

import streamlit as st
import pandas as pd
import boto3
from io import StringIO

# 从Streamlit Secrets加载AWS配置（生产环境推荐）
aws_access_key = st.secrets["AWS_ACCESS_KEY_ID"]
aws_secret_key = st.secrets["AWS_SECRET_ACCESS_KEY"]
s3_bucket = st.secrets["S3_BUCKET_NAME"]

# 初始化S3客户端
s3 = boto3.client(
    's3',
    aws_access_key_id=aws_access_key,
    aws_secret_access_key=aws_secret_key
)

st.subheader("从S3存储桶选择CSV文件")
try:
    # 获取桶内所有CSV文件
    response = s3.list_objects_v2(Bucket=s3_bucket)
    csv_files = [obj['Key'] for obj in response.get('Contents', []) if obj['Key'].endswith('.csv')]
    
    if csv_files:
        selected_file = st.selectbox("选择要加载的CSV", csv_files)
        if st.button("加载文件"):
            # 读取S3文件内容
            s3_obj = s3.get_object(Bucket=s3_bucket, Key=selected_file)
            csv_str = s3_obj['Body'].read().decode('utf-8')
            df = pd.read_csv(StringIO(csv_str))
            st.dataframe(df)
            # 后续可添加数据插入逻辑
    else:
        st.info("当前S3桶内没有CSV文件")
except Exception as e:
    st.error(f"访问S3失败: {str(e)}")

三、数据插入数据表示例 #

以SQLite为例，将解析后的DataFrame插入数据表：

import sqlite3

def insert_to_database(df, table_name):
    conn = sqlite3.connect('app_database.db')
    df.to_sql(table_name, conn, if_exists='append', index=False)
    conn.close()
    st.success(f"数据已成功插入`{table_name}`表")

# 在加载完DataFrame后调用
if 'df' in locals():
    if st.button("插入到数据表"):
        insert_to_database(df, 'uploaded_csv_data')

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