Merge branch 'master' of http://git.choozmo.com:3000/ai-anchor/video-maker

backend/app/app/api/api_v1/endpoints/images.py

@@ -6,11 +6,13 @@ from fastapi import APIRouter, Depends, HTTPException
 from fastapi import WebSocket, BackgroundTasks
 from sqlalchemy.orm import Session
 import asyncio
+import time
 import shutil
 import app.crud as crud
 import app.models as models
 import app.schemas as schemas 
 from app.api import deps
+from datetime import datetime
 
 from app.core.celery_app import celery_app
 from app.core.config import settings
@@ -27,7 +29,7 @@ router = APIRouter()
 sr_clients = {}
 
 @router.post("/sr")
-async def supser_resolution(
+def supser_resolution(
     *,
     db: Session = Depends(deps.get_db),
     current_user: models.User = Depends(deps.get_current_active_user),
@@ -37,6 +39,7 @@ async def supser_resolution(
     """
     Super Resolution.
     """
+    print("api start: "+str(datetime.now()))
     filenames = [random_name(20)+Path(file.filename).suffix for file in upload_files]
     stemnames = [Path(filename).stem for filename in filenames]
     new_dir = random_name(10)
@@ -52,10 +55,9 @@ async def supser_resolution(
           return {"error": str(e)}
       finally:
           upload_files[i].file.close()
-
-
+          
     background_tasks.add_task(wait_finish, new_dir, filenames)
-    
+
     print(filenames)
     return JSONResponse({"filenames": filenames}, background=background_tasks)
 
@@ -111,10 +113,12 @@ async def websocket_endpoint(websocket: WebSocket):
     await websocket.accept()
     key = websocket.headers.get('sec-websocket-key')
     sr_clients[key] = websocket
+    print(f"new comer: {key}")
     try:
         while True:
             data = await websocket.receive_text()
-            if not data.startswith("subscribe"):
+            print(f"{key}:{data}")
+            if data.startswith("unsubscribe"):
               del sr_clients[key]
               #for client in sr_clients.values():
               #      await client.send_text(f"ID: {key} | Message: {data}")
@@ -122,4 +126,3 @@ async def websocket_endpoint(websocket: WebSocket):
     except:
         # 接続が切れた場合、当該クライアントを削除する
         del sr_clients[key]
-

backend/app/app/api/api_v1/endpoints/videos.py

@@ -1,10 +1,10 @@
 from typing import Any, List, Optional
 import subprocess
 from fastapi import UploadFile, File, Form
-from fastapi.responses import FileResponse
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi.responses import FileResponse, JSONResponse
+from fastapi import APIRouter, Depends, HTTPException, BackgroundTasks, WebSocket
 from sqlalchemy.orm import Session
-
+from fastapi.encoders import jsonable_encoder
 import app.crud as crud
 import app.models as models
 import app.schemas as schemas 
@@ -13,14 +13,16 @@ from app.api import deps
 from app.core.celery_app import celery_app
 from app.core.config import settings
 from pathlib import Path
+from app.db.session import SessionLocal
 
-from app.core.celery_app import celery_app
+import asyncio
 
 BACKEND_ZIP_STORAGE = Path("/app").joinpath(settings.BACKEND_ZIP_STORAGE)
 LOCAL_ZIP_STORAGE = Path("/").joinpath(settings.LOCAL_ZIP_STORAGE)
 
 
 router = APIRouter()
+video_clients = {}
 
 @router.get("/", response_model=List[schemas.Video])
 def get_video_list(
@@ -40,6 +42,20 @@ def get_video_list(
         )
     return videos
 
+@router.post("/test")
+def test(
+    *,
+    title: str,
+    anchor_id: int,
+    lang_id: int,
+    current_user: models.User = Depends(deps.get_current_active_user),
+) -> Any:
+    video_data = {"title":title, "anchor_id":anchor_id, "lang_id":lang_id}
+    print(video_data)
+    task = celery_app.send_task("app.worker.make_video_test",  kwargs=video_data, )
+    print(task)
+    return "ok"
+
 @router.post("/", response_model=schemas.Video)
 def upload_plot(
     *,
@@ -49,18 +65,17 @@ def upload_plot(
     lang_id: int=Form(...),
     upload_file: UploadFile=File(),
     current_user: models.User = Depends(deps.get_current_active_user),
+    background_tasks: BackgroundTasks,
 ) -> Any:
     """
     Create new video.
     """
     print(title)
     print(upload_file.filename)
-    file_name = crud.video.generate_file_name(db=db, n=20)
-    video_create = schemas.VideoCreate(title=title, progress_state="waiting", stored_file_name=file_name)
-    video = crud.video.create_with_owner(db=db, obj_in=video_create, owner_id=current_user.id)
-
+    filename = crud.video.generate_file_name(db=db, n=20)
+    
     try:
-        with open(str(Path(BACKEND_ZIP_STORAGE).joinpath(video.stored_file_name+".zip")), 'wb') as f:
+        with open(str(Path(BACKEND_ZIP_STORAGE).joinpath(filename+".zip")), 'wb') as f:
             while contents := upload_file.file.read(1024 * 1024):
                 f.write(contents)
     except Exception as e:
@@ -68,13 +83,76 @@ def upload_plot(
         return {"error": str(e)}
     finally:
         upload_file.file.close()
+    
+
+    '''
     zip_filename = video.stored_file_name+".zip"
     print(str(BACKEND_ZIP_STORAGE/zip_filename))
     r = subprocess.run(["sshpass", "-p", "choozmo9", 
                     "scp", "-P", "5722", "-o", "StrictHostKeyChecking=no", f"{str(BACKEND_ZIP_STORAGE/zip_filename)}", f"root@172.104.93.163:{str(LOCAL_ZIP_STORAGE/zip_filename)}"])
     print(r.returncode)
     celery_app.send_task("app.worker.make_video", args=[video.id, video.stored_file_name, current_user.id, anchor_id, current_user.membership_status, current_user.available_time])
-    return video
+    '''
+    video_create = schemas.VideoCreate(title=title, progress_state="PENDING", stored_filename=filename)
+    video = crud.video.create_with_owner(db=db, obj_in=video_create, owner_id=current_user.id)
+    return_msg = {"video_message":"accepted"}
+    video_data = jsonable_encoder(video)
+    video_data['membership_status'] = current_user.membership_status
+    video_data['available_time'] = current_user.available_time
+    video_data['video_id'] = video_data['id']
+    background_tasks.add_task(wait_finish, video_data)
+    return  JSONResponse(return_msg, background=background_tasks)
+
+async def wait_finish(video_data:dict): 
+    zip_filename = video_data['stored_filename']+".zip"
+    process = await asyncio.create_subprocess_exec("sshpass", "-p", "choozmo9", 
+                    "scp", "-P", "5722", "-o", "StrictHostKeyChecking=no", f"{str(BACKEND_ZIP_STORAGE/zip_filename)}", f"root@172.104.93.163:{str(LOCAL_ZIP_STORAGE)}")
+    await process.wait()
+
+    task = celery_app.send_task("app.worker.make_video", kwargs=video_data)
+    while True:
+       await asyncio.sleep(1)
+       if task.state != "PENDING":
+           break
+       
+    db = SessionLocal()
+    video = db.query(models.Video).get(video_data['id'])
+    video.progress_state = "STARTED" 
+    db.commit()
+    db.close()
+    msg_data = f"{video_data['stored_filename']}:STARTED"
+    await publish(msg_data)
+
+    while True:
+        await asyncio.sleep(1)
+        if task.state != "STARTED":
+            break
+
+    if task.state == "SUCCESS":
+        db = SessionLocal()
+        video = db.query(models.Video).get(video_data['id'])
+        user = db.query(models.User).get(video_data['owner_id'])
+        video.progress_state = "SUCCESS" 
+        if time := task.result:
+            user.available_time -= int(time)
+            video.length = int(time)
+        db.commit()
+        db.close()
+        msg_data = f"{video_data['stored_filename']}:SUCCESS:{int(time)}"
+
+    elif task.state == "FAILURE":
+        db = SessionLocal()
+        video = db.query(models.Video).get(video_data['id'])
+        video.progress_state = "FAILURE" 
+        db.commit()
+        db.close()
+        msg_data = f"{video_data['stored_filename']}:FAILURE"
+
+    await publish(msg_data)
+
+async def publish(data):
+    for video_client in video_clients.values():
+        await video_client.send_text(f"{data}")
 
 @router.get("/{id}")
 def download_video(
@@ -86,3 +164,21 @@ def download_video(
     
     return {"message":"address"}
 
+@router.websocket("")
+async def websocket_endpoint(websocket: WebSocket):
+    await websocket.accept()
+    key = websocket.headers.get('sec-websocket-key')
+    video_clients[key] = websocket
+    try:
+        while True:
+            data = await websocket.receive_text()
+            if not data.startswith("subscribe"):
+              del video_clients[key]
+              #for client in sr_clients.values():
+              #      await client.send_text(f"ID: {key} | Message: {data}")
+
+    except:
+        # 接続が切れた場合、当該クライアントを削除する
+        del video_clients[key]
+
+

backend/app/app/core/celery_app.py

@@ -4,4 +4,4 @@ celery_app = Celery("worker", broker="redis://172.104.93.163:16379/0", backend="
 
 
 
-celery_app.conf.task_routes = {"app.worker.make_video": "video", "app.worker.super_resolution": "image"}
+celery_app.conf.task_routes = {"app.worker.make_video": "video", "app.worker.super_resolution": "image", "app.worker.make_video_test": "video"}

backend/app/app/crud/crud_video.py

@@ -35,9 +35,9 @@ class CRUDVideo(CRUDBase[Video, VideoCreate, VideoUpdate]):
       self, db: Session, *, n:int
     ) -> bool:
         while True:
-            file_name = random_name(n)
-            if not db.query(self.model).filter(Video.stored_file_name==file_name).first():
-              return file_name
+            filename = random_name(n)
+            if not db.query(self.model).filter(Video.stored_filename==filename).first():
+              return filename
 
 
 video = CRUDVideo(Video)

backend/app/app/schemas/video.py

@@ -8,13 +8,13 @@ from fastapi import UploadFile, File
 class VideoBase(BaseModel):
     title: Optional[str] = None
     progress_state: Optional[str] = None
-    stored_file_name: Optional[str] = None
+    stored_filename: Optional[str] = None
 
 # Properties to receive on video creation
 class VideoCreate(VideoBase):
     title: str
     progress_state: str
-    stored_file_name: str
+    stored_filename: str
 
 # Properties to receive on video update
 class VideoUpdate(VideoBase):

backend/app/app/worker.py

@@ -6,11 +6,13 @@ import requests
 from pathlib import Path
 from urllib.parse import urlparse, urljoin
 from .aianchor import  make_video_from_zip
+import gc
 #client_sentry = Client(settings.SENTRY_DSN)
 import dataset
 from app.db.session import SessionLocal
 from app.models import video
 from app import crud
+import gc
 download_to_local_url = urljoin(settings.SERVER_HOST, settings.API_V1_STR+"/videos/worker")
 upload_to_server_url = urljoin(settings.SERVER_HOST, settings.API_V1_STR+"/videos/worker")
 
@@ -20,11 +22,18 @@ CELERY_VIDEO_STORAGE = Path("/").joinpath(settings.LOCAL_VIDEO_STORAGE)
 LOCAL_VIDEO_STORAGE = Path("/").joinpath(settings.LOCAL_VIDEO_STORAGE)
 
 STORAGE_IP = '192.168.192.252'#os.getenv('STORAGE_IP')
+BACKEND_IP = '172.'
 if not STORAGE_IP:
     raise Exception
 
-@celery_app.task(acks_late=True)
-def make_video(video_id, filename, user_id) -> str:
+@celery_app.task(acks_late=True, bind=True, track_started=True)
+def make_video_test(self, title=None, anchor_id=None, lang_id=None)->int:
+   print(title, anchor_id, lang_id)
+   return 0
+
+@celery_app.task(acks_late=True, bind=True, track_started=True)
+def make_video(self, *, video_id, stored_filename=None, user_id=None, anchor_id=None, membership=None, available_time=None, **others) -> int:
+    
     #video_id, zip_filename, user_id = args
     # download 
     '''
@@ -34,40 +43,55 @@ def make_video(video_id, filename, user_id) -> str:
         for chunk in r.iter_content(chunk_size=1024):
             f.write(chunk)
     '''
-    db = SessionLocal()
-    db.execute("SELECT 1")
-    zip_filename = filename + ".zip"
+    # db = SessionLocal()
+    # db.execute("SELECT 1")
+    zip_filename = stored_filename + ".zip"
     r = subprocess.run(["sshpass", "-p", "choozmo9", 
                         "scp", "-o", "StrictHostKeyChecking=no", f"root@{STORAGE_IP}:{str(LOCAL_ZIP_STORAGE/zip_filename)}", f"{str(CELERY_ZIP_STORAGE/zip_filename)}"])
     print(f'get from local storage: {r.returncode}')
-    print(f"video_id: {video_id}, file name: {filename}")
-    db.execute(f"UPDATE video SET progress_state='processing' where id={video_id}")
-    db.commit()
+    print(f"video_id: {video_id}, file name: {stored_filename}")
+    
+    # db.execute(f"UPDATE video SET progress_state='processing' where id={video_id}")
+    # db.commit()
     # make video
+    watermark_path='medias/logo_watermark.jpg'
+    content_time = 0
     try:
-      make_video_from_zip(working_dir=CELERY_ZIP_STORAGE,style=Path("app/style/choozmo"),  inputfile=zip_filename,opening=False, ending=False)
+      if membership=="infinite":
+         watermark_path=None
+      content_time = make_video_from_zip(working_dir=CELERY_ZIP_STORAGE,style=Path("app/style/choozmo"),  
+                                         inputfile=zip_filename,
+                                         opening=False, 
+                                         ending=False, 
+                                         watermark_path=watermark_path, 
+                                         available_time=available_time)
     except Exception as e:
       print(f"error:{e}")
-      db.execute(f"UPDATE video SET progress_state='failed' where id={video_id}")
-      db.commit()
+      # db.execute(f"UPDATE video SET progress_state='failed' where id={video_id}")
+      # db.commit()
     else:
       # 
-      video_filename = filename + ".mp4"
+      video_filename = stored_filename + ".mp4"
       r = subprocess.run(["sshpass", "-p", "choozmo9", 
                           "scp", "-o", "StrictHostKeyChecking=no", f"{str(CELERY_ZIP_STORAGE/'output.mp4')}", f"root@{STORAGE_IP}:{'/var/www/videos/'+video_filename}"])
       print(f"return to local storage: {r.returncode}")
-      print(f"video_id: {video_id}, file name: {filename}")
+      print(f"video_id: {video_id}, file name: {stored_filename}")
 
-      db.execute(f"UPDATE video SET progress_state='completed' where id={video_id}")
-      db.commit()
+      # db.execute(f"UPDATE video SET progress_state='completed', length={int(content_time)} where id={video_id}")
+      # db.commit()
 
       
-      return "complete"
+      gc_num = gc.collect()
+      print(f"gc_num: {gc_num}")
+      return int(content_time)
 
-@celery_app.task(acks_late=True)
-def super_resolution(filenames):
-   source = [f"root@{STORAGE_IP}:{str(LOCAL_ZIP_STORAGE/filename)}" for filename in filenames]
-   r = subprocess.run(["sshpass", "-p", "choozmo9", 
-                        "scp", "-o", "StrictHostKeyChecking=no", f"root@{STORAGE_IP}:{str(LOCAL_ZIP_STORAGE/zip_filename)}", f"{str(CELERY_ZIP_STORAGE/zip_filename)}"])
+@celery_app.task(acks_late=True, bind=True, track_started=True)
+def super_resolution(self, dirname, filenames):
    
-
+    r = subprocess.run(["sshpass", "-p", "choozmo9", 
+                        "scp", "-o", "StrictHostKeyChecking=no", "-r", f"root@{STORAGE_IP}:{str(LOCAL_ZIP_STORAGE/dirname)}", f"{str(CELERY_ZIP_STORAGE)}"])
+   
+    r = subprocess.run(["python", "/root/github/GFPGAN/inference_gfpgan.py", "-i", f"{str(CELERY_ZIP_STORAGE/dirname)}", "-o", f"{str(CELERY_ZIP_STORAGE/dirname)}", "-v", "1.4", "-s", "2"])
+   
+    r = subprocess.run(["sshpass", "-p", "choozmo9", 
+                        "scp", "-o", "StrictHostKeyChecking=no", "-r", f"{str(CELERY_ZIP_STORAGE/dirname)}", f"root@{STORAGE_IP}:{str(LOCAL_ZIP_STORAGE)}"])

frontend/src/api.ts

@@ -15,14 +15,12 @@ export const api = {
     const params = new URLSearchParams();
     params.append("username", username);
     params.append("password", password);
-
     return axios.post(`${apiUrl}/api/v1/login/access-token`, params);
   },
   async qrLogInGetToken(username: string, password: string,ser_no: string) {
     const params = new URLSearchParams();
     params.append("username", username);
     params.append("password", password);
-
     return axios.post(`${apiUrl}/api/v1/login/access-token?add_time_code=${ser_no}`, params);
   },
   async googleLogin(username: string) {

frontend/src/interfaces/index.ts

@@ -39,6 +39,7 @@ export interface MainState {
   notifications: AppNotification[];
   videos: Video[];
   images: Image[];
+  videosWebSocket: WebSocket;
 };
 
 export interface Video {
@@ -71,4 +72,4 @@ export interface Image {
 export interface ImageDownload {
   file_name: string;
   stored_file_name: string;
-}
+}

frontend/src/stores/main.ts

@@ -18,6 +18,7 @@ const defaultState: MainState = {
   notifications: [],
   videos: [],
   images: [],
+  videosWebSocket: new WebSocket(`${wsUrl}/api/v1/videos`)
 };
 
 export const useMainStore = defineStore("MainStoreId", {

frontend/src/views/main/Dashboard.vue

@@ -11,7 +11,6 @@ const userProfile = mainStoreRef.readUserProfile;
 
 const greetedUser = computed(() => {
   const userProfile = mainStoreRef.readUserProfile;
-  console.log("userProfile", userProfile);
 
   if (userProfile.value) {
     if (userProfile.value!.full_name) {
@@ -26,7 +25,7 @@ const greetedUser = computed(() => {
 <template>
   <v-container fluid>
     <v-row dense>
-      <v-col cols="12">
+      <v-col cols="6">
         <v-card class="ma-3 pa-3">
           <v-card-title primary-title>
             <h3>Welcome {{ greetedUser }}</h3>
@@ -67,7 +66,7 @@ const greetedUser = computed(() => {
       </v-card-actions> -->
         </v-card>
       </v-col>
-      <v-col cols="6">
+      <!-- <v-col cols="6">
         <v-card class="ma-3 pa-3 second-item">
           <v-card-title primary-title>
             <h3>已使用秒數</h3>
@@ -86,16 +85,9 @@ const greetedUser = computed(() => {
             >
               查看詳情
             </v-btn>
-            <!-- <v-btn
-              to="/main/profile/password"
-              variant="flat"
-              color="primary"
-              style="padding: 0 15px"
-              >{{ t("changePassword") }}</v-btn
-            > -->
           </v-card-actions>
         </v-card>
-      </v-col>
+      </v-col> -->
       <v-col cols="6">
         <v-card class="ma-3 pa-3 second-item">
           <v-card-title primary-title>
@@ -142,6 +134,9 @@ h4 {
 }
 
 .second-item {
+  height: 255px;
+  display: flex;
+  flex-direction: column;
   strong {
     font-size: 30px;
     font-weight: 500;

frontend/src/views/main/Progress.vue

@@ -8,6 +8,7 @@ const { t } = useI18n();
 const mainStore = useMainStore();
 const mainStoreRef = storeToRefs(mainStore);
 const videos = mainStoreRef.readVideos;
+const WS = mainStore.videosWebSocket;
 
 const headers = [
   {
@@ -34,7 +35,20 @@ function previewVideo(name: String) {
 
 onMounted(async () => {
   await mainStore.actionGetVideos();
+  WS.onmessage = function (e) {
+    console.log('onmessage',e);
+    const data = e.data.split(':');
+    mainStore.videos.map(e=>{
+      if(e.stored_filename===data[0]) {
+        e.progress_state = data[1]; // 影片狀態
+        if (mainStore.userProfile && data[2]) {
+          mainStore.userProfile.available_time = mainStore.userProfile.available_time - parseInt(data[2]); // 使用秒數
+        }
+      }
+    })
+  };
 });
+
 </script>
 
 <template>
@@ -50,15 +64,15 @@ onMounted(async () => {
       :sort-by="[{ key: 'id', order: 'desc' }]"
     >
       <template v-slot:item.progress_state="{ item }">
-        <span v-if="item.raw.progress_state === 'completed'">
+        <span v-if="item.raw.progress_state === 'SUCCESS'">
           <v-icon icon="check_circle" color="success" />
           完成
         </span>
-        <span v-else-if="item.raw.progress_state === 'waiting'">
+        <span v-else-if="item.raw.progress_state === 'PENDING'">
           <v-icon icon="pending" color="warning" />
           等待中
         </span>
-        <span v-else-if="item.raw.progress_state === 'processing'">
+        <span v-else-if="item.raw.progress_state === 'STARTED'">
           <v-progress-circular
             indeterminate
             color="info"
@@ -74,8 +88,8 @@ onMounted(async () => {
       <template v-slot:item.id="{ item }">
         <v-btn
           flat
-          :disabled="item.raw.progress_state !== 'completed'"
-          @click="previewVideo(item.raw.stored_file_name)"
+          :disabled="item.raw.progress_state !== 'SUCCESS'"
+          @click="previewVideo(item.raw.stored_filename)"
         >
           <v-icon icon="play_circle" />
         </v-btn>

frontend/src/views/main/Upload.vue

@@ -3,6 +3,7 @@ import { ref, reactive, watch, computed } from "vue";
 import { useMainStore } from "@/stores/main";
 import { required } from "@/utils";
 import { useI18n } from "vue-i18n";
+import { wsUrl } from "@/env";
 import type { VideoCreate } from "@/interfaces";
 import router from "@/router";
 import Dialog from "@/components/Dialog.vue";
@@ -15,6 +16,8 @@ let dialog = reactive({
 });
 
 const { t } = useI18n();
+const mainStore = useMainStore();
+const WS = mainStore.videosWebSocket;
 const valid = ref(true);
 const title = ref("");
 const zipFiles = ref();
@@ -164,17 +167,16 @@ const getImageUrl = (imgFolder: string, name: string) => {
     .href;
 };
 
-const mainStore = useMainStore();
-
 watch(dialog, (newVal, oldVal) => {
   if (!newVal.show) {
     setTimeout(() => {
       router.push("/main/progress");
-    }, 500);
+    }, 1000);
   }
 });
 
 async function Submit() {
+  WS.send("subscribe");
   setTimeout(() => {
     dialog.show = true;
   }, 2000);
@@ -189,6 +191,7 @@ async function Submit() {
     };
 
     await mainStore.uploadPlot(video_data, zipFiles.value[0]);
+    valid.value = true;
     // (Form as any).value.reset();
   }
 }