feat(fusion_accounting_followup): level_resolver service

Made-with: Cursor

fusion_accounting_followup/__manifest__.py

@@ -1,6 +1,6 @@
 {
     'name': 'Fusion Accounting Follow-up',
-    'version': '19.0.1.0.1',
+    'version': '19.0.1.0.2',
     'category': 'Accounting/Accounting',
     'summary': 'AI-augmented customer follow-ups (dunning) for unpaid invoices.',
     'description': """

fusion_accounting_followup/services/__init__.py

@@ -1 +1,2 @@
 from . import overdue_aging
+from . import level_resolver

fusion_accounting_followup/services/level_resolver.py

@@ -0,0 +1,52 @@
+"""Level resolver: which follow-up level should fire for this partner?
+
+Pure-Python: caller passes the aging report + the configured levels list,
+and we pick the highest-numbered level whose threshold is met."""
+
+from dataclasses import dataclass
+
+
+@dataclass
+class FollowupLevelSpec:
+    sequence: int
+    name: str
+    delay_days: int
+    tone: str
+
+    def __post_init__(self):
+        if self.tone not in ('gentle', 'firm', 'legal'):
+            raise ValueError(f"Invalid tone: {self.tone}")
+
+
+def resolve_level(*, aging_report, levels: list[FollowupLevelSpec]) -> FollowupLevelSpec | None:
+    """Pick the highest-sequence level whose delay_days has been crossed by
+    the most-overdue line in the aging report. Returns None if no overdue
+    lines or no levels configured."""
+    if not levels or not aging_report:
+        return None
+    max_days_overdue = _max_days_overdue(aging_report)
+    if max_days_overdue <= 0:
+        return None
+    levels_sorted = sorted(levels, key=lambda l: l.sequence, reverse=True)
+    for level in levels_sorted:
+        if level.delay_days <= max_days_overdue:
+            return level
+    return None
+
+
+def _max_days_overdue(aging_report) -> int:
+    """Return the actual max days-overdue tracked on the report, falling
+    back to the highest populated bucket's lower bound when an older
+    aging report (without `max_days_overdue`) is passed in."""
+    tracked = getattr(aging_report, 'max_days_overdue', 0) or 0
+    if tracked:
+        return tracked
+    max_days = 0
+    for b in aging_report.buckets:
+        if b.name == 'current' or b.amount <= 0:
+            continue
+        if b.days_max is None:
+            max_days = max(max_days, b.days_min)
+        else:
+            max_days = max(max_days, b.days_min)
+    return max_days

fusion_accounting_followup/services/overdue_aging.py

@@ -33,6 +33,7 @@ class AgingReport:
     total_amount: float = 0.0
     total_overdue_amount: float = 0.0
     line_count: int = 0
+    max_days_overdue: int = 0
 
     def to_dict(self):
         return {
@@ -40,6 +41,7 @@ class AgingReport:
             'total_amount': self.total_amount,
             'total_overdue_amount': self.total_overdue_amount,
             'line_count': self.line_count,
+            'max_days_overdue': self.max_days_overdue,
             'buckets': [{
                 'name': b.name, 'days_min': b.days_min, 'days_max': b.days_max,
                 'amount': b.amount, 'line_count': b.line_count,
@@ -70,6 +72,8 @@ def compute_aging(*, move_lines: list[dict], as_of: date | None = None) -> Aging
         report.total_amount += amount
         if days_overdue > 0:
             report.total_overdue_amount += amount
+            if days_overdue > report.max_days_overdue:
+                report.max_days_overdue = days_overdue
         report.line_count += 1
 
     return report

fusion_accounting_followup/tests/__init__.py

@@ -1 +1,2 @@
 from . import test_overdue_aging
+from . import test_level_resolver

fusion_accounting_followup/tests/test_level_resolver.py

@@ -0,0 +1,58 @@
+from datetime import date, timedelta
+from odoo.tests.common import TransactionCase
+from odoo.tests import tagged
+from odoo.addons.fusion_accounting_followup.services.level_resolver import (
+    FollowupLevelSpec, resolve_level,
+)
+from odoo.addons.fusion_accounting_followup.services.overdue_aging import compute_aging
+
+
+@tagged('post_install', '-at_install')
+class TestLevelResolver(TransactionCase):
+
+    def setUp(self):
+        super().setUp()
+        self.levels = [
+            FollowupLevelSpec(sequence=1, name='Reminder', delay_days=7, tone='gentle'),
+            FollowupLevelSpec(sequence=2, name='Warning', delay_days=30, tone='firm'),
+            FollowupLevelSpec(sequence=3, name='Legal Notice', delay_days=60, tone='legal'),
+        ]
+
+    def test_no_overdue_returns_none(self):
+        as_of = date(2026, 4, 19)
+        lines = [{'date_maturity': as_of + timedelta(days=10), 'amount_residual': 100}]
+        report = compute_aging(move_lines=lines, as_of=as_of)
+        result = resolve_level(aging_report=report, levels=self.levels)
+        self.assertIsNone(result)
+
+    def test_15_days_overdue_picks_reminder(self):
+        as_of = date(2026, 4, 19)
+        lines = [{'date_maturity': as_of - timedelta(days=15), 'amount_residual': 100}]
+        report = compute_aging(move_lines=lines, as_of=as_of)
+        result = resolve_level(aging_report=report, levels=self.levels)
+        self.assertEqual(result.name, 'Reminder')
+
+    def test_45_days_overdue_picks_warning(self):
+        as_of = date(2026, 4, 19)
+        lines = [{'date_maturity': as_of - timedelta(days=45), 'amount_residual': 200}]
+        report = compute_aging(move_lines=lines, as_of=as_of)
+        result = resolve_level(aging_report=report, levels=self.levels)
+        self.assertEqual(result.name, 'Warning')
+
+    def test_75_days_overdue_picks_legal(self):
+        as_of = date(2026, 4, 19)
+        lines = [{'date_maturity': as_of - timedelta(days=75), 'amount_residual': 300}]
+        report = compute_aging(move_lines=lines, as_of=as_of)
+        result = resolve_level(aging_report=report, levels=self.levels)
+        self.assertEqual(result.name, 'Legal Notice')
+
+    def test_no_levels_returns_none(self):
+        as_of = date(2026, 4, 19)
+        lines = [{'date_maturity': as_of - timedelta(days=30), 'amount_residual': 100}]
+        report = compute_aging(move_lines=lines, as_of=as_of)
+        result = resolve_level(aging_report=report, levels=[])
+        self.assertIsNone(result)
+
+    def test_invalid_tone_raises(self):
+        with self.assertRaises(ValueError):
+            FollowupLevelSpec(sequence=1, name='X', delay_days=7, tone='invalid')