generated from coulomb/repo-seed
tegwick
7b84d34ea6
Update payment_records to 8.99 EUR gross, 0.44 EUR fees, 8.55 EUR net payout to binky-hedgehog. Link member tegwick in membership ledger and add Stripe reference catalog.
145 lines
4.4 KiB
Python
145 lines
4.4 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
|
|
from .ledger import build_monthly_ledger
|
|
from .models import (
|
|
Budget,
|
|
ExpenseRecord,
|
|
MembershipRecord,
|
|
MonthlyPlatformCost,
|
|
PaymentRecord,
|
|
PricingModel,
|
|
Product,
|
|
)
|
|
|
|
|
|
def _money(value: str | int | float) -> Decimal:
|
|
return Decimal(str(value))
|
|
|
|
|
|
def _read_json(path: Path) -> dict:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
|
|
|
|
def default_data_dir() -> Path:
|
|
return Path(__file__).resolve().parent.parent / "data"
|
|
|
|
|
|
def load_product(data_dir: Path | None = None) -> Product:
|
|
raw = _read_json((data_dir or default_data_dir()) / "product.json")
|
|
return Product(
|
|
id=raw["id"],
|
|
name=raw["name"],
|
|
lifecycle_phase=raw["lifecycle_phase"],
|
|
currency=raw["currency"],
|
|
description=raw["description"],
|
|
active_pricing_model_id=raw["active_pricing_model_id"],
|
|
)
|
|
|
|
|
|
def load_budget(data_dir: Path | None = None) -> Budget:
|
|
raw = _read_json((data_dir or default_data_dir()) / "budget.json")
|
|
return Budget(
|
|
currency=raw["currency"],
|
|
initial_budget=_money(raw["initial_budget"]),
|
|
started=raw["started"],
|
|
)
|
|
|
|
|
|
def load_pricing_models(data_dir: Path | None = None) -> list[PricingModel]:
|
|
raw = _read_json((data_dir or default_data_dir()) / "pricing-models.json")
|
|
return [
|
|
PricingModel(
|
|
id=item["id"],
|
|
name=item["name"],
|
|
model_type=item["model_type"],
|
|
lifecycle_phase=item["lifecycle_phase"],
|
|
currency=item["currency"],
|
|
access_fee_amount=_money(item["access_fee_amount"]),
|
|
access_fee_cadence=item["access_fee_cadence"],
|
|
status=item["status"],
|
|
)
|
|
for item in raw["models"]
|
|
]
|
|
|
|
|
|
def load_fx_rates(data_dir: Path | None = None) -> dict[str, Decimal]:
|
|
raw = _read_json((data_dir or default_data_dir()) / "expense_records.json")
|
|
return {pair: _money(rate) for pair, rate in raw.get("fx_rates", {}).items()} if raw.get(
|
|
"fx_rates"
|
|
) else {}
|
|
|
|
|
|
def load_expense_records(data_dir: Path | None = None) -> list[ExpenseRecord]:
|
|
raw = _read_json((data_dir or default_data_dir()) / "expense_records.json")
|
|
return [
|
|
ExpenseRecord(
|
|
id=item["id"],
|
|
period=item["period"],
|
|
vendor=item["vendor"],
|
|
description=item["description"],
|
|
cost_class=item["cost_class"],
|
|
amount=_money(item["amount"]),
|
|
currency=item["currency"],
|
|
source=item["source"],
|
|
)
|
|
for item in raw["records"]
|
|
]
|
|
|
|
|
|
def load_payment_records(data_dir: Path | None = None) -> list[PaymentRecord]:
|
|
root = data_dir or default_data_dir()
|
|
path = root / "payment_records.json"
|
|
if not path.exists():
|
|
# Backward compatibility with legacy revenue.json
|
|
path = root / "revenue.json"
|
|
raw = _read_json(path)
|
|
items = raw.get("records", raw.get("entries", []))
|
|
return [
|
|
PaymentRecord(
|
|
id=item["id"],
|
|
period=item["period"],
|
|
gross_amount=_money(item["gross_amount"]),
|
|
fees_amount=_money(item["fees_amount"]),
|
|
refunds_amount=_money(item.get("refunds_amount", "0")),
|
|
net_amount=_money(item["net_amount"]),
|
|
currency=item["currency"],
|
|
source=item["source"],
|
|
member_count=item.get("member_count", 0),
|
|
member_username=item.get("member_username"),
|
|
payout_account=item.get("payout_account"),
|
|
)
|
|
for item in items
|
|
]
|
|
|
|
|
|
def load_membership(data_dir: Path | None = None) -> list[MembershipRecord]:
|
|
raw = _read_json((data_dir or default_data_dir()) / "membership.json")
|
|
return [
|
|
MembershipRecord(
|
|
id=item["id"],
|
|
status=item["status"],
|
|
joined_at=item["joined_at"],
|
|
plan_id=item["plan_id"],
|
|
churned_at=item.get("churned_at"),
|
|
)
|
|
for item in raw["members"]
|
|
]
|
|
|
|
|
|
def load_monthly_ledger(data_dir: Path | None = None) -> list[MonthlyPlatformCost]:
|
|
root = data_dir or default_data_dir()
|
|
return build_monthly_ledger(
|
|
load_budget(root),
|
|
load_expense_records(root),
|
|
load_payment_records(root),
|
|
load_membership(root),
|
|
load_fx_rates(root),
|
|
)
|
|
|
|
|
|
def latest_period(monthly_costs: list[MonthlyPlatformCost]) -> str:
|
|
return max(item.period for item in monthly_costs) |