CMM_API/apis/main.py

from fastapi import FastAPI
from fastapi import Request
from fastapi import Form
import urllib
import sys
import os
import codecs
import html
from typing import Optional

from fastapi import FastAPI
from pydantic import BaseModel
import rpyc
import zhtest
from fastapi.middleware.cors import CORSMiddleware




class Params(BaseModel):
    content: str
    title: str
    bg1: str


class FName(BaseModel):
    fname: str


class TTS(BaseModel):
    txt: str
    num: str
#    description: Optional[str] = None
#    price: float
#    tax: Optional[float] = None

app = FastAPI()

origins = [
    "https://seo.hhh.com.tw",
    "https://seo.googo.org",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.post("/tts/audio_to_video")
async def audio_to_video(item: FName):
    print(item.fname)
    fname=zhtest.to_video.delay(item.fname)
    print(fname)    
    return {"OK": "200"}



@app.post("/full_video")
async def full_video(item: Params):
    print(item)
    fulldict={'content':item.content,'title':item.title,'bg1':item.bg1}
    fname=zhtest.full_task.delay(fulldict)
    return {"OK": "200"}


@app.post("/tts")
async def run_tts(item: TTS):
    print(item.txt)
    print(item.num)
#    import zhtest
    fname=zhtest.to_wav.delay(item.txt)
    print(fname)
    return {"OK": "200"}


@app.get("/json/test")
async def json_test():
    return {'files':['a','b','c','d','e']}



@app.get("/gpu")
async def get_gpu():
    conn = rpyc.classic.connect("139.162.121.30", 12333)
    conn.execute('import os')
    ros = conn.modules.os
#    lst=ros.listdir('/var/wav')
    return {'ok':'200'}


@app.get("/tts/files")
async def get_tts_files():
    conn = rpyc.classic.connect("139.162.121.30", 12333)
    conn.execute('import os')
    ros = conn.modules.os
    lst=ros.listdir('/var/wav')
    return {'files':lst}


@app.get("/tts/videos")
async def get_video_files():
    conn = rpyc.classic.connect("139.162.121.30", 12333)
    conn.execute('import os')
    ros = conn.modules.os
    lst=ros.listdir('/var/video')
    newlst=[]
    for l in lst:
        newlst.append('http://139.162.121.30:54321/'+l)    
    return {'files':newlst}



@app.get("/tts/s1_videos")
async def s1_videos():
    conn = rpyc.classic.connect("139.162.121.30", 12333)
    conn.execute('import os')
    ros = conn.modules.os
    lst=ros.listdir('/var/video/s1')
    newlst=[]
    for l in lst:
        newlst.append('http://139.162.121.30:54321/s1/'+l)    
    return {'files':newlst}



@app.get("/")
async def root():
    return {"message": "Hello World"}