Coverage for oc_meta / lib / agent_matching.py: 94%

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

2# 

3# SPDX-License-Identifier: ISC 

4 

5from __future__ import annotations 

6 

7import unicodedata 

8from dataclasses import dataclass 

9 

10from rapidfuzz import fuzz 

11 

12 

13@dataclass(frozen=True, slots=True) 

14class PersonName: 

15 name: str = "" 

16 given: str = "" 

17 family: str = "" 

18 

19 @property 

20 def display(self) -> str: 

21 return self.name or " ".join(part for part in (self.given, self.family) if part) 

22 

23 

24@dataclass(frozen=True, slots=True) 

25class AlignmentPair: 

26 local_index: int 

27 external_index: int 

28 score: float 

29 

30 

31@dataclass(frozen=True, slots=True) 

32class AlignmentResult: 

33 pairs: tuple[AlignmentPair, ...] 

34 unmatched_local: tuple[int, ...] 

35 unmatched_external: tuple[int, ...] 

36 ambiguous: bool 

37 

38 

39def normalize_name(value: str) -> str: 

40 decomposed = unicodedata.normalize("NFKD", value).casefold() 

41 characters = ( 

42 character for character in decomposed if unicodedata.category(character) != "Mn" 

43 ) 

44 return " ".join( 

45 "".join( 

46 character if character.isalnum() else " " for character in characters 

47 ).split() 

48 ) 

49 

50 

51def script_family(value: str) -> str: 

52 families = set() 

53 for character in value: 

54 if not character.isalpha(): 

55 continue 

56 name = unicodedata.name(character, "") 

57 if "CYRILLIC" in name: 

58 families.add("cyrillic") 

59 elif "GREEK" in name: 

60 families.add("greek") 

61 elif "LATIN" in name: 

62 families.add("latin") 

63 elif "CJK" in name or "HIRAGANA" in name or "KATAKANA" in name: 

64 families.add("cjk") 

65 else: 

66 families.add("other") 

67 return next(iter(families)) if len(families) == 1 else "mixed" 

68 

69 

70def _initials_compatible(left: str, right: str) -> bool: 

71 left_tokens = normalize_name(left).split() 

72 right_tokens = normalize_name(right).split() 

73 if not left_tokens or not right_tokens: 

74 return False 

75 limit = min(len(left_tokens), len(right_tokens)) 

76 return all( 

77 left_tokens[index] == right_tokens[index] 

78 or ( 

79 left_tokens[index][0] == right_tokens[index][0] 

80 and (len(left_tokens[index]) == 1 or len(right_tokens[index]) == 1) 

81 ) 

82 for index in range(limit) 

83 ) 

84 

85 

86def name_score(left: PersonName, right: PersonName) -> float: 

87 left_display = normalize_name(left.display) 

88 right_display = normalize_name(right.display) 

89 if not left_display or not right_display: 

90 return 0.0 

91 if left_display == right_display: 

92 return 1.0 

93 if set(left_display.split()) == set(right_display.split()): 

94 return 0.95 

95 

96 left_family = normalize_name(left.family) 

97 right_family = normalize_name(right.family) 

98 if ( 

99 left_family 

100 and right_family 

101 and left_family == right_family 

102 and _initials_compatible(left.given, right.given) 

103 ): 

104 return 0.9 

105 

106 left_script = script_family(left.display) 

107 right_script = script_family(right.display) 

108 if left_script != right_script or "mixed" in {left_script, right_script}: 

109 return 0.0 

110 return fuzz.ratio(left_display, right_display) / 100 

111 

112 

113def align_names( 

114 local: list[PersonName], external: list[PersonName], gap_penalty: float = 0.45 

115) -> AlignmentResult: 

116 rows = len(local) + 1 

117 columns = len(external) + 1 

118 scores = [[0.0] * columns for _ in range(rows)] 

119 paths = [[1] * columns for _ in range(rows)] 

120 moves = [[""] * columns for _ in range(rows)] 

121 

122 for row in range(1, rows): 

123 scores[row][0] = scores[row - 1][0] - gap_penalty 

124 moves[row][0] = "local" 

125 for column in range(1, columns): 

126 scores[0][column] = scores[0][column - 1] - gap_penalty 

127 moves[0][column] = "external" 

128 

129 for row in range(1, rows): 

130 for column in range(1, columns): 

131 candidates = { 

132 "match": scores[row - 1][column - 1] 

133 + name_score(local[row - 1], external[column - 1]), 

134 "local": scores[row - 1][column] - gap_penalty, 

135 "external": scores[row][column - 1] - gap_penalty, 

136 } 

137 best = max(candidates.values()) 

138 best_moves = [ 

139 move for move, score in candidates.items() if abs(score - best) < 1e-9 

140 ] 

141 scores[row][column] = best 

142 moves[row][column] = best_moves[0] 

143 paths[row][column] = min( 

144 2, 

145 sum( 

146 paths[row - 1][column - 1] 

147 if move == "match" 

148 else paths[row - 1][column] 

149 if move == "local" 

150 else paths[row][column - 1] 

151 for move in best_moves 

152 ), 

153 ) 

154 

155 pairs = [] 

156 unmatched_local = [] 

157 unmatched_external = [] 

158 row = len(local) 

159 column = len(external) 

160 while row or column: 

161 move = moves[row][column] 

162 if move == "match": 

163 pairs.append( 

164 AlignmentPair( 

165 local_index=row - 1, 

166 external_index=column - 1, 

167 score=name_score(local[row - 1], external[column - 1]), 

168 ) 

169 ) 

170 row -= 1 

171 column -= 1 

172 elif move == "local": 

173 unmatched_local.append(row - 1) 

174 row -= 1 

175 else: 

176 unmatched_external.append(column - 1) 

177 column -= 1 

178 

179 return AlignmentResult( 

180 pairs=tuple(reversed(pairs)), 

181 unmatched_local=tuple(reversed(unmatched_local)), 

182 unmatched_external=tuple(reversed(unmatched_external)), 

183 ambiguous=paths[-1][-1] > 1, 

184 )