"""Collections statements — the send layer behind the Collections page's Statements section. Folded in from the standalone collections_app (2026-07-05). THE one sanctioned exception to the app's read-only-on-Odoo rule, unchanged from the standalone tool: WRITE is whitelisted to exactly one operation — create on mail.mail (queueing an outbound statement email) A mail.mail record with state='outgoing' is picked up by Odoo's "Mail: Email Queue Manager" cron (every ~15 min) and delivered through the company's Office 365 relay. Because we set model/res_id, each sent statement also appears in the customer's chatter in Odoo — the audit log lives where AR already works. This module has its OWN narrow XML-RPC client; the app-wide odoo_client stays hard-blocking on all writes. The UI gates the section to admin users; SAFE_MODE is enforced HERE in the data layer so the UI cannot bypass it. Env (Space secrets / .env): ODOO_URL, ODOO_DB, ODOO_USER, ODOO_API_KEY Optional: SAFE_MODE (default ON), SAFE_RECIPIENTS, SENDER_NAME, SENDER_EMAIL, REPLY_TO, COMPANY_NAME, ROYAL_MAIL_SERVER_ID, ROYAL_AUTHOR_ID """ import os import datetime as dt import xmlrpc.client from pathlib import Path try: # self-contained: load credentials from the app root .env if present (HF uses Secrets) from dotenv import load_dotenv load_dotenv(Path(__file__).resolve().parents[1] / '.env') except Exception: pass WRITE_WHITELIST = {('mail.mail', 'create')} EXCLUDE_NAMES = {'GIFTWARE DEALS'} # the Amazon channel — not part of Fisch or Royal collections DOMAIN = [('followup_reminder_type', '=', 'automatic'), ('credit', '>', 1)] COMPANY = os.environ.get('COMPANY_NAME', 'Royal Imports') # --- Sender identity (statements go out AS Royal Imports) --- # Odoo routes outbound mail to the matching ir.mail_server by from_filter, and the Office 365 # relay only accepts sends as its authenticated address. The "Office 365 - Royal" server (id 2) # authenticates as contact@royalimports.com with from_filter='contact@royalimports.com' — so the # From MUST be that address for delivery to succeed. Friendly display name; replies routed to AR. SENDER_NAME = os.environ.get('SENDER_NAME', 'Royal Imports Accounts Receivable') SENDER_EMAIL = os.environ.get('SENDER_EMAIL', 'contact@royalimports.com') REPLY_TO = os.environ.get('REPLY_TO', 'accounting@royalimports.com') ROYAL_MAIL_SERVER_ID = int(os.environ.get('ROYAL_MAIL_SERVER_ID', '2')) ROYAL_AUTHOR_ID = int(os.environ.get('ROYAL_AUTHOR_ID', '8978')) # "Royal Imports" partner SENDER_DISPLAY = f'"{SENDER_NAME}" <{SENDER_EMAIL}>' class WriteBlocked(RuntimeError): pass class SafeModeBlocked(RuntimeError): pass # --- Testing guardrail ------------------------------------------------------- # While SAFE_MODE is on, NO email can be queued to any address outside the allow-list — enforced # here in the data layer so the UI cannot bypass it. Default: ON. To go live for real customers, # set the Space secret SAFE_MODE=0. SAFE_MODE = os.environ.get('SAFE_MODE', '1').strip().lower() not in ('0', 'false', 'no', '') SAFE_RECIPIENTS = {a.strip().lower() for a in os.environ.get('SAFE_RECIPIENTS', 'farhan@teamroyalimports.com').split(',') if a.strip()} def safe_recipient_ok(addr): return (not SAFE_MODE) or (str(addr or '').strip().lower() in SAFE_RECIPIENTS) class Odoo: """Narrow client: read anything, write ONLY the whitelisted mail.mail create.""" def __init__(self): self.url = os.environ.get('ODOO_URL', '').rstrip('/') self.db = os.environ.get('ODOO_DB', '') self.user = os.environ.get('ODOO_USER', '') self.key = os.environ.get('ODOO_API_KEY', '') missing = [k for k, v in [('ODOO_URL', self.url), ('ODOO_DB', self.db), ('ODOO_USER', self.user), ('ODOO_API_KEY', self.key)] if not v] if missing: raise RuntimeError(f"Missing secrets: {', '.join(missing)}") common = xmlrpc.client.ServerProxy(f'{self.url}/xmlrpc/2/common') self.uid = common.authenticate(self.db, self.user, self.key, {}) if not self.uid: raise RuntimeError('Odoo authentication failed') self.models = xmlrpc.client.ServerProxy(f'{self.url}/xmlrpc/2/object') def _exec(self, model, method, args, kwargs=None): mutating = method in ('write', 'create', 'unlink', 'copy') or \ any(method.startswith(p) for p in ('action_', 'button_', 'do_', 'send_', 'set_', 'update_', 'process_')) if mutating and (model, method) not in WRITE_WHITELIST: raise WriteBlocked(f'{method} on {model} is not allowed from this app') return self.models.execute_kw(self.db, self.uid, self.key, model, method, args, kwargs or {}) def search_read(self, model, domain=None, fields=None, limit=None, order=None): kw = {'fields': fields or []} if limit is not None: kw['limit'] = limit if order: kw['order'] = order return self._exec(model, 'search_read', [domain or []], kw) def queue_mail(self, payload): """The single allowed write: queue an outbound email.""" return self._exec('mail.mail', 'create', [payload]) # --------------------------------------------------------------- data builders def load_collection_list(odoo): """The saved follow-up filter (Reminders=Automatic, Receivable>$1), GIFTWARE excluded, with priority tiers.""" partners = odoo.search_read('res.partner', DOMAIN, ['name', 'credit', 'total_overdue', 'followup_status', 'followup_next_action_date', 'followup_responsible_id', 'email', 'phone', 'mobile']) partners = [p for p in partners if (p['name'] or '').strip().upper() not in EXCLUDE_NAMES] pids = [p['id'] for p in partners] docs = odoo.search_read('account.move', [('move_type', 'in', ['out_invoice', 'out_refund']), ('state', '=', 'posted'), ('payment_state', 'in', ['not_paid', 'partial']), ('partner_id', 'in', pids)], ['name', 'partner_id', 'move_type', 'invoice_date', 'invoice_date_due', 'amount_total', 'amount_residual_signed']) today = dt.date.today() by_partner = {} for d in docs: pid = d['partner_id'][0] due = d.get('invoice_date_due') try: days = (today - dt.date.fromisoformat(due)).days if due else 0 except Exception: days = 0 d['days_overdue'] = max(days, 0) d['open_signed'] = d['amount_residual_signed'] by_partner.setdefault(pid, []).append(d) rows = [] for p in partners: pid = p['id'] odoo_overdue = float(p.get('total_overdue') or 0) docs_p = sorted(by_partner.get(pid, []), key=lambda x: x.get('invoice_date_due') or '') inv_overdue = sum(d['open_signed'] for d in docs_p if d['days_overdue'] > 0) oldest = max((d['days_overdue'] for d in docs_p), default=0) gap = inv_overdue - odoo_overdue reconcile = abs(gap) > 50 if odoo_overdue >= 5000 or (odoo_overdue > 0 and oldest > 90): tier = 'A-Urgent' elif odoo_overdue >= 1000 or (odoo_overdue > 0 and oldest > 30): tier = 'B-Active' elif odoo_overdue > 0: tier = 'C-Light' else: tier = 'Monitor' rows.append({ 'partner_id': pid, 'Customer': p['name'], 'Tier': tier, 'Overdue': odoo_overdue, 'Receivable': float(p.get('credit') or 0), 'Oldest (days)': oldest, 'Open Docs': len(docs_p), 'Email': p.get('email') or '', 'Phone': p.get('phone') or p.get('mobile') or '', 'Status': (p.get('followup_status') or '').replace('_', ' '), 'Reconcile?': 'YES' if reconcile else '', '_docs': docs_p, }) tier_rank = {'A-Urgent': 0, 'B-Active': 1, 'C-Light': 2, 'Monitor': 3} rows.sort(key=lambda r: (tier_rank[r['Tier']], -r['Overdue'])) return rows # --------------------------------------------------------------- statement email DEFAULT_SUBJECT = 'Statement of Account — {company} — {month}' DEFAULT_INTRO = ( 'Dear {customer},

' 'Please find below your current statement of account with {company}. ' 'According to our records, the following invoices remain open:' ) DEFAULT_FOOTER = ( 'If you have already sent payment, please disregard this notice — and thank you. ' 'For any questions about an invoice, simply reply to this email.

' 'Thank you for your business,
{company} — Accounts Receivable' ) def render_statement_html(row, intro_tpl=DEFAULT_INTRO, footer_tpl=DEFAULT_FOOTER): month = dt.date.today().strftime('%B %Y') intro = intro_tpl.format(customer=row['Customer'], company=COMPANY, month=month) footer = footer_tpl.format(customer=row['Customer'], company=COMPANY, month=month) lines = [] total_open = 0.0 for d in row['_docs']: kind = 'Credit Note' if d['move_type'] == 'out_refund' else 'Invoice' amt = d['open_signed'] total_open += amt overdue_txt = f"{d['days_overdue']}d overdue" if d['days_overdue'] > 0 else 'current' color = '#C0392B' if d['days_overdue'] > 0 else '#1F4E78' lines.append( f"{d['name']} ({kind})" f"{d.get('invoice_date') or ''}" f"{d.get('invoice_date_due') or ''}" f"{overdue_txt}" f"${amt:,.2f}") table = ( "" "" "" "" "" "" "" + ''.join(lines) + f"" f"" f"" f"" "
DocumentDateDueStatusOpen Balance
Total open${total_open:,.2f}
Of which overdue${row['Overdue']:,.2f}
") return (f"
" f"{intro}{table}{footer}
") def queue_statement(odoo, row, subject_tpl=DEFAULT_SUBJECT, intro_tpl=DEFAULT_INTRO, footer_tpl=DEFAULT_FOOTER, override_to=None): """Queue one statement email in Odoo. Returns mail.mail id. override_to: send to a different address (used by the test-send button).""" to = override_to or row['Email'] if not to: raise ValueError(f"{row['Customer']} has no email address") # Hard guardrail — refuse any recipient outside the allow-list while SAFE_MODE is on. if not safe_recipient_ok(to): raise SafeModeBlocked( f"Guardrail ON: refusing to email {to}. Only {', '.join(sorted(SAFE_RECIPIENTS))} " f"is allowed right now. (Set SAFE_MODE=0 to send to real customers.)") month = dt.date.today().strftime('%B %Y') subject = subject_tpl.format(customer=row['Customer'], company=COMPANY, month=month) payload = { 'subject': subject, 'body_html': render_statement_html(row, intro_tpl, footer_tpl), 'email_to': to, 'email_from': SENDER_DISPLAY, # From: Royal Imports 'reply_to': REPLY_TO, # replies -> AR 'mail_server_id': ROYAL_MAIL_SERVER_ID, # force Royal O365 relay 'author_id': ROYAL_AUTHOR_ID, # clean attribution in chatter 'state': 'outgoing', 'auto_delete': False, 'model': 'res.partner', 'res_id': row['partner_id'], } return odoo.queue_mail(payload)