Coverage for src / time_agnostic_library / support.py: 100%

1# SPDX-FileCopyrightText: 2021-2026 Arcangelo Massari <arcangelo.massari@unibo.it> 

2# 

3# SPDX-License-Identifier: ISC 

4 

5 

6import json 

7import re 

8from datetime import datetime, timezone 

9from functools import lru_cache 

10from pathlib import Path 

11 

12CONFIG_PATH = "./config.json" 

13 

14_TRIPLE_LEN = 3 

15 

16_NT_TERM_RE = re.compile( 

17 r"<([^>]+)>" 

18 r'|"((?:[^"\\]|\\.)*)"\^\^<([^>]+)>' 

19 r'|"((?:[^"\\]|\\.)*)"@([a-zA-Z][\w-]*)' 

20 r'|"((?:[^"\\]|\\.)*)"' 

21 r"|(_:\S+)", 

22 re.DOTALL, 

23) 

24 

25 

26def _nt_match_to_n3(match: re.Match) -> str: 

27 if match.group(1) is not None: 

28 return f"<{match.group(1)}>" 

29 if match.group(2) is not None: 

30 return f'"{match.group(2)}"^^<{match.group(3)}>' 

31 if match.group(4) is not None: 

32 return f'"{match.group(4)}"@{match.group(5)}' 

33 if match.group(6) is not None: 

34 return f'"{match.group(6)}"' 

35 return match.group(7) 

36 

37 

38def generate_config_file( 

39 config_path: str = CONFIG_PATH, 

40 dataset_urls: list | None = None, 

41 dataset_dirs: list | None = None, 

42 *, 

43 dataset_is_quadstore: bool = True, 

44 provenance_urls: list | None = None, 

45 provenance_dirs: list | None = None, 

46 provenance_is_quadstore: bool = True, 

47 blazegraph_full_text_search: bool = False, 

48 fuseki_full_text_search: bool = False, 

49 virtuoso_full_text_search: bool = False, 

50 graphdb_connector_name: str = "", 

51) -> dict: 

52 if provenance_dirs is None: 

53 provenance_dirs = [] 

54 if provenance_urls is None: 

55 provenance_urls = [] 

56 if dataset_dirs is None: 

57 dataset_dirs = [] 

58 if dataset_urls is None: 

59 dataset_urls = [] 

60 config = { 

61 "dataset": { 

62 "triplestore_urls": dataset_urls, 

63 "file_paths": dataset_dirs, 

64 "is_quadstore": dataset_is_quadstore, 

65 }, 

66 "provenance": { 

67 "triplestore_urls": provenance_urls, 

68 "file_paths": provenance_dirs, 

69 "is_quadstore": provenance_is_quadstore, 

70 }, 

71 "blazegraph_full_text_search": str(blazegraph_full_text_search).lower(), 

72 "fuseki_full_text_search": str(fuseki_full_text_search).lower(), 

73 "virtuoso_full_text_search": str(virtuoso_full_text_search).lower(), 

74 "graphdb_connector_name": graphdb_connector_name, 

75 } 

76 with Path(config_path).open("w", encoding="utf-8") as f: 

77 json.dump(config, f) 

78 return config 

79 

80 

81@lru_cache(maxsize=4096) 

82def _cached_parse(time_string: str) -> datetime: 

83 if time_string.endswith("Z"): 

84 time_string = time_string[:-1] + "+00:00" 

85 time = datetime.fromisoformat(time_string) 

86 if time.tzinfo is None: 

87 return time.replace(tzinfo=timezone.utc) 

88 return time.astimezone(timezone.utc) 

89 

90 

91def convert_to_datetime( 

92 time_string: str | None, *, stringify: bool = False 

93) -> datetime | str | None: 

94 if time_string and time_string != "None": 

95 time = _cached_parse(time_string) 

96 if stringify: 

97 return time.isoformat() 

98 return time 

99 return None 

100 

101 

102def _strip_literal_datatype(n3: str) -> str: 

103 if not n3.startswith('"'): 

104 return n3 

105 i = 1 

106 while i < len(n3): 

107 if n3[i] == "\\": 

108 i += 2 

109 continue 

110 if n3[i] == '"': 

111 rest = n3[i + 1 :] 

112 if rest.startswith("@"): 

113 return n3 

114 return n3[: i + 1] 

115 i += 1 

116 return n3 

117 

118 

119def _to_nt_sorted_list(quads) -> list | None: 

120 if quads is None: 

121 return None 

122 lines = set() 

123 for q in quads: 

124 parts = [_strip_literal_datatype(el) for el in q[:3]] 

125 lines.add(" ".join(parts)) 

126 return sorted(lines) 

127 

128 

129def _to_dict_of_nt_sorted_lists(dictionary: dict) -> dict: 

130 result = {} 

131 for key, value in dictionary.items(): 

132 if isinstance(value, set): 

133 result[key] = _to_nt_sorted_list(value) 

134 else: 

135 result.setdefault(key, {}) 

136 for snapshot, quad_set in value.items(): 

137 result[key][snapshot] = _to_nt_sorted_list(quad_set) 

138 return result 

139 

140 

141def _nt_list_to_quad_set(nt_list: list[str]) -> set[tuple[str, ...]]: 

142 result = set() 

143 for line in nt_list: 

144 if not line.strip(): 

145 continue 

146 matches = list(_NT_TERM_RE.finditer(line)) 

147 if len(matches) >= _TRIPLE_LEN: 

148 result.add(tuple(_nt_match_to_n3(m) for m in matches[:3])) 

149 return result 

150 

151 

152def _to_dict_of_quad_sets(dictionary: dict) -> dict: 

153 result = {} 

154 for key, value in dictionary.items(): 

155 if isinstance(value, list): 

156 result[key] = _nt_list_to_quad_set(value) 

157 else: 

158 result.setdefault(key, {}) 

159 for snapshot, triples in value.items(): 

160 result[key][snapshot] = _nt_list_to_quad_set(triples) 

161 return result