#!/usr/bin/env python3
"""HPRT BI client. Python 3.10+, standard library only, bounded read-only collection."""

import argparse
import json
import os
import re
import stat
import time
from decimal import Decimal
from pathlib import Path
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode, urlsplit
from urllib.request import HTTPRedirectHandler, ProxyHandler, Request, build_opener


class BiError(RuntimeError):
    def __init__(self, status, code, request_id=None):
        self.status, self.code, self.request_id = status, code, request_id
        super().__init__(f"HTTP {status}; code={code}; request_id={request_id or '-'}")


class NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        # Never forward a password or Bearer token to another origin.
        return None


class BiClient:
    PATHS = {
        "payments",
        "payment-orders",
        "payment-events",
        "usage",
        "calls",
        "auth/session",
    }

    def __init__(
        self,
        base_url,
        username=None,
        password=None,
        *,
        api_key=None,
        sleep=time.sleep,
        clock=time.monotonic,
    ):
        parsed = urlsplit(base_url)
        local = parsed.scheme == "http" and parsed.hostname in {
            "127.0.0.1",
            "localhost",
            "::1",
        }
        if (
            not (parsed.scheme == "https" or local)
            or not parsed.hostname
            or parsed.username
            or parsed.password
            or parsed.query
            or parsed.fragment
            or parsed.path not in {"", "/"}
        ):
            raise ValueError(
                "BI_API_URL must be an HTTPS origin (loopback HTTP is allowed for tests)"
            )
        if api_key and (username or password):
            raise ValueError("Use API Key or username/password, never both")
        if not api_key and (not username or not password):
            raise ValueError("BI API key file or username/password is required")
        if api_key is not None and (
            not isinstance(api_key, str)
            or not re.fullmatch(r"hprt_bi_[0-9a-f]{32}_[A-Za-z0-9_-]{43}", api_key)
        ):
            raise ValueError("Invalid BI API key format")
        self._api_key = api_key
        self.base_url, self.username, self._password = (
            base_url.rstrip("/"),
            username,
            password,
        )
        self._sleep, self._clock = sleep, clock
        self._token, self._expires = None, 0
        self._opener = build_opener(ProxyHandler({}), NoRedirect())

    def _send(self, path, *, body=None, query=None):
        url = self.base_url + "/api/v1/bi/" + path
        if query:
            url += "?" + urlencode(query)
        headers = {"Accept": "application/json"}
        if self._api_key and body is None:
            headers["X-API-Key"] = self._api_key
        if self._token and body is None:
            headers["Authorization"] = "Bearer " + self._token
        if body is not None:
            headers["Content-Type"] = "application/json"
        request = Request(
            url,
            headers=headers,
            data=json.dumps(body).encode() if body is not None else None,
        )
        try:
            with self._opener.open(request, timeout=20) as response:
                return (
                    response.status,
                    {key.lower(): value for key, value in response.headers.items()},
                    json.loads(response.read(4_000_001)),
                )
        except HTTPError as error:
            raw = error.read(16_384)
            try:
                payload = json.loads(raw)
            except (ValueError, UnicodeDecodeError):
                payload = {}
            return (
                error.code,
                {key.lower(): value for key, value in error.headers.items()},
                payload,
            )
        except (URLError, TimeoutError, OSError) as error:
            raise BiError(0, "NETWORK_UNAVAILABLE") from error
        except (ValueError, UnicodeDecodeError) as error:
            raise BiError(0, "INVALID_RESPONSE") from error

    def login(self):
        if self._api_key:
            raise ValueError("API Key mode does not use account login")
        status, headers, payload = self._send(
            "auth/login", body={"username": self.username, "password": self._password}
        )
        if status != 200:
            self._raise(status, headers, payload)
        if not isinstance(payload, dict):
            raise BiError(0, "INVALID_RESPONSE")
        if payload.get("must_change_password"):
            raise BiError(403, "ADMIN_PASSWORD_CHANGE_REQUIRED")
        token, expires = payload.get("access_token"), payload.get("expires_in")
        if not isinstance(token, str) or not token or type(expires) is not int or expires <= 0:
            raise BiError(0, "INVALID_RESPONSE")
        self._token = token
        self._expires = self._clock() + max(1, expires - 30)

    @staticmethod
    def _raise(status, headers, payload):
        error = payload.get("error", {}) if isinstance(payload, dict) else {}
        if not isinstance(error, dict):
            error = {}
        raise BiError(
            status,
            error.get("code", "HTTP_ERROR"),
            error.get("request_id") or headers.get("x-request-id"),
        )

    def get(self, path, **query):
        if path not in self.PATHS:
            raise ValueError("Only the documented BI read endpoints are allowed")
        if self._api_key and path == "auth/session":
            raise ValueError("API Key permits only the five BI data queries")
        if not self._api_key and (not self._token or self._clock() >= self._expires):
            self.login()
        refreshed = False
        for attempt in range(3):
            try:
                status, headers, payload = self._send(path, query=query)
            except BiError as error:
                if error.code != "NETWORK_UNAVAILABLE" or attempt == 2:
                    raise
                self._sleep(attempt + 1)
                continue
            if status == 200:
                if not isinstance(payload, dict):
                    raise BiError(0, "INVALID_RESPONSE")
                return payload
            if status == 401 and not refreshed and not self._api_key:
                self._token = None
                self.login()
                refreshed = True
                continue
            if status in {429, 502, 503, 504} and attempt < 2:
                try:
                    delay = float(headers.get("retry-after", attempt + 1))
                except (TypeError, ValueError):
                    self._raise(status, headers, payload)
                if not 0 < delay <= 60:
                    self._raise(status, headers, payload)
                self._sleep(delay)
                continue
            self._raise(status, headers, payload)
        raise BiError(401, "AUTH_RETRY_EXHAUSTED")

    def pages(self, path, *, max_pages=1000, **query):
        if path not in {"payment-orders", "payment-events", "calls"}:
            raise ValueError("This endpoint is not paginated")
        query = {**query, "page_size": 200}
        seen = set()
        id_key = {
            "payment-orders": "order_id",
            "payment-events": "event_id",
            "calls": "call_id",
        }[path]
        expected_total = None
        watermark = query.get("snapshot_at")
        for number in range(1, max_pages + 1):
            page = self.get(path, page=number, **query)
            if (
                not isinstance(page.get("items"), list)
                or not isinstance(page.get("snapshot_at"), str)
                or type(page.get("total")) is not int
                or page["total"] < 0
            ):
                raise BiError(0, "INVALID_RESPONSE")
            if expected_total is not None and page["total"] != expected_total:
                raise BiError(0, "PAGINATION_CHANGED")
            if watermark is not None and page["snapshot_at"] != watermark:
                raise BiError(0, "PAGINATION_CHANGED")
            expected_total, watermark = page["total"], page["snapshot_at"]
            query["snapshot_at"] = page["snapshot_at"]
            for item in page["items"]:
                if not isinstance(item, dict) or not isinstance(item.get(id_key), str):
                    raise BiError(0, "INVALID_RESPONSE")
                if item[id_key] in seen:
                    raise BiError(0, "DUPLICATE_RECORD")
                seen.add(item[id_key])
                yield item
            expected_rows = min(
                query["page_size"],
                max(0, expected_total - (number - 1) * query["page_size"]),
            )
            if len(page["items"]) != expected_rows:
                raise BiError(0, "INCOMPLETE_PAGE")
            if number * query["page_size"] >= page["total"]:
                return
        raise BiError(0, "PAGE_LIMIT_EXCEEDED")


def money(amount, scale):
    """Decimal conversion: payment scale=1000, model cost scale=1000000; null stays unknown."""
    return None if amount is None else Decimal(amount) / Decimal(scale)


def read_credential(path, label):
    credential_file = Path(path)
    if os.name != "nt" and stat.S_IMODE(credential_file.stat().st_mode) & 0o077:
        raise ValueError(f"{label} must be readable only by its owner (chmod 600)")
    return credential_file.read_text().strip()


def from_environment():
    key_file = os.environ.get("BI_API_KEY_FILE")
    if key_file:
        if os.environ.get("BI_USERNAME") or os.environ.get("BI_PASSWORD_FILE"):
            raise ValueError("Use BI_API_KEY_FILE or username/password, never both")
        return BiClient(
            os.environ["BI_API_URL"], api_key=read_credential(key_file, "BI_API_KEY_FILE")
        )
    return BiClient(
        os.environ["BI_API_URL"],
        os.environ["BI_USERNAME"],
        read_credential(os.environ["BI_PASSWORD_FILE"], "BI_PASSWORD_FILE"),
    )


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--start-date", required=True)
    parser.add_argument("--end-date", required=True)
    parser.add_argument("--currency", default="CNY")
    parser.add_argument("--output", required=True, type=Path)
    parser.add_argument("--include-details", action="store_true")
    args = parser.parse_args()
    client = from_environment()
    window = {"start_date": args.start_date, "end_date": args.end_date}
    result = {
        "payments": client.get("payments", currency=args.currency, **window),
        "usage": client.get("usage", **window),
    }
    if args.include_details:
        result.update(
            {
                name: list(
                    client.pages(
                        name,
                        **window,
                        **({"currency": args.currency} if name != "calls" else {}),
                    )
                )
                for name in ("payment-orders", "payment-events", "calls")
            }
        )
    # Refuse to overwrite an existing delivery snapshot.
    descriptor = os.open(args.output, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    with os.fdopen(descriptor, "w", encoding="utf-8") as stream:
        json.dump(result, stream, ensure_ascii=False, indent=2)
    print("BI data saved; credentials and tokens are excluded.")


if __name__ == "__main__":
    try:
        main()
    except (BiError, ValueError, KeyError, OSError) as error:
        raise SystemExit(str(error)) from None