Creating new audio to video converter
This commit is contained in:
Scott Idem
@@ -1,4 +1,4 @@
|
||||
import aiofiles, datetime, hashlib, mimetypes, os, pathlib, random, shutil, time
|
||||
import aiofiles, datetime, hashlib, mimetypes, os, pathlib, random, shutil, subprocess, shlex, tempfile, time
|
||||
from fastapi import APIRouter, Body, Depends, File, Form, Header, HTTPException, Query, Response, status, UploadFile
|
||||
from fastapi.responses import FileResponse, StreamingResponse
|
||||
# from fastapi.responses import StreamingResponse
|
||||
@@ -7,6 +7,8 @@ from fastapi.responses import FileResponse, StreamingResponse
|
||||
from pydantic import BaseModel, EmailStr, Field
|
||||
from typing import Dict, List, Optional, Set, Union
|
||||
from pdf2image import convert_from_path
|
||||
from wand.drawing import Drawing
|
||||
from wand.image import Image
|
||||
|
||||
from app.lib_general import log, logging, common_route_params, Common_Route_Params, common_route_params_min, Common_Route_Params_Min
|
||||
from app.config import settings
|
||||
@@ -1184,4 +1186,189 @@ async def convert_file(
|
||||
# log.debug(create_event_file_obj_result)
|
||||
|
||||
|
||||
# return file_info
|
||||
# return file_info
|
||||
|
||||
|
||||
# @router.post('/create_video')
|
||||
# async def create_video(
|
||||
# file: UploadFile = File(...),
|
||||
# presentation_name: str = Form(...),
|
||||
# speaker_name: str = Form(...)
|
||||
# ):
|
||||
# # Save the uploaded audio file to a temporary directory
|
||||
# with open(f'/tmp/{file.filename}', 'wb') as f:
|
||||
# f.write(await file.read())
|
||||
|
||||
# # Generate a static image using the presentation name and speaker name
|
||||
# image_name = f'{presentation_name}_{speaker_name}.jpg'
|
||||
# cmd = f"convert -size 1280x720 xc:transparent -gravity center -pointsize 72 -fill black -annotate 0 '{presentation_name}\n{speaker_name}' /tmp/{image_name}"
|
||||
# args = shlex.split(cmd)
|
||||
# try:
|
||||
# subprocess.run(args, check=True)
|
||||
# except subprocess.CalledProcessError:
|
||||
# return {"success": False}
|
||||
|
||||
# # Run the ffmpeg command to create a video file with the audio file and static image
|
||||
# video_name = f'{presentation_name}_{speaker_name}.mp4'
|
||||
# cmd = f"ffmpeg -loop 1 -i /tmp/{image_name} -i /tmp/{file.filename} -c:a copy -c:v libx264 -shortest /tmp/{video_name}"
|
||||
# args = shlex.split(cmd)
|
||||
# try:
|
||||
# subprocess.run(args, check=True)
|
||||
# except subprocess.CalledProcessError:
|
||||
# return {"success": False}
|
||||
|
||||
# # Return a JSON response indicating success
|
||||
# return {"success": True}
|
||||
|
||||
|
||||
|
||||
|
||||
def run_ffmpeg(cmd):
|
||||
"""Runs an ffmpeg command in a non-blocking way.
|
||||
|
||||
Args:
|
||||
cmd: The ffmpeg command to run.
|
||||
|
||||
Returns:
|
||||
A Popen object representing the ffmpeg process.
|
||||
"""
|
||||
|
||||
args = shlex.split(cmd)
|
||||
|
||||
# process = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||||
# return process
|
||||
|
||||
try:
|
||||
subprocess.run(args, check=True, capture_output=True, text=True)
|
||||
# subprocess.run(args, check=True, capture_output=False, text=True)
|
||||
# subprocess.run(args, check=True, capture_output=True, text=True, stdin=subprocess.PIPE)
|
||||
# subprocess.run(args)
|
||||
log.debug(result.stdout)
|
||||
|
||||
# subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||||
# log.debug(result.stdout)
|
||||
except subprocess.CalledProcessError as e:
|
||||
log.exception('Error running ffmpeg command')
|
||||
return {'success': False, 'status_message': f'Error running ffmpeg command: {e}'}
|
||||
|
||||
return True
|
||||
|
||||
|
||||
@router.post('/create_video')
|
||||
async def create_video(
|
||||
file: UploadFile = File(...),
|
||||
presentation_name: str = Form(...),
|
||||
speaker_name: str = Form(...)
|
||||
):
|
||||
log.setLevel(logging.DEBUG)
|
||||
log.debug(locals())
|
||||
|
||||
# cmd = f"rm /tmp/*.jpg"
|
||||
cmd = f"ffmpeg -version"
|
||||
args = shlex.split(cmd)
|
||||
try:
|
||||
result = subprocess.run(args, check=True, capture_output=True, text=True)
|
||||
log.debug(result.stdout)
|
||||
except subprocess.CalledProcessError as e:
|
||||
log.exception('Error running test command')
|
||||
return {'success': False, 'status_message': f'Error running test command: {e}'}
|
||||
|
||||
# cmd = f"convert --version"
|
||||
cmd = f"convert -list font"
|
||||
args = shlex.split(cmd)
|
||||
try:
|
||||
result = subprocess.run(args, check=True, capture_output=True, text=True)
|
||||
log.debug(result.stdout)
|
||||
except subprocess.CalledProcessError as e:
|
||||
log.exception('Error running ImageMagick convert command')
|
||||
return {'success': False, 'status_message': f'Error running ImageMagick convert command: {e}'}
|
||||
|
||||
# Save the uploaded audio file to a temporary directory
|
||||
with tempfile.NamedTemporaryFile(delete=False, suffix='.mp3') as audio_file:
|
||||
audio_file.write(await file.read())
|
||||
audio_file_path = audio_file.name
|
||||
|
||||
# Generate a static image using the presentation name and speaker name
|
||||
with tempfile.NamedTemporaryFile(delete=False, suffix='.jpg') as image_file:
|
||||
image_name = image_file.name
|
||||
log.debug(image_name)
|
||||
|
||||
with Image(width=1280, height=720, background='transparent') as img:
|
||||
with Drawing() as ctx:
|
||||
ctx.font_family = 'DejaVu Sans' # DejaVu Sans, DejaVu Sans Mono, DejaVu Serif
|
||||
# ctx.font = 'Times New Roman' # DejaVu-Sans, DejaVu-Sans-Bold, DejaVu-Sans-Mono, DejaVu-Sans-Mono-Bold, DejaVu-Serif
|
||||
ctx.font_size = 64
|
||||
ctx.text_alignment = 'center'
|
||||
ctx.text_decoration = 'underline'
|
||||
ctx.text_kerning = -1
|
||||
ctx.fill_color = 'darkblue'
|
||||
ctx.stroke_color = 'gray'
|
||||
img.annotate(f'{presentation_name}\n{speaker_name}', ctx, left=50, baseline=0, angle=0)
|
||||
# gravity='center'
|
||||
# , font_size=72, fill='green'
|
||||
img.save(filename=image_name)
|
||||
|
||||
|
||||
|
||||
# cmd = f"convert -size 1280x720 xc:transparent -gravity center -pointsize 72 -fill green -annotate 0 '{presentation_name}\n{speaker_name}' {image_name}"
|
||||
# cmd = f"convert -size 1280x720 xc:transparent -annotate 0 'Hello World' {image_name}"
|
||||
# cmd = f"convert -size 1280x720 -background lightblue -fill darkblue -pointsize 72 label:{presentation_name} {image_name}"
|
||||
# cmd = f"convert -size 1280x720 -background lightblue -fill darkblue label:'Testing' {image_name}"
|
||||
# cmd = f"convert -size 1280x720 {image_name}"
|
||||
# args = shlex.split(cmd)
|
||||
# try:
|
||||
# subprocess.run(args, check=True)
|
||||
# except subprocess.CalledProcessError as e:
|
||||
# log.exception('Error running convert command')
|
||||
# return {'success': False, 'status_message': f'Error running convert command: {e}'}
|
||||
|
||||
cmd = f"ls -lha /tmp"
|
||||
args = shlex.split(cmd)
|
||||
try:
|
||||
result = subprocess.run(args, check=True, capture_output=True, text=True)
|
||||
log.debug(result.stdout)
|
||||
except subprocess.CalledProcessError as e:
|
||||
log.exception('Error running ls command')
|
||||
return {'success': False, 'status_message': f'Error running ls command: {e}'}
|
||||
|
||||
# Run the ffmpeg command to create a video file with the audio file and static image
|
||||
log.info('Run the ffmpeg command to create a video file with the audio file and static image')
|
||||
|
||||
with tempfile.NamedTemporaryFile(delete=False, suffix='.mp4') as video_file:
|
||||
video_name = video_file.name
|
||||
# NOTE: It seems very important that the -y argument is used with ffmpeg run by subprocess.run(). Otherwise the process will hang.
|
||||
# cmd = f"ffmpeg -nostdin -loglevel error -loop 1 -i {image_name} -i {audio_file_path} -c:a copy -c:v libx264 -shortest {video_name} &"
|
||||
cmd = f"ffmpeg -hide_banner -loglevel error -nostats -y -loop 1 -i {image_name} -i {audio_file_path} -c:a copy -c:v libx264 -shortest {video_name}"
|
||||
# cmd = f"ffmpeg -hide_banner -i {audio_file_path} -c:a aac {video_name}"
|
||||
log.debug(cmd)
|
||||
|
||||
# Run the ffmpeg command
|
||||
# process = await run_ffmpeg(cmd)
|
||||
|
||||
# # Monitor the status of the ffmpeg command
|
||||
# while process.poll() is None:
|
||||
# # Read the ffmpeg output
|
||||
# stdout, stderr = process.communicate(timeout=1)
|
||||
|
||||
# # Log the ffmpeg output
|
||||
# logging.debug(stdout)
|
||||
# logging.debug(stderr)
|
||||
|
||||
# # Wait for the ffmpeg command to finish
|
||||
# process.wait()
|
||||
|
||||
|
||||
|
||||
args = shlex.split(cmd)
|
||||
try:
|
||||
subprocess.run(args, check=True, capture_output=True, text=True)
|
||||
log.debug(result.stdout)
|
||||
|
||||
# subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||||
# log.debug(result.stdout)
|
||||
except subprocess.CalledProcessError as e:
|
||||
log.exception('Error running ffmpeg command')
|
||||
return {'success': False, 'status_message': f'Error running ffmpeg command: {e}'}
|
||||
|
||||
# Return the new mp4 file for download
|
||||
return FileResponse(video_name, media_type='video/mp4', filename=f'{presentation_name}_{speaker_name}.mp4')
|
||||
|
||||
Reference in New Issue
Block a user