Spaces:
Sleeping
Sleeping
| import __init_lib_path | |
| import logging | |
| from yaml import Loader, Dumper, load, dump | |
| from src.task import Task | |
| import openai | |
| import argparse | |
| import os | |
| from pathlib import Path | |
| from datetime import datetime | |
| import shutil | |
| from uuid import uuid4 | |
| def parse_args(): | |
| parser = argparse.ArgumentParser() | |
| parser.add_argument("--link", help="youtube video link here", default=None, type=str, required=False) | |
| parser.add_argument("--video_file", help="local video path here", default=None, type=str, required=False) | |
| parser.add_argument("--audio_file", help="local audio path here", default=None, type=str, required=False) | |
| parser.add_argument("--srt_file", help="srt file input path here", default=None, type=str, required=False) | |
| parser.add_argument("--continue", help="task_id that need to continue", default=None, type=str, required=False) # need implement | |
| parser.add_argument("--launch_cfg", help="launch config path", default='./configs/local_launch.yaml', type=str, required=False) | |
| # task config | |
| args = parser.parse_args() | |
| return args | |
| if __name__ == "__main__": | |
| # read args and configs | |
| args = parse_args() | |
| launch_cfg = load(open(args.launch_cfg), Loader=Loader) | |
| # TODO: task config | |
| # initialize dir | |
| local_dir = Path(launch_cfg['local_dump']) | |
| if not local_dir.exists(): | |
| local_dir.mkdir(parents=False, exist_ok=False) | |
| # get task id | |
| task_id = str(uuid4()) | |
| # create locak dir for the task | |
| task_dir = local_dir.joinpath(f"task_{task_id}") | |
| task_dir.mkdir(parents=False, exist_ok=False) | |
| task_dir.joinpath("results").mkdir(parents=False, exist_ok=False) | |
| task_dir.joinpath("logs").mkdir(parents=False, exist_ok=False) | |
| # logging | |
| logging.basicConfig(level=logging.INFO, handlers=[ | |
| logging.FileHandler( | |
| "{}/{}_{}.log".format(task_dir, f"task_{task_id}", datetime.now().strftime("%m%d%Y_%H%M%S")), | |
| 'w', encoding='utf-8')]) | |
| # TODO: write task info into log | |
| # Task create | |
| if args.link is not None: | |
| try: | |
| task = Task.fromYoutubeLink(args.link, task_id, launch_cfg) | |
| except: | |
| shutil.rmtree(task_dir) | |
| raise RuntimeError("failed to create task from youtube link") | |
| # add task to the status queue | |
| task.run_pipeline() | |