File size: 2,462 Bytes
d38fe54 9f7f597 cd84989 b628481 7ca96aa 26fc5e2 d38fe54 26fc5e2 d38fe54 26fc5e2 d38fe54 26fc5e2 d38fe54 26fc5e2 9f7f597 26fc5e2 9f7f597 d38fe54 2c03cb6 f8096b7 9e986dc 06a0cd9 9e986dc 63aa27b f8096b7 9e986dc 2c03cb6 49567b1 2c03cb6 582ad23 49567b1 582ad23 2c03cb6 49567b1 b628481 2c03cb6 b628481 2c03cb6 b628481 7ca96aa 2c03cb6 7ca96aa 33e8fe3 7ca96aa 2c03cb6 9e986dc |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 |
from fastapi import FastAPI, File, UploadFile
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse
from PIL import Image
import numpy as np
import urllib.request
import io
import os
from moviepyTest import test
from typing import *
from fastapi.responses import PlainTextResponse #执行其他py的plaintext返回
import subprocess
from fastapi import BackgroundTasks
import time
app = FastAPI()
@app.get("/inference")
def inference():
return "<p>Hello, World!</p>"
@app.get("/infer_t5")
def t5(input):
return {"output": input}
@app.get("/moviepyTest")
def t5():
result = test()
return {"output": result}
##这个比较快不用异步
@app.post("/getOriginalMangaList")
async def getOriginalMangaList(images: List[UploadFile] = File(...)):
for idx, image in enumerate(images):
img = await image.read()
image = Image.open(io.BytesIO(img)).convert("L").convert("RGB")
path_to_image = f"/manga/{idx}.jpg"
image.save(path_to_image)
return "获取图片保存成功"
##这个比较快不用异步
##这个比较快不用异步
@app.delete("/deleteFiles")
async def delete_files(directory: str):
for filename in os.listdir(directory):
file_path = os.path.join(directory, filename)
if os.path.isfile(file_path):
os.remove(file_path)
return {"message": f"成功删除{directory}目录下的所有文件"}
##这个比较快不用异步
########异步处理py文件执行接口
def file_executer(file_name:str)->"执行返回":
try:
result = subprocess.check_output(["python", f"{file_name}.py"]).decode("utf-8")
return PlainTextResponse(result)
except subprocess.CalledProcessError as e:
return PlainTextResponse(f"Error executing {file_name}.py: {e}")
@app.get("/execute_py_file/{file_name}")
async def execute_py_file(file_name: str,background_tasks: BackgroundTasks):
result = "接受到了请求{filename}任务".format(filename = file_name)
background_tasks.add_task(file_executer,file_name)
return result
########异步处理py文件执行接口
##########异步样例
def someTask():
time.sleep(20)
print("睡眠20s结束")
@app.get("/backTaskTest")
def returnRandomSubscribeUrl(background_tasks: BackgroundTasks)->str:
#返回
result = "先返回"
background_tasks.add_task(someTask)
return result
##########异步样例
|