Bridging Cloud and Accounting: Mapping AWS Spending Directly into Odoo ERP for Automated Audits

import boto3
from datetime import datetime, timedelta
 
 
def get_daily_costs(days: int = 1) -> list[dict]:
    """Fetch AWS costs for the last N days, grouped by service."""
    client = boto3.client("ce", region_name="us-east-1")
 
    end = datetime.now().strftime("%Y-%m-%d")
    start = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d")
 
    response = client.get_cost_and_usage(
        TimePeriod={"Start": start, "End": end},
        Granularity="DAILY",
        Metrics=["UnblendedCost"],
        GroupBy=[
            {"Type": "DIMENSION", "Key": "SERVICE"},
        ],
    )
 
    costs = []
    for result in response["ResultsByTime"]:
        date = result["TimePeriod"]["Start"]
        for group in result["Groups"]:
            service = group["Keys"][0]
            amount = float(
                group["Metrics"]["UnblendedCost"]["Amount"]
            )
            if amount > 0.01:  # Skip negligible charges
                costs.append(
                    {
                        "date": date,
                        "service": service,
                        "amount": round(amount, 2),
                        "currency": "USD",
                    }
                )
 
    return costs

# Map AWS service names to Odoo analytic account codes
AWS_TO_ODOO_MAP: dict[str, dict] = {
    "Amazon Elastic Compute Cloud - Compute": {
        "account_code": "6100",      # IT Infrastructure - Compute
        "analytic_account": "AWS-EC2",
        "description": "EC2 compute instances",
    },
    "Amazon Relational Database Service": {
        "account_code": "6100",      # IT Infrastructure - Compute
        "analytic_account": "AWS-RDS",
        "description": "RDS database instances",
    },
    "Amazon Simple Storage Service": {
        "account_code": "6110",      # IT Infrastructure - Storage
        "analytic_account": "AWS-S3",
        "description": "S3 object storage",
    },
    "Amazon Elastic Container Service for Kubernetes": {
        "account_code": "6100",      # IT Infrastructure - Compute
        "analytic_account": "AWS-EKS",
        "description": "EKS Kubernetes clusters",
    },
    "AWS Data Transfer": {
        "account_code": "6120",      # IT Infrastructure - Network
        "analytic_account": "AWS-TRANSFER",
        "description": "Cross-region and internet data transfer",
    },
}
 
def map_cost_to_odoo(service: str) -> dict:
    """Resolve an AWS service name to Odoo account codes."""
    return AWS_TO_ODOO_MAP.get(service, {
        "account_code": "6199",          # IT - Uncategorized
        "analytic_account": "AWS-OTHER",
        "description": f"Unmapped: {service}",
    })

import xmlrpc.client
 
 
class OdooClient:
    def __init__(self, url: str, db: str, user: str, password: str):
        self.url = url
        self.db = db
        self.uid = xmlrpc.client.ServerProxy(
            f"{url}/xmlrpc/2/common"
        ).authenticate(db, user, password, {})
        self.models = xmlrpc.client.ServerProxy(
            f"{url}/xmlrpc/2/object"
        )
        self.password = password
 
    def _execute(self, model: str, method: str, *args):
        return self.models.execute_kw(
            self.db, self.uid, self.password,
            model, method, *args,
        )
 
    def create_journal_entry(
        self,
        date: str,
        service: str,
        amount: float,
        account_code: str,
        analytic_account: str,
        description: str,
    ):
        """Create a journal entry for an AWS cost line item."""
        # Find the expense and payable account IDs
        expense_account = self._execute(
            "account.account", "search",
            [[["code", "=", account_code]]],
        )
        payable_account = self._execute(
            "account.account", "search",
            [[["code", "=", "2100"]]],  # Accounts Payable
        )
 
        if not expense_account or not payable_account:
            raise ValueError(
                f"Account {account_code} or 2100 not found in Odoo"
            )
 
        entry = self._execute(
            "account.move", "create",
            [{
                "journal_id": 2,  # Vendor Bills journal
                "date": date,
                "ref": f"AWS-{service}-{date}",
                "line_ids": [
                    (0, 0, {
                        "account_id": expense_account[0],
                        "name": f"AWS {description}",
                        "debit": amount,
                        "credit": 0,
                    }),
                    (0, 0, {
                        "account_id": payable_account[0],
                        "name": f"AWS {description} (Payable)",
                        "debit": 0,
                        "credit": amount,
                    }),
                ],
            }],
        )
 
        return entry

from aws_costs import get_daily_costs
from account_mapping import map_cost_to_odoo
from odoo_sync import OdooClient
 
 
def run_sync():
    odoo = OdooClient(
        url="https://your-odoo.com",
        db="production",
        user="api-user@company.com",
        password="your-api-key",
    )
 
    costs = get_daily_costs(days=1)
 
    for item in costs:
        mapping = map_cost_to_odoo(item["service"])
 
        odoo.create_journal_entry(
            date=item["date"],
            service=item["service"],
            amount=item["amount"],
            account_code=mapping["account_code"],
            analytic_account=mapping["analytic_account"],
            description=mapping["description"],
        )
 
        print(
            f"Synced: {item['service']} "
            f"${item['amount']} -> {mapping['account_code']}"
        )
 
 
if __name__ == "__main__":
    run_sync()