75 lines
2.7 KiB
Python
75 lines
2.7 KiB
Python
"""Aggregate per-partner reconciliation patterns from precedent rows.
|
||
|
||
Computes typical amount range, cadence, preferred strategy, common memo
|
||
tokens. Output is a dict suitable for create/write on fusion.reconcile.pattern.
|
||
"""
|
||
|
||
from collections import Counter
|
||
from statistics import median
|
||
|
||
|
||
def extract_pattern_for_partner(env, *, company_id, partner_id) -> dict:
|
||
"""Compute the pattern aggregate for one (company, partner) pair.
|
||
|
||
Returns vals dict suitable for env['fusion.reconcile.pattern'].create()."""
|
||
Precedent = env['fusion.reconcile.precedent'].sudo()
|
||
precedents = Precedent.search([
|
||
('company_id', '=', company_id),
|
||
('partner_id', '=', partner_id),
|
||
], order='reconciled_at desc', limit=200)
|
||
|
||
if not precedents:
|
||
return {
|
||
'company_id': company_id,
|
||
'partner_id': partner_id,
|
||
'reconcile_count': 0,
|
||
}
|
||
|
||
amounts = sorted(precedents.mapped('amount'))
|
||
counts = precedents.mapped('matched_move_line_count')
|
||
|
||
single_count = sum(1 for c in counts if c == 1)
|
||
multi_count = sum(1 for c in counts if c > 1)
|
||
if multi_count > single_count:
|
||
pref_strategy = 'multi_invoice'
|
||
elif _amounts_concentrated(amounts):
|
||
pref_strategy = 'exact_amount'
|
||
else:
|
||
pref_strategy = 'fifo'
|
||
|
||
reconcile_dates = sorted([p.reconciled_at for p in precedents if p.reconciled_at])
|
||
if len(reconcile_dates) >= 2:
|
||
deltas = [(reconcile_dates[i+1] - reconcile_dates[i]).days
|
||
for i in range(len(reconcile_dates) - 1)]
|
||
cadence = sum(deltas) / len(deltas) if deltas else 0.0
|
||
else:
|
||
cadence = 0.0
|
||
|
||
token_counter = Counter()
|
||
for p in precedents:
|
||
if p.memo_tokens:
|
||
for tok in p.memo_tokens.split(','):
|
||
token_counter[tok.strip()] += 1
|
||
# Keep tokens appearing in >=30% of precedents (min floor of 2 occurrences)
|
||
threshold = max(2, len(precedents) * 0.3)
|
||
common_tokens = ','.join(t for t, c in token_counter.most_common() if c >= threshold)
|
||
|
||
return {
|
||
'company_id': company_id,
|
||
'partner_id': partner_id,
|
||
'reconcile_count': len(precedents),
|
||
'typical_amount_range': f"${min(amounts):,.2f} – ${max(amounts):,.2f} (median ${median(amounts):,.2f})",
|
||
'typical_cadence_days': round(cadence, 1),
|
||
'pref_strategy': pref_strategy,
|
||
'common_memo_tokens': common_tokens,
|
||
}
|
||
|
||
|
||
def _amounts_concentrated(amounts: list[float]) -> bool:
|
||
"""True if amounts cluster around a few values (suggests exact-amount strategy)."""
|
||
if len(amounts) < 3:
|
||
return True
|
||
med = median(amounts)
|
||
within_5pct = sum(1 for a in amounts if abs(a - med) / max(med, 1) < 0.05)
|
||
return within_5pct / len(amounts) >= 0.6
|