1. Authentication & Token Cache

Client App → Auth API → token_cache.json

"""
auth_helper — Token cache to reuse tokens across script runs
=============================================================
Instead of calling authenticate() every time, this module:
  1. Loads the saved token from token_cache.json (if any)
  2. If missing → authenticates the first time and saves to file
  3. If the token is expired → refreshes and saves again
  4. If the token is still valid → uses it directly, no API call

Usage (sync):
    from auth_helper import ensure_auth
    with Auth(config) as auth:
        ensure_auth(auth)

Usage (async):
    from auth_helper import ensure_auth_async
    async with AsyncAuth(config) as auth:
        await ensure_auth_async(auth)
"""

import json
import os

from ssi_sdk.models import Token

TOKEN_FILE = os.path.join(os.path.dirname(__file__), "token_cache.json")


def load_token() -> Token | None:
    """Load the token from file; return None if the file does not exist."""
    if not os.path.exists(TOKEN_FILE):
        return None
    with open(TOKEN_FILE, "r", encoding="utf-8") as f:
        _token = json.load(f)
    return Token.from_dict(_token)


def save_token(token: Token) -> None:
    """Save the token to file."""
    with open(TOKEN_FILE, "w", encoding="utf-8") as f:
        json.dump(token.to_dict(), f, indent=2)
    print(f"Token saved to {TOKEN_FILE}")


def clean_otp(otp: str | None) -> str | None:
    """Ignore an empty OTP or a placeholder (`<OTP>`, `YOUR_OTP_CODE`) so the flow falls through to Smart OTP / manual entry."""
    if otp is None:
        return None
    otp = otp.strip()
    if not otp or (otp.startswith("<") and otp.endswith(">")) or otp.upper().startswith("YOUR_"):
        return None
    return otp


def ensure_auth(auth, otp: str | None = None) -> None:
    """Make sure auth has a valid token (sync)."""
    otp = clean_otp(otp)
    cached_token = load_token()
    if cached_token is not None:
        auth.token_manager.set_token(cached_token)
        if not auth.is_token_expired:
            print("Token still valid, using cached token.")
            return
        print("Token expired, refreshing...")
        try:
            cached_token = auth.refresh()
            save_token(cached_token)
            print("Token refresh successful.")
            return
        except Exception as e:
            print(f"Token refresh failed ({e}), authenticating again...")

    print("No valid token found, running the authenticate & OTP flow...")
    if otp:
        cached_token = auth.authenticate(otp=otp)
    else:
        print("=== Request OTP ===")
        otp_res = auth.request_otp()
        transaction_id = None
        if isinstance(otp_res, dict):
            data_map = otp_res.get("data", {})
            if isinstance(data_map, dict):
                transaction_id = data_map.get("transactionId") or data_map.get("transaction_id")
            if not transaction_id:
                transaction_id = otp_res.get("transactionId") or otp_res.get("transaction_id")

        if transaction_id:
            print("[Smart OTP] Approval request sent.")
            print("Please open the SSI iBoard app on your phone and tap APPROVE...")
            print("The SDK is polling and waiting for your approval...")
            auth.ensure_authenticated(transaction_id=transaction_id, poll_interval_ms=5_000, poll_max_retries=6)
        else:
            user_otp = input("Enter the 6-digit OTP: ").strip()
            auth.authenticate(otp=user_otp)

    if auth.token:
        save_token(auth.token)
        print("Authenticated, token saved.")


async def ensure_auth_async(auth, otp: str | None = None) -> None:
    """Make sure auth has a valid token (async). Same flow as ensure_auth, every SDK call is awaited."""
    otp = clean_otp(otp)
    cached_token = load_token()
    if cached_token is not None:
        await auth.token_manager.set_token(cached_token)
        if not auth.is_token_expired:
            print("Token still valid, using cached token.")
            return
        print("Token expired, refreshing...")
        try:
            cached_token = await auth.refresh()
            save_token(cached_token)
            print("Token refresh successful.")
            return
        except Exception as e:
            print(f"Token refresh failed ({e}), authenticating again...")

    print("No valid token found, running the authenticate & OTP flow...")
    if otp:
        cached_token = await auth.authenticate(otp=otp)
    else:
        otp_res = await auth.request_otp()
        transaction_id = None
        if isinstance(otp_res, dict):
            data_map = otp_res.get("data", {})
            if isinstance(data_map, dict):
                transaction_id = data_map.get("transactionId") or data_map.get("transaction_id")
            if not transaction_id:
                transaction_id = otp_res.get("transactionId") or otp_res.get("transaction_id")

        if transaction_id:
            print("Please open the SSI iBoard app on your phone and tap APPROVE...")
            await auth.ensure_authenticated(transaction_id=transaction_id, poll_interval_ms=5_000, poll_max_retries=6)
        else:
            user_otp = input("Enter the 6-digit OTP: ").strip()
            await auth.authenticate(otp=user_otp)

    if auth.token:
        save_token(auth.token)
        print("Authenticated, token saved.")

def enum_value(value) -> str:
    """Return .value for an enum, the value itself for a str, '-' for None."""
    if value is None:
        return "-"
    return str(getattr(value, "value", value))


def mask_account(account_no: str) -> str:
    """Mask the account number before printing (keep only the last 3 characters)."""
    if not account_no:
        return "-"
    return "*" * max(len(account_no) - 3, 0) + account_no[-3:]

from ssi_sdk import Auth, Trading, Config
from auth_helper import ensure_auth
from helpers import enum_value, mask_account

config = Config(
    client_id="<your_client_id>",
    api_key="<your_api_key>",
    api_secret="<your_api_secret>",
    private_key="<your_private_key>",
)

with Auth(config) as auth:
    # Steps 1-3: authenticate + request & verify OTP / Smart OTP
    ensure_auth(auth)
    token = auth.token
    print("Access Token :", token.access_token[:40], "...")
    print("Token Type   :", token.token_type)
    print("Expires At   :", token.expires_at)
    print("Refresh Token:", token.refresh_token[:40] if token.refresh_token else "N/A", "...")

    # Step 4: check token expiry & refresh
    if auth.is_token_expired:
        new_token = auth.refresh()
        print("New Token:", new_token.access_token[:40], "...")

    # Step 5: confirm the token works by calling an API
    with Trading(auth) as trading:
        accounts = trading.account.get_account_info()
        print(f"\nAuthenticated! Found {len(accounts)} accounts:")
        for acc in accounts:
            print(f"  - {mask_account(acc.account_no)} ({enum_value(acc.account_type)})")

import asyncio
from ssi_sdk import AsyncAuth, AsyncTrading, Config
from auth_helper import ensure_auth_async
from helpers import enum_value, mask_account

config = Config(
    client_id="<your_client_id>",
    api_key="<your_api_key>",
    api_secret="<your_api_secret>",
    private_key="<your_private_key>",
)

async def main():
    async with AsyncAuth(config) as auth:
        await ensure_auth_async(auth)
        token = auth.token
        print("Access Token :", token.access_token[:40], "...")

        if auth.is_token_expired:
            new_token = await auth.refresh()
            print("New Token:", new_token.access_token[:40], "...")

        async with AsyncTrading(auth) as trading:
            accounts = await trading.account.get_account_info()
            print(f"\nAuthenticated! Found {len(accounts)} accounts:")
            for acc in accounts:
                print(f"  - {mask_account(acc.account_no)} ({enum_value(acc.account_type)})")

asyncio.run(main())

from ssi_sdk import Auth, Config

config = Config(
    client_id="<your_client_id>",
    api_key="<your_api_key>",
    api_secret="<your_api_secret>",
    private_key="<your_private_key>",
)

with Auth(config) as auth:
    # Step 1: send the approval request to the Smart OTP app, get transaction_id
    # (transaction_id exists only for Smart OTP; it is None for SMS/Email OTP)
    otp_result = auth.request_otp_typed()
    transaction_id = otp_result.transaction_id
    print("Approval request sent. Please open the Smart OTP app and tap Approve.")

    # Step 2: the SDK polls automatically until the user approves (or poll_max_retries is reached)
    token = auth.ensure_authenticated(
        transaction_id=transaction_id,
        poll_interval_ms=5_000,
        poll_max_retries=6,
    )
    print("Approved! Access Token:", token.access_token[:40], "...")

import asyncio
from ssi_sdk import AsyncAuth, Config

config = Config(
    client_id="<your_client_id>",
    api_key="<your_api_key>",
    api_secret="<your_api_secret>",
    private_key="<your_private_key>",
)

async def main():
    async with AsyncAuth(config) as auth:
        # Step 1: send the approval request to the Smart OTP app, get transaction_id
        otp_result = await auth.request_otp_typed()
        transaction_id = otp_result.transaction_id
        print("Approval request sent. Please open the Smart OTP app and tap Approve.")

        # Step 2: the SDK polls automatically until the user approves (or poll_max_retries is reached)
        token = await auth.ensure_authenticated(
            transaction_id=transaction_id,
            poll_interval_ms=5_000,
            poll_max_retries=6,
        )
        print("Approved! Access Token:", token.access_token[:40], "...")

asyncio.run(main())