@@ -11,7 +11,8 @@ from django.core.files import File from replicate.exceptions import ModelError from messages.models import Message -from ml_model.exceptions import RequestBlocked, GenerationException +from ml_model.exceptions import RequestBlocked, GenerationException, FileExtensionNotSupported +from ml_model.services.FileService import FileProcessingService from ml_model.services.base import SimpleService from ml_model.tasks import replicate_run @@ -20,20 +21,23 @@ from payments.selectors.payment_plan_selector import PaymentPlanSelector class Seedance(SimpleService): - TOKENS_COST = { 'seedance-2.0': { - '480p': Decimal('21'), - # '480p_video': Decimal('39'), - '720p': Decimal('51'), -# '720p_video': Decimal('87') - }, # 1 second + 'non_video_in': { + '480p': Decimal('21'), + '720p': Decimal('51') + } + }, 'seedance-2.0-fast': { - '480p': Decimal('18'), -# '480p_video': Decimal('33'), - '720p': Decimal('39'), -# '720p_video': Decimal('66') - }, # 1 second + 'non_video_in': { + '480p': Decimal('18'), + '720p': Decimal('39') + }, + 'video_in': { + '480p': Decimal('33'), + '720p': Decimal('66') + } # 1 second + } } @classmethod @@ -41,11 +45,12 @@ class Seedance(SimpleService): resolution = info['resolution'] duration = info['duration'] version = info['version'] - price = cls.TOKENS_COST[version][resolution] * duration + generation_type = 'video_in' if file_exists and version == 'seedance-2.0-fast' else 'non_video_in' + price = cls.TOKENS_COST[version][generation_type][resolution] * duration return price.quantize(Decimal('0.1'), rounding='ROUND_UP') - def calculate_price(self, resolution: str, duration: int, version: str) -> Decimal: - price = self.TOKENS_COST[version][resolution] * duration + def calculate_price(self, resolution: str, duration: int, version: str, generation_type: str) -> Decimal: + price = self.TOKENS_COST[version][generation_type][resolution] * duration return price.quantize(Decimal('0.1'), rounding='ROUND_UP') def save_results(self, content: str, t: timedelta, video: str, save: bool = True) -> list[Message]: @@ -63,11 +68,31 @@ class Seedance(SimpleService): version = input_message.info.pop('version', 'seedance-2.0') resolution = input_message.info.get('resolution', '720p') duration = input_message.info.get('duration', 5) - if (balance := PaymentPlanSelector(self.store.user).get_current_balance()) < (cost := self.TOKENS_COST[version][resolution] * duration): + file = input_message.file or None + file_extension = None + generation_type = 'non_video_in' + if file: + file_bytes = file.read() + kind = filetype.guess(file_bytes[:50]) + file_extension = FileProcessingService.get_file_extension(kind.extension, file_bytes) if kind else None + available_extensions = ('JPG', 'JPEG', 'PNG', 'WEBP', 'MP4') + if not file_extension or file_extension.upper() not in available_extensions: + raise FileExtensionNotSupported(available_extensions) + elif file_extension.upper() == 'MP4': + generation_type = 'video_in' + if version == 'seedance-2.0' and generation_type == 'video_in': + available_extensions = ('JPG', 'JPEG', 'PNG', 'WEBP') + raise FileExtensionNotSupported(available_extensions) + if (balance := PaymentPlanSelector(self.store.user).get_current_balance()) < ( + cost := self.TOKENS_COST[version][generation_type][resolution] * duration): raise InsufficientBalance(balance, cost) callback_data = dict({'prompt': input_message.content, **input_message.info}) - if image := input_message.file: - callback_data.update({'image': image.url}) + if file: + reference_type = ( + 'videos' if file_extension.upper() == 'MP4' + else 'images' + ) + callback_data.update({f'reference_{reference_type}': [file.url]}) start_time = time.time() try: video = replicate_run( @@ -78,6 +103,6 @@ class Seedance(SimpleService): raise RequestBlocked raise GenerationException from exc process_time = timedelta(seconds=(time.time() - start_time)) - self.handle_invoice(input_message.content_object.model, resolution=resolution, duration=duration, version=version) + self.handle_invoice(input_message.content_object.model, resolution=resolution, duration=duration, version=version, generation_type=generation_type) msgs = self.save_results(input_message.content, process_time, video, save) return msgs