import json import logging from odoo import models, fields, api _logger = logging.getLogger(__name__) class FusionVendorTaxProfile(models.Model): _name = 'fusion.vendor.tax.profile' _description = 'Vendor Tax Profile (AI Cache)' _order = 'total_bills desc' _rec_name = 'partner_id' partner_id = fields.Many2one( 'res.partner', string='Vendor', required=True, index=True, ondelete='cascade', ) company_id = fields.Many2one( 'res.company', string='Company', default=lambda self: self.env.company, ) total_bills = fields.Integer(string='Total Bills') bills_with_hst = fields.Integer(string='Bills with HST') bills_zero_rated = fields.Integer(string='Bills Zero-Rated') avg_tax_pct = fields.Float(string='Avg Tax %', digits=(5, 2)) # Classification tax_classification = fields.Selection([ ('always_hst', 'Always HST (13%)'), ('mostly_hst', 'Mostly HST (>10%)'), ('shipping_only', 'HST on Shipping Only (<2%)'), ('never_hst', 'Never HST (0%)'), ('mixed', 'Mixed / Inconsistent'), ], string='Tax Classification') # Most common expense account primary_account_id = fields.Many2one( 'account.account', string='Primary Expense Account', ) primary_account_code = fields.Char( related='primary_account_id.code', string='Account Code', store=True, ) # AI-readable note tax_note = fields.Text( string='Tax Note (AI-Readable)', help='Plain English note the AI reads to understand tax treatment.', ) # PO-tracked vendor — bills come from purchase orders, never from bank recon is_po_vendor = fields.Boolean( string='PO-Tracked Vendor', help='Bills for this vendor are created from Purchase Orders. ' 'Do NOT create bills during bank reconciliation — just match to existing bills.', ) po_count = fields.Integer(string='Purchase Orders') # Vendor details for matching is_foreign = fields.Boolean(string='Foreign Vendor') vendor_country = fields.Char(string='Vendor Country') # Timestamps last_computed = fields.Datetime(string='Last Computed') _sql_constraints = [ ('partner_company_uniq', 'unique(partner_id, company_id)', 'One tax profile per vendor per company'), ] def _rebuild_all_profiles(self, min_bills=3): """Rebuild all vendor tax profiles from posted bill history. Called by cron or manually.""" _logger.info("Rebuilding vendor tax profiles (min_bills=%d)...", min_bills) companies = self.env['res.company'].search([]) total_created = 0 total_updated = 0 for company in companies: # Find all vendors with enough bills self.env.cr.execute(""" SELECT m.partner_id, count(*) as bill_count, SUM(CASE WHEN m.amount_tax > 0.01 THEN 1 ELSE 0 END) as with_tax, SUM(CASE WHEN m.amount_tax <= 0.01 THEN 1 ELSE 0 END) as no_tax, COALESCE(AVG(CASE WHEN m.amount_untaxed > 0 THEN m.amount_tax / m.amount_untaxed * 100 ELSE 0 END), 0) as avg_tax_pct FROM account_move m WHERE m.move_type = 'in_invoice' AND m.state = 'posted' AND m.company_id = %s AND m.partner_id IS NOT NULL GROUP BY m.partner_id HAVING count(*) >= %s """, (company.id, min_bills)) vendor_stats = self.env.cr.dictfetchall() for vs in vendor_stats: partner = self.env['res.partner'].browse(vs['partner_id']) if not partner.exists(): continue # Classify avg_pct = round(vs['avg_tax_pct'], 2) total = vs['bill_count'] with_tax = vs['with_tax'] no_tax = vs['no_tax'] if no_tax == total: classification = 'never_hst' note = f'{partner.name} NEVER charges HST. All {total} bills are zero-rated. Do NOT apply HST.' elif avg_pct >= 12.0: classification = 'always_hst' note = f'{partner.name} consistently charges HST at ~{avg_pct}%. Apply HST PURCHASE (13%) to all product lines.' elif avg_pct >= 10.0: classification = 'mostly_hst' note = f'{partner.name} usually charges HST (~{avg_pct}%). {no_tax} of {total} bills had no tax. Apply HST by default but verify zero-rated items.' elif avg_pct < 2.0 and with_tax > 0: classification = 'shipping_only' note = ( f'{partner.name} products are zero-rated (avg tax only {avg_pct}% of subtotal). ' f'HST applies ONLY to shipping/freight charges, NOT to product lines. ' f'When creating a bill, use NO TAX PURCHASE on product lines and HST PURCHASE (13%) only on shipping lines.' ) else: classification = 'mixed' note = ( f'{partner.name} has mixed tax treatment ({with_tax} bills with HST, {no_tax} without, avg {avg_pct}%). ' f'Check each bill individually — some items may be zero-rated while others have HST.' ) # Find primary expense account self.env.cr.execute(""" SELECT aml.account_id, count(*) as cnt FROM account_move_line aml JOIN account_move m ON aml.move_id = m.id WHERE m.partner_id = %s AND m.move_type = 'in_invoice' AND m.state = 'posted' AND m.company_id = %s AND aml.display_type = 'product' GROUP BY aml.account_id ORDER BY count(*) DESC LIMIT 1 """, (vs['partner_id'], company.id)) acct_row = self.env.cr.fetchone() primary_account_id = acct_row[0] if acct_row else False # Check if foreign vendor is_foreign = False country = '' if partner.country_id: country = partner.country_id.name is_foreign = partner.country_id.code != 'CA' elif partner.vat and not partner.vat.startswith('CA'): is_foreign = True # Only override to never_hst if foreign AND bills actually confirm no tax # (Don't override if bill data shows they DO charge HST — e.g., Amazon Canada) if is_foreign and avg_pct < 1.0 and no_tax > with_tax: classification = 'never_hst' note = f'{partner.name} is a FOREIGN vendor ({country or "non-Canadian"}) and bills confirm no HST. Do NOT apply any Canadian tax.' # Check if this is a PO-tracked vendor (has confirmed purchase orders) is_po_vendor = False vendor_po_count = 0 try: self.env.cr.execute(""" SELECT count(*) FROM purchase_order WHERE partner_id = %s AND state IN ('purchase', 'done') AND company_id = %s """, (vs['partner_id'], company.id)) po_row = self.env.cr.fetchone() vendor_po_count = po_row[0] if po_row else 0 is_po_vendor = vendor_po_count >= 3 except Exception: pass # purchase module may not be installed if is_po_vendor: note = ( f'PO-TRACKED VENDOR ({vendor_po_count} purchase orders). ' f'Bills are created from Purchase Orders — do NOT create bills during bank reconciliation. ' f'Instead, find the existing unpaid bill and match the bank payment to it. ' f'Tax treatment: {note}' ) # Upsert existing = self.search([ ('partner_id', '=', vs['partner_id']), ('company_id', '=', company.id), ], limit=1) vals = { 'partner_id': vs['partner_id'], 'company_id': company.id, 'total_bills': total, 'bills_with_hst': with_tax, 'bills_zero_rated': no_tax, 'avg_tax_pct': avg_pct, 'tax_classification': classification, 'primary_account_id': primary_account_id, 'tax_note': note, 'is_po_vendor': is_po_vendor, 'po_count': vendor_po_count, 'is_foreign': is_foreign, 'vendor_country': country, 'last_computed': fields.Datetime.now(), } if existing: existing.write(vals) total_updated += 1 else: self.create(vals) total_created += 1 _logger.info( "Vendor tax profiles rebuilt: %d created, %d updated", total_created, total_updated, ) return {'created': total_created, 'updated': total_updated}