できました(AI)
# kai_points.py
#
# かい鯖グループポイント Python Binding
#
# Requirements:
# pip install requests
#
# Python 3.9+
from __future__ import annotations
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional
import requests
DEFAULT_BASE_URL = "https://points.bac0n.f5.si"
# ============================================================
# Exceptions
# ============================================================
class KaiPointsError(Exception):
"""かい鯖グループポイントAPIの基底例外"""
def __init__(
self,
message: str,
*,
status_code: Optional[int] = None,
response: Optional[requests.Response] = None,
):
super().__init__(message)
self.message = message
self.status_code = status_code
self.response = response
def __str__(self) -> str:
if self.status_code is not None:
return f"[HTTP {self.status_code}] {self.message}"
return self.message
class AuthenticationError(KaiPointsError):
"""401: APIキーが無効または未指定"""
class PaymentRequiredError(KaiPointsError):
"""402: 残高不足"""
class ForbiddenError(KaiPointsError):
"""403: 権限不足 / 取引上限超過など"""
class NotFoundError(KaiPointsError):
"""404: リソースが見つからない"""
class ConflictError(KaiPointsError):
"""409: 現在の状態では操作できない"""
class TooEarlyError(KaiPointsError):
"""425: 課金タイミングがまだ来ていない"""
class RateLimitError(KaiPointsError):
"""429: リクエスト過多"""
class ServerError(KaiPointsError):
"""5xx: サーバー側エラー"""
class InvalidResponseError(KaiPointsError):
"""レスポンスが想定したJSON形式でない"""
# ============================================================
# Data models
# ============================================================
@dataclass(frozen=True)
class Player:
username: str
minecraft_id: str
points: int
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "Player":
return cls(
username=str(data["username"]),
minecraft_id=str(data["minecraft_id"]),
points=int(data["points"]),
)
@dataclass(frozen=True)
class Product:
id: int
name: str
price: int
description: Optional[str] = None
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "Product":
return cls(
id=int(data["id"]),
name=str(data["name"]),
price=int(data["price"]),
description=data.get("description"),
)
@dataclass(frozen=True)
class TransactionInitiated:
tx_token: str
amount: int
item_name: str
expires_at: Optional[datetime]
web_url: str
message: str
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "TransactionInitiated":
return cls(
tx_token=str(data["tx_token"]),
amount=int(data["amount"]),
item_name=str(data["item_name"]),
expires_at=_parse_datetime(data.get("expires_at")),
web_url=str(data["web_url"]),
message=str(data.get("message", "")),
)
@dataclass(frozen=True)
class Transaction:
status: str
raw: Dict[str, Any]
@property
def pending_buyer(self) -> bool:
return self.status == "pending_buyer"
@property
def pending_seller(self) -> bool:
return self.status == "pending_seller"
@property
def completed(self) -> bool:
return self.status == "completed"
@property
def rejected(self) -> bool:
return self.status == "rejected"
@property
def expired(self) -> bool:
return self.status == "expired"
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "Transaction":
return cls(
status=str(data["status"]),
raw=dict(data),
)
@dataclass(frozen=True)
class SubscriptionInitiated:
subscription_id: int
username: str
product_name: str
amount: int
interval_days: int
status: str
web_url: str
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "SubscriptionInitiated":
return cls(
subscription_id=int(data["subscription_id"]),
username=str(data["username"]),
product_name=str(data["product_name"]),
amount=int(data["amount"]),
interval_days=int(data["interval_days"]),
status=str(data["status"]),
web_url=str(data["web_url"]),
)
@dataclass(frozen=True)
class Subscription:
id: int
status: str
next_charge_at: Optional[datetime]
raw: Dict[str, Any]
@property
def active(self) -> bool:
return self.status == "active"
@property
def pending(self) -> bool:
return self.status == "pending"
@property
def suspended(self) -> bool:
return self.status == "suspended"
@property
def cancelled(self) -> bool:
return self.status == "cancelled"
@property
def charge_due(self) -> bool:
if self.status != "active":
return False
if self.next_charge_at is None:
return False
return datetime.now(timezone.utc) >= self.next_charge_at
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "Subscription":
return cls(
id=int(data["id"]),
status=str(data["status"]),
next_charge_at=_parse_datetime(data.get("next_charge_at")),
raw=dict(data),
)
@dataclass(frozen=True)
class SubscriptionCharge:
status: str
amount: int
next_charge_at: Optional[datetime]
raw: Dict[str, Any]
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "SubscriptionCharge":
return cls(
status=str(data["status"]),
amount=int(data["amount"]),
next_charge_at=_parse_datetime(data.get("next_charge_at")),
raw=dict(data),
)
# ============================================================
# Utilities
# ============================================================
def _parse_datetime(value: Any) -> Optional[datetime]:
if value is None:
return None
if isinstance(value, datetime):
dt = value
else:
text = str(value)
if text.endswith("Z"):
text = text[:-1] + "+00:00"
try:
dt = datetime.fromisoformat(text)
except ValueError as exc:
raise InvalidResponseError(
f"不正な日時形式です: {value!r}"
) from exc
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt
# ============================================================
# Client
# ============================================================
class KaiPointsClient:
def __init__(
self,
api_key: str,
*,
base_url: str = DEFAULT_BASE_URL,
timeout: float = 10.0,
session: Optional[requests.Session] = None,
):
if not api_key:
raise ValueError("api_key は必須です")
self.api_key = api_key
self.base_url = base_url.rstrip("/")
self.timeout = timeout
self.session = session or requests.Session()
self.session.headers.update({
"X-API-Key": self.api_key,
"Accept": "application/json",
"User-Agent": "kai-points-python/1.0",
})
# --------------------------------------------------------
# Internal HTTP methods
# --------------------------------------------------------
def _request(
self,
method: str,
path: str,
*,
json: Optional[Dict[str, Any]] = None,
params: Optional[Dict[str, Any]] = None,
authenticated: bool = True,
) -> Any:
url = f"{self.base_url}{path}"
headers: Dict[str, str] = {}
if json is not None:
headers["Content-Type"] = "application/json"
if not authenticated:
headers["X-API-Key"] = ""
try:
response = self.session.request(
method=method,
url=url,
json=json,
params=params,
headers=headers,
timeout=self.timeout,
)
except requests.Timeout as exc:
raise KaiPointsError(
f"APIリクエストがタイムアウトしました: {url}"
) from exc
except requests.RequestException as exc:
raise KaiPointsError(
f"APIへの接続に失敗しました: {exc}"
) from exc
try:
payload = response.json()
except ValueError:
payload = None
if not response.ok:
message = self._extract_error_message(
response=response,
payload=payload,
)
self._raise_http_error(
response=response,
message=message,
)
if not isinstance(payload, dict):
raise InvalidResponseError(
"APIからJSONオブジェクト以外のレスポンスが返されました",
status_code=response.status_code,
response=response,
)
if payload.get("success") is False:
raise KaiPointsError(
str(payload.get("error", "APIエラーが発生しました")),
status_code=response.status_code,
response=response,
)
if "data" not in payload:
raise InvalidResponseError(
"APIレスポンスに data がありません",
status_code=response.status_code,
response=response,
)
return payload["data"]
@staticmethod
def _extract_error_message(
*,
response: requests.Response,
payload: Any,
) -> str:
if isinstance(payload, dict):
error = payload.get("error")
if error:
return str(error)
text = response.text.strip()
if text:
return text
return f"HTTP {response.status_code}"
@staticmethod
def _raise_http_error(
*,
response: requests.Response,
message: str,
) -> None:
status = response.status_code
exception_map = {
401: AuthenticationError,
402: PaymentRequiredError,
403: ForbiddenError,
404: NotFoundError,
409: ConflictError,
425: TooEarlyError,
429: RateLimitError,
}
exc_class = exception_map.get(status)
if exc_class is None:
if status >= 500:
exc_class = ServerError
else:
exc_class = KaiPointsError
raise exc_class(
message,
status_code=status,
response=response,
)
# --------------------------------------------------------
# Account / Player
# --------------------------------------------------------
def get_balance(self) -> Any:
"""
サービスアカウント側の残高情報を取得します。
GET /api/server/balance
詳細なレスポンス形式が固定されていないため、
data をそのまま返します。
"""
return self._request(
"GET",
"/api/server/balance",
)
def get_player(self, minecraft_id: str) -> Player:
"""
Minecraft IDからユーザー情報とポイント残高を取得。
GET /api/server/player/:minecraft_id
"""
if not minecraft_id:
raise ValueError("minecraft_id は必須です")
data = self._request(
"GET",
f"/api/server/player/{requests.utils.quote(minecraft_id, safe='')}",
)
return Player.from_dict(data)
def check_points(self, username: str) -> Any:
"""
フォーラムユーザー名からポイントを確認。
GET /api/points/check?username=...
認証不要。
このエンドポイントのdata形式が仕様上明示されていないため
data をそのまま返します。
"""
if not username:
raise ValueError("username は必須です")
return self._request(
"GET",
"/api/points/check",
params={"username": username},
authenticated=False,
)
# --------------------------------------------------------
# Products
# --------------------------------------------------------
def get_products(self) -> List[Product]:
"""
GET /api/server/products
"""
data = self._request(
"GET",
"/api/server/products",
)
if not isinstance(data, list):
raise InvalidResponseError(
"商品一覧が配列ではありません"
)
return [
Product.from_dict(product)
for product in data
]
# --------------------------------------------------------
# Transactions
# --------------------------------------------------------
def initiate_transaction(
self,
mc_id: str,
product_id: int,
) -> TransactionInitiated:
"""
ポイント取引を開始。
POST /api/server/tx/initiate
"""
if not mc_id:
raise ValueError("mc_id は必須です")
if product_id <= 0:
raise ValueError("product_id は1以上である必要があります")
data = self._request(
"POST",
"/api/server/tx/initiate",
json={
"mc_id": mc_id,
"product_id": product_id,
},
)
return TransactionInitiated.from_dict(data)
def get_transaction(
self,
tx_token: str,
) -> Transaction:
"""
取引ステータスを取得。
GET /api/server/tx/:tx_token
"""
if not tx_token:
raise ValueError("tx_token は必須です")
data = self._request(
"GET",
f"/api/server/tx/{requests.utils.quote(tx_token, safe='')}",
)
return Transaction.from_dict(data)
def approve_transaction(
self,
tx_token: str,
) -> Any:
"""
アイテム付与後、ポイント消費を確定。
POST /api/server/tx/:tx_token/approve
注意:
実際にアイテムや権限を付与した後に呼んでください。
"""
if not tx_token:
raise ValueError("tx_token は必須です")
return self._request(
"POST",
f"/api/server/tx/{requests.utils.quote(tx_token, safe='')}/approve",
)
def wait_for_transaction(
self,
tx_token: str,
*,
interval: float = 5.0,
timeout: Optional[float] = 300.0,
) -> Transaction:
"""
pending_buyer から状態が変わるまでポーリングします。
通常:
pending_buyer
↓ ユーザー承認
pending_seller
Trusted Mode + Auto Approveの場合:
最初から pending_seller の場合があります。
pending_seller / completed / rejected / expired
のいずれかになったら返ります。
"""
if interval <= 0:
raise ValueError("interval は0より大きい必要があります")
started_at = time.monotonic()
while True:
transaction = self.get_transaction(tx_token)
if transaction.status in {
"pending_seller",
"completed",
"rejected",
"expired",
}:
return transaction
if timeout is not None:
elapsed = time.monotonic() - started_at
if elapsed >= timeout:
raise TimeoutError(
f"取引承認待ちがタイムアウトしました: {tx_token}"
)
time.sleep(interval)
# --------------------------------------------------------
# Subscriptions
# --------------------------------------------------------
def initiate_subscription(
self,
username: str,
product_id: int,
interval_days: int,
) -> SubscriptionInitiated:
"""
サブスクリプション登録開始。
POST /api/server/subscription/initiate
"""
if not username:
raise ValueError("username は必須です")
if product_id <= 0:
raise ValueError("product_id は1以上である必要があります")
if interval_days <= 0:
raise ValueError("interval_days は1以上である必要があります")
data = self._request(
"POST",
"/api/server/subscription/initiate",
json={
"username": username,
"product_id": product_id,
"interval_days": interval_days,
},
)
return SubscriptionInitiated.from_dict(data)
def get_subscription(
self,
subscription_id: int,
) -> Subscription:
"""
GET /api/server/subscription/:id
"""
if subscription_id <= 0:
raise ValueError("subscription_id は1以上である必要があります")
data = self._request(
"GET",
f"/api/server/subscription/{subscription_id}",
)
return Subscription.from_dict(data)
def get_subscriptions(self) -> List[Subscription]:
"""
GET /api/server/subscription
"""
data = self._request(
"GET",
"/api/server/subscription",
)
if not isinstance(data, list):
raise InvalidResponseError(
"サブスクリプション一覧が配列ではありません"
)
return [
Subscription.from_dict(subscription)
for subscription in data
]
def charge_subscription(
self,
subscription_id: int,
) -> SubscriptionCharge:
"""
定期課金を実行。
POST /api/server/subscription/:id/charge
Raises:
TooEarlyError:
next_charge_atがまだ未来
PaymentRequiredError:
残高不足。サブスクはsuspendedへ移行済み
ConflictError:
active以外の状態
"""
if subscription_id <= 0:
raise ValueError("subscription_id は1以上である必要があります")
data = self._request(
"POST",
f"/api/server/subscription/{subscription_id}/charge",
)
return SubscriptionCharge.from_dict(data)
def charge_subscription_if_due(
self,
subscription_id: int,
) -> Optional[SubscriptionCharge]:
"""
next_charge_atを確認し、
課金期限が到来している場合のみ /charge を呼びます。
課金不要の場合は None。
"""
subscription = self.get_subscription(subscription_id)
if not subscription.active:
return None
if not subscription.charge_due:
return None
return self.charge_subscription(subscription_id)
def charge_due_subscriptions(
self,
) -> Dict[int, Any]:
"""
サービスの全サブスクを確認し、
課金期限が到来しているactiveサブスクを課金します。
サービス起動時や定期ジョブから呼ぶ用途を想定。
戻り値:
{
subscription_id: SubscriptionCharge | KaiPointsError | None
}
None:
課金対象外
KaiPointsError:
課金中にエラー
"""
results: Dict[int, Any] = {}
subscriptions = self.get_subscriptions()
for subscription in subscriptions:
if not subscription.active:
results[subscription.id] = None
continue
if not subscription.charge_due:
results[subscription.id] = None
continue
try:
results[subscription.id] = self.charge_subscription(
subscription.id
)
except KaiPointsError as exc:
results[subscription.id] = exc
return results
def cancel_subscription(
self,
subscription_id: int,
) -> Any:
"""
サービス側からサブスクリプションをキャンセル。
DELETE /api/server/subscription/:id
"""
if subscription_id <= 0:
raise ValueError("subscription_id は1以上である必要があります")
return self._request(
"DELETE",
f"/api/server/subscription/{subscription_id}",
)
def wait_for_subscription(
self,
subscription_id: int,
*,
interval: float = 5.0,
timeout: Optional[float] = 300.0,
) -> Subscription:
"""
サブスクリプションの初回承認を待ちます。
active / suspended / cancelled
になった時点で返ります。
"""
if interval <= 0:
raise ValueError("interval は0より大きい必要があります")
started_at = time.monotonic()
while True:
subscription = self.get_subscription(subscription_id)
if subscription.status in {
"active",
"suspended",
"cancelled",
}:
return subscription
if timeout is not None:
elapsed = time.monotonic() - started_at
if elapsed >= timeout:
raise TimeoutError(
"サブスクリプション承認待ちが"
f"タイムアウトしました: {subscription_id}"
)
time.sleep(interval)
# --------------------------------------------------------
# Context Manager
# --------------------------------------------------------
def close(self) -> None:
self.session.close()
def __enter__(self) -> "KaiPointsClient":
return self
def __exit__(self, exc_type, exc_val, exc_tb) -> None:
self.close()
# ============================================================
# Basic usage
# ============================================================
#
# from kai_points import (
# KaiPointsClient,
# PaymentRequiredError,
# TooEarlyError,
# )
#
# client = KaiPointsClient("skp_...")
#
#
# # プレイヤー残高
# player = client.get_player("Steve")
# print(player.points)
#
#
# # 商品一覧
# for product in client.get_products():
# print(product.id, product.name, product.price)
#
#
# # 通常購入
# tx = client.initiate_transaction(
# mc_id="Steve",
# product_id=1,
# )
#
# print(tx.web_url)
#
# status = client.wait_for_transaction(tx.tx_token)
#
# if status.pending_seller:
# # ここでMinecraftサーバーなどにアイテムを付与する
# #
# # give_item(...)
# #
# # 付与が成功した場合のみapprove
# client.approve_transaction(tx.tx_token)
#
#
# # サブスク登録
# sub = client.initiate_subscription(
# username="User123",
# product_id=1,
# interval_days=30,
# )
#
# print(sub.web_url)
#
# active_sub = client.wait_for_subscription(
# sub.subscription_id
# )
#
# if active_sub.active:
# print("サブスク有効")
# print(active_sub.next_charge_at)
#
#
# # 定期実行
# try:
# charge = client.charge_subscription_if_due(
# sub.subscription_id
# )
#
# if charge is not None:
# print("課金成功")
# print(charge.amount)
# print(charge.next_charge_at)
#
# except PaymentRequiredError:
# print("ポイント残高不足")
#
# except TooEarlyError:
# # 複数プロセス等が同時に判定した場合など
# # GET時点では期限到来していても、
# # 別プロセスが先に課金した可能性がある
# print("まだ課金時期ではありません")
#
#
# # サービス起動時などに全サブスクをチェック
# results = client.charge_due_subscriptions()
#
# for subscription_id, result in results.items():
# if isinstance(result, SubscriptionCharge):
# print(
# subscription_id,
# "課金成功",
# result.next_charge_at,
# )
# elif isinstance(result, KaiPointsError):
# print(
# subscription_id,
# "課金失敗",
# result,
# )
というかサービスアカウントの仮想作成作るか
サービスアカウントを論理的に分離する
ポイントとられたのに商品もらえなかったり商品もらえたのにポイントとられなかったりしないようにテストはしておいた方がいいよ
まあねー
というか公式がPythonバインディング作りゃいい話でしょ
kai_points_mock.py.txt (30.2 KB)
というわけでモックサーバーを用意しました！