-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathutils.py
More file actions
113 lines (89 loc) · 4.19 KB
/
Copy pathutils.py
File metadata and controls
113 lines (89 loc) · 4.19 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
"""
Utility functions for the Process Builder.
"""
import re
from typing import Dict, Set
from pathlib import Path
import csv
from .models import ProcessStep, ProcessNote
def sanitize_id(id_str: str) -> str:
"""Sanitize a string to make it a valid Mermaid ID."""
# Keep meaningful characters while ensuring safe node IDs
safe_id = re.sub(r'[^a-zA-Z0-9_\s-]', '', id_str)
safe_id = re.sub(r'[\s-]+', '_', safe_id)
# Handle common keywords in step names
if any(word in safe_id.lower() for word in ['success', 'failure', 'error', 'end']):
safe_id = f"step_{safe_id}"
# Ensure ID starts with a letter (Mermaid requirement)
if not safe_id or not safe_id[0].isalpha():
safe_id = 'node_' + safe_id
return safe_id
def validate_process_flow(steps: list[ProcessStep]) -> list[str]:
"""Validate the entire process flow and return a list of issues."""
issues = []
# Check for at least one step and end points
if not steps:
issues.append("Process must have at least one step")
return issues
has_end = any(step.next_step_success.lower() == 'end' or
step.next_step_failure.lower() == 'end' for step in steps)
if not has_end:
issues.append("Process must have at least one path that leads to 'End'")
# Check all paths for circular references and missing steps
if steps:
first_step = steps[0]
# Helper function to check a path
def check_path(start_id: str, path_name: str) -> None:
visited: Set[str] = set()
current = start_id
path: list[str] = []
while current is not None and current.lower() != 'end':
path.append(current)
if current in visited:
issues.append(f"Circular reference detected in {path_name} path: {' -> '.join(path)}")
break
visited.add(current)
step = next((s for s in steps if s.step_id == current), None)
if step is None:
issues.append(f"Step name '{current}' referenced in {path_name} path not found")
break
if path_name == "success":
current = step.next_step_success
else:
current = step.next_step_failure
# Check both success and failure paths starting from the first step
check_path(first_step.step_id, "success")
check_path(first_step.step_id, "failure")
# Check for disconnected steps
all_step_ids = {step.step_id for step in steps}
referenced_steps = set()
for step in steps:
if step.next_step_success.lower() != 'end':
referenced_steps.add(step.next_step_success)
if step.next_step_failure.lower() != 'end':
referenced_steps.add(step.next_step_failure)
# Get the first step ID which doesn't need to be referenced
first_step_id = steps[0].step_id if steps else ""
disconnected = all_step_ids - referenced_steps - {first_step_id} # First step doesn't need to be referenced
if disconnected:
issues.append(f"Disconnected step names found: {', '.join(disconnected)}")
return issues
def validate_notes(notes: list[ProcessNote], steps: list[ProcessStep]) -> list[str]:
"""Validate notes and return a list of issues."""
issues = []
# Check for duplicate note IDs
note_ids = [note.note_id for note in notes]
if len(note_ids) != len(set(note_ids)):
issues.append("Duplicate note IDs found")
# Check for orphaned notes
step_ids = {step.step_id for step in steps}
for note in notes:
if note.related_step_id not in step_ids:
issues.append(f"Note {note.note_id} references non-existent step name '{note.related_step_id}'")
return issues
def write_csv(data: list[dict], filepath: Path, fieldnames: list[str]) -> None:
"""Write data to a CSV file."""
with open(filepath, 'w', newline='') as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(data)