# oropay 0.1.0 — single-file build (Python 3.8+, stdlib only: import oropay). MIT License. https://oropay.app/merchant
# Generated by sdk/build.sh from sdk/python/src/oropay/__init__.py — do not edit by hand.
"""oropay -- client for the OroPay Merchant API (https://oropay.app/merchant).

Pure standard library (urllib, hmac, hashlib, json), Python 3.8+. MIT License.

    from oropay import OroPay, verify_webhook

    oropay = OroPay(api_key=os.environ["OROPAY_API_KEY"])
    invoice = oropay.create_invoice("5.00", order_id="order-1")
    # send the buyer to invoice["payUrl"] (Telegram) or invoice["checkoutUrl"] (website)

Responses are plain dicts with the API's own (camelCase) keys. Money amounts
are strings ("100.00"), dates are ISO 8601 strings in UTC.

This file is also distributed as a single-file drop-in (oropay.py): put it next
to your code and `import oropay`.
"""

from __future__ import annotations

import hashlib
import hmac
import http.client
import json
import re
import socket
import time
import urllib.error
import urllib.parse
import urllib.request
from decimal import Decimal
from typing import Any, Dict, Iterable, List, Mapping, Optional, Sequence, Union

__version__ = "0.1.0"

__all__ = [
    "OroPay",
    "OroPayError",
    "WebhookVerificationError",
    "verify_webhook",
    "sign_webhook",
    "DEFAULT_BASE_URL",
    "__version__",
]

DEFAULT_BASE_URL = "https://oropay.app/api/merchant"

#: A money amount. Strings are recommended ("9.99"); int/float/Decimal are converted with str().
Amount = Union[str, int, float, Decimal]


# ─────────────────────────── errors ───────────────────────────


class OroPayError(Exception):
    """Any failed API call.

    status -- HTTP status (0 -- no response: network error or timeout).
    code   -- stable machine code of /v1/... methods ("DAILY_LIMIT_EXCEEDED",
              "SPEND_ID_CONFLICT", ...), "TIMEOUT" / "NETWORK_ERROR" for transport
              failures, or None (old invoice methods return only text).
    data   -- the full response body (extra fields like "leftUsd", "available").
    """

    def __init__(self, message: str, status: int = 0, code: Optional[str] = None, data: Any = None) -> None:
        super().__init__(message)
        self.message = message
        self.status = status
        self.code = code
        self.data = data

    def __repr__(self) -> str:
        return "OroPayError(status=%r, code=%r, message=%r)" % (self.status, self.code, self.message)


class WebhookVerificationError(ValueError):
    """The webhook did not pass verification. Respond 4xx and do nothing.

    reason -- MISSING_SIGNATURE, MISSING_TIMESTAMP, INVALID_TIMESTAMP,
              INVALID_SIGNATURE, STALE_TIMESTAMP or INVALID_BODY.
    """

    def __init__(self, reason: str, message: str) -> None:
        super().__init__(message)
        self.reason = reason


# ─────────────────────────── client ───────────────────────────


def _compact(d: Mapping[str, Any]) -> Dict[str, Any]:
    return {k: v for k, v in d.items() if v is not None}


def _amount(v: Optional[Amount]) -> Optional[str]:
    return None if v is None else str(v)


def _seg(v: Any, what: str) -> str:
    s = "" if v is None else str(v).strip()
    if not s:
        raise ValueError("oropay: %s is required" % what)
    return urllib.parse.quote(s, safe="")


def _without_ok(data: Dict[str, Any]) -> Dict[str, Any]:
    return {k: v for k, v in data.items() if k != "ok"}


class OroPay:
    """OroPay Merchant API client.

    api_key  -- "cxc_..." from the shop dashboard. Keep it on the server only.
    base_url -- default https://oropay.app/api/merchant
    timeout  -- seconds per request, default 30.
    """

    def __init__(
        self,
        api_key: str,
        base_url: str = DEFAULT_BASE_URL,
        timeout: float = 30.0,
        user_agent: Optional[str] = None,
    ) -> None:
        if not isinstance(api_key, str) or not api_key.strip():
            raise ValueError("oropay: api_key is required (cxc_... from the shop dashboard)")
        self._api_key = api_key.strip()
        self.base_url = (base_url or DEFAULT_BASE_URL).rstrip("/")
        self.timeout = timeout
        # The site firewall rejects the default "Python-urllib/x.y" User-Agent
        # (403 "error code: 1010"), so we always send our own.
        self.user_agent = user_agent or "oropay-python/%s" % __version__

    def __repr__(self) -> str:
        return "OroPay(base_url=%r, api_key=%r)" % (self.base_url, self._api_key[:8] + "...")

    # Low-level call: returns the parsed JSON body or raises OroPayError.
    def request(
        self,
        method: str,
        path: str,
        query: Optional[Mapping[str, Any]] = None,
        body: Optional[Mapping[str, Any]] = None,
    ) -> Dict[str, Any]:
        url = self.base_url + path
        if query:
            q = {k: str(v) for k, v in query.items() if v is not None and v != ""}
            if q:
                url += "?" + urllib.parse.urlencode(q)
        headers = {
            "Authorization": "Bearer " + self._api_key,
            "Accept": "application/json",
            "User-Agent": self.user_agent,
        }
        data = None
        if body is not None:
            headers["Content-Type"] = "application/json"
            data = json.dumps(_compact(body), ensure_ascii=False).encode("utf-8")
        req = urllib.request.Request(url, data=data, headers=headers, method=method)

        try:
            with urllib.request.urlopen(req, timeout=self.timeout) as resp:
                status = resp.status
                text = resp.read().decode("utf-8", "replace")
        except urllib.error.HTTPError as e:
            status = e.code
            try:
                text = e.read().decode("utf-8", "replace")
            except Exception:
                text = ""
        except (socket.timeout, TimeoutError) as e:
            raise OroPayError("OroPay did not answer in %s s" % self.timeout, 0, "TIMEOUT") from e
        except urllib.error.URLError as e:
            if isinstance(e.reason, (socket.timeout, TimeoutError)):
                raise OroPayError("OroPay did not answer in %s s" % self.timeout, 0, "TIMEOUT") from e
            raise OroPayError("Network error: %s" % (e.reason,), 0, "NETWORK_ERROR") from e
        except (OSError, ValueError, http.client.HTTPException) as e:  # connection reset, bad URL, ...
            raise OroPayError("Network error: %s" % (e,), 0, "NETWORK_ERROR") from e

        try:
            parsed = json.loads(text) if text else None
        except ValueError:
            parsed = None
        if status >= 400 or not isinstance(parsed, dict) or parsed.get("ok") is False:
            if isinstance(parsed, dict) and isinstance(parsed.get("error"), str):
                message = parsed["error"]
            else:
                message = "HTTP %d%s" % (status, (": " + text[:200]) if text else "")
            code = parsed.get("code") if isinstance(parsed, dict) and isinstance(parsed.get("code"), str) else None
            raise OroPayError(message, status, code, parsed if parsed is not None else text)
        return parsed

    # ───────────── invoices ─────────────

    def create_invoice(
        self,
        amount: Amount,
        currency: Optional[str] = None,
        order_id: Optional[str] = None,
        description: Optional[str] = None,
        fee_mode: Optional[str] = None,
        callback_url: Optional[str] = None,
        return_url: Optional[str] = None,
        payment_methods: Optional[Sequence[str]] = None,
        test: Optional[bool] = None,
    ) -> Dict[str, Any]:
        """Create an invoice; send the buyer to invoice["payUrl"] (Telegram) or invoice["checkoutUrl"] (website).

        amount          -- order amount, > 0, at most 2 decimals ("100.00").
        currency        -- USD (default), EUR, RUB or UAH.
        order_id        -- your order number = idempotency key (up to 128 chars): the same
                           order_id + amount + currency returns the same invoice; another amount
                           or an expired/cancelled invoice -> 409. Store it, do not regenerate on retry.
        fee_mode        -- "SHOP" or "BUYER" (default: app setting).
        callback_url    -- webhook URL for this invoice (default: app webhook URL).
        return_url      -- https://your.site/... or https://t.me/your_bot?start=... after payment.
        payment_methods -- ["telegram", "address"] (default both).
        test            -- test invoice, no real money; pay with "Pay (test)" on checkoutUrl.

        Fee percent is fixed per invoice ("feePercent"): new shops have 0% for
        30 days (60 with a promo code) -- never hard-code 0.75.
        """
        if amount is None or amount == "":
            raise ValueError("oropay: amount is required")
        body = {
            "amount": _amount(amount),
            "currency": currency,
            "orderId": order_id,
            "description": description,
            "feeMode": fee_mode,
            "callbackUrl": callback_url,
            "returnUrl": return_url,
            "paymentMethods": list(payment_methods) if payment_methods is not None else None,
            "test": test,
        }
        return self.request("POST", "/invoice", body=body)["invoice"]

    def get_invoice(self, code: str) -> Dict[str, Any]:
        """Current state of an invoice (fallback when you have no webhook)."""
        return self.request("GET", "/invoice/" + _seg(code, "code"))["invoice"]

    def cancel_invoice(self, code: str) -> Dict[str, Any]:
        """Cancel an unpaid invoice. Cancelling a cancelled one is not an error."""
        return self.request("POST", "/invoice/" + _seg(code, "code") + "/cancel")["invoice"]

    def list_invoices(
        self, limit: Optional[int] = None, status: Optional[str] = None, cursor: Optional[str] = None
    ) -> Dict[str, Any]:
        """{"invoices": [...], "nextCursor": "PR..." or None} -- newest first, cursor pagination."""
        data = self.request("GET", "/invoices", query={"limit": limit, "status": status, "cursor": cursor})
        return {"invoices": data.get("invoices", []), "nextCursor": data.get("nextCursor")}

    def get_networks(self) -> List[Dict[str, Any]]:
        """Networks for paying an invoice to an address: [{"network": "TRON:USDT", "label", "coin", "minUsd", "ttlMin"}]."""
        return self.request("GET", "/networks")["networks"]

    def create_address_payment(self, code: str, network: str) -> Dict[str, Any]:
        """Address + exact amount ("payAmount") to pay the invoice from any wallet. "qr" is made absolute."""
        data = self.request("POST", "/invoice/" + _seg(code, "code") + "/address", body={"network": network})
        return self._absolute_qr(data["payment"])

    def list_address_payments(self, code: str) -> List[Dict[str, Any]]:
        data = self.request("GET", "/invoice/" + _seg(code, "code") + "/address")
        return [self._absolute_qr(p) for p in data.get("payments", [])]

    def _absolute_qr(self, p: Dict[str, Any]) -> Dict[str, Any]:
        qr = p.get("qr") if isinstance(p, dict) else None
        if isinstance(qr, str) and qr.startswith("/"):
            p = dict(p)
            p["qr"] = urllib.parse.urljoin(self.base_url, qr)
        return p

    # ───────────── payouts and refunds ─────────────

    def get_balance(self) -> Dict[str, Any]:
        """{"balances": [...], "payouts": {...limits...}} -- any valid key."""
        return _without_ok(self.request("GET", "/v1/balance"))

    def create_cheque(
        self,
        spend_id: str,
        asset: str,
        amount: Amount,
        activations: Optional[int] = None,
        network: Optional[str] = None,
        description: Optional[str] = None,
        expires_in_days: Optional[int] = None,
        password: Optional[str] = None,
        new_users_only: Optional[bool] = None,
        channel: Optional[str] = None,
        captcha: Optional[bool] = None,
    ) -> Dict[str, Any]:
        """{"cheque": {...link...}, "repeated": bool}. Needs the payouts scope and payouts enabled."""
        body = {
            "spend_id": spend_id,
            "asset": asset,
            "amount": _amount(amount),
            "activations": activations,
            "network": network,
            "description": description,
            "expires_in_days": expires_in_days,
            "password": password,
            "new_users_only": new_users_only,
            "channel": channel,
            "captcha": captcha,
        }
        return _without_ok(self.request("POST", "/v1/cheques", body=body))

    def list_cheques(
        self, status: Optional[str] = None, count: Optional[int] = None, offset: Optional[int] = None
    ) -> List[Dict[str, Any]]:
        return self.request("GET", "/v1/cheques", query={"status": status, "count": count, "offset": offset})["cheques"]

    def cancel_cheque(self, id_or_code: Union[str, int]) -> Dict[str, Any]:
        """{"cheque": {...}, "repeated": bool} -- unclaimed activations return to the balance."""
        return _without_ok(self.request("DELETE", "/v1/cheques/" + _seg(id_or_code, "cheque id")))

    def transfer(
        self,
        spend_id: str,
        user_id: Union[str, int],
        asset: str,
        amount: Amount,
        comment: Optional[str] = None,
        network: Optional[str] = None,
    ) -> Dict[str, Any]:
        """{"transfer": {...}, "repeated": bool} -- instant transfer to an OroPay user by Telegram ID."""
        body = {
            "spend_id": spend_id,
            "user_id": None if user_id is None else str(user_id),
            "asset": asset,
            "amount": _amount(amount),
            "comment": comment,
            "network": network,
        }
        return _without_ok(self.request("POST", "/v1/transfers", body=body))

    def refund(
        self, code: str, refund_id: str, amount: Optional[Amount] = None, reason: Optional[str] = None
    ) -> Dict[str, Any]:
        """{"refund", "repeated", "refundedTotal", "refundable", "invoice"}; no amount -- the whole remainder."""
        body = {"refund_id": refund_id, "amount": _amount(amount), "reason": reason}
        return _without_ok(self.request("POST", "/v1/invoices/" + _seg(code, "code") + "/refund", body=body))

    def list_refunds(self, code: str) -> Dict[str, Any]:
        """{"refunds", "refundedTotal", "refundable", "invoice"}."""
        return _without_ok(self.request("GET", "/v1/invoices/" + _seg(code, "code") + "/refunds"))

    # ───────────── subscriptions ─────────────

    def list_plans(self) -> List[Dict[str, Any]]:
        return self.request("GET", "/v1/plans")["plans"]

    def create_plan(
        self,
        name: str,
        asset: str,
        price: Amount,
        period_days: int,
        trial_days: Optional[int] = None,
        description: Optional[str] = None,
        plan_id: Optional[str] = None,
    ) -> Dict[str, Any]:
        """{"plan": {..."link"...}, "repeated": bool}. asset USDT/USDC, period 7/30/90/365, trial 0/3/7."""
        body = {
            "name": name,
            "asset": asset,
            "price": _amount(price),
            "period_days": period_days,
            "trial_days": trial_days,
            "description": description,
            "plan_id": plan_id,
        }
        return _without_ok(self.request("POST", "/v1/plans", body=body))

    def disable_plan(self, code: str) -> Dict[str, Any]:
        """Stop accepting new subscribers (existing ones keep renewing)."""
        return self.request("POST", "/v1/plans/" + _seg(code, "plan code") + "/disable")["plan"]

    def enable_plan(self, code: str) -> Dict[str, Any]:
        return self.request("POST", "/v1/plans/" + _seg(code, "plan code") + "/enable")["plan"]

    def list_subscriptions(
        self,
        user_id: Optional[Union[str, int]] = None,
        plan: Optional[str] = None,
        status: Optional[str] = None,
        limit: Optional[int] = None,
        offset: Optional[int] = None,
    ) -> Dict[str, Any]:
        """{"subscriptions": [...]} plus top-level "active" when user_id is given."""
        query = {"user_id": user_id, "plan": plan, "status": status, "limit": limit, "offset": offset}
        return _without_ok(self.request("GET", "/v1/subscriptions", query=query))

    def has_access(self, user_id: Union[str, int], plan: Optional[str] = None) -> bool:
        """Does this Telegram user have paid access now (optionally by a given plan code)."""
        return self.list_subscriptions(user_id=user_id, plan=plan).get("active") is True

    def get_subscription(self, subscription_id: Union[str, int]) -> Dict[str, Any]:
        return self.request("GET", "/v1/subscriptions/" + _seg(subscription_id, "subscription id"))["subscription"]

    def cancel_subscription(self, subscription_id: Union[str, int]) -> Dict[str, Any]:
        """{"subscription": {...}, "repeated": bool} -- access stays until currentPeriodEnd."""
        path = "/v1/subscriptions/" + _seg(subscription_id, "subscription id") + "/cancel"
        return _without_ok(self.request("POST", path))


# ─────────────────────────── webhooks ───────────────────────────


def _first(v: Any) -> Optional[str]:
    if isinstance(v, (list, tuple)):
        v = v[0] if v else None
    if isinstance(v, bytes):
        v = v.decode("latin-1")
    return None if v is None else str(v)


def _header(headers: Any, name: str) -> Optional[str]:
    """Case-insensitive lookup in a dict, aiohttp/Flask/Django/Starlette headers or a list of pairs."""
    if headers is None:
        return None
    get = getattr(headers, "get", None)
    if callable(get):
        for variant in (name, name.title(), name.upper()):
            try:
                v = get(variant)
            except Exception:
                v = None
            if v is not None:
                return _first(v)
    items: Iterable[Any] = headers.items() if hasattr(headers, "items") else headers
    try:
        for k, v in items:
            if str(k).lower() == name:
                return _first(v)
    except (TypeError, ValueError):
        pass
    return None


def _bytes(raw_body: Any) -> bytes:
    if isinstance(raw_body, bytes):
        return raw_body
    if isinstance(raw_body, (bytearray, memoryview)):
        return bytes(raw_body)
    if isinstance(raw_body, str):
        return raw_body.encode("utf-8")
    raise TypeError(
        "oropay: raw_body must be the raw request body (bytes), not a parsed object -- "
        "e.g. await request.read() in aiohttp, request.get_data() in Flask, request.body in Django"
    )


def sign_webhook(secret: Union[str, bytes], timestamp: Union[str, int], raw_body: Union[bytes, str]) -> str:
    """HMAC-SHA256 hex of b"{timestamp}\\n" + raw_body -- exactly how OroPay signs webhooks."""
    key = secret.encode("utf-8") if isinstance(secret, str) else secret
    return hmac.new(key, str(timestamp).encode("ascii") + b"\n" + _bytes(raw_body), hashlib.sha256).hexdigest()


_TS = re.compile(r"[0-9]{10,16}")


def verify_webhook(
    raw_body: Union[bytes, str],
    headers: Any,
    secret: Union[str, bytes],
    tolerance: float = 300,
    now: Optional[float] = None,
) -> Dict[str, Any]:
    """Verify an OroPay webhook and return the parsed event (dict).

    raw_body  -- the body exactly as received (bytes), before any JSON parsing.
    headers   -- request headers (any mapping; lookup is case-insensitive).
    secret    -- your webhook secret (NOT the API key).
    tolerance -- max difference between x-cxc-timestamp (Unix ms) and now, seconds, either way.
    now       -- current Unix time in seconds (for tests); default time.time().

    Checks the HMAC-SHA256 signature from x-cxc-signature with a constant-time
    compare and rejects stale timestamps. Raises WebhookVerificationError.
    The same event can arrive twice -- dedupe by invoice["code"],
    refund["refundId"] or the subscription event "id".
    """
    if not secret:
        raise ValueError("oropay: webhook secret is required")
    body = _bytes(raw_body)
    ts = (_header(headers, "x-cxc-timestamp") or "").strip()
    sig = (_header(headers, "x-cxc-signature") or "").strip().lower()
    if not ts:
        raise WebhookVerificationError("MISSING_TIMESTAMP", "x-cxc-timestamp header is missing")
    if not sig:
        raise WebhookVerificationError("MISSING_SIGNATURE", "x-cxc-signature header is missing")
    if not _TS.fullmatch(ts):
        raise WebhookVerificationError("INVALID_TIMESTAMP", "x-cxc-timestamp must be Unix time in milliseconds")

    expected = sign_webhook(secret, ts, body)
    if not hmac.compare_digest(expected.encode("ascii"), sig.encode("utf-8")):
        raise WebhookVerificationError(
            "INVALID_SIGNATURE", "Signature mismatch -- wrong secret or the body was modified (use the raw body)"
        )

    now_ms = (time.time() if now is None else now) * 1000
    if abs(now_ms - int(ts)) > tolerance * 1000:
        raise WebhookVerificationError(
            "STALE_TIMESTAMP",
            "Webhook timestamp is more than %s s away from now -- replay or clock skew (sync NTP)" % tolerance,
        )

    try:
        event = json.loads(body.decode("utf-8"))
    except ValueError:
        raise WebhookVerificationError("INVALID_BODY", "Webhook body is not valid JSON") from None
    if not isinstance(event, dict):
        raise WebhookVerificationError("INVALID_BODY", "Webhook body is not a JSON object")
    return event