from concurrent.futures import ThreadPoolExecutor from typing import Dict, List, Optional, Tuple from rapidfuzz import fuzz from sklearn.metrics.pairwise import cosine_similarity from sentence_transformers import SentenceTransformer import re import itertools # ---------- Model Store ---------- MODEL_STORE = { "model1": SentenceTransformer("sentence-transformers/all-mpnet-base-v2"), "model2": SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2"), } # ---------- Text Preprocessing ---------- def preprocess_for_matching(text: str) -> str: """Standardize text for matching""" if not text or text in ["-", " ", ""]: return "" return text.upper().strip() # ---------- Core Matching Functions ---------- def calculate_fuzzy_scores(input1: str, input2: str) -> Dict[str, float]: """Calculate fuzzy matching scores using RapidFuzz""" return { "simple_ratio": fuzz.ratio(input1, input2), "token_set_ratio": fuzz.token_set_ratio(input1, input2), "w_ratio": fuzz.WRatio(input1, input2), "partial_ratio": fuzz.partial_ratio(input1, input2), } def calculate_semantic_similarity(model_name: str, input1: str, input2: str) -> float: """Calculate semantic similarity using sentence transformers""" model = MODEL_STORE[model_name] embedding1 = model.encode([input1]) embedding2 = model.encode([input2]) print("name1",input1) print("name2",input2) return cosine_similarity(embedding1, embedding2)[0][0] def calculate_final_score(fuzzy_scores: Dict[str, float], semantic_score: float) -> float: """Calculate weighted final score""" weights = { "simple_ratio": 0.15, "token_set_ratio": 0.40, "partial_ratio": 0.20, "w_ratio": 0.05, "semantic_score": 0.20, } normalized_scores = { "simple_ratio": fuzzy_scores.get("simple_ratio", 0), "token_set_ratio": fuzzy_scores.get("token_set_ratio", 0), "partial_ratio": fuzzy_scores.get("partial_ratio", 0), "w_ratio": fuzzy_scores.get("w_ratio", 0), "semantic_score": semantic_score * 100, } weighted_sum = sum(normalized_scores[key] * weight for key, weight in weights.items()) return max(0, min(100, weighted_sum)) def calculate_overall_similarity(score1: float, score2: float) -> float: """Calculate overall similarity from two model scores""" return score1 * 0.6 + score2 * 0.4 # # ---------- Main Matching Function ---------- # def match_entities(value1: str, value2: str) -> Dict: # """ # Match two entities using fuzzy + semantic similarity # Returns: {"Result": "Match"/"No Match", "Confidence Score": int, "Overall Similarity": float} # """ # standardized_input1 = preprocess_for_matching(value1) # standardized_input2 = preprocess_for_matching(value2) # if not standardized_input1 or not standardized_input2: # return {"Result": "missing value", "Confidence Score": 100, "Overall Similarity": 0} # # Calculate fuzzy scores # fuzzy_match_scores = calculate_fuzzy_scores(standardized_input1, standardized_input2) # # Calculate semantic similarity using both models in parallel # with ThreadPoolExecutor() as executor: # f1 = executor.submit(calculate_semantic_similarity, "model1", standardized_input1, standardized_input2) # f2 = executor.submit(calculate_semantic_similarity, "model2", standardized_input1, standardized_input2) # cosine1 = f1.result() # cosine2 = f2.result() # ff1 = executor.submit(calculate_final_score, fuzzy_match_scores, cosine1) # ff2 = executor.submit(calculate_final_score, fuzzy_match_scores, cosine2) # final1 = ff1.result() # final2 = ff2.result() # overall_similarity = calculate_overall_similarity(final1, final2) # similarity_threshold = 85 # match_status = "Match" if overall_similarity >= similarity_threshold else "No Match" # # Confidence: 100 if both agree, 50 if they disagree # confidence = 100 if (final1 >= similarity_threshold and final2 >= similarity_threshold) or \ # (final1 < similarity_threshold and final2 < similarity_threshold) else 50 # return round(overall_similarity, 2) # # ---------- Name Matching Logic ---------- # def concatenate_name_parts(firstname: str, middlename: str, lastname: str) -> str: # """Concatenate name parts in alphabetical order""" # parts = [] # if firstname and firstname not in ["-", " ", ""]: # parts.append(preprocess_for_matching(firstname)) # if middlename and middlename not in ["-", " ", ""]: # parts.append(preprocess_for_matching(middlename)) # if lastname and lastname not in ["-", " ", ""]: # parts.append(preprocess_for_matching(lastname)) # if not parts: # return "" # # Sort alphabetically and concatenate # parts.sort() # return " ".join(parts) def check_substring_match(str1: str, str2: str) -> bool: """Check if one string is a substring of another""" if not str1 or not str2: return False return str1 in str2 or str2 in str1 def check_individual_name_matches(name_full: str, fname: str, mname: str, lname: str) -> Tuple[bool, bool, bool]: """ Check if full name contains first, middle, or last name as substring Returns: (first_match, middle_match, last_match) """ f_match = check_substring_match(name_full, fname) if fname else False m_match = check_substring_match(name_full, mname) if mname else False l_match = check_substring_match(name_full, lname) if lname else False return f_match, m_match, l_match # def calculate_similarity_with_models(text1: str, text2: str) -> Dict: # """ # Calculate similarity using fuzzy scores and embedding models # Returns complete match result with similarity percentage # """ # if not text1 or not text2: # return 0 # # Calculate fuzzy scores # fuzzy_scores = { # "simple_ratio": fuzz.ratio(text1, text2), # "token_set_ratio": fuzz.token_set_ratio(text1, text2), # "w_ratio": fuzz.WRatio(text1, text2), # "partial_ratio": fuzz.partial_ratio(text1, text2), # } # # Calculate semantic similarity using both models # with ThreadPoolExecutor() as executor: # model1 = MODEL_STORE["model1"] # model2 = MODEL_STORE["model2"] # f1 = executor.submit(lambda: cosine_similarity( # model1.encode([text1]), model1.encode([text2]))[0][0]) # f2 = executor.submit(lambda: cosine_similarity( # model2.encode([text1]), model2.encode([text2]))[0][0]) # cosine1 = f1.result() # cosine2 = f2.result() # # Calculate final scores # weights = { # "simple_ratio": 0.15, # "token_set_ratio": 0.40, # "partial_ratio": 0.20, # "w_ratio": 0.05, # "semantic_score": 0.20, # } # def calc_final(fuzzy, semantic): # normalized = { # "simple_ratio": fuzzy["simple_ratio"], # "token_set_ratio": fuzzy["token_set_ratio"], # "partial_ratio": fuzzy["partial_ratio"], # "w_ratio": fuzzy["w_ratio"], # "semantic_score": semantic * 100, # } # return sum(normalized[k] * weights[k] for k in weights.keys()) # final1 = calc_final(fuzzy_scores, cosine1) # final2 = calc_final(fuzzy_scores, cosine2) # overall_similarity = final1 * 0.6 + final2 * 0.4 # similarity_threshold = 85 # match_status = "Match" if overall_similarity >= similarity_threshold else "No Match" # confidence = 100 if (final1 >= similarity_threshold and final2 >= similarity_threshold) or \ # (final1 < similarity_threshold and final2 < similarity_threshold) else 50 # return round(overall_similarity, 2) def concatenate_name_parts(firstname: str, middlename: str, lastname: str) -> str: """Concatenate name parts""" parts = [] if firstname and firstname not in ["-", " ", ""]: parts.append(firstname.upper().strip()) if middlename and middlename not in ["-", " ", ""]: parts.append(middlename.upper().strip()) if lastname and lastname not in ["-", " ", ""]: parts.append(lastname.upper().strip()) if not parts: return "" parts.sort() return " ".join(parts) # def match_names_cross_records(r1_name: str, r1_firstname: str, r1_lastname: str, r1_middlename: str, # r2_name: str, r2_firstname: str, r2_lastname: str, r2_middlename: str) -> Dict: # """ # Match names between two records with three cases: # Case 1: Both have full names - ignore F/M/L fields # Case 2: One has full name, other has F/M/L # Case 3: Both have F/M/L # """ # # Preprocess all inputs # r1_name_proc = r1_name.upper().strip() if r1_name and r1_name not in ["-", " ", ""] else "" # r2_name_proc = r2_name.upper().strip() if r2_name and r2_name not in ["-", " ", ""] else "" # # Determine which case we're in # r1_has_fullname = bool(r1_name_proc) # r2_has_fullname = bool(r2_name_proc) # # CASE 1: Both records have full names # # If both have full names, ignore F/M/L fields completely # if r1_has_fullname and r2_has_fullname: # return handle_case1(r1_name_proc, r2_name_proc) # # Only process F/M/L fields if we're not in Case 1 # r1_fname = r1_firstname.upper().strip() if r1_firstname and r1_firstname not in ["-", " ", ""] else "" # r1_mname = r1_middlename.upper().strip() if r1_middlename and r1_middlename not in ["-", " ", ""] else "" # r1_lname = r1_lastname.upper().strip() if r1_lastname and r1_lastname not in ["-", " ", ""] else "" # r2_fname = r2_firstname.upper().strip() if r2_firstname and r2_firstname not in ["-", " ", ""] else "" # r2_mname = r2_middlename.upper().strip() if r2_middlename and r2_middlename not in ["-", " ", ""] else "" # r2_lname = r2_lastname.upper().strip() if r2_lastname and r2_lastname not in ["-", " ", ""] else "" # r1_concat = concatenate_name_parts(r1_fname, r1_mname, r1_lname) # r2_concat = concatenate_name_parts(r2_fname, r2_mname, r2_lname) # # CASE 2: One has full name, other has F/M/L # if r1_has_fullname and not r2_has_fullname and r2_concat: # return handle_case2(r1_name_proc, r2_fname, r2_mname, r2_lname, r2_concat) # elif r2_has_fullname and not r1_has_fullname and r1_concat: # return handle_case2(r2_name_proc, r1_fname, r1_mname, r1_lname, r1_concat) # # CASE 3: Both have F/M/L # elif not r1_has_fullname and not r2_has_fullname and r1_concat and r2_concat: # return handle_case3(r1_fname, r1_mname, r1_lname, r1_concat, # r2_fname, r2_mname, r2_lname, r2_concat) # # Missing data # return 0 # ---------- helpers used only inside the new logic ---------- def _normalize_and_sort(name: str) -> str: """ 1. Split on any non-alphanumeric character (space, underscore, comma, etc.) 2. Remove empty tokens 3. Upper-case 4. Sort alphabetically 5. Re-join with single space """ tokens = re.split(r'[^A-Za-z0-9]+', name.strip()) tokens = [t.upper() for t in tokens if t] return ' '.join(sorted(tokens)) def _all_name_combinations(fname: str, mname: str, lname: str) -> list[str]: """ Return every possible ordering of the supplied parts, dropping any empty/blank components. """ parts = [] for p in (fname, mname, lname): if p and p.strip() not in ('-', '', ' '): parts.append(p.strip().upper()) if not parts: return [] # itertools.permutations gives every ordering return [' '.join(order) for order in itertools.permutations(parts)] # ----------------------------------------------------------- # def handle_case1(full_name1: str, full_name2: str) -> Dict: # """ # Case-1 (both records supply a full name) # New rule: alphabetically-sort the tokens of each name before comparison. # """ # if not full_name1 or not full_name2: # return 0 # # 1. normalise + alphabetically sort each full name # sorted1 = _normalize_and_sort(full_name1) # sorted2 = _normalize_and_sort(full_name2) # # 2. fast substring check on the *sorted* version # if check_substring_match(sorted1, sorted2): # result = calculate_similarity_with_models(sorted1, sorted2) # return result # # 3. fall back to full model evaluation # return calculate_similarity_with_models(sorted1, sorted2) # def handle_case2(full_name: str, # fname: str, mname: str, lname: str, # concat_name: str) -> Dict: # """ # Case-2 (one side has full name, the other has F/M/L) # Step-0: build every possible ordering of F/M/L. # If **any** ordering == full_name → treat as identical. # Otherwise continue with the original individual-token checks. # """ # # 0. try every permutation of F/M/L # for permuted in _all_name_combinations(fname, mname, lname): # if permuted == full_name.upper().strip(): # # we consider the names identical # return {"Result": "Match", "Confidence Score": 100, "Overall Similarity": 100.0} # # 1. no permutation matched → proceed with original logic # f_match, m_match, l_match = check_individual_name_matches( # full_name.upper().strip(), # fname.upper().strip() if fname else "", # mname.upper().strip() if mname else "", # lname.upper().strip() if lname else "" # ) # if not f_match and not m_match and not l_match: # result = calculate_similarity_with_models(full_name, concat_name) # if result["Result"] == "No Match": # return calculate_similarity_with_models(full_name, concat_name) # return result # if f_match and not m_match and not l_match: # result = calculate_similarity_with_models(full_name, concat_name) # result["Result"] = "Match (family match)" # return result # if m_match or l_match: # result = calculate_similarity_with_models(full_name, concat_name) # result["Result"] = "Match (partial match)" # return result # return calculate_similarity_with_models(full_name, concat_name) # def handle_case3(r1_fname: str, r1_mname: str, r1_lname: str, r1_concat: str, # r2_fname: str, r2_mname: str, r2_lname: str, r2_concat: str) -> Dict: # """ # Handle Case 3: Both records have F/M/L # """ # # Check substring matches for each component # f_match = check_substring_match(r1_fname, r2_fname) if r1_fname and r2_fname else False # m_match = check_substring_match(r1_mname, r2_mname) if r1_mname and r2_mname else False # l_match = check_substring_match(r1_lname, r2_lname) if r1_lname and r2_lname else False # # Rule 1: No matches at all # if not f_match and not m_match and not l_match: # # Try substring comparison first, then models # result = calculate_similarity_with_models(r1_concat, r2_concat) # if result["Result"] == "No Match": # print("concat name1 in case3",r1_concat) # print("concat name2 in case3",r2_concat) # return calculate_similarity_with_models(r1_concat, r2_concat) # return result # # Rule 2: Only first name matches (family match) # if f_match and not m_match and not l_match: # result = calculate_similarity_with_models(r1_concat, r2_concat) # result["Result"] = "Match (family match)" # return result # # Rule 3: First name + (middle or last) matches (partial match) # if (m_match or l_match): # result = calculate_similarity_with_models(r1_concat, r2_concat) # result["Result"] = "Match (partial match)" # return result # # Fallback: pass to models # return calculate_similarity_with_models(r1_concat, r2_concat) # def match_name(name: str, firstname: str, lastname: str, middlename: str) -> Dict: # """ # Match name with logic: # 1. If NAME == concat(F,M,L alphabetically) -> use NAME # 2. If NAME is null/"-"/" " -> use concat(F,M,L) # 3. If concat(F,M,L) is null/"-"/" " -> use NAME # 4. If NAME != concat(F,M,L) -> pass both to model, match if either matches # """ # name_processed = preprocess_for_matching(name) # concat_name = concatenate_name_parts(firstname, middlename, lastname) # # Case 1: NAME matches concatenated name # if name_processed and concat_name and name_processed == concat_name: # return 100 # # Case 2: NAME is empty, use concatenated # if not name_processed and concat_name: # return 100 # # Case 3: Concat is empty, use NAME # if name_processed and not concat_name: # return 100 # # Case 4: Both exist but different - use model # if name_processed and concat_name and name_processed != concat_name: # # Pass both to model for fuzzy matching # return match_entities(name_processed, concat_name) # # Both empty # return 0 # # ---------- Address Matching Logic (1:N) ---------- # def match_addresses_1_to_n(addresses_r1: List[str], addresses_r2: List[str]) -> Dict: # """ # Match addresses 1:N - if any address in R1 matches any in R2, return Match # Returns the best match found # """ # valid_addr1 = [preprocess_for_matching(addr) for addr in addresses_r1 if addr and addr not in ["-", " ", ""]] # valid_addr2 = [preprocess_for_matching(addr) for addr in addresses_r2 if addr and addr not in ["-", " ", ""]] # if not valid_addr1 or not valid_addr2: # return 0 # best_result = {"Result": "No Match", "Confidence Score": 0, "Overall Similarity": 0} # # Compare each address in R1 with each in R2 # for addr1 in valid_addr1: # for addr2 in valid_addr2: # result = match_entities(addr1, addr2) # # if result["Overall Similarity"] > best_result["Overall Similarity"]: # # best_result = result # return result # # ---------- Single Field Matching ---------- # def match_single_field(value1: str, value2: str) -> Dict: # """Match single fields like SPOUSENAME, MOTHERNAME, etc.""" # return match_entities(value1, value2) def match_entities(value1: str, value2: str) -> float: """ Match two entities using fuzzy + semantic similarity Returns: similarity score as float (0-100) """ standardized_input1 = preprocess_for_matching(value1) standardized_input2 = preprocess_for_matching(value2) if not standardized_input1 or not standardized_input2: return 0 # Calculate fuzzy scores fuzzy_match_scores = calculate_fuzzy_scores(standardized_input1, standardized_input2) # Calculate semantic similarity using both models in parallel with ThreadPoolExecutor() as executor: f1 = executor.submit(calculate_semantic_similarity, "model1", standardized_input1, standardized_input2) f2 = executor.submit(calculate_semantic_similarity, "model2", standardized_input1, standardized_input2) cosine1 = f1.result() cosine2 = f2.result() ff1 = executor.submit(calculate_final_score, fuzzy_match_scores, cosine1) ff2 = executor.submit(calculate_final_score, fuzzy_match_scores, cosine2) final1 = ff1.result() final2 = ff2.result() overall_similarity = calculate_overall_similarity(final1, final2) return round(overall_similarity, 2) def calculate_similarity_with_models(text1: str, text2: str) -> float: """ Calculate similarity using fuzzy scores and embedding models Returns similarity percentage as float """ if not text1 or not text2: return 0 # Calculate fuzzy scores fuzzy_scores = { "simple_ratio": fuzz.ratio(text1, text2), "token_set_ratio": fuzz.token_set_ratio(text1, text2), "w_ratio": fuzz.WRatio(text1, text2), "partial_ratio": fuzz.partial_ratio(text1, text2), } # Calculate semantic similarity using both models with ThreadPoolExecutor() as executor: model1 = MODEL_STORE["model1"] model2 = MODEL_STORE["model2"] f1 = executor.submit(lambda: cosine_similarity( model1.encode([text1]), model1.encode([text2]))[0][0]) f2 = executor.submit(lambda: cosine_similarity( model2.encode([text1]), model2.encode([text2]))[0][0]) cosine1 = f1.result() cosine2 = f2.result() # Calculate final scores weights = { "simple_ratio": 0.15, "token_set_ratio": 0.40, "partial_ratio": 0.20, "w_ratio": 0.05, "semantic_score": 0.20, } def calc_final(fuzzy, semantic): normalized = { "simple_ratio": fuzzy["simple_ratio"], "token_set_ratio": fuzzy["token_set_ratio"], "partial_ratio": fuzzy["partial_ratio"], "w_ratio": fuzzy["w_ratio"], "semantic_score": semantic * 100, } return sum(normalized[k] * weights[k] for k in weights.keys()) final1 = calc_final(fuzzy_scores, cosine1) final2 = calc_final(fuzzy_scores, cosine2) overall_similarity = final1 * 0.6 + final2 * 0.4 return round(overall_similarity, 2) def handle_case1(full_name1: str, full_name2: str) -> float: """ Case-1 (both records supply a full name) Returns similarity score as float """ if not full_name1 or not full_name2: return 0 # 1. normalise + alphabetically sort each full name sorted1 = _normalize_and_sort(full_name1) sorted2 = _normalize_and_sort(full_name2) # 2. fast substring check on the *sorted* version if check_substring_match(sorted1, sorted2): result = calculate_similarity_with_models(sorted1, sorted2) return result # 3. fall back to full model evaluation return calculate_similarity_with_models(sorted1, sorted2) # # def handle_case2(full_name: str, # # fname: str, mname: str, lname: str, # # concat_name: str) -> float: # # """ # # Case-2 (one side has full name, the other has F/M/L) # # Returns similarity score as float # # """ # # # 0. try every permutation of F/M/L # # for permuted in _all_name_combinations(fname, mname, lname): # # if permuted == full_name.upper().strip(): # # # we consider the names identical # # return 100.0 # # # 1. no permutation matched → proceed with original logic # # f_match, m_match, l_match = check_individual_name_matches( # # full_name.upper().strip(), # # fname.upper().strip() if fname else "", # # mname.upper().strip() if mname else "", # # lname.upper().strip() if lname else "" # # ) # # if not f_match and not m_match and not l_match: # # result = calculate_similarity_with_models(full_name, concat_name) # # return result # # if f_match and not m_match and not l_match: # # result = calculate_similarity_with_models(full_name, concat_name) # # return result # # if m_match or l_match: # # result = calculate_similarity_with_models(full_name, concat_name) # # return result # # return calculate_similarity_with_models(full_name, concat_name) # def handle_case3(r1_fname: str, r1_mname: str, r1_lname: str, r1_concat: str, # r2_fname: str, r2_mname: str, r2_lname: str, r2_concat: str) -> float: # """ # Handle Case 3: Both records have F/M/L # Returns similarity score as float # """ # # Check substring matches for each component # f_match = check_substring_match(r1_fname, r2_fname) if r1_fname and r2_fname else False # m_match = check_substring_match(r1_mname, r2_mname) if r1_mname and r2_mname else False # l_match = check_substring_match(r1_lname, r2_lname) if r1_lname and r2_lname else False # # Rule 1: No matches at all # if not f_match and not m_match and not l_match: # # Try substring comparison first, then models # result = calculate_similarity_with_models(r1_concat, r2_concat) # return result # # Rule 2: Only first name matches (family match) # if f_match and not m_match and not l_match: # result = calculate_similarity_with_models(r1_concat, r2_concat) # return result # # Rule 3: First name + (middle or last) matches (partial match) # if (m_match or l_match): # result = calculate_similarity_with_models(r1_concat, r2_concat) # return result # # Fallback: pass to models # return calculate_similarity_with_models(r1_concat, r2_concat) def handle_case2(full_name: str, fname: str, mname: str, lname: str, concat_name: str) -> float: """ Case-2 (one side has full name, the other has F/M/L) Returns similarity score as float UPDATED: Checks lastname for family match instead of firstname """ # 0. try every permutation of F/M/L for permuted in _all_name_combinations(fname, mname, lname): if permuted == full_name.upper().strip(): # we consider the names identical return 100.0 # 1. no permutation matched → proceed with original logic f_match, m_match, l_match = check_individual_name_matches( full_name.upper().strip(), fname.upper().strip() if fname else "", mname.upper().strip() if mname else "", lname.upper().strip() if lname else "" ) # NEW LOGIC: If no matches at all, compute similarity if not f_match and not m_match and not l_match: result = calculate_similarity_with_models(full_name, concat_name) return result # NEW LOGIC: If ONLY lastname matches → family match # This is the key change - checking lastname instead of firstname if l_match and not f_match and not m_match: result = calculate_similarity_with_models(full_name, concat_name) # Return higher score for family match or boost the score return max(result, 85.0) # Ensure minimum 85% for family match # If firstname or middlename matches (partial match) if f_match or m_match: result = calculate_similarity_with_models(full_name, concat_name) return result return calculate_similarity_with_models(full_name, concat_name) def handle_case3(r1_fname: str, r1_mname: str, r1_lname: str, r1_concat: str, r2_fname: str, r2_mname: str, r2_lname: str, r2_concat: str) -> float: """ Handle Case 3: Both records have F/M/L Returns similarity score as float UPDATED: Checks lastname for family match instead of firstname """ # Check substring matches for each component f_match = check_substring_match(r1_fname, r2_fname) if r1_fname and r2_fname else False m_match = check_substring_match(r1_mname, r2_mname) if r1_mname and r2_mname else False l_match = check_substring_match(r1_lname, r2_lname) if r1_lname and r2_lname else False # Rule 1: No matches at all if not f_match and not m_match and not l_match: result = calculate_similarity_with_models(r1_concat, r2_concat) return result # NEW LOGIC: Rule 2 - Only lastname matches (family match) # This is the key change - checking lastname instead of firstname if l_match and not f_match and not m_match: result = calculate_similarity_with_models(r1_concat, r2_concat) # Return higher score for family match or boost the score return max(result, 85.0) # Ensure minimum 85% for family match # Rule 3: Lastname + (firstname or middle) matches (partial match) # Strong indicator of same person if l_match and (f_match or m_match): result = calculate_similarity_with_models(r1_concat, r2_concat) return max(result, 90.0) # Higher confidence when lastname + another field matches # Rule 4: Only firstname or middlename matches (weaker match) if f_match or m_match: result = calculate_similarity_with_models(r1_concat, r2_concat) return result # Fallback: pass to models return calculate_similarity_with_models(r1_concat, r2_concat) def match_name(name: str, firstname: str, lastname: str, middlename: str) -> float: """ Match name with logic Returns similarity score as float or "missing value" """ name_processed = preprocess_for_matching(name) concat_name = concatenate_name_parts(firstname, middlename, lastname) # Case 1: NAME matches concatenated name if name_processed and concat_name and name_processed == concat_name: return 100 # Case 2: NAME is empty, use concatenated if not name_processed and concat_name: return 100 # Case 3: Concat is empty, use NAME if name_processed and not concat_name: return 100 # Case 4: Both exist but different - use model if name_processed and concat_name and name_processed != concat_name: # Pass both to model for fuzzy matching return match_entities(name_processed, concat_name) # Both empty return 0 def match_names_cross_records(r1_name: str, r1_firstname: str, r1_lastname: str, r1_middlename: str, r2_name: str, r2_firstname: str, r2_lastname: str, r2_middlename: str) -> float: """ Match names between two records with three cases Returns similarity score as float or "missing value" """ # Preprocess all inputs r1_name_proc = r1_name.upper().strip() if r1_name and r1_name not in ["-", " ", ""] else "" r2_name_proc = r2_name.upper().strip() if r2_name and r2_name not in ["-", " ", ""] else "" # Determine which case we're in r1_has_fullname = bool(r1_name_proc) r2_has_fullname = bool(r2_name_proc) # CASE 1: Both records have full names if r1_has_fullname and r2_has_fullname: return handle_case1(r1_name_proc, r2_name_proc) # Only process F/M/L fields if we're not in Case 1 r1_fname = r1_firstname.upper().strip() if r1_firstname and r1_firstname not in ["-", " ", ""] else "" r1_mname = r1_middlename.upper().strip() if r1_middlename and r1_middlename not in ["-", " ", ""] else "" r1_lname = r1_lastname.upper().strip() if r1_lastname and r1_lastname not in ["-", " ", ""] else "" r2_fname = r2_firstname.upper().strip() if r2_firstname and r2_firstname not in ["-", " ", ""] else "" r2_mname = r2_middlename.upper().strip() if r2_middlename and r2_middlename not in ["-", " ", ""] else "" r2_lname = r2_lastname.upper().strip() if r2_lastname and r2_lastname not in ["-", " ", ""] else "" r1_concat = concatenate_name_parts(r1_fname, r1_mname, r1_lname) r2_concat = concatenate_name_parts(r2_fname, r2_mname, r2_lname) # CASE 2: One has full name, other has F/M/L if r1_has_fullname and not r2_has_fullname and r2_concat: return handle_case2(r1_name_proc, r2_fname, r2_mname, r2_lname, r2_concat) elif r2_has_fullname and not r1_has_fullname and r1_concat: return handle_case2(r2_name_proc, r1_fname, r1_mname, r1_lname, r1_concat) # CASE 3: Both have F/M/L elif not r1_has_fullname and not r2_has_fullname and r1_concat and r2_concat: return handle_case3(r1_fname, r1_mname, r1_lname, r1_concat, r2_fname, r2_mname, r2_lname, r2_concat) # Missing data return 0 def match_addresses_1_to_n(addresses_r1: List[str], addresses_r2: List[str]) -> float: """ Match addresses 1:N - if any address in R1 matches any in R2 Returns similarity score as float or "missing value" """ valid_addr1 = [preprocess_for_matching(addr) for addr in addresses_r1 if addr and addr not in ["-", " ", ""]] valid_addr2 = [preprocess_for_matching(addr) for addr in addresses_r2 if addr and addr not in ["-", " ", ""]] if not valid_addr1 or not valid_addr2: return 0 best_score = 0 # Compare each address in R1 with each in R2 for addr1 in valid_addr1: for addr2 in valid_addr2: result = match_entities(addr1, addr2) # Convert to float to handle numpy types try: score = float(result) if score > best_score: best_score = score except (TypeError, ValueError): # If conversion fails, skip this result continue return best_score def match_single_field(value1: str, value2: str) -> float: """ Match single fields like SPOUSENAME, MOTHERNAME, etc. Returns similarity score as float or "missing value" """ return match_entities(value1, value2)