Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions src/mytnb/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
CustomerAccount,
DailyUsage,
Metric,
PaymentHistoryEntry,
TariffBlock,
TariffBlockLegendGroup,
TariffBlockLegendItem,
Expand All @@ -29,6 +30,7 @@
"CustomerAccount",
"DailyUsage",
"Metric",
"PaymentHistoryEntry",
"TariffBlock",
"TariffBlockLegendGroup",
"TariffBlockLegendItem",
Expand Down
92 changes: 80 additions & 12 deletions src/mytnb/client/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,12 +21,30 @@
BillHistoryEntry,
BREligibility,
CustomerAccount,
PaymentHistoryEntry,
SMRAccount,
)

logger = logging.getLogger(__name__)


def _resolve_account(
account: str | CustomerAccount,
is_owner: bool | None = None,
) -> tuple[str, bool, str]:
"""Resolve account identifier + metadata from a string or CustomerAccount.

Returns ``(account_number, is_owner, account_type)``.
"""
if isinstance(account, CustomerAccount):
return (
account.account_number,
is_owner if is_owner is not None else account.is_owned_bool,
"RE" if account.account_category_id == "2" else "UTIL",
)
return (account, is_owner if is_owner is not None else True, "UTIL")


class MyTNBClient:
"""Client for interacting with the myTNB API.

Expand Down Expand Up @@ -163,14 +181,15 @@ async def get_eligibility_icons(self) -> dict:

async def get_account_usage_smart(
self,
account_number: str,
account: str | CustomerAccount,
*,
is_owner: bool = True,
is_owner: bool | None = None,
) -> AccountUsage:
"""Get smart meter account usage data."""
acc_no, owner, _ = _resolve_account(account, is_owner)
data = {
"contractAccount": account_number,
"isOwner": "true" if is_owner else "false",
"contractAccount": acc_no,
"isOwner": "true" if owner else "false",
"metercode": "",
"usrInf": self._legacy_transport.base_user_info(),
}
Expand Down Expand Up @@ -236,29 +255,31 @@ async def get_customer_accounts(self) -> list[CustomerAccount]:

async def get_account_due_amount(
self,
account_number: str,
account: str | CustomerAccount,
*,
is_owner: bool = True,
is_owner: bool | None = None,
) -> AccountDueAmount:
"""Get account due amount as a typed model."""
acc_no, owner, _ = _resolve_account(account, is_owner)
data = {
"contractAccount": account_number,
"isOwnedAccount": "true" if is_owner else "false",
"contractAccount": acc_no,
"isOwnedAccount": "true" if owner else "false",
"usrInf": self._legacy_transport.base_user_info(),
}
result = await self._legacy_transport.post("GetAccountDueAmount", data)
return AccountDueAmount.from_api_response(result.get("data", result))

async def get_bill_history(
self,
account_number: str,
account: str | CustomerAccount,
*,
is_owner: bool = True,
is_owner: bool | None = None,
) -> list[BillHistoryEntry]:
"""Get bill payment history as typed models (most recent first)."""
acc_no, owner, _ = _resolve_account(account, is_owner)
data = {
"contractAccount": account_number,
"isOwnedAccount": "true" if is_owner else "false",
"contractAccount": acc_no,
"isOwnedAccount": "true" if owner else "false",
"usrInf": self._legacy_transport.base_user_info(),
}
result = await self._legacy_transport.post("GetBillHistory", data)
Expand All @@ -279,6 +300,53 @@ async def get_bill_history(
)
return []

async def get_payment_history(
self,
account: str | CustomerAccount,
*,
is_owner: bool | None = None,
account_type: str | None = None,
) -> list[PaymentHistoryEntry]:
"""Get bill & payment history as typed models (most recent first).

Uses the GetAccountBillPayHistoryV4 endpoint which returns both bills
and payments. Each entry has a ``history_type`` field ("PAYMENT",
"BILL", or "ADVICE") so callers can filter as needed.

When ``account`` is a ``CustomerAccount``, ``is_owner`` and
``account_type`` are derived automatically (unless explicitly
overridden).
"""
acc_no, owner, derived_type = _resolve_account(account, is_owner)
if account_type is None:
account_type = derived_type
data = {
"contractAccount": acc_no,
"isOwnedAccount": owner,
"accountType": account_type,
"usrInf": self._legacy_transport.base_user_info(),
}
Comment on lines +323 to +328
result = await self._legacy_transport.post(
"GetAccountBillPayHistoryV4", data
)
inner = result.get("data", result)
histories: list[PaymentHistoryEntry] = []
if isinstance(inner, dict):
for group in inner.get("BillPayHistories", []):
if not isinstance(group, dict):
continue
for item in group.get("BillPayHistoryData", []):
if isinstance(item, dict):
histories.append(PaymentHistoryEntry.model_validate(item))
elif isinstance(inner, list):
for group in inner:
if not isinstance(group, dict):
continue
for item in group.get("BillPayHistoryData", []):
if isinstance(item, dict):
histories.append(PaymentHistoryEntry.model_validate(item))
return histories
Comment thread
danieyal marked this conversation as resolved.

async def get_current_usage(self, account_number: str) -> dict:
"""Get a simplified summary of current usage."""
usage = await self.get_account_usage_smart(account_number)
Expand Down
59 changes: 56 additions & 3 deletions src/mytnb/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,18 +2,22 @@

from __future__ import annotations

import re
from datetime import date as date_cls
from datetime import datetime
from datetime import datetime, timezone
from typing import Any, Optional

from pydantic import AliasChoices, BaseModel, Field, field_validator

# myTNB returns dates as DD/MM/YYYY; keep ISO variants as fallbacks.
_API_DATE_FORMATS = ("%d/%m/%Y", "%Y-%m-%d", "%d-%m-%Y")

# Microsoft JSON date format: /Date(1782921600000)/
_MS_DATE_RE = re.compile(r"/Date\((\d+)\)/")
Comment on lines +15 to +16


def parse_api_date(value: Any) -> Optional[date_cls]:
"""Parse a myTNB date string (typically DD/MM/YYYY) into a date.
"""Parse a myTNB date string (DD/MM/YYYY or /Date(ms)/) into a date.

Returns None for empty/None/unparseable input so a single bad field never
breaks an entire response.
Expand All @@ -24,6 +28,15 @@ def parse_api_date(value: Any) -> Optional[date_cls]:
return value
if not isinstance(value, str) or not value.strip():
return None
# Microsoft /Date(1782921600000)/ format (ms since epoch UTC)
ms_match = _MS_DATE_RE.fullmatch(value.strip())
if ms_match:
try:
return datetime.fromtimestamp(
int(ms_match.group(1)) / 1000, tz=timezone.utc
).date()
except (ValueError, OSError):
return None
Comment on lines +34 to +39
for fmt in _API_DATE_FORMATS:
try:
return datetime.strptime(value.strip(), fmt).date()
Expand Down Expand Up @@ -436,7 +449,7 @@ def is_paid_bool(self) -> bool:


class BillHistoryEntry(BaseModel):
"""A single bill payment history entry from GetBillHistory.
"""A single bill entry from GetBillHistory (bills *issued*, not payments).

Raw shape: {"DtBill": "31/05/2026", "AmPayable": "87.50", "BillingNo": "12345"}
"""
Expand All @@ -463,6 +476,46 @@ def _coerce_billing_no(cls, v: Any) -> str:
return "" if v is None else str(v)


class PaymentHistoryEntry(BaseModel):
"""A single payment-or-bill entry from GetAccountBillPayHistoryV4.

Raw shape:
{"BillOrPaymentDate":"15/01/2026","HistoryType":"Payment","Amount":"100.50",…}
"""

date: Optional[date_cls] = Field(default=None, alias="BillOrPaymentDate")
history_type: str = Field(default="", alias="HistoryType")
date_and_history_type: str = Field(default="", alias="DateAndHistoryType")
amount: Optional[float] = Field(default=None, alias="Amount")
reference_number: str = Field(default="", alias="DetailedInfoNumber")
paid_via: str = Field(default="", alias="PaidVia")
history_type_text: str = Field(default="", alias="HistoryTypeText")
is_payment_pending: bool = Field(default=False, alias="IsPaymentPending")

model_config = {"populate_by_name": True, "extra": "ignore"}

@field_validator("date", mode="before")
@classmethod
def _parse_date(cls, v: Any) -> Optional[date_cls]:
return parse_api_date(v)

@field_validator("amount", mode="before")
@classmethod
def _parse_amount(cls, v: Any) -> Optional[float]:
return _parse_optional_float(v)
Comment thread
danieyal marked this conversation as resolved.
Comment on lines +479 to +505

@field_validator("history_type", "date_and_history_type", "reference_number",
"paid_via", "history_type_text", mode="before")
@classmethod
def _coerce_str(cls, v: Any) -> str:
return "" if v is None else str(v)

@property
def is_payment(self) -> bool:
"""True when this entry is a payment (not a bill or advice)."""
return self.history_type.upper() == "PAYMENT" # pylint: disable=no-member


class AccountDueAmount(BaseModel):
"""Outstanding balance for an account from GetAccountDueAmount.

Expand Down
Loading