api.py

from fastapi import FastAPI, File, UploadFile,Body,Query,Response
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
from bilibili_api import sync, video_uploader, Credential#bili上传部分
import asyncio #保证后台任务的先后
import json #对bili_meta_data进行dumps ->str和loads—>dict
import bilibili_api


sessdata = os.getenv('sessdata')
bili_jct = os.getenv('bili_jct')
buvid3 = os.getenv('buvid3')
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") #增加指定保存路径功能,默认是/manga路径
def getOriginalMangaList(images: List[UploadFile] = File(...), save_path: str = "/manga"):
    saved_files = []
    for idx, image in enumerate(images):
        img = image.file.read()
        image_data = Image.open(io.BytesIO(img)).convert("L").convert("RGB")
        path_to_image = os.path.join(save_path, f"{idx}.jpg")
        image_data.save(path_to_image)
        saved_files.append(path_to_image)
    return {"message": "Images saved successfully", "saved_files": saved_files}
##这个比较快不用异步

##这个比较快不用异步
@app.delete("/deleteFiles")
async def delete_all_files_and_folders(directory:str):
    for root, dirs, files in os.walk(directory, topdown=False):
        for name in files:
            file_path = os.path.join(root, name)
            print("删除指定目录函数检查到文件,删除文件:", file_path)
            os.remove(file_path)
        for name in dirs:
            dir_path = os.path.join(root, name)
            print("删除指定目录函数检查到目录,删除目录:", dir_path)
            os.rmdir(dir_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文件执行接口




#保证既要提交队列后返回给用户,又要先完成前面的video生成步骤再执行后面的submit函数,还有兼顾allow_submit的值
##能立刻返回的
@app.get("/execute_all_task")
async def execute_all_task(background_tasks: BackgroundTasks, bili_meta_data,file_list: List[str] = Query(["0filterImage","1removeMask", "2magiDialogCut", "3mergeDialogToVideo"]),
                             mp4_out_file: str = 'mp4_out/output.mp4',
                           allow_submit: bool = False, cover_url: str = ''
                           ):
    #cover_url因为本地上传有协程问题,所以采用cover_url然后转换为Picture类型即可->bilibili_api.utls.picture.Picture.from_url(str)->str
    print("获取到的用户bili_meta_data是:", bili_meta_data)
    #这里需要先转换bili_meta_data从str变成dict才能继续
    bili_meta_data = json.loads(bili_meta_data)
    print("获取到的file_list是",file_list)
    print("获取到的allow_submit是",allow_submit)
    
    background_tasks.add_task(process_tasks_and_upload, file_list, bili_meta_data, cover_url,allow_submit,background_tasks)
    return {"message": "提交成功"}

def process_tasks_and_upload(file_list, bili_meta_data,  cover_url,allow_submit,background_tasks):
    tasks = []
    for file_name in file_list:
        tasks.append(file_executer(file_name))

    if allow_submit:
        #因为采用cover_url上传,所以不需要读取内容然后保存本地->这样也会引起协程问题
        print("开始对cover_url是",cover_url,"的提交进程进行加入后台任务")
        background_tasks.add_task(upload_video,bili_meta_data, cover_url)
##尝试能立刻返回的








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





async def upload_video(meta:dict,cover_url:str):
    credential = Credential(sessdata=sessdata,
                            bili_jct=bili_jct,
                            buvid3=buvid3)

    #cover = bilibili_api.utils.picture.Picture.from_url(cover_url)
    cover_url = "https://rogerxavier-moviepy-with-manga-test.hf.space/cover.jpg"
    cover = bilibili_api.utils.picture.Picture.from_url(cover_url)#准备用cover/0.jpg作为图片
    
    
    
    page = video_uploader.VideoUploaderPage(path='mp4_out/output_video.mp4', title=meta['title'], description=meta['desc'])
    print("上传前的cover_url是",cover_url)
    
    uploader = video_uploader.VideoUploader([page], meta, credential, cover=cover)

    @uploader.on("__ALL__")
    async def ev(data):
        print(data)

    await uploader.start()


#检测output.mp4下载查看生成效果(b不过审时用)
@app.get("/get_video")
async def get_video(dir:str = "/mp4_out/output.mp4")->"video file in header":
    video_file_path = dir
    video_file = open(video_file_path, "rb")
    video_data = video_file.read()
    video_file.close()
    
    response = Response(content=video_data, media_type="video/mp4")
    response.headers["Content-Disposition"] = "attachment; filename=output.mp4"
    
    return response


#将cover目录下面的0.jpg返回图片作为cover地址
@app.get("/cover.jpg")
async def get_cover_image():
    cover_dir = "cover"
    file_path = os.path.join(cover_dir, "0.jpg")
    return FileResponse(file_path)