我正在尝试使用 Streamlit 应用程序在 s3 上上传文件，但收到错误文件名必须是路径

我尝试打印路径并发现 Streamlit 暂时存储文件，但我无法获取路径 临时文件已存储

我无法获取文件的路径。我什至尝试打印它，但是 没有路径。我之前尝试通过指定文件路径来上传 本地机器并且代码运行良好

import streamlit as st
import requests
from dotenv import load_dotenv
import boto3
from botocore.exceptions import NoCredentialsError, PartialCredentialsError,ClientError
import io
import os
import time

def upload_to_s3(file_path,bucket_name,object_name=None):
    
    s3=boto3.client("s3")
    if object_name is None:
        object_name=file_path
    try:
     s3.upload_file(file_path,bucket_name,object_name)
    except ClientError as e:
        print("Client error occured {e}")


def main():
    bucket="shellkode1111"
    st.set_page_config(page_title="apptech",layout="wide")
    st.title("Apple")
    uploaded_file=st.file_uploader("Upload file",type=["mp4"])
    s3=boto3.client("s3")
    sts=boto3.client("sts")
    response = s3.head_bucket(Bucket="shellkode1111")
    try:
        s3.upload_file(uploaded_file,bucket,uploaded_file.name)
    except ClientError as e:
        print("Client Error occured {e}")
    print(response)
    bucket="shellkode1111"
    print("------------------------------")
    print(uploaded_file)
    print("----------------------------------")
    print("Working1")
    if(uploaded_file):
        print("Working")
        response=upload_to_s3(uploaded_file,bucket,uploaded_file)


if __name__== "__main__":
    main()

这个错误信息“文件名必须是路径”表明你在尝试将 UploadedFile 对象直接传递给 s3.upload_file 函数，而该函数期望的是一个字符串形式的文件路径。

要解决这个问题，你需要先将上传的文件保存到一个临时位置，然后将该临时文件的路径传递给 s3.upload_file 函数。以下是如何修改你的代码来实现这个功能：

import streamlit as st
import boto3
from botocore.exceptions import ClientError
import os
import tempfile

def upload_to_s3(file_object, bucket_name, object_name=None):
    """
    将文件上传到 S3

    Args:
        file_object (UploadedFile): 要上传的文件对象。
        bucket_name (str): S3 存储桶名称。
        object_name (str, optional): 对象在 S3 中的名称。 
                                     如果未提供，将使用上传文件的原始名称。

    Returns:
        bool: 上传成功返回 True，否则返回 False。
    """

    s3 = boto3.client("s3")
    if object_name is None:
        object_name = file_object.name

    # 使用 tempfile 创建一个安全的临时文件
    with tempfile.NamedTemporaryFile(delete=False) as temp_file:
        temp_file.write(file_object.read())
        temp_file_path = temp_file.name

        try:
            s3.upload_file(temp_file_path, bucket_name, object_name)
            return True
        except ClientError as e:
            print(f"Client Error occured: {e}")
            return False
        finally:
            #  确保删除临时文件
            os.remove(temp_file_path)


def main():
    bucket = "shellkode1111"
    st.set_page_config(page_title="apptech", layout="wide")
    st.title("Apple")
    uploaded_file = st.file_uploader("Upload file", type=["mp4"])

    if uploaded_file:
        if upload_to_s3(uploaded_file, bucket):
            st.success(f"File '{uploaded_file.name}' uploaded successfully to S3 bucket '{bucket}'.")
        else:
            st.error(f"Error uploading file '{uploaded_file.name}' to S3.")


if __name__ == "__main__":
    main()

这个代码做了以下几件事：

1. 使用 tempfile.NamedTemporaryFile 创建临时文件 : 这个函数会在安全的位置创建一个临时文件，并将上传文件的内容写入其中。它还提供了临时文件的路径 ( temp_file_path )，以便你可以将其传递给 s3.upload_file 。
2. 将临时文件路径传递给 s3.upload_file : 在 upload_to_s3 函数中，我们现在将 temp_file_path 传递给 s3.upload_file 来上传文件。
3. 删除临时文件 : 在 finally 块中，我们使用 os.remove(temp_file_path) 删除临时文件，以确保不会留下任何残留文件。
4. 改进错误处理 : 代码现在会捕获 ClientError 异常，并打印更详细的错误信息，帮助你调试上传过程中可能出现的任何问题。

通过这些更改，你的代码应该能够成功将文件上传到 S3。