File size: 5,891 Bytes
e28a7b2 76af018 e28a7b2 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 |
"""Core duplicate-detection logic."""
from __future__ import annotations
import logging
from collections import defaultdict, deque
from datetime import datetime
from decimal import Decimal
from typing import Dict, Iterable, List, Sequence, Set, Tuple
from .config import settings
from .merchant_alias import MerchantAliasResolver, normalize_merchant
from .models import DuplicateCluster, Expense, MergeSuggestion
from .repositories import MergeSuggestionRepository
logger = logging.getLogger("DuplicateDetector")
def _pct_delta(a: Decimal, b: Decimal) -> float:
if a == 0:
return float("inf")
return abs(float((a - b) / a * Decimal(100)))
def _minutes_delta(a: datetime, b: datetime) -> float:
return abs((a - b).total_seconds() / 60)
class DuplicateDetector:
def __init__(
self,
*,
alias_resolver: MerchantAliasResolver,
suggestions_repo: MergeSuggestionRepository,
amount_tolerance_pct: float | None = None,
time_tolerance_minutes: int | None = None,
) -> None:
self.alias_resolver = alias_resolver
self.suggestions_repo = suggestions_repo
self.amount_tolerance_pct = amount_tolerance_pct or float(settings.amount_tolerance_pct)
self.time_tolerance_minutes = time_tolerance_minutes or settings.time_tolerance_minutes
def _build_graph(self, expenses: Sequence[Expense]) -> Dict[int, Set[int]]:
adjacency: Dict[int, Set[int]] = defaultdict(set)
for i, exp_a in enumerate(expenses):
for j in range(i + 1, len(expenses)):
exp_b = expenses[j]
if exp_a.user_id and exp_b.user_id and exp_a.user_id != exp_b.user_id:
continue
delta_minutes = _minutes_delta(exp_a.expense_time, exp_b.expense_time)
if delta_minutes > self.time_tolerance_minutes:
break
amount_delta_pct = _pct_delta(exp_a.amount, exp_b.amount)
if amount_delta_pct > self.amount_tolerance_pct:
continue
alias_match, alias_rule = self.alias_resolver.are_aliases(
exp_a.merchant,
exp_b.merchant,
)
if alias_match:
adjacency[i].add(j)
adjacency[j].add(i)
return adjacency
def _clusters_from_graph(
self,
adjacency: Dict[int, Set[int]],
expenses: Sequence[Expense],
) -> List[DuplicateCluster]:
visited: Set[int] = set()
clusters: List[DuplicateCluster] = []
for node in range(len(expenses)):
if node in visited or node not in adjacency:
continue
component_nodes: List[int] = []
queue: deque[int] = deque([node])
while queue:
current = queue.popleft()
if current in visited:
continue
visited.add(current)
component_nodes.append(current)
for neighbor in adjacency[current]:
if neighbor not in visited:
queue.append(neighbor)
if len(component_nodes) <= 1:
continue
component_nodes.sort()
component_expenses = [expenses[idx] for idx in component_nodes]
amounts = [exp.amount for exp in component_expenses]
times = [exp.expense_time for exp in component_expenses]
amount_delta_pct = _pct_delta(min(amounts), max(amounts))
time_delta_minutes = _minutes_delta(min(times), max(times))
merchant_rule = self._merchant_rule(component_expenses)
clusters.append(
DuplicateCluster(
expenses=component_expenses,
amount_delta_pct=amount_delta_pct,
time_delta_minutes=time_delta_minutes,
merchant_rule=merchant_rule,
),
)
return clusters
def _merchant_rule(self, expenses: Sequence[Expense]) -> str:
normalized = {normalize_merchant(exp.merchant) for exp in expenses}
if len(normalized) == 1:
return "exact"
return "alias"
def find_clusters(self, expenses: Sequence[Expense]) -> List[DuplicateCluster]:
if not expenses:
return []
sorted_expenses = sorted(expenses, key=lambda e: e.expense_time)
graph = self._build_graph(sorted_expenses)
clusters = self._clusters_from_graph(graph, sorted_expenses)
logger.info("Evaluated %d expenses, found %d clusters", len(expenses), len(clusters))
return clusters
def persist_suggestions(self, clusters: Iterable[DuplicateCluster]) -> List[str]:
suggestion_ids: List[str] = []
for cluster in clusters:
candidate_ids = [expense.expense_id for expense in cluster.expenses]
tie_breaker = "same purchase?" if len(candidate_ids) > 2 else None
details = cluster.to_details()
if tie_breaker:
details["tie_breaker"] = tie_breaker
suggestion = MergeSuggestion(
candidate_ids=candidate_ids,
message="These seem similar. Would you like to merge them?",
details=details,
audit={
"generated_by": settings.service_name,
"generated_at": datetime.utcnow(),
"rule_version": "v1.0",
},
)
suggestion_id = self.suggestions_repo.insert_soft_merge(suggestion)
suggestion_ids.append(suggestion_id)
logger.info(
"Recorded merge suggestion %s for candidates %s",
suggestion_id,
candidate_ids,
)
return suggestion_ids
|