feat(fusion_accounting_followup): LLM text cache model

Made-with: Cursor

fusion_accounting_followup/__manifest__.py

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

fusion_accounting_followup/models/__init__.py

@@ -1,2 +1,3 @@
 from . import fusion_followup_level
 from . import fusion_followup_run
+from . import fusion_followup_text_cache

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/security/ir.model.access.csv

@@ -3,3 +3,5 @@ access_fusion_followup_level_user,fusion.followup.level.user,model_fusion_follow
 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

@@ -5,3 +5,4 @@ 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

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)