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

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 qlever_full_text_search: bool = False, 

52) -> dict: 

53 if provenance_dirs is None: 

54 provenance_dirs = [] 

55 if provenance_urls is None: 

56 provenance_urls = [] 

57 if dataset_dirs is None: 

58 dataset_dirs = [] 

59 if dataset_urls is None: 

60 dataset_urls = [] 

61 config = { 

62 "dataset": { 

63 "triplestore_urls": dataset_urls, 

64 "file_paths": dataset_dirs, 

65 "is_quadstore": dataset_is_quadstore, 

66 }, 

67 "provenance": { 

68 "triplestore_urls": provenance_urls, 

69 "file_paths": provenance_dirs, 

70 "is_quadstore": provenance_is_quadstore, 

71 }, 

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

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

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

75 "graphdb_connector_name": graphdb_connector_name, 

76 "qlever_full_text_search": str(qlever_full_text_search).lower(), 

77 } 

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

79 json.dump(config, f) 

80 return config 

81 

82 

83@lru_cache(maxsize=4096) 

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

85 if time_string.endswith("Z"): 

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

87 time = datetime.fromisoformat(time_string) 

88 if time.tzinfo is None: 

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

90 return time.astimezone(timezone.utc) 

91 

92 

93def convert_to_datetime( 

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

95) -> datetime | str | None: 

96 if time_string and time_string != "None": 

97 time = _cached_parse(time_string) 

98 if stringify: 

99 return time.isoformat() 

100 return time 

101 return None 

102 

103 

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

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

106 return n3 

107 i = 1 

108 while i < len(n3): 

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

110 i += 2 

111 continue 

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

113 rest = n3[i + 1 :] 

114 if rest.startswith("@"): 

115 return n3 

116 return n3[: i + 1] 

117 i += 1 

118 return n3 

119 

120 

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

122 if quads is None: 

123 return None 

124 lines = set() 

125 for q in quads: 

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

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

128 return sorted(lines) 

129 

130 

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

132 result = {} 

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

134 if isinstance(value, set): 

135 result[key] = _to_nt_sorted_list(value) 

136 else: 

137 result.setdefault(key, {}) 

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

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

140 return result 

141 

142 

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

144 result = set() 

145 for line in nt_list: 

146 if not line.strip(): 

147 continue 

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

149 if len(matches) >= _TRIPLE_LEN: 

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

151 return result 

152 

153 

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

155 result = {} 

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

157 if isinstance(value, list): 

158 result[key] = _nt_list_to_quad_set(value) 

159 else: 

160 result.setdefault(key, {}) 

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

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

163 return result