diff --git a/src/mytnb/__init__.py b/src/mytnb/__init__.py index 4779582..805295b 100644 --- a/src/mytnb/__init__.py +++ b/src/mytnb/__init__.py @@ -11,6 +11,7 @@ CustomerAccount, DailyUsage, Metric, + PaymentHistoryEntry, TariffBlock, TariffBlockLegendGroup, TariffBlockLegendItem, @@ -29,6 +30,7 @@ "CustomerAccount", "DailyUsage", "Metric", + "PaymentHistoryEntry", "TariffBlock", "TariffBlockLegendGroup", "TariffBlockLegendItem", diff --git a/src/mytnb/client/client.py b/src/mytnb/client/client.py index 7f60b2f..70e67ea 100644 --- a/src/mytnb/client/client.py +++ b/src/mytnb/client/client.py @@ -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. @@ -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(), } @@ -236,14 +255,15 @@ 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) @@ -251,14 +271,15 @@ async def get_account_due_amount( 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) @@ -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(), + } + 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 + 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) diff --git a/src/mytnb/models.py b/src/mytnb/models.py index e5ab3ce..8e2bd3d 100644 --- a/src/mytnb/models.py +++ b/src/mytnb/models.py @@ -2,8 +2,9 @@ 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 @@ -11,9 +12,12 @@ # 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+)\)/") + 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. @@ -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 for fmt in _API_DATE_FORMATS: try: return datetime.strptime(value.strip(), fmt).date() @@ -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"} """ @@ -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) + + @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.