| 1 | import asyncio |
| 2 | from typing import List, Literal, Optional, Union |
| 3 | from PIL import Image |
| 4 | |
| 5 | from utils.video import download_video |
| 6 | |
| 7 | |
| 8 | class VideoOutput: |
| 9 | fmt: Literal["url", "bytes"] |
| 10 | ext: str = "mp4" |
| 11 | data: Union[str, bytes] |
| 12 | |
| 13 | def __init__( |
| 14 | self, |
| 15 | fmt: Literal["url", "bytes"], |
| 16 | ext: str, |
| 17 | data: Union[str, bytes], |
| 18 | ): |
| 19 | self.fmt = fmt |
| 20 | self.ext = ext |
| 21 | self.data = data |
| 22 | |
| 23 | def save_url(self, path: str) -> None: |
| 24 | """Download and save a video from a URL to the specified path. |
| 25 | |
| 26 | Args: |
| 27 | path (str): Path where the video will be saved. |
| 28 | """ |
| 29 | download_video(self.data, path) |
| 30 | |
| 31 | def save_bytes(self, path: str) -> None: |
| 32 | """Save a bytes object to the specified path. |
| 33 | |
| 34 | Args: |
| 35 | path (str): Path where the video will be saved. |
| 36 | """ |
| 37 | with open(path, 'wb') as f: |
| 38 | f.write(self.data) |
| 39 | |
| 40 | def save(self, path: str) -> None: |
| 41 | save_func = getattr(self, f"save_{self.fmt}") |
| 42 | save_func(path) |
| 43 | |
| 44 |