52 lines
1.8 KiB
Python
52 lines
1.8 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.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"
|
||
|
|
|
||
|
|
|
||
|
|
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"
|
||
|
|
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,
|
||
|
|
)
|
||
|
|
)
|
||
|
|
return rows
|