66 lines
2.5 KiB
Python
66 lines
2.5 KiB
Python
"""HostEurope invoice CSV ingestion."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import date
|
|
from pathlib import Path
|
|
|
|
from fin_hub.attribution import optional_attribution
|
|
from fin_hub.ingest._csv import parse_amount, pick, read_csv_rows
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class HostEuropeCostRow:
|
|
service_id: str
|
|
title: str
|
|
amount: float
|
|
currency: str
|
|
period_month: str
|
|
incurred_on: date | None
|
|
environment: str = "production"
|
|
source: str = "hosteurope"
|
|
client_id: str | None = None
|
|
application_id: str | None = None
|
|
app_instance_id: str | None = None
|
|
cost_attribution_key: str | None = None
|
|
|
|
|
|
def parse_hosteurope_csv(path: Path, *, default_currency: str = "EUR") -> list[HostEuropeCostRow]:
|
|
rows: list[HostEuropeCostRow] = []
|
|
for row in read_csv_rows(path):
|
|
title = pick(row, "product", "description", "service", "title")
|
|
amount_raw = pick(row, "amount", "net", "total", "price")
|
|
if not title or not amount_raw:
|
|
continue
|
|
service_id = pick(row, "service_id", "product_id") or title.lower().replace(" ", "-")[:128]
|
|
period = pick(row, "period_month", "month", "billing_period")
|
|
incurred_raw = pick(row, "date", "invoice_date", "incurred_on")
|
|
incurred_on = date.fromisoformat(incurred_raw[:10]) if incurred_raw else None
|
|
if not period and incurred_on:
|
|
period = incurred_on.strftime("%Y-%m")
|
|
if not period:
|
|
continue
|
|
currency = pick(row, "currency") or default_currency
|
|
environment = pick(row, "environment", "env") or "production"
|
|
attribution = optional_attribution(
|
|
pick(row, "client_id", "client") or None,
|
|
pick(row, "application_id", "app_id", "application") or None,
|
|
pick(row, "app_instance_id", "instance_id", "instance") or None,
|
|
)
|
|
rows.append(
|
|
HostEuropeCostRow(
|
|
service_id=service_id,
|
|
title=title,
|
|
amount=parse_amount(amount_raw),
|
|
currency=currency.upper()[:3],
|
|
period_month=period[:7],
|
|
incurred_on=incurred_on,
|
|
environment=environment,
|
|
client_id=attribution.client_id if attribution else None,
|
|
application_id=attribution.application_id if attribution else None,
|
|
app_instance_id=attribution.app_instance_id if attribution else None,
|
|
cost_attribution_key=attribution.key if attribution else None,
|
|
)
|
|
)
|
|
return rows
|