feat(fusion_accounting_followup): inherit account.move.line for level tracking

Made-with: Cursor

fusion_accounting_followup/__manifest__.py

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

fusion_accounting_followup/models/__init__.py

@@ -0,0 +1,5 @@
+from . import fusion_followup_level
+from . import fusion_followup_run
+from . import fusion_followup_text_cache
+from . import res_partner
+from . import account_move_line

fusion_accounting_followup/models/account_move_line.py

@@ -0,0 +1,14 @@
+"""Inherit account.move.line: track last follow-up level."""
+
+from odoo import _, api, fields, models
+
+
+class AccountMoveLine(models.Model):
+    _inherit = "account.move.line"
+
+    fusion_followup_level_id = fields.Many2one(
+        'fusion.followup.level', copy=False,
+        help="Last follow-up level at which this line was contacted.")
+    fusion_followup_last_run_date = fields.Datetime(
+        copy=False,
+        help="When the line was most-recently included in a follow-up.")

fusion_accounting_followup/models/fusion_followup_level.py

@@ -0,0 +1,42 @@
+"""Follow-up level definition (e.g. Reminder at 7 days, Warning at 30, Legal at 60)."""
+
+from odoo import _, api, fields, models
+
+
+TONE_SELECTION = [
+    ('gentle', 'Gentle'),
+    ('firm', 'Firm'),
+    ('legal', 'Legal'),
+]
+
+
+class FusionFollowupLevel(models.Model):
+    _name = "fusion.followup.level"
+    _description = "Fusion Follow-up Level"
+    _order = "sequence, id"
+
+    name = fields.Char(required=True, translate=True)
+    sequence = fields.Integer(required=True, default=10,
+                              help="Order in which levels escalate (1, 2, 3...).")
+    delay_days = fields.Integer(required=True,
+                                help="Min days overdue to trigger this level.")
+    tone = fields.Selection(TONE_SELECTION, required=True, default='gentle')
+    description = fields.Text()
+    company_id = fields.Many2one('res.company', default=lambda self: self.env.company)
+
+    mail_template_id = fields.Many2one('mail.template',
+                                       domain=[('model', '=', 'res.partner')])
+
+    requires_manual_review = fields.Boolean(default=False,
+        help="If True, follow-ups at this level need human approval before send.")
+
+    active = fields.Boolean(default=True)
+
+    _check_delay_positive = models.Constraint(
+        'CHECK(delay_days >= 0)',
+        'delay_days must be non-negative.',
+    )
+    _unique_sequence_per_company = models.Constraint(
+        'UNIQUE(company_id, sequence)',
+        'Sequence must be unique per company.',
+    )

fusion_accounting_followup/models/fusion_followup_run.py

@@ -0,0 +1,54 @@
+"""Audit record of one follow-up execution (per partner per level)."""
+
+from odoo import _, api, fields, models
+
+
+STATE_SELECTION = [
+    ('draft', 'Draft'),
+    ('sent', 'Sent'),
+    ('manual_review', 'Manual Review'),
+    ('skipped', 'Skipped'),
+    ('failed', 'Failed'),
+]
+
+
+class FusionFollowupRun(models.Model):
+    _name = "fusion.followup.run"
+    _description = "Fusion Follow-up Run (Per-Partner Audit)"
+    _order = "execution_date desc, id desc"
+
+    partner_id = fields.Many2one('res.partner', required=True, ondelete='cascade')
+    company_id = fields.Many2one('res.company', required=True,
+                                 default=lambda self: self.env.company)
+    level_id = fields.Many2one('fusion.followup.level', ondelete='restrict')
+
+    execution_date = fields.Datetime(default=fields.Datetime.now, required=True)
+    state = fields.Selection(STATE_SELECTION, default='draft', required=True)
+
+    overdue_amount = fields.Float()
+    longest_overdue_days = fields.Integer()
+
+    risk_score = fields.Integer()
+    risk_band = fields.Selection([
+        ('low', 'Low'), ('medium', 'Medium'),
+        ('high', 'High'), ('critical', 'Critical'),
+    ])
+
+    subject = fields.Char()
+    body = fields.Text()
+    tone_used = fields.Selection([
+        ('gentle', 'Gentle'), ('firm', 'Firm'), ('legal', 'Legal'),
+    ])
+    sent_to_email = fields.Char()
+
+    text_was_ai_generated = fields.Boolean(default=False)
+    ai_provider = fields.Char(help="LLM provider name (openai, claude, etc.) if AI was used.")
+
+    notes = fields.Text()
+    error_message = fields.Text()
+
+    def action_mark_sent(self):
+        self.write({'state': 'sent'})
+
+    def action_mark_failed(self, error: str = ''):
+        self.write({'state': 'failed', 'error_message': error})

fusion_accounting_followup/models/fusion_followup_text_cache.py

@@ -0,0 +1,60 @@
+"""Cache of AI-generated follow-up text to avoid LLM cost on repeats."""
+
+import hashlib
+
+from odoo import _, api, fields, models
+
+
+class FusionFollowupTextCache(models.Model):
+    _name = "fusion.followup.text.cache"
+    _description = "Cache of AI-generated follow-up text"
+    _order = "generated_at desc"
+
+    partner_id = fields.Many2one('res.partner', required=True, ondelete='cascade')
+    level_id = fields.Many2one('fusion.followup.level', ondelete='cascade')
+    company_id = fields.Many2one('res.company', required=True,
+                                 default=lambda self: self.env.company)
+
+    fingerprint = fields.Char(required=True, index=True,
+                              help="SHA-256 of input parameters")
+
+    subject = fields.Char()
+    body = fields.Text()
+    tone_used = fields.Selection([
+        ('gentle', 'Gentle'), ('firm', 'Firm'), ('legal', 'Legal'),
+    ])
+    key_points = fields.Json()
+
+    generated_at = fields.Datetime(default=fields.Datetime.now, required=True)
+    expires_at = fields.Datetime()
+    use_count = fields.Integer(default=0)
+    provider = fields.Char()
+
+    @api.model
+    def compute_fingerprint(self, *, partner_id: int, level_id: int,
+                            overdue_amount: float, longest_overdue_days: int,
+                            invoice_count: int, tone: str) -> str:
+        """Stable hash of the inputs that determine the generated text."""
+        s = f"{partner_id}|{level_id}|{round(overdue_amount, 2)}|" \
+            f"{longest_overdue_days}|{invoice_count}|{tone}"
+        return hashlib.sha256(s.encode('utf-8')).hexdigest()
+
+    @api.model
+    def lookup(self, *, partner_id: int, level_id: int,
+               overdue_amount: float, longest_overdue_days: int,
+               invoice_count: int, tone: str):
+        """Find a cached entry matching these inputs, or empty recordset."""
+        fp = self.compute_fingerprint(
+            partner_id=partner_id, level_id=level_id,
+            overdue_amount=overdue_amount,
+            longest_overdue_days=longest_overdue_days,
+            invoice_count=invoice_count, tone=tone,
+        )
+        return self.search([
+            ('partner_id', '=', partner_id),
+            ('fingerprint', '=', fp),
+        ], limit=1)
+
+    def action_increment_use(self):
+        for rec in self:
+            rec.use_count += 1

fusion_accounting_followup/models/res_partner.py

@@ -0,0 +1,52 @@
+"""Inherit res.partner: add follow-up state fields."""
+
+from odoo import _, api, fields, models
+
+
+FOLLOWUP_STATUS = [
+    ('no_action', 'No Action Needed'),
+    ('action_due', 'Action Due'),
+    ('paused', 'Paused'),
+    ('blocked', 'Blocked'),
+    ('with_credit_team', 'With Credit Team'),
+]
+
+
+class ResPartner(models.Model):
+    _inherit = "res.partner"
+
+    fusion_followup_status = fields.Selection(
+        FOLLOWUP_STATUS, default='no_action', tracking=True,
+        help="Current follow-up status as computed by the engine.")
+    fusion_followup_paused_until = fields.Date(
+        tracking=True,
+        help="Pause follow-ups for this partner until this date.")
+    fusion_followup_last_level_id = fields.Many2one(
+        'fusion.followup.level',
+        help="The most-recent follow-up level this partner has been contacted at.")
+    fusion_followup_last_run_date = fields.Datetime(readonly=True)
+    fusion_followup_run_ids = fields.One2many(
+        'fusion.followup.run', 'partner_id', string='Follow-up History')
+    fusion_followup_run_count = fields.Integer(
+        compute='_compute_fusion_followup_run_count')
+    fusion_followup_risk_score = fields.Integer(
+        readonly=True, default=0,
+        help="Latest computed payment risk (0-100). Updated by cron.")
+    fusion_followup_risk_band = fields.Selection([
+        ('low', 'Low'), ('medium', 'Medium'),
+        ('high', 'High'), ('critical', 'Critical'),
+    ], default='low', readonly=True)
+
+    def _compute_fusion_followup_run_count(self):
+        for partner in self:
+            partner.fusion_followup_run_count = len(partner.fusion_followup_run_ids)
+
+    def action_view_followup_history(self):
+        self.ensure_one()
+        return {
+            'type': 'ir.actions.act_window',
+            'res_model': 'fusion.followup.run',
+            'view_mode': 'list,form',
+            'domain': [('partner_id', '=', self.id)],
+            'context': {'default_partner_id': self.id},
+        }

fusion_accounting_followup/security/ir.model.access.csv

@@ -1 +1,7 @@
 id,name,model_id:id,group_id:id,perm_read,perm_write,perm_create,perm_unlink
+access_fusion_followup_level_user,fusion.followup.level.user,model_fusion_followup_level,base.group_user,1,0,0,0
+access_fusion_followup_level_admin,fusion.followup.level.admin,model_fusion_followup_level,fusion_accounting_core.group_fusion_accounting_admin,1,1,1,1
+access_fusion_followup_run_user,fusion.followup.run.user,model_fusion_followup_run,base.group_user,1,0,0,0
+access_fusion_followup_run_admin,fusion.followup.run.admin,model_fusion_followup_run,fusion_accounting_core.group_fusion_accounting_admin,1,1,1,1
+access_fusion_followup_text_cache_user,fusion.followup.text.cache.user,model_fusion_followup_text_cache,base.group_user,1,0,0,0
+access_fusion_followup_text_cache_admin,fusion.followup.text.cache.admin,model_fusion_followup_text_cache,fusion_accounting_core.group_fusion_accounting_admin,1,1,1,1

fusion_accounting_followup/tests/__init__.py

@@ -3,3 +3,8 @@ from . import test_level_resolver
 from . import test_risk_scorer
 from . import test_tone_selector
 from . import test_followup_text_generator
+from . import test_fusion_followup_level
+from . import test_fusion_followup_run
+from . import test_fusion_followup_text_cache
+from . import test_res_partner_inherit
+from . import test_account_move_line_inherit

fusion_accounting_followup/tests/test_account_move_line_inherit.py

@@ -0,0 +1,34 @@
+from odoo import fields as odoo_fields
+from odoo.tests.common import TransactionCase
+from odoo.tests import tagged
+
+
+@tagged('post_install', '-at_install')
+class TestAccountMoveLineFollowup(TransactionCase):
+    """Verify follow-up tracking fields are added to account.move.line."""
+
+    def test_fields_exist_on_model(self):
+        """Both new fields are declared on account.move.line."""
+        AML = self.env['account.move.line']
+        self.assertIn('fusion_followup_level_id', AML._fields)
+        self.assertIn('fusion_followup_last_run_date', AML._fields)
+        self.assertEqual(
+            AML._fields['fusion_followup_level_id'].comodel_name,
+            'fusion.followup.level',
+        )
+
+    def test_assign_level_and_date_on_existing_line(self):
+        """We can write the new fields onto an existing move line."""
+        line = self.env['account.move.line'].search([], limit=1)
+        if not line:
+            self.skipTest("No account.move.line records present in DB to test against.")
+        level = self.env['fusion.followup.level'].create({
+            'name': 'Reminder', 'sequence': 601, 'delay_days': 7, 'tone': 'gentle',
+        })
+        when = odoo_fields.Datetime.now()
+        line.write({
+            'fusion_followup_level_id': level.id,
+            'fusion_followup_last_run_date': when,
+        })
+        self.assertEqual(line.fusion_followup_level_id, level)
+        self.assertEqual(line.fusion_followup_last_run_date, when)

fusion_accounting_followup/tests/test_fusion_followup_level.py

@@ -0,0 +1,42 @@
+from odoo.tests.common import TransactionCase
+from odoo.tests import tagged
+
+
+@tagged('post_install', '-at_install')
+class TestFusionFollowupLevel(TransactionCase):
+
+    def test_create_minimal(self):
+        level = self.env['fusion.followup.level'].create({
+            'name': 'Reminder', 'sequence': 1, 'delay_days': 7, 'tone': 'gentle',
+        })
+        self.assertEqual(level.name, 'Reminder')
+        self.assertTrue(level.active)
+
+    def test_negative_delay_rejected(self):
+        with self.assertRaises(Exception):
+            self.env['fusion.followup.level'].create({
+                'name': 'Bad', 'sequence': 1, 'delay_days': -5, 'tone': 'gentle',
+            })
+
+    def test_duplicate_sequence_rejected(self):
+        self.env['fusion.followup.level'].create({
+            'name': 'A', 'sequence': 100, 'delay_days': 7, 'tone': 'gentle',
+        })
+        with self.assertRaises(Exception):
+            self.env['fusion.followup.level'].create({
+                'name': 'B', 'sequence': 100, 'delay_days': 30, 'tone': 'firm',
+            })
+
+    def test_three_levels_escalate(self):
+        for seq, name, days, tone in [(1, 'R', 7, 'gentle'),
+                                      (2, 'W', 30, 'firm'),
+                                      (3, 'L', 60, 'legal')]:
+            self.env['fusion.followup.level'].create({
+                'name': name, 'sequence': seq + 200,
+                'delay_days': days, 'tone': tone,
+            })
+        levels = self.env['fusion.followup.level'].search([
+            ('sequence', '>', 200),
+        ], order='sequence')
+        self.assertEqual(len(levels), 3)
+        self.assertEqual(levels.mapped('tone'), ['gentle', 'firm', 'legal'])

fusion_accounting_followup/tests/test_fusion_followup_run.py

@@ -0,0 +1,44 @@
+from odoo.tests.common import TransactionCase
+from odoo.tests import tagged
+
+
+@tagged('post_install', '-at_install')
+class TestFusionFollowupRun(TransactionCase):
+
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.partner = cls.env['res.partner'].create({'name': 'Run Test Partner'})
+        cls.level = cls.env['fusion.followup.level'].create({
+            'name': 'Reminder', 'sequence': 301, 'delay_days': 7, 'tone': 'gentle',
+        })
+
+    def test_create_minimal(self):
+        run = self.env['fusion.followup.run'].create({
+            'partner_id': self.partner.id,
+            'level_id': self.level.id,
+        })
+        self.assertEqual(run.state, 'draft')
+        self.assertTrue(run.execution_date)
+
+    def test_action_mark_sent(self):
+        run = self.env['fusion.followup.run'].create({
+            'partner_id': self.partner.id,
+            'level_id': self.level.id,
+        })
+        run.action_mark_sent()
+        self.assertEqual(run.state, 'sent')
+
+    def test_action_mark_failed_records_error(self):
+        run = self.env['fusion.followup.run'].create({
+            'partner_id': self.partner.id,
+        })
+        run.action_mark_failed(error='SMTP unreachable')
+        self.assertEqual(run.state, 'failed')
+        self.assertEqual(run.error_message, 'SMTP unreachable')
+
+    def test_partner_required(self):
+        with self.assertRaises(Exception):
+            self.env['fusion.followup.run'].create({
+                'level_id': self.level.id,
+            })

fusion_accounting_followup/tests/test_fusion_followup_text_cache.py

@@ -0,0 +1,60 @@
+from odoo.tests.common import TransactionCase
+from odoo.tests import tagged
+
+
+@tagged('post_install', '-at_install')
+class TestFusionFollowupTextCache(TransactionCase):
+
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.partner = cls.env['res.partner'].create({'name': 'Cache Test Partner'})
+        cls.level = cls.env['fusion.followup.level'].create({
+            'name': 'Reminder', 'sequence': 401, 'delay_days': 7, 'tone': 'gentle',
+        })
+        cls.cache = cls.env['fusion.followup.text.cache']
+
+    def _kwargs(self, **overrides):
+        base = dict(
+            partner_id=self.partner.id, level_id=self.level.id,
+            overdue_amount=1234.56, longest_overdue_days=10,
+            invoice_count=3, tone='gentle',
+        )
+        base.update(overrides)
+        return base
+
+    def test_fingerprint_stable_and_unique(self):
+        fp1 = self.cache.compute_fingerprint(**self._kwargs())
+        fp2 = self.cache.compute_fingerprint(**self._kwargs())
+        fp3 = self.cache.compute_fingerprint(**self._kwargs(tone='firm'))
+        self.assertEqual(fp1, fp2)
+        self.assertNotEqual(fp1, fp3)
+        self.assertEqual(len(fp1), 64)
+
+    def test_lookup_returns_empty_when_missing(self):
+        result = self.cache.lookup(**self._kwargs())
+        self.assertFalse(result)
+
+    def test_lookup_finds_cached_entry(self):
+        kwargs = self._kwargs()
+        fp = self.cache.compute_fingerprint(**kwargs)
+        entry = self.cache.create({
+            'partner_id': self.partner.id,
+            'level_id': self.level.id,
+            'fingerprint': fp,
+            'subject': 'Hi',
+            'body': 'Please pay.',
+            'tone_used': 'gentle',
+        })
+        found = self.cache.lookup(**kwargs)
+        self.assertEqual(found.id, entry.id)
+
+    def test_action_increment_use(self):
+        entry = self.cache.create({
+            'partner_id': self.partner.id,
+            'fingerprint': 'abc123',
+        })
+        self.assertEqual(entry.use_count, 0)
+        entry.action_increment_use()
+        entry.action_increment_use()
+        self.assertEqual(entry.use_count, 2)

fusion_accounting_followup/tests/test_res_partner_inherit.py

@@ -0,0 +1,27 @@
+from odoo.tests.common import TransactionCase
+from odoo.tests import tagged
+
+
+@tagged('post_install', '-at_install')
+class TestResPartnerFollowup(TransactionCase):
+
+    def test_default_status_no_action(self):
+        partner = self.env['res.partner'].create({'name': 'Default Status'})
+        self.assertEqual(partner.fusion_followup_status, 'no_action')
+        self.assertEqual(partner.fusion_followup_risk_band, 'low')
+        self.assertEqual(partner.fusion_followup_risk_score, 0)
+
+    def test_run_count_reflects_history(self):
+        partner = self.env['res.partner'].create({'name': 'History Partner'})
+        self.assertEqual(partner.fusion_followup_run_count, 0)
+        for _ in range(3):
+            self.env['fusion.followup.run'].create({'partner_id': partner.id})
+        partner.invalidate_recordset(['fusion_followup_run_count', 'fusion_followup_run_ids'])
+        self.assertEqual(partner.fusion_followup_run_count, 3)
+
+    def test_action_view_followup_history_returns_action(self):
+        partner = self.env['res.partner'].create({'name': 'Action Partner'})
+        action = partner.action_view_followup_history()
+        self.assertEqual(action['res_model'], 'fusion.followup.run')
+        self.assertEqual(action['domain'], [('partner_id', '=', partner.id)])
+        self.assertEqual(action['context']['default_partner_id'], partner.id)