feat: complete LunaBot deck integration API

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "allium-deck-python"
-version = "0.0.1"
+version = "0.0.2"
 edition = "2021"
 license = "MIT"
 publish = false
@@ -10,7 +10,7 @@ name = "_native"
 crate-type = ["cdylib"]
 
 [dependencies]
-allium-deck = { git = "https://github.com/empty-sekai/allium-deck.git", rev = "2cf7e77736c1d545f0858ec96711feae9e6fcbed" }
+allium-deck = { version = "0.0.5", git = "https://github.com/empty-sekai/allium-deck.git", rev = "b799502a44a71488e82463ccfa607b2143600b23" }
 pyo3 = { version = "0.23", features = ["abi3-py310"] }
 serde_json = "1"
 

README.md

@@ -41,6 +41,23 @@ from sekai_deck_recommend_cpp import (
 Masterdata, music metadata, and user data remain runtime inputs and are not
 bundled into the wheel. The recommendation engine uses Allium's DFS search.
 
+## API coverage
+
+The `sekai_deck_recommend_cpp` interface includes the complete LunaBot deck
+workflow:
+
+- mutable option, user-data, card, deck, support-deck, and result objects
+- single and batch recommendation
+- World Bloom support-deck calculation
+- area-item upgrade recommendation
+- per-music score and event-point calculation
+- note-level exact live calculation
+- configurable batch worker count
+
+Each recommendation result includes `cost_ms`, the wall-clock time spent in the
+native search itself. Batch results report this value independently for every
+request.
+
 ## License
 
 MIT

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "maturin"
 
 [project]
 name = "allium-sekai-deck"
-version = "0.0.1"
+version = "0.0.2"
 description = "Precompiled Rust deck recommendation engine for Python and LunaBot integrations"
 requires-python = ">=3.10,<4"
 readme = "README.md"

python/sekai_deck_recommend_cpp/__init__.py

@@ -1,6 +1,9 @@
 from __future__ import annotations
 
 import json
+import os
+from collections.abc import Sequence, Set
+from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 
 from ._models import (
@@ -16,6 +19,51 @@
     RecommendSupportDeckCard,
 )
 
+_VALID_REGIONS = {"jp", "en", "tw", "kr", "cn"}
+_VALID_LIVE_TYPES = {
+    "multi",
+    "solo",
+    "auto",
+    "challenge",
+    "challenge_auto",
+    "mysekai",
+}
+_engine_thread_count = 0
+
+
+def _native_sequence(value, name: str):
+    if isinstance(value, (str, bytes)) or not isinstance(value, (Sequence, Set)):
+        raise TypeError(f"{name} must be a sequence")
+    return list(value)
+
+
+def _raise_batch_error(index: int, error: Exception):
+    if isinstance(error, RuntimeError):
+        raise RuntimeError(f"batch item {index} failed: {error}") from error
+    raise ValueError(f"batch item {index}: {error}") from error
+
+
+def set_engine_thread_count(threads: int) -> None:
+    global _engine_thread_count
+    if not isinstance(threads, int):
+        raise TypeError("threads must be an int")
+    if threads < 0:
+        raise ValueError("threads must be non-negative")
+    _engine_thread_count = min(threads, os.cpu_count() or 1)
+
+
+def _effective_engine_thread_count() -> int:
+    if _engine_thread_count > 0:
+        return _engine_thread_count
+    raw = os.environ.get("DECK_ENGINE_THREADS", "")
+    try:
+        configured = int(raw)
+    except ValueError:
+        configured = 1
+    if configured <= 0:
+        return 1
+    return min(configured, os.cpu_count() or 1)
+
 
 class SekaiDeckRecommend:
     def __init__(self) -> None:
@@ -31,52 +79,159 @@ def _require_native(self):
             raise RuntimeError("_allium_deck_native is not installed")
         return self._native
 
+    @staticmethod
+    def _validate_region(region: str) -> str:
+        if region not in _VALID_REGIONS:
+            raise ValueError(f"Invalid region: {region}")
+        return region
+
+    @staticmethod
+    def _resolve_user_data(options: DeckRecommendOptions) -> DeckRecommendUserData:
+        user_data = options.user_data
+        if user_data is None and options.user_data_file_path:
+            user_data = DeckRecommendUserData()
+            user_data.load_from_file(options.user_data_file_path)
+        if user_data is None and options.user_data_str is not None:
+            user_data = DeckRecommendUserData()
+            user_data.load_from_bytes(options.user_data_str)
+        if not isinstance(user_data, DeckRecommendUserData) or user_data._native is None:
+            raise ValueError(
+                "Either user_data / user_data_file_path / user_data_str is required."
+            )
+        return user_data
+
+    @classmethod
+    def _validate_options(cls, options: DeckRecommendOptions) -> str:
+        if not isinstance(options, DeckRecommendOptions):
+            raise TypeError("options must be DeckRecommendOptions")
+        if options.region is None:
+            raise ValueError("region is required.")
+        return cls._validate_region(options.region)
+
     def update_masterdata(self, base_dir: str, region: str) -> None:
-        self._require_native().update_masterdata(str(Path(base_dir)), region)
+        self._require_native().update_masterdata(
+            str(Path(base_dir)), self._validate_region(region)
+        )
 
     def update_masterdata_from_strings(self, data, region: str) -> None:
+        if not isinstance(data, dict):
+            raise TypeError("data must be a dict")
         normalized = {
             (name if str(name).endswith(".json") else f"{name}.json"): (
                 value.decode("utf-8") if isinstance(value, bytes) else str(value)
             )
             for name, value in data.items()
         }
-        self._require_native().update_masterdata_from_strings(normalized, region)
+        self._require_native().update_masterdata_from_strings(
+            normalized, self._validate_region(region)
+        )
 
     def update_musicmetas(self, path: str, region: str) -> None:
-        self._require_native().update_musicmetas(Path(path).read_text("utf-8"), region)
+        self._require_native().update_musicmetas(
+            Path(path).read_text("utf-8"), self._validate_region(region)
+        )
 
     def update_musicmetas_from_string(self, data, region: str) -> None:
         text = data.decode("utf-8") if isinstance(data, bytes) else str(data)
-        self._require_native().update_musicmetas(text, region)
+        self._require_native().update_musicmetas(text, self._validate_region(region))
 
     def recommend(self, options: DeckRecommendOptions) -> DeckRecommendResult:
-        if not isinstance(options, DeckRecommendOptions):
-            raise TypeError("options must be DeckRecommendOptions")
-        user_data = options.user_data
-        if user_data is None and options.user_data_file_path:
-            user_data = DeckRecommendUserData()
-            user_data.load_from_file(options.user_data_file_path)
-        if user_data is None and options.user_data_str is not None:
-            user_data = DeckRecommendUserData()
-            user_data.load_from_bytes(options.user_data_str)
-        if not isinstance(user_data, DeckRecommendUserData) or user_data._native is None:
-            raise ValueError("options.user_data is required")
+        region = self._validate_options(options)
+        user_data = self._resolve_user_data(options)
         payload = self._require_native().recommend(
-            options.region,
-            json.dumps(options.to_native_dict(), separators=(",", ":")),
+            region,
+            json.dumps(options._to_native_dict(), separators=(",", ":")),
             user_data._native,
         )
         return DeckRecommendResult.from_dict(json.loads(payload))
 
-    def recommend_area_items(self, *args, **kwargs):
-        raise NotImplementedError("recommend_area_items is not available in 0.0.1")
+    def recommend_batch(
+        self, options_list: list[DeckRecommendOptions]
+    ) -> list[DeckRecommendResult]:
+        options_list = _native_sequence(options_list, "options_list")
+        for index, options in enumerate(options_list):
+            if not isinstance(options, DeckRecommendOptions):
+                raise TypeError(f"batch item {index}: expected DeckRecommendOptions")
+        workers = min(_effective_engine_thread_count(), max(1, len(options_list)))
+        if workers <= 1 or len(options_list) <= 1:
+            result = []
+            for index, options in enumerate(options_list):
+                try:
+                    result.append(self.recommend(options))
+                except Exception as error:
+                    _raise_batch_error(index, error)
+            return result
+        with ThreadPoolExecutor(max_workers=workers) as executor:
+            futures = [executor.submit(self.recommend, options) for options in options_list]
+            result = []
+            for index, future in enumerate(futures):
+                try:
+                    result.append(future.result())
+                except Exception as error:
+                    _raise_batch_error(index, error)
+            return result
+
+    def get_world_bloom_support_cards(
+        self, options: DeckRecommendOptions
+    ) -> list[RecommendSupportDeckCard]:
+        region = self._validate_options(options)
+        user_data = self._resolve_user_data(options)
+        payload = self._require_native().get_world_bloom_support_cards(
+            region,
+            json.dumps(options._to_native_dict(), separators=(",", ":")),
+            user_data._native,
+        )
+        return [RecommendSupportDeckCard.from_dict(item) for item in json.loads(payload)]
+
+    def recommend_area_items(
+        self, options: DeckRecommendOptions, card_ids: list[int]
+    ) -> list[dict]:
+        region = self._validate_options(options)
+        user_data = self._resolve_user_data(options)
+        card_ids = _native_sequence(card_ids, "card_ids")
+        if not 1 <= len(card_ids) <= 5:
+            raise ValueError("cardIds must contain 1 to 5 cards")
+        payload = self._require_native().recommend_area_items(
+            region, card_ids, user_data._native
+        )
+        return json.loads(payload)
 
-    def recommend_music(self, *args, **kwargs):
-        raise NotImplementedError("recommend_music is not available in 0.0.1")
+    def recommend_music(
+        self, options: DeckRecommendOptions, deck: RecommendDeck
+    ) -> list[dict]:
+        region = self._validate_options(options)
+        if not isinstance(deck, RecommendDeck):
+            raise TypeError("deck must be RecommendDeck")
+        payload = self._require_native().recommend_music(
+            region,
+            json.dumps(options._to_native_dict(), separators=(",", ":")),
+            json.dumps(deck.to_dict(), separators=(",", ":")),
+        )
+        return json.loads(payload)
 
-    def calculate_exact_live(self, *args, **kwargs):
-        raise NotImplementedError("calculate_exact_live is not available in 0.0.1")
+    def calculate_exact_live(
+        self,
+        region: str,
+        power: int,
+        skills: list[float],
+        live_type: str,
+        music_score_json: str,
+        multi_sum_power: int = 0,
+        fever_music_score_json: str | None = None,
+    ) -> dict:
+        skills = _native_sequence(skills, "skills")
+        if live_type not in _VALID_LIVE_TYPES or live_type == "mysekai":
+            raise ValueError(f"Invalid live type: {live_type}")
+        payload = self._require_native().calculate_exact_live(
+            self._validate_region(region),
+            power,
+            skills,
+            live_type,
+            music_score_json,
+            multi_sum_power,
+            fever_music_score_json,
+        )
+        return json.loads(payload)
 
 
 __all__ = [
@@ -91,4 +246,5 @@ def calculate_exact_live(self, *args, **kwargs):
     "RecommendDeck",
     "RecommendSupportDeckCard",
     "SekaiDeckRecommend",
+    "set_engine_thread_count",
 ]