File size: 6,323 Bytes
e454383 32ca700 c1bb824 e454383 32ca700 e454383 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 | import os
import sys
import subprocess
from pathlib import Path
from typing import List
from dotenv import load_dotenv
ROOT = Path(__file__).resolve().parents[2]
load_dotenv(str(Path(__file__).resolve().parents[1] / ".env"))
sys.path.append(str(Path(__file__).resolve().parents[1])) # Add parent directory to path
import typer
try:
import questionary # type: ignore
except ImportError:
questionary = None # Fallback if not installed
from docker_builder import build_dockerfile, write_dockerfile
from graphrag_builder import build_graphrag
from hf_deployer import deploy_space
from vector_db_builder import build_vector_db_for_dir
app = typer.Typer(help="Interactive build & deploy tool")
BACKEND_DIR = ROOT / "_backend"
DATA_DIR = BACKEND_DIR / "data"
DEFAULT_DOCKERFILE = ROOT / "Dockerfile.generated"
def list_data_files() -> List[Path]:
# Listing all the directories in the data directory
folders = [p for p in DATA_DIR.iterdir() if p.is_dir()]
if not folders:
return []
return sorted(folders)
def run_command(cmd: List[str], cwd: Path | None = None):
typer.echo(f"Running: {' '.join(cmd)}")
res = subprocess.run(cmd, cwd=cwd, text=True)
if res.returncode != 0:
raise typer.Exit(code=res.returncode)
def ensure_questionary():
if questionary is None:
typer.echo("questionary not installed. Please add it to requirements and install.")
raise typer.Exit(code=1)
@app.command()
def interactive(
graphrag: bool = typer.Option(False, help="Build GraphRAG artifacts"),
deploy: bool = typer.Option(False, help="Deploy to Hugging Face after build"),
dockerfile_out: Path = typer.Option(DEFAULT_DOCKERFILE, help="Path for generated Dockerfile"),
non_interactive: bool = typer.Option(False, help="Skip confirmations where possible")
):
"""Run the interactive (or semi-interactive) build workflow."""
if not non_interactive:
ensure_questionary()
files = list_data_files()
if not files:
typer.echo("No data files found under _backend/data.")
raise typer.Exit(code=1)
if non_interactive:
selected_paths = files # take all
else:
choices = [f.relative_to(DATA_DIR).as_posix() for f in files]
selected_rel = questionary.checkbox(
"Select data files to include:", choices=choices
).ask()
if not selected_rel:
typer.echo("No files selected.")
raise typer.Exit()
selected_paths = [DATA_DIR / r for r in selected_rel]
# Optional vector DB build using new utility
do_reset = True if non_interactive else questionary.confirm(
"Build/reset vector database(s) for selected data directories?"
).ask()
if do_reset:
for data_path in selected_paths:
# Prompt for file suffix and collection name
if non_interactive:
file_suffix = None
collection_name = data_path.name
else:
file_suffix = questionary.text(
f"File suffix to include for {data_path.name} (e.g. .md, .srt, leave blank for all):",
default=""
).ask()
file_suffix = file_suffix if file_suffix.strip() else None
collection_name = questionary.text(
f"Collection name for {data_path.name}:",
default=data_path.name
).ask()
build_vector_db_for_dir(
data_dir=str(data_path),
collection_name=collection_name,
file_suffix=file_suffix,
verbose=True
)
graphrag_artifact_paths: List[str] = []
build_graphrag_flag = graphrag or (False if non_interactive else questionary.confirm(
"Build GraphRAG artifacts?"
).ask())
if build_graphrag_flag:
out_dir = BACKEND_DIR / "graphrag_artifacts"
build_graphrag(selected_paths, out_dir)
graphrag_artifact_paths.append(str(out_dir.relative_to(ROOT)))
frontend_dir = Path(os.getenv("FRONTEND_DIR", "frontend"))
build_cmd = os.getenv("FRONTEND_BUILD_CMD", "npm run build")
build_out = os.getenv("FRONTEND_BUILD_OUTPUT", "dist")
prebuild_frontend = True if non_interactive else questionary.confirm(
"Pre-build frontend locally first?"
).ask()
if prebuild_frontend:
if frontend_dir.exists():
run_command(build_cmd.split(), cwd=ROOT / frontend_dir)
else:
typer.echo(f"Frontend directory {frontend_dir} not found; skipping pre-build.")
docker_content = build_dockerfile(
frontend_build_cmd=build_cmd,
frontend_build_output=build_out,
graphrag_artifact_paths=graphrag_artifact_paths or None,
)
write_dockerfile(dockerfile_out, docker_content)
typer.echo(f"Wrote Dockerfile to {dockerfile_out}")
build_image = True if non_interactive else questionary.confirm(
"Build Docker image locally?"
).ask()
if build_image:
tag = os.getenv("IMAGE_TAG", "interview-synth:latest") if non_interactive else questionary.text(
"Image tag:", default="interview-synth:latest"
).ask()
run_command(["docker", "build", "-f", str(dockerfile_out), "-t", tag, "."], cwd=ROOT)
do_deploy = deploy or (False if non_interactive else questionary.confirm(
"Deploy to Hugging Face Spaces?"
).ask())
if do_deploy:
space_id = os.getenv("HF_SPACE_ID") or ("" if non_interactive else questionary.text(
"Enter HF Space (user/space):"
).ask())
token = os.getenv("HF_TOKEN") or ("" if non_interactive else questionary.password(
"Enter Hugging Face token:"
).ask())
if not space_id or not token:
typer.echo("Space ID or token missing; aborting deploy.")
else:
# Write the generated Dockerfile to project root for Spaces
(ROOT / "Dockerfile").write_text(docker_content, encoding="utf-8")
deploy_space(space_id, ROOT, token)
typer.echo("Done.")
if __name__ == "__main__":
app()
|