|
import os |
|
from abc import ABC |
|
|
|
from shortGPT.audio.voice_module import VoiceModule |
|
from shortGPT.config.languages import Language |
|
from shortGPT.config.path_utils import get_program_path |
|
from shortGPT.database.content_database import ContentDatabase |
|
|
|
CONTENT_DB = ContentDatabase() |
|
|
|
|
|
class AbstractContentEngine(ABC): |
|
def __init__(self, short_id: str, content_type: str, language: Language, voiceModule: VoiceModule): |
|
if short_id: |
|
self.dataManager = CONTENT_DB.getContentDataManager( |
|
short_id, content_type |
|
) |
|
else: |
|
self.dataManager = CONTENT_DB.createContentDataManager(content_type) |
|
self.id = str(self.dataManager._getId()) |
|
self.initializeMagickAndFFMPEG() |
|
self.prepareEditingPaths() |
|
self._db_language = language.value |
|
self.voiceModule = voiceModule |
|
self.stepDict = {} |
|
self.default_logger = lambda _: None |
|
self.logger = self.default_logger |
|
|
|
def __getattr__(self, name): |
|
if name.startswith('_db_'): |
|
db_path = name[4:] |
|
cache_attr = '_' + name |
|
if not hasattr(self, cache_attr): |
|
setattr(self, cache_attr, self.dataManager.get(db_path)) |
|
return getattr(self, cache_attr) |
|
else: |
|
return super().__getattr__(name) |
|
|
|
def __setattr__(self, name, value): |
|
if name.startswith('_db_'): |
|
db_path = name[4:] |
|
cache_attr = '_' + name |
|
setattr(self, cache_attr, value) |
|
self.dataManager.save(db_path, value) |
|
else: |
|
super().__setattr__(name, value) |
|
|
|
def prepareEditingPaths(self): |
|
self.dynamicAssetDir = f".editing_assets/{self.dataManager.contentType}_assets/{self.id}/" |
|
if not os.path.exists(self.dynamicAssetDir): |
|
os.makedirs(self.dynamicAssetDir) |
|
|
|
def verifyParameters(*args, **kargs): |
|
keys = list(kargs.keys()) |
|
for key in keys: |
|
if not kargs[key]: |
|
print(kargs) |
|
raise Exception(f"Parameter :{key} is null") |
|
|
|
def isShortDone(self): |
|
return self._db_ready_to_upload |
|
|
|
def makeContent(self): |
|
while (not self.isShortDone()): |
|
currentStep = self._db_last_completed_step + 1 |
|
if currentStep not in self.stepDict: |
|
raise Exception(f'Incorrect step {currentStep}') |
|
if self.stepDict[currentStep].__name__ == "_editAndRenderShort": |
|
yield currentStep, f'Current step ({currentStep} / {self.get_total_steps()}) : ' + "Preparing rendering assets..." |
|
else: |
|
yield currentStep, f'Current step ({currentStep} / {self.get_total_steps()}) : ' + self.stepDict[currentStep].__name__ |
|
if self.logger is not self.default_logger: |
|
print(f'Step {currentStep} {self.stepDict[currentStep].__name__}') |
|
self.stepDict[currentStep]() |
|
self._db_last_completed_step = currentStep |
|
|
|
def get_video_output_path(self): |
|
return self._db_video_path |
|
|
|
def get_total_steps(self): |
|
return len(self.stepDict) |
|
|
|
def set_logger(self, logger): |
|
self.logger = logger |
|
|
|
def initializeMagickAndFFMPEG(self): |
|
ffmpeg_path = get_program_path("ffmpeg") |
|
if not ffmpeg_path: |
|
raise Exception("FFmpeg, a program used for automated editing within ShortGPT was not found on your computer. Please go back to the README and follow the instructions to install FFMPEG") |
|
ffprobe_path = get_program_path("ffprobe") |
|
if not ffprobe_path: |
|
raise Exception("FFProbe, a dependecy of FFmpeg was not found. Please go back to the README and follow the instructions to install FFMPEG") |
|
convert_path = get_program_path("convert") |
|
if not convert_path: |
|
raise Exception("ImageMagick, a program required for making Captions with ShortGPT was not found on your computer. Please go back to the README and follow the instructions to install ImageMagick") |
|
|
