Spaces:
Running
Running
| import graphviz | |
| import json | |
| from tempfile import NamedTemporaryFile | |
| import os | |
| def generate_network_graph(json_input: str, output_format: str) -> str: | |
| try: | |
| if not json_input.strip(): | |
| return "Error: Empty input" | |
| data = json.loads(json_input) | |
| if 'nodes' not in data or 'connections' not in data: | |
| raise ValueError("Missing required fields: nodes or connections") | |
| dot = graphviz.Graph( | |
| name='NetworkGraph', | |
| format='png', | |
| engine='neato', | |
| graph_attr={ | |
| 'overlap': 'false', | |
| 'splines': 'true', | |
| 'bgcolor': 'white', | |
| 'pad': '0.5', | |
| 'layout': 'neato' | |
| }, | |
| node_attr={ | |
| 'fixedsize': 'false' | |
| } | |
| ) | |
| base_color = '#19191a' | |
| type_colors = { | |
| 'server': base_color, | |
| 'service': '#4a90e2', | |
| 'database': '#a0a0a0', | |
| 'user': '#f39c12', | |
| 'default': base_color | |
| } | |
| nodes = data.get('nodes', []) | |
| connections = data.get('connections', []) | |
| for node in nodes: | |
| node_id = node.get('id') | |
| label = node.get('label') | |
| node_type = node.get('type', 'default') | |
| if not all([node_id, label]): | |
| raise ValueError(f"Invalid node: {node}") | |
| node_color = type_colors.get(node_type, type_colors['default']) | |
| if node_type == 'database' or node_color == '#a0a0a0': | |
| font_color = 'black' | |
| elif node_color == '#f39c12': | |
| font_color = 'black' | |
| else: | |
| font_color = 'white' | |
| if node_type == 'server': | |
| shape = 'box' | |
| style = 'filled,rounded' | |
| elif node_type == 'database': | |
| shape = 'cylinder' | |
| style = 'filled' | |
| elif node_type == 'user': | |
| shape = 'ellipse' | |
| style = 'filled' | |
| elif node_type == 'service': | |
| shape = 'hexagon' | |
| style = 'filled' | |
| else: | |
| shape = 'circle' | |
| style = 'filled' | |
| dot.node( | |
| node_id, | |
| label, | |
| shape=shape, | |
| style=style, | |
| fillcolor=node_color, | |
| fontcolor=font_color, | |
| fontsize='12' | |
| ) | |
| for connection in connections: | |
| from_node = connection.get('from') | |
| to_node = connection.get('to') | |
| label = connection.get('label', '') | |
| weight = connection.get('weight', 1) | |
| if not all([from_node, to_node]): | |
| raise ValueError(f"Invalid connection: {connection}") | |
| penwidth = str(max(1, min(5, weight))) | |
| dot.edge( | |
| from_node, | |
| to_node, | |
| label=label, | |
| color='#4a4a4a', | |
| fontcolor='#4a4a4a', | |
| fontsize='10', | |
| penwidth=penwidth | |
| ) | |
| with NamedTemporaryFile(delete=False, suffix=f'.{output_format}') as tmp: | |
| dot.render(tmp.name, format=output_format, cleanup=True) | |
| return f"{tmp.name}.{output_format}" | |
| except json.JSONDecodeError: | |
| return "Error: Invalid JSON format" | |
| except Exception as e: | |
| return f"Error: {str(e)}" |