#!/usr/bin/env python3 import argparse import json import os import sqlite3 import subprocess from datetime import datetime, timezone from decimal import Decimal, ROUND_HALF_UP from pathlib import Path from uuid import uuid4 TWOPLACES = Decimal('0.01') SANITY_THRESHOLD = Decimal('0.05') def money(value): if value is None or value == '': return Decimal('0.00') text = str(value).strip().replace('$', '').replace(',', '') if text.startswith('(') and text.endswith(')'): text = '-' + text[1:-1] return Decimal(text).quantize(TWOPLACES, rounding=ROUND_HALF_UP) def fmt(value: Decimal) -> str: return str(value.quantize(TWOPLACES, rounding=ROUND_HALF_UP)) def ensure_schema(conn): conn.executescript(''' CREATE TABLE IF NOT EXISTS purchase_orders ( id INTEGER PRIMARY KEY AUTOINCREMENT, order_id TEXT UNIQUE, merchant TEXT, order_date TEXT, fulfillment_type TEXT, order_total TEXT, item_subtotal TEXT, adjustment_total TEXT, reconciliation_delta TEXT, synced_at TEXT DEFAULT (datetime('now')) ); CREATE TABLE IF NOT EXISTS purchase_items ( id INTEGER PRIMARY KEY AUTOINCREMENT, order_id TEXT, item_name TEXT, quantity TEXT, unit_price TEXT, item_total TEXT, is_business_expense INTEGER DEFAULT 0, FOREIGN KEY (order_id) REFERENCES purchase_orders(order_id) ); CREATE TABLE IF NOT EXISTS purchase_adjustments ( id INTEGER PRIMARY KEY AUTOINCREMENT, order_id TEXT, adjustment_name TEXT, adjustment_amount TEXT, adjustment_type TEXT, FOREIGN KEY (order_id) REFERENCES purchase_orders(order_id) ); CREATE TABLE IF NOT EXISTS purchase_sync_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, synced_at TEXT DEFAULT (datetime('now')), merchant TEXT, orders_found INTEGER, items_written INTEGER, status TEXT, message TEXT ); CREATE TABLE IF NOT EXISTS purchase_order_exceptions ( id INTEGER PRIMARY KEY AUTOINCREMENT, run_id TEXT NOT NULL, merchant TEXT NOT NULL, order_id TEXT NOT NULL, route_family TEXT, order_date TEXT, visible_order_total TEXT, extracted_subtotal TEXT, reconciliation_delta TEXT, failure_reason TEXT NOT NULL, failure_detail TEXT, detail_url TEXT, artifact_refs_json TEXT, status TEXT NOT NULL DEFAULT 'open', retry_count INTEGER NOT NULL DEFAULT 0, last_retry_at TEXT, created_at TEXT NOT NULL DEFAULT (datetime('now')), updated_at TEXT NOT NULL DEFAULT (datetime('now')) ); CREATE INDEX IF NOT EXISTS idx_purchase_order_exceptions_merchant_status_created ON purchase_order_exceptions (merchant, status, created_at); CREATE INDEX IF NOT EXISTS idx_purchase_order_exceptions_order ON purchase_order_exceptions (order_id); ''' ) def load_classification(path): p = Path(path) if not p.exists(): return {} return json.loads(p.read_text(encoding='utf-8')) def merchant_name_from_classification(classification, requested): merchants = classification.get('merchants', {}) if requested in merchants: return requested for name in merchants.keys(): if name.lower() == requested.lower(): return name return requested def is_business_expense(name, rules): low = name.lower() for rule in rules: if str(rule).lower() in low: return True return False def normalize_order(raw, merchant_name, expense_rules): items = raw.get('items', []) adjustments = raw.get('adjustments', []) item_subtotal = sum((money(item.get('item_total')) for item in items), Decimal('0.00')) adjustment_total = sum((money(adj.get('adjustment_amount')) for adj in adjustments), Decimal('0.00')) order_total = money(raw.get('order_total')) delta = order_total - item_subtotal - adjustment_total sanity_failure = (abs(delta) / order_total) > SANITY_THRESHOLD if order_total != 0 else False normalized_items = [] flagged_business = 0 for item in items: item_name = str(item.get('item_name', '')).strip() flagged = 1 if is_business_expense(item_name, expense_rules) else 0 flagged_business += flagged normalized_items.append({ 'name': item_name, 'qty': str(item.get('quantity', '')).strip(), 'unit_price': fmt(money(item.get('unit_price'))), 'item_total': fmt(money(item.get('item_total'))), 'is_business_expense': flagged, }) return { 'order_id': str(raw['order_id']).strip(), 'merchant': merchant_name, 'order_date': str(raw.get('order_date', '')).strip(), 'fulfillment_type': str(raw.get('fulfillment_type', '')).strip(), 'route_family': str(raw.get('route_family', '')).strip() or None, 'detail_url': str(raw.get('detail_url', '')).strip() or None, 'artifact_refs': raw.get('artifact_refs') or raw.get('artifacts') or [], 'visible_order_total': fmt(money(raw.get('visible_order_total', raw.get('order_total')))), 'order_total': fmt(order_total), 'item_subtotal': fmt(item_subtotal), 'adjustment_total': fmt(adjustment_total), 'reconciliation_delta': fmt(delta), 'sanity_failure': sanity_failure, 'flagged_business': flagged_business, 'items': normalized_items, 'adjustments': [ { 'name': str(adj.get('adjustment_name', '')).strip(), 'amount': fmt(money(adj.get('adjustment_amount'))), 'type': str(adj.get('adjustment_type', '')).strip(), } for adj in adjustments ], } def record_partial_sync_exception(cur, run_id, order): reason = 'sanity_delta_exceeds_threshold' detail = ( f"abs(delta)/order_total exceeded {fmt(SANITY_THRESHOLD)}; " f"order_total={order['order_total']}; item_subtotal={order['item_subtotal']}; " f"adjustment_total={order['adjustment_total']}; delta={order['reconciliation_delta']}" ) artifact_refs = order.get('artifact_refs', []) cur.execute( 'INSERT INTO purchase_order_exceptions ' '(run_id, merchant, order_id, route_family, order_date, visible_order_total, extracted_subtotal, ' 'reconciliation_delta, failure_reason, failure_detail, detail_url, artifact_refs_json, status, retry_count, ' 'created_at, updated_at) ' 'VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, datetime(\'now\'), datetime(\'now\'))', ( run_id, order['merchant'], order['order_id'], order.get('route_family'), order.get('order_date'), order.get('visible_order_total'), order.get('item_subtotal'), order.get('reconciliation_delta'), reason, detail, order.get('detail_url'), json.dumps(artifact_refs, ensure_ascii=False), 'open', ) ) def maybe_send_telegram_alert(merchant, partial_sync_count, threshold, telegram_target): if partial_sync_count <= threshold: return {'attempted': False, 'sent': False, 'reason': 'below_threshold'} if not telegram_target: return {'attempted': False, 'sent': False, 'reason': 'missing_telegram_target'} message_text = ( f"⚠️ Purchase tracker partial_sync alert\n" f"Merchant: {merchant}\n" f"Count this sync run: {partial_sync_count}\n" f"Action: check purchase_order_exceptions in purchases.db" ) cmd = [ 'openclaw', 'message', 'send', '--channel', 'telegram', '--target', telegram_target, '--message', message_text, ] try: subprocess.run(cmd, check=True, capture_output=True, text=True) return {'attempted': True, 'sent': True, 'reason': 'sent'} except Exception as exc: return {'attempted': True, 'sent': False, 'reason': f'send_failed: {exc}'} def append_dorian_log(log_path, merchant, stats, sanity_failures): p = Path(log_path) p.parent.mkdir(parents=True, exist_ok=True) ts = datetime.now(timezone.utc).isoformat() line = ( f"[{ts}] merchant={merchant} orders_seen={stats['orders_seen']} " f"orders_inserted={stats['orders_inserted']} items_written={stats['items_written']} " f"adjustments_written={stats['adjustments_written']} flagged_business={stats['flagged_business']} " f"sanity_failures={len(sanity_failures)} status={stats['status']}" ) with p.open('a', encoding='utf-8') as fh: fh.write(line + '\n') def main(): parser = argparse.ArgumentParser() parser.add_argument('--db-path', required=True) parser.add_argument('--input-json', required=True) parser.add_argument('--merchant', default='Target') parser.add_argument('--classification-json', default='/Users/openclaw/outputs/purchase-tracker/classification.json') parser.add_argument('--log-path', default='/Users/openclaw/outputs/purchase-tracker/logs/dorian_log.txt') parser.add_argument('--alert-threshold', type=int, default=2) parser.add_argument('--telegram-target', default=os.environ.get('PURCHASE_TRACKER_TELEGRAM_TARGET', '')) args = parser.parse_args() payload = json.loads(Path(args.input_json).read_text(encoding='utf-8')) classification = load_classification(args.classification_json) merchant_name = merchant_name_from_classification(classification, args.merchant) expense_rules = classification.get('expense_rules', {}).get(merchant_name, []) + classification.get('expense_rules', {}).get('global', []) orders = [normalize_order(order, merchant_name, expense_rules) for order in payload.get('orders', [])] sync_run_id = f"{merchant_name.lower()}-{datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')}-{uuid4().hex[:8]}" conn = sqlite3.connect(args.db_path) ensure_schema(conn) cur = conn.cursor() orders_inserted = 0 items_written = 0 adjustments_written = 0 flagged_business = 0 sanity_failures = [] partial_sync_skipped = 0 for order in orders: if order['sanity_failure']: sanity_failures.append({ 'order_id': order['order_id'], 'order_total': order['order_total'], 'delta': order['reconciliation_delta'], }) partial_sync_skipped += 1 record_partial_sync_exception(cur, sync_run_id, order) continue cur.execute( 'INSERT OR IGNORE INTO purchase_orders ' '(order_id, merchant, order_date, fulfillment_type, order_total, item_subtotal, adjustment_total, reconciliation_delta, synced_at) ' 'VALUES (?, ?, ?, ?, ?, ?, ?, ?, datetime(\'now\'))', ( order['order_id'], order['merchant'], order['order_date'], order['fulfillment_type'], order['order_total'], order['item_subtotal'], order['adjustment_total'], order['reconciliation_delta'] ) ) inserted = cur.rowcount > 0 if inserted: orders_inserted += 1 for item in order['items']: cur.execute( 'INSERT INTO purchase_items (order_id, item_name, quantity, unit_price, item_total, is_business_expense) VALUES (?, ?, ?, ?, ?, ?)', (order['order_id'], item['name'], item['qty'], item['unit_price'], item['item_total'], item['is_business_expense']) ) flagged_business += item['is_business_expense'] items_written += 1 for adj in order['adjustments']: cur.execute( 'INSERT INTO purchase_adjustments (order_id, adjustment_name, adjustment_amount, adjustment_type) VALUES (?, ?, ?, ?)', (order['order_id'], adj['name'], adj['amount'], adj['type']) ) adjustments_written += 1 alert_result = maybe_send_telegram_alert( merchant_name, partial_sync_skipped, args.alert_threshold, args.telegram_target, ) status = 'complete' if not sanity_failures else 'partial_sync' message = ( f"{merchant_name} sync complete. {len(orders)} orders seen, {orders_inserted} inserted, " f"{items_written} items, {adjustments_written} adjustments, " f"{len(sanity_failures)} sanity flags, {partial_sync_skipped} skipped as partial-sync, " f"run_id={sync_run_id}, alert={alert_result['reason']}." ) cur.execute( 'INSERT INTO purchase_sync_log (merchant, orders_found, items_written, status, message) VALUES (?, ?, ?, ?, ?)', (merchant_name, len(orders), items_written, status, message) ) conn.commit() conn.close() stats = { 'merchant': merchant_name, 'orders_seen': len(orders), 'orders_inserted': orders_inserted, 'items_written': items_written, 'adjustments_written': adjustments_written, 'flagged_business': flagged_business, 'sanity_failures': sanity_failures, 'partial_sync_skipped': partial_sync_skipped, 'status': status, 'sync_run_id': sync_run_id, 'alert_threshold': args.alert_threshold, 'alert': alert_result, } append_dorian_log(args.log_path, merchant_name, stats, sanity_failures) print(json.dumps(stats)) if __name__ == '__main__': main()