Spaces:
Sleeping
Sleeping
| from shapely.geometry import box, Point | |
| from collections import defaultdict, deque | |
| def map_arrows(nodes, arrows): | |
| for node in nodes: | |
| node["shape"] = box(*node["bbox"]) | |
| edges = [] | |
| for arrow in arrows: | |
| tail_point = Point(arrow["tail"]) | |
| head_point = Point(arrow["head"]) | |
| source = None | |
| target = None | |
| for node in nodes: | |
| if node["shape"].contains(tail_point): | |
| source = node["id"] | |
| if node["shape"].contains(head_point): | |
| target = node["id"] | |
| label = arrow.get("label", "") | |
| if source and target and source != target: | |
| edges.append((source, target, label)) | |
| return edges | |
| def detect_node_type(text): | |
| text_lower = text.lower() | |
| if "start" in text_lower: | |
| return "start" | |
| if "end" in text_lower or "full" in text_lower: | |
| return "end" | |
| if "?" in text or "yes" in text_lower or "no" in text_lower: | |
| return "decision" | |
| return "process" | |
| def build_flowchart_json(nodes, edges): | |
| graph = {} | |
| reverse_links = defaultdict(list) | |
| edge_labels = defaultdict(list) | |
| for node in nodes: | |
| raw_text = node.get("text", "").strip() | |
| node_type = node.get("type") or detect_node_type(raw_text) | |
| graph[node["id"]] = { | |
| "text": raw_text, | |
| "type": node_type, | |
| "next": [] | |
| } | |
| for source, target, label in edges: | |
| graph[source]["next"].append(target) | |
| reverse_links[target].append(source) | |
| edge_labels[(source, target)] = label.lower().strip() | |
| start_nodes = [nid for nid in graph if len(reverse_links[nid]) == 0] | |
| flowchart_json = { | |
| "start": start_nodes[0] if start_nodes else None, | |
| "steps": [] | |
| } | |
| visited = set() | |
| queue = deque(start_nodes) | |
| while queue: | |
| current = queue.popleft() | |
| if current in visited: | |
| continue | |
| visited.add(current) | |
| info = graph[current] | |
| step = { | |
| "id": current, | |
| "text": info["text"], | |
| "type": info["type"] | |
| } | |
| parents = reverse_links[current] | |
| if len(parents) == 1: | |
| step["parent"] = parents[0] | |
| elif len(parents) > 1: | |
| step["parents"] = parents | |
| next_nodes = info["next"] | |
| if info["type"] == "decision" and len(next_nodes) >= 2: | |
| branches = {} | |
| for target in next_nodes: | |
| label = edge_labels.get((current, target), "") | |
| if "yes" in label: | |
| branches["yes"] = target | |
| elif "no" in label: | |
| branches["no"] = target | |
| else: | |
| branches.setdefault("unknown", []).append(target) | |
| step["branches"] = branches | |
| queue.extend(next_nodes) | |
| elif len(next_nodes) == 1: | |
| step["next"] = next_nodes[0] | |
| queue.append(next_nodes[0]) | |
| elif len(next_nodes) > 1: | |
| step["next"] = next_nodes | |
| queue.extend(next_nodes) | |
| flowchart_json["steps"].append(step) | |
| return flowchart_json | |
| if __name__ == "__main__": | |
| nodes = [ | |
| {"id": "node1", "bbox": [100, 100, 200, 150], "text": "Start"}, | |
| {"id": "node2", "bbox": [300, 100, 400, 150], "text": "Is valid?"}, | |
| {"id": "node3", "bbox": [500, 50, 600, 100], "text": "Approve"}, | |
| {"id": "node4", "bbox": [500, 150, 600, 200], "text": "Reject"} | |
| ] | |
| arrows = [ | |
| {"id": "arrow1", "tail": (200, 125), "head": (300, 125), "label": ""}, | |
| {"id": "arrow2", "tail": (400, 125), "head": (500, 75), "label": "Yes"}, | |
| {"id": "arrow3", "tail": (400, 125), "head": (500, 175), "label": "No"} | |
| ] | |
| edges = map_arrows(nodes, arrows) | |
| flowchart_json = build_flowchart_json(nodes, edges) | |
| import json | |
| print(json.dumps(flowchart_json, indent=2)) |