Coverage for src / graphable / parsers / utils.py: 87%

1from pathlib import Path 

2from re import search 

3from typing import Any 

4 

5from ..graph import Graph 

6from ..graphable import Graphable 

7 

8 

9def build_graph_from_data( 

10 nodes_data: list[dict[str, Any]], 

11 edges_data: list[dict[str, Any]], 

12 reference_type: type = str, 

13) -> Graph[Any]: 

14 """ 

15 Standardized helper to construct a Graph from raw data. 

16 

17 Args: 

18 nodes_data: List of node dictionaries (id, tags, duration, status). 

19 edges_data: List of edge dictionaries (source, target, attributes). 

20 reference_type: Type to cast the reference string to. 

21 

22 Returns: 

23 Graph: The constructed Graph instance. 

24 """ 

25 node_map: dict[str, Graphable[Any]] = {} 

26 

27 # 1. Create nodes 

28 for node_entry in nodes_data: 

29 node_id = str(node_entry["id"]) 

30 # Use provided reference if available, otherwise fallback to id 

31 ref_val = node_entry.get("reference", node_id) 

32 if reference_type is not str: 

33 try: 

34 ref_val = reference_type(ref_val) 

35 except (ValueError, TypeError): 

36 pass 

37 

38 node = Graphable(ref_val) 

39 

40 # Metadata 

41 for tag in node_entry.get("tags", []): 

42 node.add_tag(tag) 

43 

44 if "duration" in node_entry: 

45 node.duration = float(node_entry["duration"]) 

46 if "status" in node_entry: 

47 node.status = str(node_entry["status"]) 

48 

49 node_map[node_id] = node 

50 

51 # 2. Link edges 

52 g = Graph(set(node_map.values())) 

53 for edge_entry in edges_data: 

54 u_id = str(edge_entry["source"]) 

55 v_id = str(edge_entry["target"]) 

56 

57 if u_id in node_map and v_id in node_map: 

58 # All other keys are attributes 

59 attrs = { 

60 k: v for k, v in edge_entry.items() if k not in ("source", "target") 

61 } 

62 g.add_edge(node_map[u_id], node_map[v_id], **attrs) 

63 

64 return g 

65 

66 

67def is_path(source: str | Path) -> bool: 

68 """ 

69 Check if the given source is likely a path to a file. 

70 

71 Args: 

72 source: The source to check (string or Path object). 

73 

74 Returns: 

75 bool: True if it's a Path object or a string that points to an existing file. 

76 """ 

77 if isinstance(source, Path): 

78 return True 

79 

80 if isinstance(source, str) and "\n" not in source and len(source) < 1024: 

81 try: 

82 return Path(source).exists() 

83 except OSError: 

84 pass 

85 

86 return False 

87 

88 

89def extract_checksum(source: str | Path) -> str | None: 

90 """ 

91 Extract a blake2b checksum from the first few lines of a source. 

92 Looks for the pattern 'blake2b: <hash>'. 

93 

94 Args: 

95 source: The source to check (string or Path object). 

96 

97 Returns: 

98 str | None: The hash if found, otherwise None. 

99 """ 

100 content = "" 

101 if is_path(source): 

102 try: 

103 # Read only the first 1KB to find comments 

104 with open(source, "r") as f: 

105 content = f.read(1024) 

106 except Exception: 

107 return None 

108 else: 

109 content = str(source)[:1024] 

110 

111 # Match blake2b: followed by hex chars (usually 128 for blake2b) 

112 match = search(r"blake2b:\s*([a-fA-F0-9]+)", content) 

113 if match: 

114 return match.group(1) 

115 

116 return None