maimai_py.api

  1import logging
  2import os
  3from importlib.util import find_spec
  4from logging import warning
  5from typing import Annotated, Any, Callable, Literal, Optional, Union
  6from urllib.parse import unquote, urlparse
  7
  8from pydantic import PydanticUndefinedAnnotation
  9
 10from maimai_py.maimai import MaimaiClient, MaimaiClientMultithreading, MaimaiPlates, MaimaiSongs, _UnsetSentinel
 11from maimai_py.models import *
 12from maimai_py.providers import *
 13from maimai_py.providers.hybrid import HybridProvider
 14
 15PlateAttrs = Literal["remained", "cleared", "played", "all"]
 16
 17
 18def xstr(s: Optional[str]) -> str:
 19    return "" if s is None else str(s).lower()
 20
 21
 22def istr(i: Optional[list]) -> str:
 23    return "" if i is None else "".join(i).lower()
 24
 25
 26def pagination(page_size, page, data):
 27    total_pages = (len(data) + page_size - 1) // page_size
 28    if page < 1 or page > total_pages:
 29        return []
 30
 31    start = (page - 1) * page_size
 32    end = page * page_size
 33    return data[start:end]
 34
 35
 36def get_filters(functions: dict[Any, Callable[..., bool]]):
 37    union = [flag for cond, flag in functions.items() if cond is not None]
 38    filter = lambda obj: all([flag(obj) for flag in union])
 39    return filter
 40
 41
 42if find_spec("fastapi"):
 43    from fastapi import APIRouter, Depends, FastAPI, Query, Request
 44    from fastapi.openapi.utils import get_openapi
 45    from fastapi.responses import JSONResponse
 46
 47    class MaimaiRoutes:
 48        _client: MaimaiClient
 49        _with_curves: bool
 50
 51        _lxns_token: Optional[str] = None
 52        _divingfish_token: Optional[str] = None
 53        _arcade_proxy: Optional[str] = None
 54
 55        def __init__(
 56            self,
 57            client: MaimaiClient,
 58            lxns_token: Optional[str] = None,
 59            divingfish_token: Optional[str] = None,
 60            arcade_proxy: Optional[str] = None,
 61            with_curves: bool = False,
 62        ):
 63            self._client = client
 64            self._lxns_token = lxns_token
 65            self._divingfish_token = divingfish_token
 66            self._arcade_proxy = arcade_proxy
 67            self._with_curves = with_curves
 68
 69        def _dep_lxns_player(self, credentials: Optional[str] = None, friend_code: Optional[int] = None, qq: Optional[int] = None):
 70            return PlayerIdentifier(credentials=credentials, qq=qq, friend_code=friend_code)
 71
 72        def _dep_divingfish_player(self, username: Optional[str] = None, credentials: Optional[str] = None, qq: Optional[int] = None):
 73            return PlayerIdentifier(qq=qq, credentials=credentials, username=username)
 74
 75        def _dep_arcade_player(self, credentials: str):
 76            return PlayerIdentifier(credentials=credentials)
 77
 78        def _dep_divingfish(self) -> IProvider:
 79            return DivingFishProvider(developer_token=self._divingfish_token)
 80
 81        def _dep_lxns(self) -> IProvider:
 82            return LXNSProvider(developer_token=self._lxns_token)
 83
 84        def _dep_arcade(self) -> IProvider:
 85            return ArcadeProvider(http_proxy=self._arcade_proxy)
 86
 87        def _dep_hybrid(self) -> IProvider:
 88            return HybridProvider()
 89
 90        def get_router(self, dep_provider: Callable, dep_player: Optional[Callable] = None, skip_base: bool = True) -> APIRouter:
 91            router = APIRouter()
 92
 93            def try_add_route(func: Callable, router: APIRouter, dep_provider: Callable):
 94                provider_type = func.__annotations__.get("provider")
 95                if provider_type and isinstance(dep_provider(), provider_type):
 96                    method = "GET" if "get_" in func.__name__ else "POST"
 97                    response_model = func.__annotations__.get("return")
 98                    router.add_api_route(
 99                        f"/{func.__name__.split('_')[-1]}",
100                        func,
101                        name=f"{func.__name__}",
102                        methods=[method],
103                        response_model=response_model,
104                        description=func.__doc__,
105                    )
106
107            async def _get_songs(
108                id: Optional[int] = None,
109                title: Optional[str] = None,
110                artist: Optional[str] = None,
111                genre: Optional[Genre] = None,
112                bpm: Optional[int] = None,
113                map: Optional[str] = None,
114                version: Optional[int] = None,
115                type: Optional[SongType] = None,
116                level: Optional[str] = None,
117                versions: Optional[Version] = None,
118                keywords: Optional[str] = None,
119                page: int = Query(1, ge=1),
120                page_size: int = Query(100, ge=1),
121                provider: ISongProvider = Depends(dep_provider),
122            ) -> list[Song]:
123                curve_provider = DivingFishProvider(developer_token=self._divingfish_token) if self._with_curves else None
124                maimai_songs: MaimaiSongs = await self._client.songs(provider=provider, curve_provider=curve_provider)
125                type_func: Callable[[Song], bool] = lambda song: song.get_difficulties(type) != []  # type: ignore
126                level_func: Callable[[Song], bool] = lambda song: any([diff.level == level for diff in song.get_difficulties()])
127                versions_func: Callable[[Song], bool] = lambda song: versions.value <= song.version < all_versions[all_versions.index(versions) + 1].value  # type: ignore
128                keywords_func: Callable[[Song], bool] = lambda song: xstr(keywords) in xstr(song.title) + xstr(song.artist) + istr(song.aliases)
129                songs = await maimai_songs.filter(id=id, title=title, artist=artist, genre=genre, bpm=bpm, map=map, version=version)
130                filters = get_filters({type: type_func, level: level_func, versions: versions_func, keywords: keywords_func})
131                result = [song for song in songs if filters(song)]
132                return pagination(page_size, page, result)
133
134            async def _get_icons(
135                id: Optional[int] = None,
136                name: Optional[str] = None,
137                description: Optional[str] = None,
138                genre: Optional[str] = None,
139                keywords: Optional[str] = None,
140                page: int = Query(1, ge=1),
141                page_size: int = Query(100, ge=1),
142                provider: IItemListProvider = Depends(dep_provider),
143            ) -> list[PlayerIcon]:
144                items = await self._client.items(PlayerIcon, provider=provider)
145                if id is not None:
146                    return [item] if (item := await items.by_id(id)) else []
147                keyword_func = lambda icon: xstr(keywords) in (xstr(icon.name) + xstr(icon.description) + xstr(icon.genre))
148                filters = get_filters({keywords: keyword_func})
149                result = [x for x in await items.filter(name=name, description=description, genre=genre) if filters(x)]
150                return pagination(page_size, page, result)
151
152            async def _get_nameplates(
153                id: Optional[int] = None,
154                name: Optional[str] = None,
155                description: Optional[str] = None,
156                genre: Optional[str] = None,
157                keywords: Optional[str] = None,
158                page: int = Query(1, ge=1),
159                page_size: int = Query(100, ge=1),
160                provider: IItemListProvider = Depends(dep_provider),
161            ) -> list[PlayerNamePlate]:
162                items = await self._client.items(PlayerNamePlate, provider=provider)
163                if id is not None:
164                    return [item] if (item := await items.by_id(id)) else []
165                keyword_func = lambda icon: xstr(keywords) in (xstr(icon.name) + xstr(icon.description) + xstr(icon.genre))
166                filters = get_filters({keywords: keyword_func})
167                result = [x for x in await items.filter(name=name, description=description, genre=genre) if filters(x)]
168                return pagination(page_size, page, result)
169
170            async def _get_frames(
171                id: Optional[int] = None,
172                name: Optional[str] = None,
173                description: Optional[str] = None,
174                genre: Optional[str] = None,
175                keywords: Optional[str] = None,
176                page: int = Query(1, ge=1),
177                page_size: int = Query(100, ge=1),
178                provider: IItemListProvider = Depends(dep_provider),
179            ) -> list[PlayerFrame]:
180                items = await self._client.items(PlayerFrame, provider=provider)
181                if id is not None:
182                    return [item] if (item := await items.by_id(id)) else []
183                keyword_func = lambda icon: xstr(keywords) in (xstr(icon.name) + xstr(icon.description) + xstr(icon.genre))
184                filters = get_filters({keywords: keyword_func})
185                result = [x for x in await items.filter(name=name, description=description, genre=genre) if filters(x)]
186                return pagination(page_size, page, result)
187
188            async def _get_trophies(
189                id: Optional[int] = None,
190                name: Optional[str] = None,
191                color: Optional[str] = None,
192                keywords: Optional[str] = None,
193                page: int = Query(1, ge=1),
194                page_size: int = Query(100, ge=1),
195                provider: IItemListProvider = Depends(dep_provider),
196            ) -> list[PlayerTrophy]:
197                items = await self._client.items(PlayerTrophy, provider=provider)
198                if id is not None:
199                    return [item] if (item := await items.by_id(id)) else []
200                keyword_func = lambda icon: xstr(keywords) in (xstr(icon.name) + xstr(icon.color))
201                filters = get_filters({keywords: keyword_func})
202                result = [x for x in await items.filter(name=name, color=color) if filters(x)]
203                return pagination(page_size, page, result)
204
205            async def _get_charas(
206                id: Optional[int] = None,
207                name: Optional[str] = None,
208                keywords: Optional[str] = None,
209                page: int = Query(1, ge=1),
210                page_size: int = Query(100, ge=1),
211                provider: IItemListProvider = Depends(dep_provider),
212            ) -> list[PlayerChara]:
213                items = await self._client.items(PlayerChara, provider=provider)
214                if id is not None:
215                    return [item] if (item := await items.by_id(id)) else []
216                keyword_func = lambda chara: xstr(keywords) in xstr(chara.name)
217                filters = get_filters({keywords: keyword_func})
218                result = [x for x in await items.filter(name=name) if filters(x)]
219                return pagination(page_size, page, result)
220
221            async def _get_partners(
222                id: Optional[int] = None,
223                name: Optional[str] = None,
224                keywords: Optional[str] = None,
225                page: int = Query(1, ge=1),
226                page_size: int = Query(100, ge=1),
227                provider: IItemListProvider = Depends(dep_provider),
228            ) -> list[PlayerPartner]:
229                items = await self._client.items(PlayerPartner, provider=provider)
230                if id is not None:
231                    return [item] if (item := await items.by_id(id)) else []
232                keyword_func = lambda partner: xstr(keywords) in xstr(partner.name)
233                filters = get_filters({keywords: keyword_func})
234                result = [x for x in await items.filter(name=name) if filters(x)]
235                return pagination(page_size, page, result)
236
237            async def _get_areas(
238                lang: Literal["ja", "zh"] = "ja",
239                id: Optional[str] = None,
240                name: Optional[str] = None,
241                keywords: Optional[str] = None,
242                page: int = Query(1, ge=1),
243                page_size: int = Query(100, ge=1),
244                provider: IAreaProvider = Depends(dep_provider),
245            ) -> list[Area]:
246                areas = await self._client.areas(lang, provider=provider)
247                if id is not None:
248                    return [area] if (area := await areas.by_id(id)) else []
249                if name is not None:
250                    return [area] if (area := await areas.by_name(name)) else []
251                keyword_func = lambda area: xstr(keywords) in (xstr(area.name) + xstr(area.comment))
252                filters = get_filters({keywords: keyword_func})
253                result = [x for x in await areas.get_all() if filters(x)]
254                return pagination(page_size, page, result)
255
256            async def _get_scores(
257                provider: IScoreProvider = Depends(dep_provider),
258                player: PlayerIdentifier = Depends(dep_player),
259            ) -> list[ScoreExtend]:
260                scores = await self._client.scores(player, provider=provider)
261                return scores.scores
262
263            async def _get_regions(
264                provider: IRegionProvider = Depends(dep_provider),
265                player: PlayerIdentifier = Depends(dep_player),
266            ) -> list[PlayerRegion]:
267                return await self._client.regions(player, provider=provider)
268
269            async def _get_players(
270                provider: IPlayerProvider = Depends(dep_provider),
271                player: PlayerIdentifier = Depends(dep_player),
272            ) -> Union[Player, DivingFishPlayer, LXNSPlayer, ArcadePlayer]:
273                return await self._client.players(player, provider=provider)
274
275            async def _get_bests(
276                provider: IScoreProvider = Depends(dep_provider),
277                player: PlayerIdentifier = Depends(dep_player),
278            ) -> PlayerBests:
279                maimai_scores = await self._client.bests(player, provider=provider)
280                return maimai_scores.get_player_bests()
281
282            async def _post_scores(
283                scores: list[Score],
284                provider: IScoreUpdateProvider = Depends(dep_provider),
285                player: PlayerIdentifier = Depends(dep_player),
286            ) -> None:
287                await self._client.updates(player, scores, provider=provider)
288
289            async def _get_plates(
290                plate: str,
291                attr: Literal["remained", "cleared", "played", "all"] = "remained",
292                provider: IScoreProvider = Depends(dep_provider),
293                player: PlayerIdentifier = Depends(dep_player),
294            ) -> list[PlateObject]:
295                plates: MaimaiPlates = await self._client.plates(player, plate, provider=provider)
296                return await getattr(plates, f"get_{attr}")()
297
298            async def _get_minfo(
299                id: Optional[int] = None,
300                title: Optional[str] = None,
301                keywords: Optional[str] = None,
302                provider: IScoreProvider = Depends(dep_provider),
303                player: PlayerIdentifier = Depends(dep_player),
304            ) -> Optional[PlayerSong]:
305                song_trait = id if id is not None else title if title is not None else keywords if keywords is not None else None
306                identifier = None if player._is_empty() else player
307                if song_trait is not None:
308                    return await self._client.minfo(song_trait, identifier, provider=provider)
309
310            async def _get_identifiers(
311                code: str,
312                provider: IPlayerIdentifierProvider = Depends(dep_provider),
313            ) -> PlayerIdentifier:
314                return await self._client.identifiers(code, provider=provider)
315
316            bases: list[Callable] = [_get_songs, _get_icons, _get_nameplates, _get_frames, _get_trophies, _get_charas, _get_partners, _get_areas]
317            players: list[Callable] = [_get_scores, _get_regions, _get_players, _get_bests, _post_scores, _get_plates, _get_minfo, _get_identifiers]
318
319            all = players + (bases if not skip_base else [])
320            try:
321                [try_add_route(func, router, dep_provider) for func in all]
322            except PydanticUndefinedAnnotation:
323                warning(
324                    "Current pydantic version does not support maimai.py API annotations"
325                    "MaimaiRoutes may not work properly."
326                    "Please upgrade pydantic to 2.7+."
327                )
328
329            return router
330
331
332if all([find_spec(p) for p in ["fastapi", "uvicorn", "typer"]]):
333    import typer
334    import uvicorn
335    from fastapi import APIRouter, Depends, FastAPI, Request
336    from fastapi.openapi.utils import get_openapi
337    from fastapi.responses import JSONResponse
338
339    # prepare for ASGI app
340    asgi_app = FastAPI(title="maimai.py API", description="The definitive python wrapper for MaimaiCN related development.")
341    routes = MaimaiRoutes(MaimaiClientMultithreading())  # type: ignore
342
343    # register routes and middlewares
344    asgi_app.include_router(routes.get_router(routes._dep_hybrid, skip_base=False), tags=["base"])
345    asgi_app.include_router(routes.get_router(routes._dep_divingfish, routes._dep_divingfish_player), prefix="/divingfish", tags=["divingfish"])
346    asgi_app.include_router(routes.get_router(routes._dep_lxns, routes._dep_lxns_player), prefix="/lxns", tags=["lxns"])
347    asgi_app.include_router(routes.get_router(routes._dep_arcade, routes._dep_arcade_player), prefix="/arcade", tags=["arcade"])
348
349    def main(
350        host: Annotated[str, typer.Option(help="The host address to bind to.")] = "127.0.0.1",
351        port: Annotated[int, typer.Option(help="The port number to bind to.")] = 8000,
352        redis: Annotated[Optional[str], typer.Option(help="Redis server address, for example: redis://localhost:6379/0.")] = None,
353        lxns_token: Annotated[Optional[str], typer.Option(help="LXNS developer token for LXNS API.")] = None,
354        divingfish_token: Annotated[Optional[str], typer.Option(help="DivingFish developer token for DivingFish API.")] = None,
355        arcade_proxy: Annotated[Optional[str], typer.Option(help="HTTP proxy for Arcade API.")] = None,
356        with_curves: Annotated[bool, typer.Option(help="Whether to fetch curves from Divingfish.")] = False,
357    ):
358        # prepare for redis cache backend
359        redis_backend = UNSET
360        if redis and find_spec("redis"):
361            from aiocache import RedisCache
362            from aiocache.serializers import PickleSerializer
363
364            redis_url = urlparse(redis)
365            redis_backend = RedisCache(
366                serializer=PickleSerializer(),
367                endpoint=unquote(redis_url.hostname or "localhost"),
368                port=redis_url.port or 6379,
369                password=redis_url.password,
370                db=int(unquote(redis_url.path).replace("/", "")),
371            )
372
373        # override the default maimai.py client
374        routes._client._cache = routes._client._cache if isinstance(redis_backend, _UnsetSentinel) else redis_backend
375        routes._lxns_token = lxns_token or os.environ.get("LXNS_DEVELOPER_TOKEN")
376        routes._divingfish_token = divingfish_token or os.environ.get("DIVINGFISH_DEVELOPER_TOKEN")
377        routes._arcade_proxy = arcade_proxy
378        routes._with_curves = with_curves
379
380        @asgi_app.exception_handler(MaimaiPyError)
381        async def exception_handler(request: Request, exc: MaimaiPyError):
382            return JSONResponse(
383                status_code=400,
384                content={"message": f"Oops! There goes a maimai.py error {exc}.", "details": repr(exc)},
385            )
386
387        @asgi_app.get("/", include_in_schema=False)
388        async def root():
389            return {"message": "Hello, maimai.py! Check /docs for more information."}
390
391        @asgi_app.on_event("startup")
392        async def startup_event():
393            if routes._with_curves:
394                curve_provider = DivingFishProvider(developer_token=routes._divingfish_token)
395                logging.info("with_curves is enabled, pre-fetching curves from DivingFish.")
396                await routes._client.songs(provider=HybridProvider(), curve_provider=curve_provider)
397
398        # run the ASGI app with uvicorn
399        uvicorn.run(asgi_app, host=host, port=port)
400
401    def openapi():
402        specs = get_openapi(
403            title=asgi_app.title,
404            version=asgi_app.version,
405            openapi_version=asgi_app.openapi_version,
406            description=asgi_app.description,
407            routes=asgi_app.routes,
408        )
409        with open(f"openapi.json", "w") as f:
410            json.dump(specs, f)
411
412    if __name__ == "__main__":
413        typer.run(main)
414
415
416if find_spec("maimai_ffi") and find_spec("nuitka"):
417    import json
418
419    import cryptography
420    import cryptography.fernet
421    import cryptography.hazmat.backends
422    import cryptography.hazmat.primitives.ciphers
423    import maimai_ffi
424    import maimai_ffi.model
425    import maimai_ffi.request
426    import redis