|
|
import os |
|
|
import shutil |
|
|
from typing import Any |
|
|
|
|
|
import click |
|
|
|
|
|
from taskweaver.cli.util import CliContext |
|
|
|
|
|
|
|
|
def validate_empty_workspace(ctx: click.Context, param: Any, value: Any) -> str: |
|
|
ctx_obj: CliContext = ctx.obj |
|
|
value = ( |
|
|
value if value is not None else ctx_obj.workspace_param if ctx_obj.workspace_param is not None else os.getcwd() |
|
|
) |
|
|
value_str = str(value) |
|
|
is_cur_empty: bool = not os.path.exists(value) or (os.path.isdir(value) and len(os.listdir(value_str)) == 0) |
|
|
if ctx_obj.is_workspace_valid: |
|
|
if value == ctx_obj.workspace: |
|
|
click.echo( |
|
|
"The current directory has already been initialized. No need to do it again.", |
|
|
) |
|
|
else: |
|
|
click.echo( |
|
|
"The current directory is under a configured workspace.", |
|
|
) |
|
|
ctx.exit(1) |
|
|
if not is_cur_empty: |
|
|
click.echo( |
|
|
f"The directory {click.format_filename(value)} is not empty. " |
|
|
"Please change the working directory to an empty directory for initializing a new workspace. " |
|
|
"Refer to --help for more information.", |
|
|
) |
|
|
ctx.exit(1) |
|
|
return value |
|
|
|
|
|
|
|
|
@click.command(short_help="Initialize TaskWeaver project") |
|
|
@click.pass_context |
|
|
@click.option( |
|
|
"--project", |
|
|
"-p", |
|
|
type=click.Path(file_okay=False, dir_okay=True, resolve_path=True), |
|
|
required=False, |
|
|
default=None, |
|
|
is_eager=True, |
|
|
callback=validate_empty_workspace, |
|
|
) |
|
|
def init( |
|
|
ctx: click.Context, |
|
|
project: str, |
|
|
): |
|
|
"""Initialize TaskWeaver environment""" |
|
|
click.echo( |
|
|
f"Initializing TaskWeaver in directory {project}...", |
|
|
) |
|
|
if not os.path.exists(project): |
|
|
os.mkdir(project) |
|
|
|
|
|
def get_dir(*dir: str): |
|
|
return os.path.join(project, *dir) |
|
|
|
|
|
dir_list = [ |
|
|
"codeinterpreter_examples", |
|
|
"planner_examples", |
|
|
"plugins", |
|
|
"config", |
|
|
"workspace", |
|
|
] |
|
|
for dir in dir_list: |
|
|
dir_path = get_dir(dir) |
|
|
if not os.path.exists(dir_path): |
|
|
os.mkdir(dir_path) |
|
|
|
|
|
init_temp_dir = get_dir("init") |
|
|
import zipfile |
|
|
from pathlib import Path |
|
|
|
|
|
tpl_dir = os.path.join(init_temp_dir, "template") |
|
|
ext_zip_file = Path(__file__).parent / "taskweaver-ext.zip" |
|
|
if os.path.exists(ext_zip_file): |
|
|
with zipfile.ZipFile(ext_zip_file, "r") as zip_ref: |
|
|
|
|
|
zip_ref.extractall(tpl_dir) |
|
|
|
|
|
tpl_planner_example_dir = os.path.join(tpl_dir, "taskweaver-ext", "planner_examples") |
|
|
tpl_ci_example_dir = os.path.join(tpl_dir, "taskweaver-ext", "codeinterpreter_examples") |
|
|
tpl_plugin_dir = os.path.join(tpl_dir, "taskweaver-ext", "plugins") |
|
|
tpl_config_dir = os.path.join(tpl_dir, "taskweaver-ext") |
|
|
planner_example_dir = get_dir("planner_examples") |
|
|
ci_example_dir = get_dir("codeinterpreter_examples") |
|
|
plugin_dir = get_dir("plugins") |
|
|
copy_files(tpl_planner_example_dir, planner_example_dir) |
|
|
copy_files(tpl_ci_example_dir, ci_example_dir) |
|
|
copy_files(tpl_plugin_dir, plugin_dir) |
|
|
copy_file(tpl_config_dir, "taskweaver_config.json", get_dir("")) |
|
|
|
|
|
try: |
|
|
shutil.rmtree(init_temp_dir) |
|
|
except Exception: |
|
|
click.secho("Failed to remove temporary directory", fg="yellow") |
|
|
click.secho( |
|
|
f"TaskWeaver project has been initialized successfully at {click.format_filename(project)}.", |
|
|
fg="green", |
|
|
) |
|
|
|
|
|
|
|
|
def copy_files(src_dir: str, dst_dir: str): |
|
|
|
|
|
files = os.listdir(src_dir) |
|
|
|
|
|
|
|
|
for file in files: |
|
|
if os.path.isfile(os.path.join(src_dir, file)): |
|
|
copy_file(src_dir, file, dst_dir) |
|
|
|
|
|
|
|
|
def copy_file(src_dir: str, filename: str, dst_dir: str): |
|
|
shutil.copy( |
|
|
os.path.join(src_dir, filename), |
|
|
os.path.join(dst_dir, filename), |
|
|
) |
|
|
|
