Coverage for src\sembra\core.py: 99%
249 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-26 12:45 +0700
1from __future__ import annotations
3import time
4from typing import Any, Dict, List, Optional, Set, Tuple
6from sembra.types import (
7 MatchInfo,
8 PairwiseResult,
9 SembraConfig,
10 SimilarityResult,
11)
12from sembra.utils import (
13 count_nodes,
14 get_json_type,
15 jaro_winkler,
16 number_similarity,
17)
20def sembra(a: Any, b: Any, config: Optional[SembraConfig] = None) -> float:
21 cfg = config or SembraConfig()
22 cache: Dict[int, Dict[int, float]] = {} if cfg.use_cache else None # type: ignore
23 return _similarity(a, b, cfg, 0, cache)
26def sembra_report(a: Any, b: Any, config: Optional[SembraConfig] = None) -> SimilarityResult:
27 cfg = config or SembraConfig()
28 cache: Dict[int, Dict[int, float]] = {} if cfg.use_cache else None # type: ignore
29 matches: List[MatchInfo] = []
31 start = time.perf_counter_ns()
32 score = _similarity(a, b, cfg, 0, cache, matches, "")
33 duration = time.perf_counter_ns() - start
35 return SimilarityResult(score=score, matches=matches, duration_ns=duration)
38def sembra_pairwise(
39 objects_a: List[Any],
40 objects_b: List[Any],
41 config: Optional[SembraConfig] = None,
42 labels_a: Optional[List[str]] = None,
43 labels_b: Optional[List[str]] = None,
44) -> PairwiseResult:
45 cfg = config or SembraConfig()
46 n, m = len(objects_a), len(objects_b)
47 matrix: List[List[float]] = [[0.0] * m for _ in range(n)]
49 start = time.perf_counter_ns()
50 for i in range(n):
51 for j in range(m):
52 matrix[i][j] = sembra(objects_a[i], objects_b[j], cfg)
53 duration = time.perf_counter_ns() - start
55 return PairwiseResult(
56 matrix=matrix,
57 labels_a=labels_a or [str(i) for i in range(n)],
58 labels_b=labels_b or [str(j) for j in range(m)],
59 duration_ns=duration,
60 )
63def _similarity(
64 a: Any,
65 b: Any,
66 cfg: SembraConfig,
67 depth: int,
68 cache: Optional[Dict[int, Dict[int, float]]],
69 matches: Optional[List[MatchInfo]] = None,
70 path: str = "",
71) -> float:
72 if depth > cfg.max_depth:
73 return 0.5
75 if cfg.early_exit and a is b:
76 return 1.0
78 type_a = get_json_type(a)
79 type_b = get_json_type(b)
81 if type_a != type_b:
82 return _cross_type_similarity(a, b, type_a, type_b, cfg)
84 if cache is not None:
85 id_a, id_b = id(a), id(b)
86 if id_a in cache and id_b in cache[id_a]:
87 return cache[id_a][id_b]
89 result = _type_dispatch(a, b, type_a, cfg, depth, cache, matches, path)
91 if cache is not None:
92 id_a, id_b = id(a), id(b)
93 if id_a not in cache:
94 cache[id_a] = {}
95 cache[id_a][id_b] = result
97 return result
100def _cross_type_similarity(a: Any, b: Any, type_a: str, type_b: str, cfg: SembraConfig) -> float:
101 a_null = type_a == "null"
102 b_null = type_b == "null"
103 if a_null and b_null:
104 return 1.0
105 if a_null or b_null:
106 return 0.0
108 if type_a == "number" and type_b == "string":
109 try:
110 return number_similarity(a, float(b))
111 except (ValueError, TypeError):
112 return 0.0
113 if type_a == "string" and type_b == "number":
114 try:
115 return number_similarity(float(a), b)
116 except (ValueError, TypeError):
117 return 0.0
119 if type_a == "boolean" and type_b == "number":
120 return number_similarity(1.0 if a else 0.0, b)
121 if type_a == "number" and type_b == "boolean":
122 return number_similarity(a, 1.0 if b else 0.0)
124 if type_a == "string" and type_b == "boolean":
125 str_b = "true" if b else "false"
126 if cfg.string_similarity:
127 return jaro_winkler(a, str_b)
128 return 1.0 if a == str_b else 0.0
129 if type_a == "boolean" and type_b == "string":
130 str_a = "true" if a else "false"
131 if cfg.string_similarity:
132 return jaro_winkler(str_a, b)
133 return 1.0 if str_a == b else 0.0
135 return 0.0
138def _type_dispatch(
139 a: Any,
140 b: Any,
141 t: str,
142 cfg: SembraConfig,
143 depth: int,
144 cache: Optional[Dict[int, Dict[int, float]]],
145 matches: Optional[List[MatchInfo]],
146 path: str,
147) -> float:
148 if t == "null":
149 return 1.0
150 if t == "boolean":
151 if cfg.boolean_partial:
152 return 1.0 if a == b else 0.0
153 return 1.0 if a is b else 0.0
154 if t == "number":
155 if not cfg.number_similarity:
156 return 1.0 if a == b else 0.0
157 val = number_similarity(a, b)
158 if val >= cfg.number_threshold:
159 return 1.0
160 return val
161 if t == "string":
162 if a == b:
163 return 1.0
164 if not cfg.string_similarity:
165 return 0.0
166 return jaro_winkler(a, b)
167 if t == "array":
168 return _array_similarity(a, b, cfg, depth + 1, cache, matches, path)
169 if t == "object":
170 return _object_similarity(a, b, cfg, depth + 1, cache, matches, path)
172 return 0.0
175def _array_similarity(
176 arr1: List[Any],
177 arr2: List[Any],
178 cfg: SembraConfig,
179 depth: int,
180 cache: Optional[Dict[int, Dict[int, float]]],
181 matches: Optional[List[MatchInfo]],
182 path: str,
183) -> float:
184 if not arr1 and not arr2:
185 return 1.0
186 if not arr1 or not arr2:
187 return 0.0
189 if not cfg.array_matching:
190 return _array_positional(arr1, arr2, cfg, depth, cache, matches, path)
192 m, n = len(arr1), len(arr2)
194 if m > cfg.max_array_size or n > cfg.max_array_size:
195 return _array_type_bucketed(arr1, arr2, cfg, depth, cache, matches, path)
197 return _array_optimal_match(arr1, arr2, cfg, depth, cache, matches, path)
200def _array_positional(
201 arr1: List[Any],
202 arr2: List[Any],
203 cfg: SembraConfig,
204 depth: int,
205 cache: Optional[Dict[int, Dict[int, float]]],
206 matches: Optional[List[MatchInfo]],
207 path: str,
208) -> float:
209 n = min(len(arr1), len(arr2))
210 if n == 0:
211 return 0.0
212 total = 0.0
213 for i in range(n):
214 child_path = f"{path}[{i}]"
215 total += _similarity(arr1[i], arr2[i], cfg, depth, cache, matches, child_path)
216 return total / max(len(arr1), len(arr2))
219def _greedy_directional(
220 rows: int, cols: int, matrix: List[List[float]], threshold: float
221) -> float:
222 used = set()
223 total = 0.0
224 for i in range(rows):
225 best_j = -1
226 best_sim = threshold
227 for j in range(cols):
228 if j in used:
229 continue
230 if matrix[i][j] > best_sim:
231 best_sim = matrix[i][j]
232 best_j = j
233 if best_j != -1:
234 used.add(best_j)
235 total += best_sim
236 return total
239def _array_optimal_match(
240 arr1: List[Any],
241 arr2: List[Any],
242 cfg: SembraConfig,
243 depth: int,
244 cache: Optional[Dict[int, Dict[int, float]]],
245 matches: Optional[List[MatchInfo]],
246 path: str,
247) -> float:
248 m, n = len(arr1), len(arr2)
249 matrix = [[0.0] * n for _ in range(m)]
251 for i in range(m):
252 for j in range(n):
253 child_path = f"{path}[{i}]<->[{j}]"
254 matrix[i][j] = _similarity(arr1[i], arr2[j], cfg, depth, cache, matches, child_path)
256 score_ab = _greedy_directional(m, n, matrix, cfg.array_match_threshold)
257 transposed = [[matrix[i][j] for i in range(m)] for j in range(n)]
258 score_ba = _greedy_directional(n, m, transposed, cfg.array_match_threshold)
260 score = (score_ab / max(m, 1) + score_ba / max(n, 1)) / 2.0
262 if matches is not None:
263 used_cols: Set[int] = set()
264 for i in range(m):
265 best_j = -1
266 best_sim = cfg.array_match_threshold
267 for j in range(n):
268 if j in used_cols:
269 continue
270 if matrix[i][j] > best_sim:
271 best_sim = matrix[i][j]
272 best_j = j
273 if best_j != -1:
274 used_cols.add(best_j)
275 matches.append(
276 MatchInfo(
277 path_a=f"{path}[{i}]",
278 path_b=f"{path}[{best_j}]",
279 similarity=best_sim,
280 match_type="array_element",
281 )
282 )
284 return score
287def _array_type_bucketed(
288 arr1: List[Any],
289 arr2: List[Any],
290 cfg: SembraConfig,
291 depth: int,
292 cache: Optional[Dict[int, Dict[int, float]]],
293 matches: Optional[List[MatchInfo]],
294 path: str,
295) -> float:
296 buckets1: Dict[str, List[Any]] = {}
297 buckets2: Dict[str, List[Any]] = {}
298 for item in arr1:
299 t = get_json_type(item)
300 buckets1.setdefault(t, []).append(item)
301 for item in arr2:
302 t = get_json_type(item)
303 buckets2.setdefault(t, []).append(item)
305 all_types = set(buckets1.keys()) | set(buckets2.keys())
306 total_sim = 0.0
307 max_possible = max(len(arr1), len(arr2))
309 for t in all_types:
310 b1 = buckets1.get(t, [])
311 b2 = buckets2.get(t, [])
312 if b1 and b2:
313 bucket_sim = _array_optimal_match(b1, b2, cfg, depth, cache, matches, f"{path}[{t}]")
314 total_sim += bucket_sim * max(len(b1), len(b2))
316 return total_sim / max_possible if max_possible > 0 else 0.0
319def _object_similarity(
320 obj1: Dict[str, Any],
321 obj2: Dict[str, Any],
322 cfg: SembraConfig,
323 depth: int,
324 cache: Optional[Dict[int, Dict[int, float]]],
325 matches: Optional[List[MatchInfo]],
326 path: str,
327) -> float:
328 if not obj1 and not obj2:
329 return 1.0
330 if not obj1 or not obj2:
331 return 0.0
333 keys1 = set(obj1.keys())
334 keys2 = set(obj2.keys())
336 exact_match_keys = keys1 & keys2
337 remaining1 = keys1 - keys2
338 remaining2 = keys2 - keys1
340 total_weight = 0.0
341 weighted_sum = 0.0
343 for key in exact_match_keys:
344 child_path = f"{path}.{key}"
345 weight = _key_weight(key, obj1[key], obj2[key], cfg)
346 sim = _similarity(obj1[key], obj2[key], cfg, depth, cache, matches, child_path)
347 weighted_sum += weight * sim
348 total_weight += weight
349 if matches is not None:
350 matches.append(
351 MatchInfo(
352 path_a=child_path,
353 path_b=child_path,
354 similarity=sim,
355 match_type="exact_key",
356 )
357 )
359 matched_rename_count = 0
360 if cfg.object_key_rename and remaining1 and remaining2:
361 matched_renamed = _detect_key_renames(
362 obj1, obj2, remaining1, remaining2, cfg, depth, cache, matches, path
363 )
364 matched_rename_count = len(matched_renamed)
365 for sim, w in matched_renamed:
366 rename_discount = 0.9
367 weighted_sum += w * sim * rename_discount
368 total_weight += w
370 unmatched_count = (len(remaining1) - matched_rename_count) + (
371 len(remaining2) - matched_rename_count
372 )
373 unmatched_penalty = max(0, unmatched_count) * 0.5
374 total_weight += unmatched_penalty
376 return weighted_sum / total_weight if total_weight > 0 else 0.0
379def _key_weight(key: str, val1: Any, val2: Any, cfg: SembraConfig) -> float:
380 if cfg.key_weight_mode == "uniform":
381 return 1.0
382 if cfg.key_weight_mode == "depth":
383 return 1.0
384 if cfg.key_weight_mode == "size":
385 return 1.0 + 0.1 * min(count_nodes(val1), count_nodes(val2))
386 return 1.0
389def _detect_key_renames(
390 obj1: Dict[str, Any],
391 obj2: Dict[str, Any],
392 remaining1: Set[str],
393 remaining2: Set[str],
394 cfg: SembraConfig,
395 depth: int,
396 cache: Optional[Dict[int, Dict[int, float]]],
397 matches: Optional[List[MatchInfo]],
398 path: str,
399) -> List[Tuple[float, float]]:
400 r1_list = list(remaining1)
401 r2_list = list(remaining2)
403 results: List[Tuple[float, float]] = []
404 used_r2: Set[int] = set()
406 for k1 in r1_list:
407 best_j = -1
408 best_sim = cfg.object_rename_threshold
409 v1 = obj1[k1]
410 for j, k2 in enumerate(r2_list):
411 if j in used_r2:
412 continue
413 v2 = obj2[k2]
414 sim = _similarity(v1, v2, cfg, depth, cache, None, f"{path}.{k1}~{k2}")
415 if sim > best_sim:
416 best_sim = sim
417 best_j = j
419 if best_j != -1:
420 used_r2.add(best_j)
421 results.append((best_sim, 1.0))
422 if matches is not None:
423 k2 = r2_list[best_j]
424 matches.append(
425 MatchInfo(
426 path_a=f"{path}.{k1}",
427 path_b=f"{path}.{k2}",
428 similarity=best_sim,
429 match_type="key_rename",
430 )
431 )
433 return results