import re from typing import Any from pydantic import ValidationError from app.agents.visual_tools.icon_manifest import hydrate_icon from app.agents.visual_tools.schemas import VisualPlan FALLBACK_COLORS = ["#06b6d4", "#8b5cf6", "#10b981", "#f59e0b", "#ef4444", "#f97316"] VALID_HEX = re.compile(r"^#[0-9a-fA-F]{6}$") STOPWORDS = { "about", "after", "again", "also", "and", "application", "are", "because", "before", "being", "between", "can", "clear", "does", "each", "exactly", "from", "have", "into", "like", "main", "must", "need", "needs", "showing", "that", "the", "their", "then", "there", "these", "this", "through", "user", "users", "using", "where", "which", "will", "with", "within", "your", } def _title_words(text: str) -> list[str]: words = re.findall(r"[A-Za-z][A-Za-z0-9-]{2,}", text) picked: list[str] = [] seen: set[str] = set() for word in words: key = word.lower() if key in STOPWORDS or key in seen: continue seen.add(key) picked.append(word) return picked def _labelize(words: list[str], start: int, default: str) -> str: chunk = words[start : start + 2] if not chunk: return default return " ".join(word.capitalize() for word in chunk)[:28] def fallback_plan(text: str) -> VisualPlan: words = _title_words(text) lower = text.lower() if any(term in lower for term in (" vs ", " versus ", "compare", "difference")): intent = "comparison" nodes = [ {"id": "n1", "label": _labelize(words, 0, "Option A"), "type": "concept", "icon": "concept", "role": "left", "color": "#06b6d4"}, {"id": "n2", "label": _labelize(words, 2, "Trait A"), "type": "process", "icon": "process", "role": "left", "color": "#06b6d4"}, {"id": "n3", "label": _labelize(words, 4, "Option B"), "type": "concept", "icon": "concept", "role": "right", "color": "#f59e0b"}, {"id": "n4", "label": _labelize(words, 6, "Trait B"), "type": "process", "icon": "process", "role": "right", "color": "#f59e0b"}, ] edges = [] elif any(term in lower for term in ("cycle", "loop", "repeat", "feedback")): intent = "cyclic" nodes = [ {"id": f"n{i + 1}", "label": _labelize(words, i * 2, default), "type": "process", "icon": "process", "color": FALLBACK_COLORS[i % len(FALLBACK_COLORS)]} for i, default in enumerate(("Trigger", "Action", "Feedback", "Adjustment")) ] edges = [ {"source": "n1", "target": "n2", "label": "starts"}, {"source": "n2", "target": "n3", "label": "creates"}, {"source": "n3", "target": "n4", "label": "guides"}, {"source": "n4", "target": "n1", "label": "repeats"}, ] elif any(term in lower for term in ("pipeline", "workflow", "process", "flow", "stage", "step")): intent = "sequential" defaults = ("Input", "Workspace", "Processing", "Preview", "Output") nodes = [ {"id": f"n{i + 1}", "label": _labelize(words, i * 2, default), "type": "process", "icon": "process", "color": FALLBACK_COLORS[i % len(FALLBACK_COLORS)]} for i, default in enumerate(defaults) ] edges = [ {"source": f"n{i + 1}", "target": f"n{i + 2}", "label": label} for i, label in enumerate(("feeds", "builds", "checks", "delivers")) ] else: intent = "radial" defaults = ("Core Idea", "Key Part", "Cause", "Effect", "Example") nodes = [ {"id": f"n{i + 1}", "label": _labelize(words, i * 2, default), "type": "concept", "icon": "concept", "color": FALLBACK_COLORS[i % len(FALLBACK_COLORS)]} for i, default in enumerate(defaults) ] edges = [ {"source": "n1", "target": f"n{i}", "label": "relates"} for i in range(2, len(nodes) + 1) ] return VisualPlan( title=" ".join(word.capitalize() for word in words[:4]) or "Concept Map", intent=intent, nodes=nodes, edges=edges, ) def validate_plan(raw: Any, fallback_text: str) -> VisualPlan: try: plan = VisualPlan.model_validate(raw) except (ValidationError, TypeError, ValueError): return fallback_plan(fallback_text) seen: set[str] = set() cleaned_nodes = [] for idx, node in enumerate(plan.nodes[:10]): node.id = re.sub(r"[^a-zA-Z0-9_-]", "_", node.id or f"n{idx + 1}") if node.id in seen: node.id = f"{node.id}_{idx}" seen.add(node.id) node.label = (node.label or f"Node {idx + 1}")[:60] node.color = node.color if VALID_HEX.match(node.color) else FALLBACK_COLORS[idx % len(FALLBACK_COLORS)] if hydrate_icon(node.icon) is None: node.icon = node.type if hydrate_icon(node.type) else "concept" cleaned_nodes.append(node) if not cleaned_nodes: return fallback_plan(fallback_text) valid_ids = {node.id for node in cleaned_nodes} cleaned_edges = [ edge for edge in plan.edges[:30] if edge.source in valid_ids and edge.target in valid_ids and edge.source != edge.target ] plan.nodes = cleaned_nodes plan.edges = cleaned_edges return plan