from collections import defaultdict from typing import Optional import pandas as pd from backend.utils.addressMatch import AddressMatch def resolve_group_ambiguity( matches: list[tuple[Optional[str], str]], ) -> list[tuple[Optional[str], str]]: """Resolve cross-row UPRN ambiguity within one postcode group (ADR-0057). ``matches`` is ``(uprn, normalised_address)`` per row, in order. Each row is matched independently, so nothing stops one UPRN being the best match for two *different* addresses — almost always a coarse EPC record absorbing several real addresses (e.g. flats in a block matched to a flat-less record). Withhold that UPRN on every such row, so a distinct address is never coerced onto a shared UPRN: the downstream ``property`` identity insert (``on_conflict_do_nothing`` on ``(portfolio_id, uprn)``) would otherwise silently merge them, and the ``property_overrides`` upsert would then collide on ``(property_id, override_component, building_part)``. A UPRN shared only by rows with the *same* normalised address is a genuine re-listing of one property and is kept. Returns ``(uprn, status)`` per row in input order, where status is ``"matched"`` (kept), ``"ambiguous_duplicate"`` (withheld → uprn ``None``), or ``"unmatched"`` (input uprn was already ``None``). Withheld rows keep their lexiscore upstream for triage on the confirmation page. """ distinct_addresses: dict[str, set[str]] = defaultdict(set) for uprn, norm_address in matches: if uprn: distinct_addresses[uprn].add(norm_address) resolved: list[tuple[Optional[str], str]] = [] for uprn, _norm_address in matches: if not uprn: resolved.append((None, "unmatched")) elif len(distinct_addresses[uprn]) > 1: resolved.append((None, "ambiguous_duplicate")) else: resolved.append((uprn, "matched")) return resolved def all_uprns_match( df: pd.DataFrame, target_uprn: str, column: str = "uprn", ) -> bool: if column not in df.columns: return False uprns = df[column].dropna().astype(str).str.strip().unique() if len(uprns) == 0: return False return len(uprns) == 1 and uprns[0] == str(target_uprn) def rank_address_similarity( address_list_df: pd.DataFrame, user_address: str, address_column: str = "address", uprn_column: str = "uprn", ) -> pd.DataFrame: """ Annotate EPC results with lexicographical similarity scores and ranks. Returns a DataFrame sorted by descending lexiscore. DOES NOT choose or return a UPRN. """ if address_column not in address_list_df.columns: raise ValueError(f"Missing column: {address_column}") if uprn_column not in address_list_df.columns: raise ValueError(f"Missing column: {uprn_column}") out = address_list_df.copy() user_norm = AddressMatch.normalise_address(user_address) out["lexiscore"] = out[address_column].apply( lambda x: AddressMatch.levenshtein(user_norm, x) ) out[uprn_column] = out[uprn_column].astype(str).str.replace(r"\.0$", "", regex=True) out["lexirank"] = out["lexiscore"].rank(method="dense", ascending=False).astype(int) return out.sort_values( ["lexirank", "lexiscore"], ascending=[True, False], )