"""dpg — entry point. Subcommands: generate ingest → PRD → backend → frontend edit agentic edit on an existing run Examples: uv run python main.py generate --prompt "A warehouse management system" uv run python main.py generate --prompt "..." --images-dir 01-specs/images/ uv run python main.py edit uv run python main.py edit --prompt "make the header blue" """ from __future__ import annotations import argparse import asyncio import json import logging import os import sys from pathlib import Path # ── Path setup ──────────────────────────────────────────────────────────────── ROOT_DIR = Path(__file__).resolve().parent MASTER_PIPELINE_DIR = ROOT_DIR / "pipeline" / "master-pipeline" / "pipeline" STEP01_PIPELINE_DIR = ROOT_DIR / "pipeline" / "step-01-input-ingestion" / "pipeline" for _p in (ROOT_DIR, MASTER_PIPELINE_DIR, STEP01_PIPELINE_DIR): sp = str(_p) if sp not in sys.path: sys.path.insert(0, sp) # Load .env early so AWS / Bedrock model IDs are visible. try: from dotenv import load_dotenv # type: ignore load_dotenv(ROOT_DIR / ".env", override=True) except ImportError: pass # ── Logging ─────────────────────────────────────────────────────────────────── from shared.config import ( # noqa: E402 DEFAULT_API_BASE_URL, LOG_DATE_FORMAT, LOG_FORMAT, LOG_LEVEL_DEFAULT, ) logging.basicConfig( level=LOG_LEVEL_DEFAULT, format=LOG_FORMAT, datefmt=LOG_DATE_FORMAT, ) _DIVIDER = "─" * 55 # ══════════════════════════════════════════════════════════════════════════════ # GENERATE # ══════════════════════════════════════════════════════════════════════════════ def cmd_generate(args: argparse.Namespace) -> int: from orchestrator import generate_project from shared.config import RUNS_DIR output_dir = Path(args.output).expanduser().resolve() if args.output else RUNS_DIR output_dir.mkdir(parents=True, exist_ok=True) if args.file: from file_to_images import prepare_file images_dir = prepare_file(Path(args.file).expanduser().resolve()) elif args.images_dir: images_dir = Path(args.images_dir).expanduser().resolve() else: images_dir = None db_info: dict | None = None if args.db_info: try: db_info = json.loads(args.db_info) except json.JSONDecodeError as exc: print(f"\nERROR: --db-info is not valid JSON: {exc}", file=sys.stderr) return 2 # Prepend domain context to the prompt so downstream LLM steps get the full picture. user_prompt = args.prompt context_lines: list[str] = [] if args.domain: context_lines.append(f"Domain: {args.domain}") if args.domain_url: context_lines.append(f"Domain URL: {args.domain_url}") if context_lines and user_prompt: user_prompt = "\n".join(context_lines) + "\n\n" + user_prompt elif context_lines: user_prompt = "\n".join(context_lines) print(_DIVIDER) print(" Pipeline: ingest → PRD → backend → frontend") print(_DIVIDER) print(f" Inputs dir : {args.inputs_dir or ROOT_DIR / '01-specs'}") print(f" Output : {output_dir}") if args.name: print(f" Name : {args.name}") if args.domain: print(f" Domain : {args.domain!r}") if args.domain_url: print(f" Domain URL : {args.domain_url!r}") if db_info: print(f" DB host : {db_info.get('host')} db={db_info.get('database')}") if user_prompt: print(f" Prompt : {user_prompt!r}") print(_DIVIDER) try: project_root = generate_project( user_prompt=user_prompt, images_dir=images_dir, inputs_dir=None, output_dir=output_dir, project_name=args.name, api_base_url=args.api_base_url, run_id=args.run_id, domain_url=args.domain_url, db_info=db_info, ) except (FileNotFoundError, ValueError) as exc: print(f"\nERROR: {exc}", file=sys.stderr) return 2 except KeyboardInterrupt: print("\nInterrupted.", file=sys.stderr) return 130 print() print(_DIVIDER) print(f" Done. Project root: {project_root}") print(_DIVIDER) print(" Backend:") print(f" cd {project_root / 'backend'} && uv sync && uv run uvicorn main:app --reload --port 8000") print(" Frontend:") print(f" cd {project_root / 'frontend'} && npm install && npm run dev") return 0 # ══════════════════════════════════════════════════════════════════════════════ # EDIT # ══════════════════════════════════════════════════════════════════════════════ # def _resolve_edit_run_dir(run_id: str | None, run_dir_str: str | None) -> Path: # from shared.config import RUNS_DIR # if run_dir_str: # p = Path(run_dir_str).expanduser() # p = p if p.is_absolute() else ROOT_DIR / p # if not p.exists(): # raise FileNotFoundError(f"Run directory not found: {p}") # return p # if run_id: # cand = RUNS_DIR / run_id # if cand.exists(): # return cand # raise FileNotFoundError(f"Run not found: {run_id}") # if RUNS_DIR.exists(): # candidates = [ # d for d in RUNS_DIR.iterdir() # if d.is_dir() and (d / "frontend" / "src" / "App.tsx").exists() # ] # if candidates: # latest = max(candidates, key=lambda d: d.stat().st_mtime) # print(f" Using latest project: {latest.name}", flush=True) # return latest # raise FileNotFoundError("No runs found. Run `python main.py generate` first.") # def cmd_edit(args: argparse.Namespace) -> int: # if args.model: # os.environ["AGENT_EDIT_MODEL"] = args.model # try: # run_dir = _resolve_edit_run_dir(args.run_id, args.run_dir) # except FileNotFoundError as exc: # print(f"ERROR: {exc}", file=sys.stderr) # return 1 # # The edit pipeline operates on the React source tree. # edit_run_dir = run_dir / "frontend" / "src" # if not edit_run_dir.exists(): # print(f"ERROR: frontend/src not found in {run_dir}", file=sys.stderr) # return 1 # from edit import run_agentic_edit_pipeline, store as session_store # async def _run_one(request: str) -> None: # if not session_store._ok: # await session_store.initialize() # print(flush=True) # async for event in run_agentic_edit_pipeline(request_text=request, run_dir=edit_run_dir): # etype = event.get("type") # if etype == "progress": # print(f" {event.get('message', '')}", flush=True) # elif etype == "page_done": # print( # f" [page_done] {event.get('component_name')} ({event.get('page_id')})", # flush=True, # ) # elif etype == "done": # print(f"\n Done — {event.get('summary', '')}", flush=True) # edited = event.get("edited_files", []) # if edited: # print(f" Files: {', '.join(edited)}", flush=True) # elif etype == "error": # print(f"\n ERROR: {event.get('message', '')}", file=sys.stderr, flush=True) # print(f"\n{_DIVIDER}") # print(f" Agentic edit mode | run: {run_dir.name}") # print(f' Type an edit request, or "exit" to quit.') # print(_DIVIDER) # pending = args.prompt # rc = 0 # while True: # if pending is not None: # raw = pending # pending = None # print(f"\n[agent] > {raw}") # else: # try: # raw = input("\n[agent] > ").strip() # except (EOFError, KeyboardInterrupt): # print("\n Bye!") # break # if not raw: # continue # if raw.lower() in ("exit", "quit"): # print(" Bye!") # break # try: # asyncio.run(_run_one(raw)) # except Exception as exc: # print(f"\n ERROR: {exc}", file=sys.stderr) # rc = 1 # try: # if session_store._ok: # asyncio.run(session_store.shutdown()) # except Exception: # pass # return rc # ══════════════════════════════════════════════════════════════════════════════ # CLI # ══════════════════════════════════════════════════════════════════════════════ def _build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="Dynamic page generation pipeline.", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__, ) parser.add_argument("--log-level", default=LOG_LEVEL_DEFAULT) sub = parser.add_subparsers(dest="command", required=True, metavar="{generate,edit}") pg = sub.add_parser("generate", help="Run the full ingest → PRD (from images) → backend → frontend pipeline.") pg.add_argument("--images-dir", default=None) pg.add_argument("--inputs-dir", default=None, help="Directory with ddl/, images/ subfolders (default: 01-specs/).") pg.add_argument("--output", default=None, help="Output directory (default: runs/outputs/).") pg.add_argument("--name", default=None) pg.add_argument("--prompt", default=None) pg.add_argument("--domain", default=None, help="Business domain (e.g. 'Healthcare', 'Finance').") pg.add_argument("--domain-url", default=None, help="Website/URL domain (e.g. 'myapp.com').") pg.add_argument("--db-info", default=None, help="JSON string with DB credentials from domain-info API.") pg.add_argument("--file", default=None, help="Single image, PDF, or PPTX file to use as visual input.") pg.add_argument("--api-base-url", default=DEFAULT_API_BASE_URL) pg.add_argument("--run-id", default=None) pe = sub.add_parser("edit", help="Edit an existing generated run.") pe.add_argument("--prompt", default=None) pe.add_argument("--model", default=None) pe.add_argument("--run-id", default=None) pe.add_argument("--run-dir", default=None) return parser def main(argv: list[str] | None = None) -> int: args = _build_parser().parse_args(argv) logging.getLogger().setLevel(args.log_level) if args.command == "generate": return cmd_generate(args) # if args.command == "edit": # return cmd_edit(args) return 2 if __name__ == "__main__": raise SystemExit(main())