File size: 3,890 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
bc7dec9
96888b9
 
 
 
 
 
 
 
 
 
 
 
 
 
bc7dec9
49f01e6
b628481
 
bc7dec9
49f01e6
b628481
2c03cb6
 
 
 
 
 
 
 
 
 
bc7dec9
 
 
 
 
 
 
 
 
 
 
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
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
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:
        print("开始执行py任务",file_name)
        # result = subprocess.check_output(["python", f"{file_name}.py"], stderr=subprocess.STDOUT).decode("utf-8")#执行完成后显示运行py的print
        ##########test 边执行py边显示print+++++++++++++++++++++++++++++++++++++
        # 开始执行Python 脚本
        output = []
        process = subprocess.Popen(["python", f"{file_name}.py"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
        # 逐行读取脚本的输出并显示
        for line in iter(process.stdout.readline, b''):
            line_text = line.decode('utf-8').strip()
            print(line_text)
            output.append(line_text)
        process.wait()
        result = '\n'.join(output)
        ##########test 边执行py边显示print+++++++++++++++++++++++++++++++++++++
        
        print("执行完成py任务:",file_name,"结果如下")
        print(result)
        return PlainTextResponse(result)
    except subprocess.CalledProcessError as e:
        print("执行py任务失败",file_name,"原因如下")
        print(f"Error executing {file_name}.py: {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文件执行接口



#收到图片后为了后台任务的顺序,单独设定一个函数依次将1去水印,2裁剪,3合并按照顺序加入后台任务队列
@app.get("/execute_all_task")
async def execute_all_task(background_tasks: BackgroundTasks):
    background_tasks.add_task(file_executer,"1removeMask")
    background_tasks.add_task(file_executer,"2magiDialogCut")
    background_tasks.add_task(file_executer,"3mergeDialogToVideo")
    return {"message": "Tasks added to the queue"}






##########异步样例
def someTask():
    time.sleep(20)
    print("睡眠20s结束")
    
@app.get("/backTaskTest")
def returnRandomSubscribeUrl(background_tasks: BackgroundTasks)->str:
    #返回
    result = "先返回"
    background_tasks.add_task(someTask)
    return result
##########异步样例