如何将通过 FastAPI 上传的视频传递给 OpenCV VideoCapture？

Answer 1

您正在尝试传递文件 contents (bytes) 或 UploadFile 对象；但是，VideoCapture() 接受视频filename、捕获设备或 IP 视频流。

UploadFile 基本上是一个SpooledTemporaryFile（一个file-like 对象），其操作类似于TemporaryFile。但是，它在文件系统中没有可见的名称。正如您提到的，在处理完文件后您不会将文件保留在服务器上，您可以将文件内容复制到“在文件系统中具有可见名称，可用于打开文件”的 NamedTemporaryFile （使用name 属性），如here 和here 所述。根据documentation：

name 是否可以再次打开文件，而 命名的临时文件仍然打开，因平台而异（可以是 所以在 Unix 上使用；它不能在 Windows 上）。如果删除是true（ 默认），文件一关闭就被删除。

因此，在 Windows 上，您需要在实例化 NamedTemporaryFile 时将 delete 参数设置为 False，一旦完成，您可以使用 os.remove() 或 os.unlink() 方法手动删除它。

下面给出了如何做到这一点的两个选项。选项1使用def 端点实现解决方案，而选项 2使用async def 端点（利用aiofiles 库）。 def和async def的区别请看this answer。如果您希望用户上传大小不适合内存的相当大的文件，请查看this 和this 的答案，了解如何以块的形式读取上传的视频文件。

选项 1 - 使用 def 端点

from fastapi import FastAPI, File, UploadFile
from tempfile import NamedTemporaryFile
import os

@app.post("/video/detect-faces")
def detect_faces(file: UploadFile = File(...)):
    temp = NamedTemporaryFile(delete=False)
    try:
        try:
            contents = file.file.read()
            with temp as f:
                f.write(contents);
        except Exception:
            return {"message": "There was an error uploading the file"}
        finally:
            file.file.close()
        
        res = process_video(temp.name)  # Pass temp.name to VideoCapture()
    except Exception:
        return {"message": "There was an error processing the file"}
    finally:
        #temp.close()  # the `with` statement above takes care of closing the file
        os.remove(temp.name)
        
    return res

选项 2 - 使用 async def 端点

from fastapi import FastAPI, File, UploadFile
from tempfile import NamedTemporaryFile
from fastapi.concurrency import run_in_threadpool
import aiofiles
import asyncio
import os

@app.post("/video/detect-faces")
async def detect_faces(file: UploadFile = File(...)):
    try:
        async with aiofiles.tempfile.NamedTemporaryFile("wb", delete=False) as temp:
            try:
                contents = await file.read()
                await temp.write(contents)
            except Exception:
                return {"message": "There was an error uploading the file"}
            finally:
                await file.close()
        
        res = await run_in_threadpool(process_video, temp.name)  # Pass temp.name to VideoCapture()
    except Exception:
        return {"message": "There was an error processing the file"}
    finally:
        os.remove(temp.name)

    return res