| | 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") |
| |
|
