File size: 4,753 Bytes
90c6b42 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 | from datetime import datetime, timezone
import logging
from typing import Optional
from accounting.models import Account, AccountType, EntryType, JournalEntry, Transaction
from ecommerce.models import EcommerceOrder
from sqlalchemy.orm import Session
from core.identity_resolver import CustomerResolutionEngine
logger = logging.getLogger(__name__)
class OrderToLedgerMapper:
def __init__(self, db: Session):
self.db = db
self.resolver = CustomerResolutionEngine(db)
def process_order(self, order_id: str) -> Optional[str]:
"""
Takes an EcommerceOrder and creates appropriate accounting entries.
Returns the ID of the created Transaction.
"""
order = self.db.query(EcommerceOrder).filter(EcommerceOrder.id == order_id).first()
if not order:
logger.error(f"Order {order_id} not found for ledger mapping")
return None
if order.is_ledger_synced:
logger.info(f"Order {order_id} already synced to ledger")
return order.ledger_transaction_id
# 1. Resolve Identity to get Accounting Entity
customer = self.resolver.resolve_customer(
order.workspace_id,
order.customer.email,
order.customer.first_name,
order.customer.last_name
)
# 2. Find/Prepare Accounts
# In a real system, these would be configured per workspace.
# We'll use defaults or find by type.
cash_account = self._get_account(order.workspace_id, AccountType.ASSET, "Bank")
revenue_account = self._get_account(order.workspace_id, AccountType.REVENUE, "Service Revenue")
tax_account = self._get_account(order.workspace_id, AccountType.LIABILITY, "Sales Tax Payable")
shipping_account = self._get_account(order.workspace_id, AccountType.REVENUE, "Shipping Income")
# 3. Create Aggregate Transaction
tx = Transaction(
workspace_id=order.workspace_id,
transaction_date=order.created_at or datetime.now(timezone.utc),
description=f"Shopify Order #{order.order_number}",
amount=order.total_price,
source="ecommerce"
)
self.db.add(tx)
self.db.flush()
# 4. Create Journal Entries (Double Entry)
# DEBIT Cash (Asset)
self.db.add(JournalEntry(
transaction_id=tx.id,
account_id=cash_account.id,
type=EntryType.DEBIT,
amount=order.total_price
))
# CREDIT Revenue
if order.subtotal_price > 0:
self.db.add(JournalEntry(
transaction_id=tx.id,
account_id=revenue_account.id,
type=EntryType.CREDIT,
amount=order.subtotal_price
))
# CREDIT Tax Liability
if order.total_tax > 0:
self.db.add(JournalEntry(
transaction_id=tx.id,
account_id=tax_account.id,
type=EntryType.CREDIT,
amount=order.total_tax
))
# CREDIT Shipping Income
if order.total_shipping > 0:
self.db.add(JournalEntry(
transaction_id=tx.id,
account_id=shipping_account.id,
type=EntryType.CREDIT,
amount=order.total_shipping
))
# 5. Link back to Order
order.ledger_transaction_id = tx.id
order.is_ledger_synced = True
self.db.commit()
logger.info(f"Successfully synced Order {order_id} to Ledger Transaction {tx.id}")
return tx.id
def _get_account(self, workspace_id: str, account_type: AccountType, default_name: str) -> Account:
"""Helper to find or create a default account for a given type and name."""
account = self.db.query(Account).filter(
Account.workspace_id == workspace_id,
Account.type == account_type,
Account.name == default_name
).first()
if not account:
# Fallback to any account of that type if name match fails
account = self.db.query(Account).filter(
Account.workspace_id == workspace_id,
Account.type == account_type
).first()
if not account:
account = Account(
workspace_id=workspace_id,
name=default_name,
type=account_type,
code=f"{account_type.value[:1]}000-AUTO"
)
self.db.add(account)
self.db.flush()
logger.info(f"Created default {account_type} account: {default_name}")
return account
|