from fastapi import FastAPI,Cookie, Depends, Query, status,File, UploadFile,Request,Response,HTTPException from fastapi.templating import Jinja2Templates from fastapi.responses import HTMLResponse, RedirectResponse, JSONResponse from typing import List, Optional from os.path import isfile, isdir, join import threading import zhtts import os import urllib import requests from bs4 import BeautifulSoup from PIL import Image,ImageDraw,ImageFont import pyttsx3 import rpyc import random import time import math import hashlib import re import asyncio import urllib.request from fastapi.responses import FileResponse from fastapi.middleware.cors import CORSMiddleware import dataset from datetime import datetime, timedelta from util.swap_face import swap_face from fastapi.staticfiles import StaticFiles import shutil import io from first import first from passlib.context import CryptContext from jose import JWTError, jwt from fastapi_jwt_auth import AuthJWT from fastapi_jwt_auth.exceptions import AuthJWTException from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm import models import pymysql from first import first pymysql.install_as_MySQLdb() app = FastAPI() app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) SECRET_KEY = "df2f77bd544240801a048bd4293afd8eeb7fff3cb7050e42c791db4b83ebadcd" ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 300 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") app.mount("/static", StaticFiles(directory="static"), name="static") app.mount("/static/img", StaticFiles(directory="static/img"), name="static/img") app.mount("/templates", StaticFiles(directory="templates"), name="templates") templates = Jinja2Templates(directory="templates") oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") tmp_video_dir = '../OpenshotService/tmp_video/' tmp_avatar_dir = '../../face_swap/tmp_avatar/' #change source face path here video_sub_folder = 'ai_anchor_video/' avatar_sub_folder = 'swap_save/' tmp_img_sub_folder = 'tmp_img/' img_upload_folder = '/var/www/html/'+tmp_img_sub_folder video_dest = '/var/www/html/'+video_sub_folder avatar_dest = '/var/www/html/'+avatar_sub_folder @app.get("/index2") async def index2(): return FileResponse('static/index2.html') @app.get("/index_eng") async def index2(): return FileResponse('static/index_eng.html') # home page @app.get("/index", response_class=HTMLResponse) async def get_home_page(request: Request, response: Response): return templates.TemplateResponse("index.html", {"request": request, "response": response}) @app.get("/", response_class=HTMLResponse) async def get_home_page(request: Request, response: Response): return templates.TemplateResponse("index.html", {"request": request, "response": response}) @app.get("/make_video", response_class=HTMLResponse) async def get_home_page(request: Request, response: Response): return templates.TemplateResponse("make_video.html", {"request": request, "response": response}) # login & register page @app.get("/login", response_class=HTMLResponse) async def get_login_and_register_page(request: Request): return templates.TemplateResponse("login.html", {"request": request}) @app.post("/login") async def login_for_access_token(request: Request, form_data: OAuth2PasswordRequestForm = Depends(), Authorize: AuthJWT = Depends()): db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/AI_anchor?charset=utf8mb4') user = authenticate_user(form_data.username, form_data.password) if not user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect username or password", headers={"WWW-Authenticate": "Bearer"}, ) access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) access_token = create_access_token( data={"sub": user.username}, expires_delta=access_token_expires ) table = db['users'] user.token = access_token table.update(dict(user), ['username']) access_token = Authorize.create_access_token(subject=user.username) refresh_token = Authorize.create_refresh_token(subject=user.username) Authorize.set_access_cookies(access_token) Authorize.set_refresh_cookies(refresh_token) #return templates.TemplateResponse("index.html", {"request": request, "msg": 'Login'}) return {"access_token": access_token, "token_type": "bearer"} @app.post("/token") async def access_token(form_data: OAuth2PasswordRequestForm = Depends(), Authorize: AuthJWT = Depends()): db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/AI_anchor?charset=utf8mb4') user = authenticate_user(form_data.username, form_data.password) if not user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect username or password", headers={"WWW-Authenticate": "Bearer"}, ) access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) access_token = create_access_token( data={"sub": user.username}, expires_delta=access_token_expires ) return {"access_token": access_token, "token_type": "bearer"} @app.post("/register") async def register(request: Request): user = models.User(**await request.form()) user_register(user) return templates.TemplateResponse("login.html", {'request': request,"success": True}, status_code=status.HTTP_302_FOUND) @app.get('/user_profile', response_class=HTMLResponse) def protected(request: Request, Authorize: AuthJWT = Depends()): Authorize.jwt_required() current_user = Authorize.get_jwt_subject() return current_user @app.get('/logout') def logout(request: Request, Authorize: AuthJWT = Depends()): Authorize.jwt_required() Authorize.unset_jwt_cookies() return {"msg": "Successfully logout"} @app.get("/gen_avatar") async def avatar(): return FileResponse('static/gen_avatar.html') @app.post("/swapFace") async def swapFace(req:models.swap_req): if 'http' not in req.imgurl: req.imgurl= 'http://'+req.imgurl try: im = Image.open(requests.get(req.imgurl, stream=True).raw) im= im.convert("RGB") except: return {'msg':"無法辨別圖片網址"+req.imgurl} name_hash = str(time.time()).replace('.','') x = threading.Thread(target=gen_avatar, args=(name_hash,req.imgurl)) x.start() return {'msg':'人物生成中，請稍候'} @app.post("/uploadfile/") async def create_upload_file(file: UploadFile = File(...)): img_name = str(time.time()).replace('.','') try: contents = await file.read() image = Image.open(io.BytesIO(contents)) image= image.convert("RGB") image.save(img_upload_folder+img_name+'.jpg') except: return {'msg':'檔案無法使用'} return {"msg": 'www.choozmo.com:8168/'+tmp_img_sub_folder+img_name+'.jpg'} @app.post("/make_anchor_video_v2") async def make_anchor_video_v2(req:models.request): if len(req.image_urls) != len(req.text_content): return {'msg':'副標題數量、圖片(影片)數量以及台詞數量必須一致'} for idx in range(len(req.image_urls)): if 'http' not in req.image_urls[idx]: req.image_urls[idx] = 'http://'+req.image_urls[idx] for txt in req.text_content: if re.search('[a-zA-Z]', txt) !=None: return {'msg':'輸入字串不能包含英文字!'} name_hash = str(time.time()).replace('.','') for imgu in req.image_urls: try: if get_url_type(imgu) =='video/mp4': r=requests.get(imgu) else: im = Image.open(requests.get(imgu, stream=True).raw) im= im.convert("RGB") except: return {'msg':"無法辨別圖片網址"+imgu} save_history(req,name_hash) x = threading.Thread(target=gen_video_queue, args=(name_hash,req.name, req.text_content, req.image_urls,int(req.avatar))) x.start() return {"msg":"製作影片需要時間，請您耐心等候，成果會傳送至LINE群組中"} @app.post("/make_anchor_video_eng") async def make_anchor_video_eng(req:models.request_eng): if len(req.image_urls) != len(req.sub_titles) or len(req.sub_titles) != len(req.text_content): return {'msg':'副標題數量、圖片(影片)數量以及台詞數量必須一致'} for idx in range(len(req.image_urls)): if 'http' not in req.image_urls[idx]: req.image_urls[idx] = 'http://'+req.image_urls[idx] name_hash = str(time.time()).replace('.','') for imgu in req.image_urls: try: if get_url_type(imgu) =='video/mp4': r=requests.get(imgu) else: im = Image.open(requests.get(imgu, stream=True).raw) im= im.convert("RGB") except: return {'msg':"無法辨別圖片網址"+imgu} save_history(req,name_hash) x = threading.Thread(target=gen_video_queue_eng, args=(name_hash,req.name, req.text_content, req.image_urls,req.sub_titles,int(req.avatar))) x.start() return {"msg":"製作影片需要時間，請您耐心等候，成果會傳送至LINE群組中"} @app.get("/history_input") async def history_input(): db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/AI_anchor?charset=utf8mb4') statement = 'SELECT * FROM history_input ORDER BY timestamp DESC LIMIT 50' logs = [] for row in db.query(statement): logs.append({'id':row['id'],'name':row['name'],'text_content':row['text_content'].split(','),'link':row['link'],'image_urls':row['image_urls'].split(',')}) return logs @AuthJWT.load_config def get_config(): return models.Settings() @app.exception_handler(AuthJWTException) def authjwt_exception_handler(request: Request, exc: AuthJWTException): return JSONResponse( status_code=exc.status_code, content={"detail": exc.message} ) async def get_current_user(token: str = Depends(oauth2_scheme)): db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/AI_anchor?charset=utf8mb4') credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) username: str = payload.get("sub") if username is None: raise credentials_exception token_data = TokenData(username=username) except JWTError: raise credentials_exception user = get_user(username=token_data.username) if user is None: raise credentials_exception return user def check_user_exists(username): db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/AI_anchor?charset=utf8mb4') if int(next(iter(db.query('SELECT COUNT(*) FROM AI_anchor.users WHERE username = "'+username+'"')))['COUNT(*)']) > 0: return True else: return False def get_user(username: str): db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/AI_anchor?charset=utf8mb4') if not check_user_exists(username): # if user don't exist return False user_dict = next( iter(db.query('SELECT * FROM AI_anchor.users where username ="'+username+'"'))) user = User(**user_dict) return user def user_register(user): db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/AI_anchor?charset=utf8mb4') table = db['users'] user.password = get_password_hash(user.password) table.insert(dict(user)) def get_password_hash(password): return pwd_context.hash(password) def verify_password(plain_password, hashed_password): return pwd_context.verify(plain_password, hashed_password) def authenticate_user(username: str, password: str): db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/AI_anchor?charset=utf8mb4') if not check_user_exists(username): # if user don't exist return False user_dict = next(iter(db.query('SELECT * FROM AI_anchor.users where username ="'+username+'"'))) user = models.User(**user_dict) if not verify_password(password, user.password): return False return user def create_access_token(data: dict, expires_delta: Optional[timedelta] = None): to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=15) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt def save_history(req,name_hash): db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/AI_anchor?charset=utf8mb4') log_table = db['history_input'] txt_content_seperate_by_dot = '' for txt in req.text_content: txt_content_seperate_by_dot += txt+"," txt_content_seperate_by_dot = txt_content_seperate_by_dot[:-1] img_urls_seperate_by_dot = '' for iurl in req.image_urls: img_urls_seperate_by_dot += iurl+"," img_urls_seperate_by_dot = img_urls_seperate_by_dot[:-1] time_stamp = datetime.fromtimestamp(time.time()) time_stamp = time_stamp.strftime("%Y-%m-%d %H:%M:%S") pk = log_table.insert({'name':req.name,'text_content':txt_content_seperate_by_dot,'image_urls':img_urls_seperate_by_dot,'link':'www.choozmo.com:8168/'+video_sub_folder+name_hash+'.mp4','timestamp':time_stamp}) def get_url_type(url): req = urllib.request.Request(url, method='HEAD', headers={'User-Agent': 'Mozilla/5.0'}) r = urllib.request.urlopen(req) contentType = r.getheader('Content-Type') return contentType def notify_group(msg): glist=['WekCRfnAirSiSxALiD6gcm0B56EejsoK89zFbIaiZQD'] for gid in glist: headers = { "Authorization": "Bearer " + gid, "Content-Type": "application/x-www-form-urlencoded" } params = {"message": msg} r = requests.post("https://notify-api.line.me/api/notify",headers=headers, params=params) def gen_video(name_hash,name,text_content, image_urls,avatar): c = rpyc.connect("localhost", 8878) c._config['sync_request_timeout'] = None remote_svc = c.root my_answer = remote_svc.call_video(name_hash,name,text_content, image_urls,avatar) # method call shutil.copy(tmp_video_dir+name_hash+'.mp4',video_dest+name_hash+'.mp4') os.remove(tmp_video_dir+name_hash+'.mp4') def gen_video_eng(name_hash,name,text_content, image_urls,sub_titles,avatar): c = rpyc.connect("localhost", 8878) c._config['sync_request_timeout'] = None remote_svc = c.root my_answer = remote_svc.call_video_eng(name_hash,name,text_content, image_urls,sub_titles,avatar) # method call shutil.copy(tmp_video_dir+name_hash+'.mp4',video_dest+name_hash+'.mp4') os.remove(tmp_video_dir+name_hash+'.mp4') def gen_video_queue(name_hash,name,text_content, image_urls,avatar): db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/AI_anchor?charset=utf8mb4') time_stamp = datetime.fromtimestamp(time.time()).strftime("%Y-%m-%d %H:%M:%S") txt_content_seperate_by_dot = '' for txt in text_content: txt_content_seperate_by_dot += txt+"," txt_content_seperate_by_dot = txt_content_seperate_by_dot[:-1] img_urls_seperate_by_dot = '' for iurl in image_urls: img_urls_seperate_by_dot += iurl+"," img_urls_seperate_by_dot = img_urls_seperate_by_dot[:-1] db['video_queue'].insert({'name_hash':name_hash,'name':name,'text_content':txt_content_seperate_by_dot,'image_urls':img_urls_seperate_by_dot,'avatar':avatar,'timestamp':time_stamp}) while True: if first(db.query('SELECT * FROM video_queue_status'))['status'] == 1:#only one row in this table, which is the id 1 one print('another process running, leave loop')#1 means already running break if first(db.query('SELECT COUNT(1) FROM video_queue'))['COUNT(1)'] == 0: print('all finish, leave loop') break top1 = first(db.query('SELECT * FROM video_queue')) try: db.query('UPDATE video_queue_status SET status = 1;') c = rpyc.connect("localhost", 8878) c._config['sync_request_timeout'] = None remote_svc = c.root my_answer = remote_svc.call_video(top1['name_hash'],top1['name'],top1['text_content'].split(','), top1['image_urls'].split(','),top1['avatar']) # method call shutil.copy(tmp_video_dir+top1['name_hash']+'.mp4',video_dest+top1['name_hash']+'.mp4') os.remove(tmp_video_dir+top1['name_hash']+'.mp4') except: print('video generation error') notify_group('影片錯誤') db['video_queue'].delete(id=top1['id']) db.query('UPDATE video_queue_status SET status = 0') def gen_video_queue_eng(name_hash,name,text_content, image_urls,sub_titles,avatar): db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/AI_anchor?charset=utf8mb4') time_stamp = datetime.fromtimestamp(time.time()).strftime("%Y-%m-%d %H:%M:%S") txt_content_seperate_by_dot = '' for txt in text_content: txt_content_seperate_by_dot += txt+"," txt_content_seperate_by_dot = txt_content_seperate_by_dot[:-1] img_urls_seperate_by_dot = '' for iurl in image_urls: img_urls_seperate_by_dot += iurl+"," img_urls_seperate_by_dot = img_urls_seperate_by_dot[:-1] subtitles_seperate_by_dot = '' for sub in sub_titles: subtitles_seperate_by_dot += sub+"," subtitles_seperate_by_dot = subtitles_seperate_by_dot[:-1] db['video_queue'].insert({'name_hash':name_hash,'name':name,'text_content':txt_content_seperate_by_dot,'image_urls':img_urls_seperate_by_dot,'subtitles':subtitles_seperate_by_dot,'avatar':avatar,'timestamp':time_stamp}) while True: if first(db.query('SELECT * FROM video_queue_status'))['status'] == 1:#only one row in this table, which is the id 1 one print('another process running, leave loop') break if first(db.query('SELECT COUNT(1) FROM video_queue'))['COUNT(1)'] == 0: print('all finish, leave loop') break top1 = first(db.query('SELECT * FROM video_queue')) try: db.query('UPDATE video_queue_status SET status = 1;') c = rpyc.connect("localhost", 8878) c._config['sync_request_timeout'] = None remote_svc = c.root my_answer = remote_svc.call_video_eng(top1['name_hash'],top1['name'],top1['text_content'].split(','), top1['image_urls'].split(','),top1['subtitles'].split(','),top1['avatar']) # method call shutil.copy(tmp_video_dir+top1['name_hash']+'.mp4',video_dest+top1['name_hash']+'.mp4') os.remove(tmp_video_dir+top1['name_hash']+'.mp4') except: print('video generation error') notify_group('影片錯誤') db['video_queue'].delete(id=top1['id']) db.query('UPDATE video_queue_status SET status = 0') def gen_avatar(name_hash, imgurl): db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/AI_anchor?charset=utf8mb4') db['avatar_queue'].insert({'name_hash':name_hash,'imgurl':imgurl}) while True: statement = 'SELECT * FROM avatar_service_status'#only one row in this table, which is the id 1 one status = -1 for row in db.query(statement): status = row['status'] if status == 1: print('leave process loop') break statement = 'SELECT * FROM avatar_queue' works = [] for row in db.query(statement): works.append({'id':row['id'],'name_hash':row['name_hash'],'imgurl':row['imgurl']}) if len(works)==0: print('leave process loop') break try: statement = 'UPDATE avatar_service_status SET status = 1 WHERE id=1;' db.query(statement) name_hash = works[0]['name_hash'] imgurl = works[0]['imgurl'] c = rpyc.connect("localhost", 8868) c._config['sync_request_timeout'] = None remote_svc = c.root my_answer = remote_svc.call_avatar(name_hash,imgurl) # method call shutil.copy(tmp_avatar_dir+name_hash+'.mp4',avatar_dest+name_hash+'.mp4') os.remove(tmp_avatar_dir+name_hash+'.mp4') except: print('gen error') notify_group('無法辨識人臉') db['avatar_queue'].delete(id=works[0]['id']) statement = 'UPDATE avatar_service_status SET status = 0 WHERE id=1;' #only one row in this table, which id 1 one db.query(statement)