import os import uuid from typing import List, Dict, Optional from pydantic import BaseModel from langgraph.graph import StateGraph from .document_parser import parse_documents from .requirements_extractor import extract_requirements from .ui_generator import generate_ui_html # ========================================================== # 🧱 Define the State Schema (Pydantic model) # ========================================================== class PipelineState(BaseModel): files: Optional[List[str]] = None text: Optional[str] = None requirements: Optional[Dict] = None html: Optional[str] = None public_url: Optional[str] = None # ========================================================== # 📄 Node 1 — Document Parser # ========================================================== async def node_parse_docs(state: PipelineState): print("📄 [Document Parser] Extracting text...") result = await parse_documents(state.files) # handle both string or dict outputs if isinstance(result, dict): text = result.get("text", "").strip() else: text = str(result).strip() if not text: raise ValueError("No text extracted from documents.") return {"text": text} # ========================================================== # 🧠 Node 2 — Requirements Extractor # ========================================================== async def node_requirements(state: PipelineState, config=None) -> Dict: print("🧩 [Requirements Agent] Extracting structured info...") if not state.text: raise ValueError("Missing 'text' field in state for requirement extraction.") requirements = await extract_requirements(state.text) if not requirements: raise ValueError("Requirements extraction returned empty.") return {"requirements": requirements} # ========================================================== # 🎨 Node 3 — UI Generator # ========================================================== async def node_ui_generator(state: PipelineState, config=None) -> Dict: print("🎨 [UI Generator] Creating PowerApps-style interface...") if not state.requirements: raise ValueError("Missing 'requirements' in state for UI generation.") # Load reference design ref_path = os.path.join("templates", "demo_qms_design.html") if not os.path.exists(ref_path): raise FileNotFoundError(f"Reference HTML not found at {ref_path}") with open(ref_path, "r", encoding="utf-8") as f: reference_html = f.read() ui = await generate_ui_html(state.requirements, reference_html) html = ui.get("html", "") if not html or "