DF.

Dockerfile

@@ -0,0 +1,18 @@
+FROM python:bullseye
+
+EXPOSE 7860
+
+ENV PORT=7860 \
+    PROCS=1 \
+    GENERAL_SERVER_HOST=0.0.0.0
+
+COPY . /hibi
+
+WORKDIR /hibi
+
+RUN pip install .
+
+CMD hibiapi run --port $PORT --workers $PROCS
+
+HEALTHCHECK --interval=30s --timeout=30s --start-period=5s --retries=3 \
+    CMD httpx --verbose --follow-redirects http://127.0.0.1:${PORT}

LICENSE

@@ -0,0 +1,201 @@
+                                 Apache License
+                           Version 2.0, January 2004
+                        http://www.apache.org/licenses/
+
+   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+   1. Definitions.
+
+      "License" shall mean the terms and conditions for use, reproduction,
+      and distribution as defined by Sections 1 through 9 of this document.
+
+      "Licensor" shall mean the copyright owner or entity authorized by
+      the copyright owner that is granting the License.
+
+      "Legal Entity" shall mean the union of the acting entity and all
+      other entities that control, are controlled by, or are under common
+      control with that entity. For the purposes of this definition,
+      "control" means (i) the power, direct or indirect, to cause the
+      direction or management of such entity, whether by contract or
+      otherwise, or (ii) ownership of fifty percent (50%) or more of the
+      outstanding shares, or (iii) beneficial ownership of such entity.
+
+      "You" (or "Your") shall mean an individual or Legal Entity
+      exercising permissions granted by this License.
+
+      "Source" form shall mean the preferred form for making modifications,
+      including but not limited to software source code, documentation
+      source, and configuration files.
+
+      "Object" form shall mean any form resulting from mechanical
+      transformation or translation of a Source form, including but
+      not limited to compiled object code, generated documentation,
+      and conversions to other media types.
+
+      "Work" shall mean the work of authorship, whether in Source or
+      Object form, made available under the License, as indicated by a
+      copyright notice that is included in or attached to the work
+      (an example is provided in the Appendix below).
+
+      "Derivative Works" shall mean any work, whether in Source or Object
+      form, that is based on (or derived from) the Work and for which the
+      editorial revisions, annotations, elaborations, or other modifications
+      represent, as a whole, an original work of authorship. For the purposes
+      of this License, Derivative Works shall not include works that remain
+      separable from, or merely link (or bind by name) to the interfaces of,
+      the Work and Derivative Works thereof.
+
+      "Contribution" shall mean any work of authorship, including
+      the original version of the Work and any modifications or additions
+      to that Work or Derivative Works thereof, that is intentionally
+      submitted to Licensor for inclusion in the Work by the copyright owner
+      or by an individual or Legal Entity authorized to submit on behalf of
+      the copyright owner. For the purposes of this definition, "submitted"
+      means any form of electronic, verbal, or written communication sent
+      to the Licensor or its representatives, including but not limited to
+      communication on electronic mailing lists, source code control systems,
+      and issue tracking systems that are managed by, or on behalf of, the
+      Licensor for the purpose of discussing and improving the Work, but
+      excluding communication that is conspicuously marked or otherwise
+      designated in writing by the copyright owner as "Not a Contribution."
+
+      "Contributor" shall mean Licensor and any individual or Legal Entity
+      on behalf of whom a Contribution has been received by Licensor and
+      subsequently incorporated within the Work.
+
+   2. Grant of Copyright License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      copyright license to reproduce, prepare Derivative Works of,
+      publicly display, publicly perform, sublicense, and distribute the
+      Work and such Derivative Works in Source or Object form.
+
+   3. Grant of Patent License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      (except as stated in this section) patent license to make, have made,
+      use, offer to sell, sell, import, and otherwise transfer the Work,
+      where such license applies only to those patent claims licensable
+      by such Contributor that are necessarily infringed by their
+      Contribution(s) alone or by combination of their Contribution(s)
+      with the Work to which such Contribution(s) was submitted. If You
+      institute patent litigation against any entity (including a
+      cross-claim or counterclaim in a lawsuit) alleging that the Work
+      or a Contribution incorporated within the Work constitutes direct
+      or contributory patent infringement, then any patent licenses
+      granted to You under this License for that Work shall terminate
+      as of the date such litigation is filed.
+
+   4. Redistribution. You may reproduce and distribute copies of the
+      Work or Derivative Works thereof in any medium, with or without
+      modifications, and in Source or Object form, provided that You
+      meet the following conditions:
+
+      (a) You must give any other recipients of the Work or
+          Derivative Works a copy of this License; and
+
+      (b) You must cause any modified files to carry prominent notices
+          stating that You changed the files; and
+
+      (c) You must retain, in the Source form of any Derivative Works
+          that You distribute, all copyright, patent, trademark, and
+          attribution notices from the Source form of the Work,
+          excluding those notices that do not pertain to any part of
+          the Derivative Works; and
+
+      (d) If the Work includes a "NOTICE" text file as part of its
+          distribution, then any Derivative Works that You distribute must
+          include a readable copy of the attribution notices contained
+          within such NOTICE file, excluding those notices that do not
+          pertain to any part of the Derivative Works, in at least one
+          of the following places: within a NOTICE text file distributed
+          as part of the Derivative Works; within the Source form or
+          documentation, if provided along with the Derivative Works; or,
+          within a display generated by the Derivative Works, if and
+          wherever such third-party notices normally appear. The contents
+          of the NOTICE file are for informational purposes only and
+          do not modify the License. You may add Your own attribution
+          notices within Derivative Works that You distribute, alongside
+          or as an addendum to the NOTICE text from the Work, provided
+          that such additional attribution notices cannot be construed
+          as modifying the License.
+
+      You may add Your own copyright statement to Your modifications and
+      may provide additional or different license terms and conditions
+      for use, reproduction, or distribution of Your modifications, or
+      for any such Derivative Works as a whole, provided Your use,
+      reproduction, and distribution of the Work otherwise complies with
+      the conditions stated in this License.
+
+   5. Submission of Contributions. Unless You explicitly state otherwise,
+      any Contribution intentionally submitted for inclusion in the Work
+      by You to the Licensor shall be under the terms and conditions of
+      this License, without any additional terms or conditions.
+      Notwithstanding the above, nothing herein shall supersede or modify
+      the terms of any separate license agreement you may have executed
+      with Licensor regarding such Contributions.
+
+   6. Trademarks. This License does not grant permission to use the trade
+      names, trademarks, service marks, or product names of the Licensor,
+      except as required for reasonable and customary use in describing the
+      origin of the Work and reproducing the content of the NOTICE file.
+
+   7. Disclaimer of Warranty. Unless required by applicable law or
+      agreed to in writing, Licensor provides the Work (and each
+      Contributor provides its Contributions) on an "AS IS" BASIS,
+      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+      implied, including, without limitation, any warranties or conditions
+      of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+      PARTICULAR PURPOSE. You are solely responsible for determining the
+      appropriateness of using or redistributing the Work and assume any
+      risks associated with Your exercise of permissions under this License.
+
+   8. Limitation of Liability. In no event and under no legal theory,
+      whether in tort (including negligence), contract, or otherwise,
+      unless required by applicable law (such as deliberate and grossly
+      negligent acts) or agreed to in writing, shall any Contributor be
+      liable to You for damages, including any direct, indirect, special,
+      incidental, or consequential damages of any character arising as a
+      result of this License or out of the use or inability to use the
+      Work (including but not limited to damages for loss of goodwill,
+      work stoppage, computer failure or malfunction, or any and all
+      other commercial damages or losses), even if such Contributor
+      has been advised of the possibility of such damages.
+
+   9. Accepting Warranty or Additional Liability. While redistributing
+      the Work or Derivative Works thereof, You may choose to offer,
+      and charge a fee for, acceptance of support, warranty, indemnity,
+      or other liability obligations and/or rights consistent with this
+      License. However, in accepting such obligations, You may act only
+      on Your own behalf and on Your sole responsibility, not on behalf
+      of any other Contributor, and only if You agree to indemnify,
+      defend, and hold each Contributor harmless for any liability
+      incurred by, or claims asserted against, such Contributor by reason
+      of your accepting any such warranty or additional liability.
+
+   END OF TERMS AND CONDITIONS
+
+   APPENDIX: How to apply the Apache License to your work.
+
+      To apply the Apache License to your work, attach the following
+      boilerplate notice, with the fields enclosed by brackets "[]"
+      replaced with your own identifying information. (Don't include
+      the brackets!)  The text should be enclosed in the appropriate
+      comment syntax for the file format. We also recommend that a
+      file or class name and description of purpose be included on the
+      same "printed page" as the copyright notice for easier
+      identification within third-party archives.
+
+   Copyright 2020-2021 Mix Technology
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.

docker-compose.yml

@@ -0,0 +1,41 @@
+version: "3.9"
+
+volumes:
+  hibi_redis: {}
+
+networks:
+  hibi_net: {}
+
+services:
+  redis:
+    image: redis:alpine
+    container_name: hibi_redis
+    healthcheck:
+      test: ["CMD-SHELL", "redis-cli ping"]
+      interval: 10s
+      timeout: 5s
+      retries: 5
+    networks:
+      - hibi_net
+    volumes:
+      - hibi_redis:/data
+    expose: [6379]
+
+  api:
+    container_name: hibiapi
+    build:
+      dockerfile: Dockerfile
+      context: .
+    restart: on-failure
+    networks:
+      - hibi_net
+    depends_on:
+      redis:
+        condition: service_healthy
+    ports:
+      - "8080:8080"
+    environment:
+      PORT: "8080"
+      FORWARDED_ALLOW_IPS: "*"
+      GENERAL_CACHE_URI: "redis://redis:6379"
+      GENERAL_SERVER_HOST: "0.0.0.0"

hibiapi/__init__.py

@@ -0,0 +1,15 @@
+r"""
+  _    _ _ _     _          _____ _____  
+ | |  | (_) |   (_)   /\   |  __ \_   _| 
+ | |__| |_| |__  _   /  \  | |__) || |   
+ |  __  | | '_ \| | / /\ \ |  ___/ | |   
+ | |  | | | |_) | |/ ____ \| |    _| |_  
+ |_|  |_|_|_.__/|_/_/    \_\_|   |_____| 
+
+A program that implements easy-to-use APIs for a variety of commonly used sites
+Repository: https://github.com/mixmoe/HibiAPI
+"""  # noqa:W291,W293
+
+from importlib.metadata import version
+
+__version__ = version("hibiapi")

hibiapi/__main__.py

@@ -0,0 +1,118 @@
+import os
+from pathlib import Path
+
+import typer
+import uvicorn
+
+from hibiapi import __file__ as root_file
+from hibiapi import __version__
+from hibiapi.utils.config import CONFIG_DIR, DEFAULT_DIR, Config
+from hibiapi.utils.log import LOG_LEVEL, logger
+
+COPYRIGHT = r"""
+<b><g>
+  _    _ _ _     _          _____ _____  
+ | |  | (_) |   (_)   /\   |  __ \_   _| 
+ | |__| |_| |__  _   /  \  | |__) || |   
+ |  __  | | '_ \| | / /\ \ |  ___/ | |   
+ | |  | | | |_) | |/ ____ \| |    _| |_  
+ |_|  |_|_|_.__/|_/_/    \_\_|   |_____| 
+</g><e>
+A program that implements easy-to-use APIs for a variety of commonly used sites
+Repository: https://github.com/mixmoe/HibiAPI
+</e></b>""".strip()  # noqa:W291
+
+
+LOG_CONFIG = {
+    "version": 1,
+    "disable_existing_loggers": False,
+    "handlers": {
+        "default": {
+            "class": "hibiapi.utils.log.LoguruHandler",
+        },
+    },
+    "loggers": {
+        "uvicorn.error": {
+            "handlers": ["default"],
+            "level": LOG_LEVEL,
+        },
+        "uvicorn.access": {
+            "handlers": ["default"],
+            "level": LOG_LEVEL,
+        },
+    },
+}
+
+RELOAD_CONFIG = {
+    "reload": True,
+    "reload_dirs": [
+        *map(str, [Path(root_file).parent.absolute(), CONFIG_DIR.absolute()])
+    ],
+    "reload_includes": ["*.py", "*.yml"],
+}
+
+
+cli = typer.Typer()
+
+
+@cli.callback(invoke_without_command=True)
+@cli.command()
+def run(
+    ctx: typer.Context,
+    host: str = Config["server"]["host"].as_str(),
+    port: int = Config["server"]["port"].as_number(),
+    workers: int = 1,
+    reload: bool = False,
+):
+    if ctx.invoked_subcommand is not None:
+        return
+
+    if ctx.info_name != (func_name := run.__name__):
+        logger.warning(
+            f"Directly usage of command <r>{ctx.info_name}</r> is <b>deprecated</b>, "
+            f"please use <g>{ctx.info_name} {func_name}</g> instead."
+        )
+
+    try:
+        terminal_width, _ = os.get_terminal_size()
+    except OSError:
+        terminal_width = 0
+    logger.warning(
+        "\n".join(i.center(terminal_width) for i in COPYRIGHT.splitlines()),
+    )
+    logger.info(f"HibiAPI version: <g><b>{__version__}</b></g>")
+
+    uvicorn.run(
+        "hibiapi.app:app",
+        host=host,
+        port=port,
+        access_log=False,
+        log_config=LOG_CONFIG,
+        workers=workers,
+        forwarded_allow_ips=Config["server"]["allowed-forward"].get_optional(str),
+        **(RELOAD_CONFIG if reload else {}),
+    )
+
+
+@cli.command()
+def config(force: bool = False):
+    total_written = 0
+    CONFIG_DIR.mkdir(parents=True, exist_ok=True)
+    for file in os.listdir(DEFAULT_DIR):
+        default_path = DEFAULT_DIR / file
+        config_path = CONFIG_DIR / file
+        if not (existed := config_path.is_file()) or force:
+            total_written += config_path.write_text(
+                default_path.read_text(encoding="utf-8"),
+                encoding="utf-8",
+            )
+            typer.echo(
+                typer.style(("Overwritten" if existed else "Created") + ": ", fg="blue")
+                + typer.style(str(config_path), fg="yellow")
+            )
+    if total_written > 0:
+        typer.echo(f"Config folder generated, {total_written=}")
+
+
+if __name__ == "__main__":
+    cli()

hibiapi/api/bika/__init__.py

@@ -0,0 +1,3 @@
+from .api import BikaEndpoints, ImageQuality, ResultSort  # noqa: F401
+from .constants import BikaConstants  # noqa: F401
+from .net import BikaLogin, NetRequest  # noqa: F401

hibiapi/api/bika/api.py

@@ -0,0 +1,206 @@
+import hashlib
+import hmac
+from datetime import timedelta
+from enum import Enum
+from time import time
+from typing import Any, Optional, cast
+
+from httpx import URL
+
+from hibiapi.api.bika.constants import BikaConstants
+from hibiapi.api.bika.net import NetRequest
+from hibiapi.utils.cache import cache_config
+from hibiapi.utils.decorators import enum_auto_doc
+from hibiapi.utils.net import catch_network_error
+from hibiapi.utils.routing import BaseEndpoint, dont_route, request_headers
+
+
+@enum_auto_doc
+class ImageQuality(str, Enum):
+    """哔咔API返回的图片质量"""
+
+    low = "low"
+    """低质量"""
+    medium = "medium"
+    """中等质量"""
+    high = "high"
+    """高质量"""
+    original = "original"
+    """原图"""
+
+
+@enum_auto_doc
+class ResultSort(str, Enum):
+    """哔咔API返回的搜索结果排序方式"""
+
+    date_descending = "dd"
+    """最新发布"""
+    date_ascending = "da"
+    """最早发布"""
+    like_descending = "ld"
+    """最多喜欢"""
+    views_descending = "vd"
+    """最多浏览"""
+
+
+class BikaEndpoints(BaseEndpoint):
+    @staticmethod
+    def _sign(url: URL, timestamp_bytes: bytes, nonce: bytes, method: bytes):
+        return hmac.new(
+            BikaConstants.DIGEST_KEY,
+            (
+                url.raw_path.lstrip(b"/")
+                + timestamp_bytes
+                + nonce
+                + method
+                + BikaConstants.API_KEY
+            ).lower(),
+            hashlib.sha256,
+        ).hexdigest()
+
+    @dont_route
+    @catch_network_error
+    async def request(
+        self,
+        endpoint: str,
+        *,
+        params: Optional[dict[str, Any]] = None,
+        body: Optional[dict[str, Any]] = None,
+        no_token: bool = False,
+    ):
+        net_client = cast(NetRequest, self.client.net_client)
+        if not no_token:
+            async with net_client.auth_lock:
+                if net_client.token is None:
+                    await net_client.login(self)
+
+        headers = {
+            "Authorization": net_client.token or "",
+            "Time": (current_time := f"{time():.0f}".encode()),
+            "Image-Quality": request_headers.get().get(
+                "X-Image-Quality", ImageQuality.medium
+            ),
+            "Nonce": (nonce := hashlib.md5(current_time).hexdigest().encode()),
+            "Signature": self._sign(
+                request_url := self._join(
+                    base=BikaConstants.API_HOST,
+                    endpoint=endpoint,
+                    params=params or {},
+                ),
+                current_time,
+                nonce,
+                b"GET" if body is None else b"POST",
+            ),
+        }
+
+        response = await (
+            self.client.get(request_url, headers=headers)
+            if body is None
+            else self.client.post(request_url, headers=headers, json=body)
+        )
+        return response.json()
+
+    @cache_config(ttl=timedelta(days=1))
+    async def collections(self):
+        return await self.request("collections")
+
+    @cache_config(ttl=timedelta(days=3))
+    async def categories(self):
+        return await self.request("categories")
+
+    @cache_config(ttl=timedelta(days=3))
+    async def keywords(self):
+        return await self.request("keywords")
+
+    async def advanced_search(
+        self,
+        *,
+        keyword: str,
+        page: int = 1,
+        sort: ResultSort = ResultSort.date_descending,
+    ):
+        return await self.request(
+            "comics/advanced-search",
+            body={
+                "keyword": keyword,
+                "sort": sort,
+            },
+            params={
+                "page": page,
+                "s": sort,
+            },
+        )
+
+    async def category_list(
+        self,
+        *,
+        category: str,
+        page: int = 1,
+        sort: ResultSort = ResultSort.date_descending,
+    ):
+        return await self.request(
+            "comics",
+            params={
+                "page": page,
+                "c": category,
+                "s": sort,
+            },
+        )
+
+    async def author_list(
+        self,
+        *,
+        author: str,
+        page: int = 1,
+        sort: ResultSort = ResultSort.date_descending,
+    ):
+        return await self.request(
+            "comics",
+            params={
+                "page": page,
+                "a": author,
+                "s": sort,
+            },
+        )
+
+    @cache_config(ttl=timedelta(days=3))
+    async def comic_detail(self, *, id: str):
+        return await self.request("comics/{id}", params={"id": id})
+
+    async def comic_recommendation(self, *, id: str):
+        return await self.request("comics/{id}/recommendation", params={"id": id})
+
+    async def comic_episodes(self, *, id: str, page: int = 1):
+        return await self.request(
+            "comics/{id}/eps",
+            params={
+                "id": id,
+                "page": page,
+            },
+        )
+
+    async def comic_page(self, *, id: str, order: int = 1, page: int = 1):
+        return await self.request(
+            "comics/{id}/order/{order}/pages",
+            params={
+                "id": id,
+                "order": order,
+                "page": page,
+            },
+        )
+
+    async def comic_comments(self, *, id: str, page: int = 1):
+        return await self.request(
+            "comics/{id}/comments",
+            params={
+                "id": id,
+                "page": page,
+            },
+        )
+
+    async def games(self, *, page: int = 1):
+        return await self.request("games", params={"page": page})
+
+    @cache_config(ttl=timedelta(days=3))
+    async def game_detail(self, *, id: str):
+        return await self.request("games/{id}", params={"id": id})

hibiapi/api/bika/constants.py

@@ -0,0 +1,19 @@
+from hibiapi.utils.config import APIConfig
+
+
+class BikaConstants:
+    DIGEST_KEY = b"~d}$Q7$eIni=V)9\\RK/P.RM4;9[7|@/CA}b~OW!3?EV`:<>M7pddUBL5n|0/*Cn"
+    API_KEY = b"C69BAF41DA5ABD1FFEDC6D2FEA56B"
+    DEFAULT_HEADERS = {
+        "API-Key": API_KEY,
+        "App-Channel": "2",
+        "App-Version": "2.2.1.2.3.3",
+        "App-Build-Version": "44",
+        "App-UUID": "defaultUuid",
+        "Accept": "application/vnd.picacomic.com.v1+json",
+        "App-Platform": "android",
+        "User-Agent": "okhttp/3.8.1",
+        "Content-Type": "application/json; charset=UTF-8",
+    }
+    API_HOST = "https://picaapi.picacomic.com/"
+    CONFIG = APIConfig("bika")

hibiapi/api/bika/net.py

@@ -0,0 +1,73 @@
+import asyncio
+from base64 import urlsafe_b64decode
+from datetime import datetime, timezone
+from functools import lru_cache
+from typing import TYPE_CHECKING, Any, Literal, Optional
+
+from pydantic import BaseModel, Field
+
+from hibiapi.api.bika.constants import BikaConstants
+from hibiapi.utils.net import BaseNetClient
+
+if TYPE_CHECKING:
+    from .api import BikaEndpoints
+
+
+class BikaLogin(BaseModel):
+    email: str
+    password: str
+
+
+class JWTHeader(BaseModel):
+    alg: str
+    typ: Literal["JWT"]
+
+
+class JWTBody(BaseModel):
+    id: str = Field(alias="_id")
+    iat: datetime
+    exp: datetime
+
+
+@lru_cache(maxsize=4)
+def load_jwt(token: str):
+    def b64pad(data: str):
+        return data + "=" * (-len(data) % 4)
+
+    head, body, _ = token.split(".")
+    head_data = JWTHeader.parse_raw(urlsafe_b64decode(b64pad(head)))
+    body_data = JWTBody.parse_raw(urlsafe_b64decode(b64pad(body)))
+    return head_data, body_data
+
+
+class NetRequest(BaseNetClient):
+    _token: Optional[str] = None
+
+    def __init__(self):
+        super().__init__(
+            headers=BikaConstants.DEFAULT_HEADERS.copy(),
+            proxies=BikaConstants.CONFIG["proxy"].as_dict(),
+        )
+        self.auth_lock = asyncio.Lock()
+
+    @property
+    def token(self) -> Optional[str]:
+        if self._token is None:
+            return None
+        _, body = load_jwt(self._token)
+        return None if body.exp < datetime.now(timezone.utc) else self._token
+
+    async def login(self, endpoint: "BikaEndpoints"):
+        login_data = BikaConstants.CONFIG["account"].get(BikaLogin)
+        login_result: dict[str, Any] = await endpoint.request(
+            "auth/sign-in",
+            body=login_data.dict(),
+            no_token=True,
+        )
+        assert login_result["code"] == 200, login_result["message"]
+        if not (
+            isinstance(login_data := login_result.get("data"), dict)
+            and "token" in login_data
+        ):
+            raise ValueError("failed to read Bika account token.")
+        self._token = login_data["token"]

hibiapi/api/bilibili/__init__.py

@@ -0,0 +1,4 @@
+# flake8:noqa:F401
+from .api import *  # noqa: F401, F403
+from .constants import BilibiliConstants
+from .net import NetRequest

hibiapi/api/bilibili/api/__init__.py

@@ -0,0 +1,4 @@
+# flake8:noqa:F401
+from .base import BaseBilibiliEndpoint, TimelineType, VideoFormatType, VideoQualityType
+from .v2 import BilibiliEndpointV2, SearchType
+from .v3 import BilibiliEndpointV3

hibiapi/api/bilibili/api/base.py

@@ -0,0 +1,278 @@
+import hashlib
+import json
+from enum import Enum, IntEnum
+from time import time
+from typing import Any, Optional, overload
+
+from httpx import URL
+
+from hibiapi.api.bilibili.constants import BilibiliConstants
+from hibiapi.utils.decorators import enum_auto_doc
+from hibiapi.utils.net import catch_network_error
+from hibiapi.utils.routing import BaseEndpoint, dont_route
+
+
+@enum_auto_doc
+class TimelineType(str, Enum):
+    """番剧时间线类型"""
+
+    CN = "cn"
+    """国产动画"""
+    GLOBAL = "global"
+    """番剧"""
+
+
+@enum_auto_doc
+class VideoQualityType(IntEnum):
+    """视频质量类型"""
+
+    VIDEO_240P = 6
+    VIDEO_360P = 16
+    VIDEO_480P = 32
+    VIDEO_720P = 64
+    VIDEO_720P_60FPS = 74
+    VIDEO_1080P = 80
+    VIDEO_1080P_PLUS = 112
+    VIDEO_1080P_60FPS = 116
+    VIDEO_4K = 120
+
+
+@enum_auto_doc
+class VideoFormatType(IntEnum):
+    """视频格式类型"""
+
+    FLV = 0
+    MP4 = 2
+    DASH = 16
+
+
+class BaseBilibiliEndpoint(BaseEndpoint):
+    def _sign(self, base: str, endpoint: str, params: dict[str, Any]) -> URL:
+        params.update(
+            {
+                **BilibiliConstants.DEFAULT_PARAMS,
+                "access_key": BilibiliConstants.ACCESS_KEY,
+                "appkey": BilibiliConstants.APP_KEY,
+                "ts": int(time()),
+            }
+        )
+        params = {k: params[k] for k in sorted(params.keys())}
+        url = self._join(base=base, endpoint=endpoint, params=params)
+        params["sign"] = hashlib.md5(url.query + BilibiliConstants.SECRET).hexdigest()
+        return URL(url, params=params)
+
+    @staticmethod
+    def _parse_json(content: str) -> dict[str, Any]:
+        try:
+            return json.loads(content)
+        except json.JSONDecodeError:
+            # NOTE: this is used to parse jsonp response
+            right, left = content.find("("), content.rfind(")")
+            return json.loads(content[right + 1 : left].strip())
+
+    @overload
+    async def request(
+        self,
+        endpoint: str,
+        *,
+        sign: bool = True,
+        params: Optional[dict[str, Any]] = None,
+    ) -> dict[str, Any]: ...
+
+    @overload
+    async def request(
+        self,
+        endpoint: str,
+        source: str,
+        *,
+        sign: bool = True,
+        params: Optional[dict[str, Any]] = None,
+    ) -> dict[str, Any]: ...
+
+    @dont_route
+    @catch_network_error
+    async def request(
+        self,
+        endpoint: str,
+        source: Optional[str] = None,
+        *,
+        sign: bool = True,
+        params: Optional[dict[str, Any]] = None,
+    ) -> dict[str, Any]:
+        host = BilibiliConstants.SERVER_HOST[source or "app"]
+        url = (self._sign if sign else self._join)(
+            base=host, endpoint=endpoint, params=params or {}
+        )
+        response = await self.client.get(url)
+        response.raise_for_status()
+        return self._parse_json(response.text)
+
+    async def playurl(
+        self,
+        *,
+        aid: int,
+        cid: int,
+        quality: VideoQualityType = VideoQualityType.VIDEO_480P,
+        type: VideoFormatType = VideoFormatType.FLV,
+    ):
+        return await self.request(
+            "x/player/playurl",
+            "api",
+            sign=False,
+            params={
+                "avid": aid,
+                "cid": cid,
+                "qn": quality,
+                "fnval": type,
+                "fnver": 0,
+                "fourk": 0 if quality >= VideoQualityType.VIDEO_4K else 1,
+            },
+        )
+
+    async def view(self, *, aid: int):
+        return await self.request(
+            "x/v2/view",
+            params={
+                "aid": aid,
+            },
+        )
+
+    async def search(self, *, keyword: str, page: int = 1, pagesize: int = 20):
+        return await self.request(
+            "x/v2/search",
+            params={
+                "duration": 0,
+                "keyword": keyword,
+                "pn": page,
+                "ps": pagesize,
+            },
+        )
+
+    async def search_hot(self, *, limit: int = 50):
+        return await self.request(
+            "x/v2/search/hot",
+            params={
+                "limit": limit,
+            },
+        )
+
+    async def search_suggest(self, *, keyword: str, type: str = "accurate"):
+        return await self.request(
+            "x/v2/search/suggest",
+            params={
+                "keyword": keyword,
+                "type": type,
+            },
+        )
+
+    async def space(self, *, vmid: int, page: int = 1, pagesize: int = 10):
+        return await self.request(
+            "x/v2/space",
+            params={
+                "vmid": vmid,
+                "ps": pagesize,
+                "pn": page,
+            },
+        )
+
+    async def space_archive(self, *, vmid: int, page: int = 1, pagesize: int = 10):
+        return await self.request(
+            "x/v2/space/archive",
+            params={
+                "vmid": vmid,
+                "ps": pagesize,
+                "pn": page,
+            },
+        )
+
+    async def favorite_video(
+        self,
+        *,
+        fid: int,
+        vmid: int,
+        page: int = 1,
+        pagesize: int = 20,
+    ):
+        return await self.request(
+            "x/v2/fav/video",
+            "api",
+            params={
+                "fid": fid,
+                "pn": page,
+                "ps": pagesize,
+                "vmid": vmid,
+                "order": "ftime",
+            },
+        )
+
+    async def event_list(
+        self,
+        *,
+        fid: int,
+        vmid: int,
+        page: int = 1,
+        pagesize: int = 20,
+    ):  # NOTE: this endpoint is not used
+        return await self.request(
+            "event/getlist",
+            "api",
+            params={
+                "fid": fid,
+                "pn": page,
+                "ps": pagesize,
+                "vmid": vmid,
+                "order": "ftime",
+            },
+        )
+
+    async def season_info(self, *, season_id: int):
+        return await self.request(
+            "pgc/view/web/season",
+            "api",
+            params={
+                "season_id": season_id,
+            },
+        )
+
+    async def bangumi_source(self, *, episode_id: int):
+        return await self.request(
+            "api/get_source",
+            "bgm",
+            params={
+                "episode_id": episode_id,
+            },
+        )
+
+    async def season_recommend(self, *, season_id: int):
+        return await self.request(
+            "pgc/season/web/related/recommend",
+            "api",
+            sign=False,
+            params={
+                "season_id": season_id,
+            },
+        )
+
+    async def timeline(self, *, type: TimelineType = TimelineType.GLOBAL):
+        return await self.request(
+            "web_api/timeline_{type}",
+            "bgm",
+            sign=False,
+            params={
+                "type": type,
+            },
+        )
+
+    async def suggest(self, *, keyword: str):  # NOTE: this endpoint is not used
+        return await self.request(
+            "main/suggest",
+            "search",
+            sign=False,
+            params={
+                "func": "suggest",
+                "suggest_type": "accurate",
+                "sug_type": "tag",
+                "main_ver": "v1",
+                "keyword": keyword,
+            },
+        )

hibiapi/api/bilibili/api/v2.py

@@ -0,0 +1,124 @@
+from collections.abc import Coroutine
+from enum import Enum
+from functools import wraps
+from typing import Callable, Optional, TypeVar
+
+from hibiapi.api.bilibili.api.base import (
+    BaseBilibiliEndpoint,
+    TimelineType,
+    VideoFormatType,
+    VideoQualityType,
+)
+from hibiapi.utils.decorators import enum_auto_doc
+from hibiapi.utils.exceptions import ClientSideException
+from hibiapi.utils.net import AsyncHTTPClient
+from hibiapi.utils.routing import BaseEndpoint
+
+_AnyCallable = TypeVar("_AnyCallable", bound=Callable[..., Coroutine])
+
+
+def process_keyerror(function: _AnyCallable) -> _AnyCallable:
+    @wraps(function)
+    async def wrapper(*args, **kwargs):
+        try:
+            return await function(*args, **kwargs)
+        except (KeyError, IndexError) as e:
+            raise ClientSideException(detail=str(e)) from None
+
+    return wrapper  # type:ignore
+
+
+@enum_auto_doc
+class SearchType(str, Enum):
+    """搜索类型"""
+
+    search = "search"
+    """综合搜索"""
+
+    suggest = "suggest"
+    """搜索建议"""
+
+    hot = "hot"
+    """热门"""
+
+
+class BilibiliEndpointV2(BaseEndpoint, cache_endpoints=False):
+    def __init__(self, client: AsyncHTTPClient):
+        super().__init__(client)
+        self.base = BaseBilibiliEndpoint(client)
+
+    @process_keyerror
+    async def playurl(
+        self,
+        *,
+        aid: int,
+        page: Optional[int] = None,
+        quality: VideoQualityType = VideoQualityType.VIDEO_480P,
+        type: VideoFormatType = VideoFormatType.MP4,
+    ):  # NOTE: not completely same with origin
+        video_view = await self.base.view(aid=aid)
+        if page is None:
+            return video_view
+        cid: int = video_view["data"]["pages"][page - 1]["cid"]
+        return await self.base.playurl(
+            aid=aid,
+            cid=cid,
+            quality=quality,
+            type=type,
+        )
+
+    async def seasoninfo(self, *, season_id: int):  # NOTE: not same with origin
+        return await self.base.season_info(season_id=season_id)
+
+    async def source(self, *, episode_id: int):
+        return await self.base.bangumi_source(episode_id=episode_id)
+
+    async def seasonrecommend(self, *, season_id: int):  # NOTE: not same with origin
+        return await self.base.season_recommend(season_id=season_id)
+
+    async def search(
+        self,
+        *,
+        keyword: str = "",
+        type: SearchType = SearchType.search,
+        page: int = 1,
+        pagesize: int = 20,
+        limit: int = 50,
+    ):
+        if type == SearchType.suggest:
+            return await self.base.search_suggest(keyword=keyword)
+        elif type == SearchType.hot:
+            return await self.base.search_hot(limit=limit)
+        else:
+            return await self.base.search(
+                keyword=keyword,
+                page=page,
+                pagesize=pagesize,
+            )
+
+    async def timeline(
+        self, *, type: TimelineType = TimelineType.GLOBAL
+    ):  # NOTE: not same with origin
+        return await self.base.timeline(type=type)
+
+    async def space(self, *, vmid: int, page: int = 1, pagesize: int = 10):
+        return await self.base.space(
+            vmid=vmid,
+            page=page,
+            pagesize=pagesize,
+        )
+
+    async def archive(self, *, vmid: int, page: int = 1, pagesize: int = 10):
+        return await self.base.space_archive(
+            vmid=vmid,
+            page=page,
+            pagesize=pagesize,
+        )
+
+    async def favlist(self, *, fid: int, vmid: int, page: int = 1, pagesize: int = 20):
+        return await self.base.favorite_video(
+            fid=fid,
+            vmid=vmid,
+            page=page,
+            pagesize=pagesize,
+        )

hibiapi/api/bilibili/api/v3.py

@@ -0,0 +1,79 @@
+from hibiapi.api.bilibili.api.base import (
+    BaseBilibiliEndpoint,
+    TimelineType,
+    VideoFormatType,
+    VideoQualityType,
+)
+from hibiapi.utils.net import AsyncHTTPClient
+from hibiapi.utils.routing import BaseEndpoint
+
+
+class BilibiliEndpointV3(BaseEndpoint, cache_endpoints=False):
+    def __init__(self, client: AsyncHTTPClient):
+        super().__init__(client)
+        self.base = BaseBilibiliEndpoint(client)
+
+    async def video_info(self, *, aid: int):
+        return await self.base.view(aid=aid)
+
+    async def video_address(
+        self,
+        *,
+        aid: int,
+        cid: int,
+        quality: VideoQualityType = VideoQualityType.VIDEO_480P,
+        type: VideoFormatType = VideoFormatType.FLV,
+    ):
+        return await self.base.playurl(
+            aid=aid,
+            cid=cid,
+            quality=quality,
+            type=type,
+        )
+
+    async def user_info(self, *, uid: int, page: int = 1, size: int = 10):
+        return await self.base.space(
+            vmid=uid,
+            page=page,
+            pagesize=size,
+        )
+
+    async def user_uploaded(self, *, uid: int, page: int = 1, size: int = 10):
+        return await self.base.space_archive(
+            vmid=uid,
+            page=page,
+            pagesize=size,
+        )
+
+    async def user_favorite(self, *, uid: int, fid: int, page: int = 1, size: int = 10):
+        return await self.base.favorite_video(
+            fid=fid,
+            vmid=uid,
+            page=page,
+            pagesize=size,
+        )
+
+    async def season_info(self, *, season_id: int):
+        return await self.base.season_info(season_id=season_id)
+
+    async def season_recommend(self, *, season_id: int):
+        return await self.base.season_recommend(season_id=season_id)
+
+    async def season_episode(self, *, episode_id: int):
+        return await self.base.bangumi_source(episode_id=episode_id)
+
+    async def season_timeline(self, *, type: TimelineType = TimelineType.GLOBAL):
+        return await self.base.timeline(type=type)
+
+    async def search(self, *, keyword: str, page: int = 1, size: int = 20):
+        return await self.base.search(
+            keyword=keyword,
+            page=page,
+            pagesize=size,
+        )
+
+    async def search_recommend(self, *, limit: int = 50):
+        return await self.base.search_hot(limit=limit)
+
+    async def search_suggestion(self, *, keyword: str):
+        return await self.base.search_suggest(keyword=keyword)

hibiapi/api/bilibili/constants.py

@@ -0,0 +1,32 @@
+from http.cookies import SimpleCookie
+from typing import Any
+
+from hibiapi.utils.config import APIConfig
+
+_CONFIG = APIConfig("bilibili")
+
+
+class BilibiliConstants:
+    SERVER_HOST: dict[str, str] = {
+        "app": "https://app.bilibili.com",
+        "api": "https://api.bilibili.com",
+        "interface": "https://interface.bilibili.com",
+        "main": "https://www.bilibili.com",
+        "bgm": "https://bangumi.bilibili.com",
+        "comment": "https://comment.bilibili.com",
+        "search": "https://s.search.bilibili.com",
+        "mobile": "https://m.bilibili.com",
+    }
+    APP_HOST: str = "http://app.bilibili.com"
+    DEFAULT_PARAMS: dict[str, Any] = {
+        "build": 507000,
+        "device": "android",
+        "platform": "android",
+        "mobi_app": "android",
+    }
+    APP_KEY: str = "1d8b6e7d45233436"
+    SECRET: bytes = b"560c52ccd288fed045859ed18bffd973"
+    ACCESS_KEY: str = "5271b2f0eb92f5f89af4dc39197d8e41"
+    COOKIES: SimpleCookie = SimpleCookie(_CONFIG["net"]["cookie"].as_str())
+    USER_AGENT: str = _CONFIG["net"]["user-agent"].as_str()
+    CONFIG: APIConfig = _CONFIG

hibiapi/api/bilibili/net.py

@@ -0,0 +1,13 @@
+from httpx import Cookies
+
+from hibiapi.utils.net import BaseNetClient
+
+from .constants import BilibiliConstants
+
+
+class NetRequest(BaseNetClient):
+    def __init__(self):
+        super().__init__(
+            headers={"user-agent": BilibiliConstants.USER_AGENT},
+            cookies=Cookies({k: v.value for k, v in BilibiliConstants.COOKIES.items()}),
+        )

hibiapi/api/netease/__init__.py

@@ -0,0 +1,4 @@
+# flake8:noqa:F401
+from .api import BitRateType, NeteaseEndpoint, RecordPeriodType, SearchType
+from .constants import NeteaseConstants
+from .net import NetRequest

hibiapi/api/netease/api.py

@@ -0,0 +1,326 @@
+import base64
+import json
+import secrets
+import string
+from datetime import timedelta
+from enum import IntEnum
+from ipaddress import IPv4Address
+from random import randint
+from typing import Annotated, Any, Optional
+
+from Cryptodome.Cipher import AES
+from Cryptodome.Util.Padding import pad
+from fastapi import Query
+
+from hibiapi.api.netease.constants import NeteaseConstants
+from hibiapi.utils.cache import cache_config
+from hibiapi.utils.decorators import enum_auto_doc
+from hibiapi.utils.exceptions import UpstreamAPIException
+from hibiapi.utils.net import catch_network_error
+from hibiapi.utils.routing import BaseEndpoint, dont_route
+
+
+@enum_auto_doc
+class SearchType(IntEnum):
+    """搜索内容类型"""
+
+    SONG = 1
+    """单曲"""
+    ALBUM = 10
+    """专辑"""
+    ARTIST = 100
+    """歌手"""
+    PLAYLIST = 1000
+    """歌单"""
+    USER = 1002
+    """用户"""
+    MV = 1004
+    """MV"""
+    LYRICS = 1006
+    """歌词"""
+    DJ = 1009
+    """主播电台"""
+    VIDEO = 1014
+    """视频"""
+
+
+@enum_auto_doc
+class BitRateType(IntEnum):
+    """歌曲码率"""
+
+    LOW = 64000
+    MEDIUM = 128000
+    STANDARD = 198000
+    HIGH = 320000
+
+
+@enum_auto_doc
+class MVResolutionType(IntEnum):
+    """MV分辨率"""
+
+    QVGA = 240
+    VGA = 480
+    HD = 720
+    FHD = 1080
+
+
+@enum_auto_doc
+class RecordPeriodType(IntEnum):
+    """听歌记录时段类型"""
+
+    WEEKLY = 1
+    """本周"""
+    ALL = 0
+    """所有时段"""
+
+
+class _EncryptUtil:
+    alphabets = bytearray(ord(char) for char in string.ascii_letters + string.digits)
+
+    @staticmethod
+    def _aes(data: bytes, key: bytes) -> bytes:
+        data = pad(data, 16) if len(data) % 16 else data
+        return base64.encodebytes(
+            AES.new(
+                key=key,
+                mode=AES.MODE_CBC,
+                iv=NeteaseConstants.AES_IV,
+            ).encrypt(data)
+        )
+
+    @staticmethod
+    def _rsa(data: bytes):
+        result = pow(
+            base=int(data.hex(), 16),
+            exp=NeteaseConstants.RSA_PUBKEY,
+            mod=NeteaseConstants.RSA_MODULUS,
+        )
+        return f"{result:0>256x}"
+
+    @classmethod
+    def encrypt(cls, data: dict[str, Any]) -> dict[str, str]:
+        secret = bytes(secrets.choice(cls.alphabets) for _ in range(16))
+        secure_key = cls._rsa(bytes(reversed(secret)))
+        return {
+            "params": cls._aes(
+                data=cls._aes(
+                    data=json.dumps(data).encode(),
+                    key=NeteaseConstants.AES_KEY,
+                ),
+                key=secret,
+            ).decode("ascii"),
+            "encSecKey": secure_key,
+        }
+
+
+class NeteaseEndpoint(BaseEndpoint):
+    def _construct_headers(self):
+        headers = self.client.headers.copy()
+        headers["X-Real-IP"] = str(
+            IPv4Address(
+                randint(
+                    int(NeteaseConstants.SOURCE_IP_SEGMENT.network_address),
+                    int(NeteaseConstants.SOURCE_IP_SEGMENT.broadcast_address),
+                )
+            )
+        )
+        return headers
+
+    @dont_route
+    @catch_network_error
+    async def request(
+        self, endpoint: str, *, params: Optional[dict[str, Any]] = None
+    ) -> dict[str, Any]:
+        params = {
+            **(params or {}),
+            "csrf_token": self.client.cookies.get("__csrf", ""),
+        }
+        response = await self.client.post(
+            self._join(
+                NeteaseConstants.HOST,
+                endpoint=endpoint,
+                params=params,
+            ),
+            headers=self._construct_headers(),
+            data=_EncryptUtil.encrypt(params),
+        )
+        response.raise_for_status()
+        if not response.text.strip():
+            raise UpstreamAPIException(
+                f"Upstream API {endpoint=} returns blank content"
+            )
+        return response.json()
+
+    async def search(
+        self,
+        *,
+        s: str,
+        search_type: SearchType = SearchType.SONG,
+        limit: int = 20,
+        offset: int = 0,
+    ):
+        return await self.request(
+            "api/cloudsearch/pc",
+            params={
+                "s": s,
+                "type": search_type,
+                "limit": limit,
+                "offset": offset,
+                "total": True,
+            },
+        )
+
+    async def artist(self, *, id: int):
+        return await self.request(
+            "weapi/v1/artist/{artist_id}",
+            params={
+                "artist_id": id,
+            },
+        )
+
+    async def album(self, *, id: int):
+        return await self.request(
+            "weapi/v1/album/{album_id}",
+            params={
+                "album_id": id,
+            },
+        )
+
+    async def detail(
+        self,
+        *,
+        id: Annotated[list[int], Query()],
+    ):
+        return await self.request(
+            "api/v3/song/detail",
+            params={
+                "c": json.dumps(
+                    [{"id": str(i)} for i in id],
+                ),
+            },
+        )
+
+    @cache_config(ttl=timedelta(minutes=20))
+    async def song(
+        self,
+        *,
+        id: Annotated[list[int], Query()],
+        br: BitRateType = BitRateType.STANDARD,
+    ):
+        return await self.request(
+            "weapi/song/enhance/player/url",
+            params={
+                "ids": [str(i) for i in id],
+                "br": br,
+            },
+        )
+
+    async def playlist(self, *, id: int):
+        return await self.request(
+            "weapi/v6/playlist/detail",
+            params={
+                "id": id,
+                "total": True,
+                "offset": 0,
+                "limit": 1000,
+                "n": 1000,
+            },
+        )
+
+    async def lyric(self, *, id: int):
+        return await self.request(
+            "weapi/song/lyric",
+            params={
+                "id": id,
+                "os": "pc",
+                "lv": -1,
+                "kv": -1,
+                "tv": -1,
+            },
+        )
+
+    async def mv(self, *, id: int):
+        return await self.request(
+            "api/v1/mv/detail",
+            params={
+                "id": id,
+            },
+        )
+
+    async def mv_url(
+        self,
+        *,
+        id: int,
+        res: MVResolutionType = MVResolutionType.FHD,
+    ):
+        return await self.request(
+            "weapi/song/enhance/play/mv/url",
+            params={
+                "id": id,
+                "r": res,
+            },
+        )
+
+    async def comments(self, *, id: int, offset: int = 0, limit: int = 1):
+        return await self.request(
+            "weapi/v1/resource/comments/R_SO_4_{song_id}",
+            params={
+                "song_id": id,
+                "offset": offset,
+                "total": True,
+                "limit": limit,
+            },
+        )
+
+    async def record(self, *, id: int, period: RecordPeriodType = RecordPeriodType.ALL):
+        return await self.request(
+            "weapi/v1/play/record",
+            params={
+                "uid": id,
+                "type": period,
+            },
+        )
+
+    async def djradio(self, *, id: int):
+        return await self.request(
+            "api/djradio/v2/get",
+            params={
+                "id": id,
+            },
+        )
+
+    async def dj(self, *, id: int, offset: int = 0, limit: int = 20, asc: bool = False):
+        # NOTE: Possible not same with origin
+        return await self.request(
+            "weapi/dj/program/byradio",
+            params={
+                "radioId": id,
+                "offset": offset,
+                "limit": limit,
+                "asc": asc,
+            },
+        )
+
+    async def detail_dj(self, *, id: int):
+        return await self.request(
+            "api/dj/program/detail",
+            params={
+                "id": id,
+            },
+        )
+
+    async def user(self, *, id: int):
+        return await self.request(
+            "weapi/v1/user/detail/{id}",
+            params={"id": id},
+        )
+
+    async def user_playlist(self, *, id: int, limit: int = 50, offset: int = 0):
+        return await self.request(
+            "weapi/user/playlist",
+            params={
+                "uid": id,
+                "limit": limit,
+                "offset": offset,
+            },
+        )

hibiapi/api/netease/constants.py

@@ -0,0 +1,33 @@
+from http.cookies import SimpleCookie
+from ipaddress import IPv4Network
+
+from hibiapi.utils.config import APIConfig
+
+_Config = APIConfig("netease")
+
+
+class NeteaseConstants:
+    AES_KEY: bytes = b"0CoJUm6Qyw8W8jud"
+    AES_IV: bytes = b"0102030405060708"
+    RSA_PUBKEY: int = int("010001", 16)
+    RSA_MODULUS: int = int(
+        "00e0b509f6259df8642dbc3566290147"
+        "7df22677ec152b5ff68ace615bb7b725"
+        "152b3ab17a876aea8a5aa76d2e417629"
+        "ec4ee341f56135fccf695280104e0312"
+        "ecbda92557c93870114af6c9d05c4f7f"
+        "0c3685b7a46bee255932575cce10b424"
+        "d813cfe4875d3e82047b97ddef52741d"
+        "546b8e289dc6935b3ece0462db0a22b8e7",
+        16,
+    )
+
+    HOST: str = "http://music.163.com"
+    COOKIES: SimpleCookie = SimpleCookie(_Config["net"]["cookie"].as_str())
+    SOURCE_IP_SEGMENT: IPv4Network = _Config["net"]["source"].get(IPv4Network)
+    DEFAULT_HEADERS: dict[str, str] = {
+        "user-agent": _Config["net"]["user-agent"].as_str(),
+        "referer": "http://music.163.com",
+    }
+
+    CONFIG: APIConfig = _Config

hibiapi/api/netease/net.py

@@ -0,0 +1,13 @@
+from httpx import Cookies
+
+from hibiapi.utils.net import BaseNetClient
+
+from .constants import NeteaseConstants
+
+
+class NetRequest(BaseNetClient):
+    def __init__(self):
+        super().__init__(
+            headers=NeteaseConstants.DEFAULT_HEADERS,
+            cookies=Cookies({k: v.value for k, v in NeteaseConstants.COOKIES.items()}),
+        )

hibiapi/api/pixiv/__init__.py

@@ -0,0 +1,13 @@
+# flake8:noqa:F401
+from .api import (
+    IllustType,
+    PixivEndpoints,
+    RankingDate,
+    RankingType,
+    SearchDurationType,
+    SearchModeType,
+    SearchNovelModeType,
+    SearchSortType,
+)
+from .constants import PixivConstants
+from .net import NetRequest, PixivAuthData

hibiapi/api/pixiv/api.py

@@ -0,0 +1,613 @@
+import json
+import re
+from datetime import date, timedelta
+from enum import Enum
+from typing import Any, Literal, Optional, Union, cast, overload
+
+from hibiapi.api.pixiv.constants import PixivConstants
+from hibiapi.api.pixiv.net import NetRequest as PixivNetClient
+from hibiapi.utils.cache import cache_config
+from hibiapi.utils.decorators import enum_auto_doc
+from hibiapi.utils.net import catch_network_error
+from hibiapi.utils.routing import BaseEndpoint, dont_route, request_headers
+
+
+@enum_auto_doc
+class IllustType(str, Enum):
+    """画作类型"""
+
+    illust = "illust"
+    """插画"""
+    manga = "manga"
+    """漫画"""
+
+
+@enum_auto_doc
+class RankingType(str, Enum):
+    """排行榜内容类型"""
+
+    day = "day"
+    """日榜"""
+    week = "week"
+    """周榜"""
+    month = "month"
+    """月榜"""
+    day_male = "day_male"
+    """男性向"""
+    day_female = "day_female"
+    """女性向"""
+    week_original = "week_original"
+    """原创周榜"""
+    week_rookie = "week_rookie"
+    """新人周榜"""
+    day_ai = "day_ai"
+    """AI日榜"""
+    day_manga = "day_manga"
+    """漫画日榜"""
+    week_manga = "week_manga"
+    """漫画周榜"""
+    month_manga = "month_manga"
+    """漫画月榜"""
+    week_rookie_manga = "week_rookie_manga"
+    """漫画新人周榜"""
+    day_r18 = "day_r18"
+    day_male_r18 = "day_male_r18"
+    day_female_r18 = "day_female_r18"
+    week_r18 = "week_r18"
+    week_r18g = "week_r18g"
+    day_r18_ai = "day_r18_ai"
+    day_r18_manga = "day_r18_manga"
+    week_r18_manga = "week_r18_manga"
+
+
+@enum_auto_doc
+class SearchModeType(str, Enum):
+    """搜索匹配类型"""
+
+    partial_match_for_tags = "partial_match_for_tags"
+    """标签部分一致"""
+    exact_match_for_tags = "exact_match_for_tags"
+    """标签完全一致"""
+    title_and_caption = "title_and_caption"
+    """标题说明文"""
+
+
+@enum_auto_doc
+class SearchNovelModeType(str, Enum):
+    """搜索匹配类型"""
+
+    partial_match_for_tags = "partial_match_for_tags"
+    """标签部分一致"""
+    exact_match_for_tags = "exact_match_for_tags"
+    """标签完全一致"""
+    text = "text"
+    """正文"""
+    keyword = "keyword"
+    """关键词"""
+
+
+@enum_auto_doc
+class SearchSortType(str, Enum):
+    """搜索排序类型"""
+
+    date_desc = "date_desc"
+    """按日期倒序"""
+    date_asc = "date_asc"
+    """按日期正序"""
+    popular_desc = "popular_desc"
+    """受欢迎降序(Premium功能)"""
+
+
+@enum_auto_doc
+class SearchDurationType(str, Enum):
+    """搜索时段类型"""
+
+    within_last_day = "within_last_day"
+    """一天内"""
+    within_last_week = "within_last_week"
+    """一周内"""
+    within_last_month = "within_last_month"
+    """一个月内"""
+
+
+class RankingDate(date):
+    @classmethod
+    def yesterday(cls) -> "RankingDate":
+        yesterday = cls.today() - timedelta(days=1)
+        return cls(yesterday.year, yesterday.month, yesterday.day)
+
+    def toString(self) -> str:
+        return self.strftime(r"%Y-%m-%d")
+
+    @classmethod
+    def new(cls, date: date) -> "RankingDate":
+        return cls(date.year, date.month, date.day)
+
+
+class PixivEndpoints(BaseEndpoint):
+    @staticmethod
+    def _parse_accept_language(accept_language: str) -> str:
+        first_language, *_ = accept_language.partition(",")
+        language_code, *_ = first_language.partition(";")
+        return language_code.lower().strip()
+
+    @overload
+    async def request(
+        self,
+        endpoint: str,
+        *,
+        params: Optional[dict[str, Any]] = None,
+        return_text: Literal[False] = False,
+    ) -> dict[str, Any]: ...
+
+    @overload
+    async def request(
+        self,
+        endpoint: str,
+        *,
+        params: Optional[dict[str, Any]] = None,
+        return_text: Literal[True],
+    ) -> str: ...
+
+    @dont_route
+    @catch_network_error
+    async def request(
+        self,
+        endpoint: str,
+        *,
+        params: Optional[dict[str, Any]] = None,
+        return_text: bool = False,
+    ) -> Union[dict[str, Any], str]:
+        headers = self.client.headers.copy()
+
+        net_client = cast(PixivNetClient, self.client.net_client)
+        async with net_client.auth_lock:
+            auth, token = net_client.get_available_user()
+            if auth is None:
+                auth = await net_client.auth(token)
+        headers["Authorization"] = f"Bearer {auth.access_token}"
+
+        if language := request_headers.get().get("Accept-Language"):
+            language = self._parse_accept_language(language)
+            headers["Accept-Language"] = language
+
+        response = await self.client.get(
+            self._join(
+                base=PixivConstants.APP_HOST,
+                endpoint=endpoint,
+                params=params or {},
+            ),
+            headers=headers,
+        )
+        if return_text:
+            return response.text
+        return response.json()
+
+    @cache_config(ttl=timedelta(days=3))
+    async def illust(self, *, id: int):
+        return await self.request("v1/illust/detail", params={"illust_id": id})
+
+    @cache_config(ttl=timedelta(days=1))
+    async def member(self, *, id: int):
+        return await self.request("v1/user/detail", params={"user_id": id})
+
+    async def member_illust(
+        self,
+        *,
+        id: int,
+        illust_type: IllustType = IllustType.illust,
+        page: int = 1,
+        size: int = 30,
+    ):
+        return await self.request(
+            "v1/user/illusts",
+            params={
+                "user_id": id,
+                "type": illust_type,
+                "offset": (page - 1) * size,
+            },
+        )
+
+    async def favorite(
+        self,
+        *,
+        id: int,
+        tag: Optional[str] = None,
+        max_bookmark_id: Optional[int] = None,
+    ):
+        return await self.request(
+            "v1/user/bookmarks/illust",
+            params={
+                "user_id": id,
+                "tag": tag,
+                "restrict": "public",
+                "max_bookmark_id": max_bookmark_id or None,
+            },
+        )
+
+    # 用户收藏的小说
+    async def favorite_novel(
+        self,
+        *,
+        id: int,
+        tag: Optional[str] = None,
+    ):
+        return await self.request(
+            "v1/user/bookmarks/novel",
+            params={
+                "user_id": id,
+                "tag": tag,
+                "restrict": "public",
+            },
+        )
+
+    async def following(self, *, id: int, page: int = 1, size: int = 30):
+        return await self.request(
+            "v1/user/following",
+            params={
+                "user_id": id,
+                "offset": (page - 1) * size,
+            },
+        )
+
+    async def follower(self, *, id: int, page: int = 1, size: int = 30):
+        return await self.request(
+            "v1/user/follower",
+            params={
+                "user_id": id,
+                "offset": (page - 1) * size,
+            },
+        )
+
+    @cache_config(ttl=timedelta(hours=12))
+    async def rank(
+        self,
+        *,
+        mode: RankingType = RankingType.week,
+        date: Optional[RankingDate] = None,
+        page: int = 1,
+        size: int = 30,
+    ):
+        return await self.request(
+            "v1/illust/ranking",
+            params={
+                "mode": mode,
+                "date": RankingDate.new(date or RankingDate.yesterday()).toString(),
+                "offset": (page - 1) * size,
+            },
+        )
+
+    async def search(
+        self,
+        *,
+        word: str,
+        mode: SearchModeType = SearchModeType.partial_match_for_tags,
+        order: SearchSortType = SearchSortType.date_desc,
+        duration: Optional[SearchDurationType] = None,
+        page: int = 1,
+        size: int = 30,
+        include_translated_tag_results: bool = True,
+        search_ai_type: bool = True,  # 搜索结果是否包含AI作品
+    ):
+        return await self.request(
+            "v1/search/illust",
+            params={
+                "word": word,
+                "search_target": mode,
+                "sort": order,
+                "duration": duration,
+                "offset": (page - 1) * size,
+                "include_translated_tag_results": include_translated_tag_results,
+                "search_ai_type": 1 if search_ai_type else 0,
+            },
+        )
+
+    # 热门插画作品预览
+    async def popular_preview(
+        self,
+        *,
+        word: str,
+        mode: SearchModeType = SearchModeType.partial_match_for_tags,
+        merge_plain_keyword_results: bool = True,
+        include_translated_tag_results: bool = True,
+        filter: str = "for_ios",
+    ):
+        return await self.request(
+            "v1/search/popular-preview/illust",
+            params={
+                "word": word,
+                "search_target": mode,
+                "merge_plain_keyword_results": merge_plain_keyword_results,
+                "include_translated_tag_results": include_translated_tag_results,
+                "filter": filter,
+            },
+        )
+
+    async def search_user(
+        self,
+        *,
+        word: str,
+        page: int = 1,
+        size: int = 30,
+    ):
+        return await self.request(
+            "v1/search/user",
+            params={"word": word, "offset": (page - 1) * size},
+        )
+
+    async def tags_autocomplete(
+        self,
+        *,
+        word: str,
+        merge_plain_keyword_results: bool = True,
+    ):
+        return await self.request(
+            "/v2/search/autocomplete",
+            params={
+                "word": word,
+                "merge_plain_keyword_results": merge_plain_keyword_results,
+            },
+        )
+
+    @cache_config(ttl=timedelta(hours=12))
+    async def tags(self):
+        return await self.request("v1/trending-tags/illust")
+
+    @cache_config(ttl=timedelta(minutes=15))
+    async def related(self, *, id: int, page: int = 1, size: int = 30):
+        return await self.request(
+            "v2/illust/related",
+            params={
+                "illust_id": id,
+                "offset": (page - 1) * size,
+            },
+        )
+
+    @cache_config(ttl=timedelta(days=3))
+    async def ugoira_metadata(self, *, id: int):
+        return await self.request(
+            "v1/ugoira/metadata",
+            params={
+                "illust_id": id,
+            },
+        )
+
+    # 大家的新作品（插画）
+    async def illust_new(
+        self,
+        *,
+        content_type: str = "illust",
+    ):
+        return await self.request(
+            "v1/illust/new",
+            params={
+                "content_type": content_type,
+                "filter": "for_ios",
+            },
+        )
+
+    # pixivision(亮点/特辑) 列表
+    async def spotlights(
+        self,
+        *,
+        category: str = "all",
+        page: int = 1,
+        size: int = 10,
+    ):
+        return await self.request(
+            "v1/spotlight/articles",
+            params={
+                "filter": "for_ios",
+                "category": category,
+                "offset": (page - 1) * size,
+            },
+        )
+
+    # 插画评论
+    async def illust_comments(
+        self,
+        *,
+        id: int,
+        page: int = 1,
+        size: int = 30,
+    ):
+        return await self.request(
+            "v3/illust/comments",
+            params={
+                "illust_id": id,
+                "offset": (page - 1) * size,
+            },
+        )
+
+    # 插画评论回复
+    async def illust_comment_replies(
+        self,
+        *,
+        id: int,
+    ):
+        return await self.request(
+            "v2/illust/comment/replies",
+            params={
+                "comment_id": id,
+            },
+        )
+
+    # 小说评论
+    async def novel_comments(
+        self,
+        *,
+        id: int,
+        page: int = 1,
+        size: int = 30,
+    ):
+        return await self.request(
+            "v3/novel/comments",
+            params={
+                "novel_id": id,
+                "offset": (page - 1) * size,
+            },
+        )
+
+    # 小说评论回复
+    async def novel_comment_replies(
+        self,
+        *,
+        id: int,
+    ):
+        return await self.request(
+            "v2/novel/comment/replies",
+            params={
+                "comment_id": id,
+            },
+        )
+
+    # 小说排行榜
+    async def rank_novel(
+        self,
+        *,
+        mode: str = "day",
+        date: Optional[RankingDate] = None,
+        page: int = 1,
+        size: int = 30,
+    ):
+        return await self.request(
+            "v1/novel/ranking",
+            params={
+                "mode": mode,
+                "date": RankingDate.new(date or RankingDate.yesterday()).toString(),
+                "offset": (page - 1) * size,
+            },
+        )
+
+    async def member_novel(self, *, id: int, page: int = 1, size: int = 30):
+        return await self.request(
+            "/v1/user/novels",
+            params={
+                "user_id": id,
+                "offset": (page - 1) * size,
+            },
+        )
+
+    async def novel_series(self, *, id: int):
+        return await self.request("/v2/novel/series", params={"series_id": id})
+
+    async def novel_detail(self, *, id: int):
+        return await self.request("/v2/novel/detail", params={"novel_id": id})
+
+    # 已被官方移除，调用 webview/v2/novel 作兼容处理
+    async def novel_text(self, *, id: int):
+        # return await self.request("/v1/novel/text", params={"novel_id": id})
+        response = await self.webview_novel(id=id)
+        return {"novel_text": response["text"] or ""}
+
+    # 获取小说 HTML 后解析 JSON
+    async def webview_novel(self, *, id: int):
+        response = await self.request(
+            "webview/v2/novel",
+            params={
+                "id": id,
+                "viewer_version": "20221031_ai",
+            },
+            return_text=True,
+        )
+
+        novel_match = re.search(r"novel:\s+(?P<data>{.+?}),\s+isOwnWork", response)
+        return json.loads(novel_match["data"] if novel_match else response)
+
+    @cache_config(ttl=timedelta(hours=12))
+    async def tags_novel(self):
+        return await self.request("v1/trending-tags/novel")
+
+    async def search_novel(
+        self,
+        *,
+        word: str,
+        mode: SearchNovelModeType = SearchNovelModeType.partial_match_for_tags,
+        sort: SearchSortType = SearchSortType.date_desc,
+        merge_plain_keyword_results: bool = True,
+        include_translated_tag_results: bool = True,
+        duration: Optional[SearchDurationType] = None,
+        page: int = 1,
+        size: int = 30,
+        search_ai_type: bool = True,  # 搜索结果是否包含AI作品
+    ):
+        return await self.request(
+            "/v1/search/novel",
+            params={
+                "word": word,
+                "search_target": mode,
+                "sort": sort,
+                "merge_plain_keyword_results": merge_plain_keyword_results,
+                "include_translated_tag_results": include_translated_tag_results,
+                "duration": duration,
+                "offset": (page - 1) * size,
+                "search_ai_type": 1 if search_ai_type else 0,
+            },
+        )
+
+    # 热门小说作品预览
+    async def popular_preview_novel(
+        self,
+        *,
+        word: str,
+        mode: SearchNovelModeType = SearchNovelModeType.partial_match_for_tags,
+        merge_plain_keyword_results: bool = True,
+        include_translated_tag_results: bool = True,
+        filter: str = "for_ios",
+    ):
+        return await self.request(
+            "v1/search/popular-preview/novel",
+            params={
+                "word": word,
+                "search_target": mode,
+                "merge_plain_keyword_results": merge_plain_keyword_results,
+                "include_translated_tag_results": include_translated_tag_results,
+                "filter": filter,
+            },
+        )
+
+    async def novel_new(self, *, max_novel_id: Optional[int] = None):
+        return await self.request(
+            "/v1/novel/new", params={"max_novel_id": max_novel_id}
+        )
+
+    # 人气直播列表
+    async def live_list(self, *, page: int = 1, size: int = 30):
+        params = {"list_type": "popular", "offset": (page - 1) * size}
+        if not params["offset"]:
+            del params["offset"]
+        return await self.request("v1/live/list", params=params)
+
+    # 相关小说作品
+    async def related_novel(self, *, id: int, page: int = 1, size: int = 30):
+        return await self.request(
+            "v1/novel/related",
+            params={
+                "novel_id": id,
+                "offset": (page - 1) * size,
+            },
+        )
+
+    # 相关用户
+    async def related_member(self, *, id: int):
+        return await self.request("v1/user/related", params={"seed_user_id": id})
+
+    # 漫画系列
+    async def illust_series(self, *, id: int, page: int = 1, size: int = 30):
+        return await self.request(
+            "v1/illust/series",
+            params={"illust_series_id": id, "offset": (page - 1) * size},
+        )
+
+    # 用户的漫画系列
+    async def member_illust_series(self, *, id: int, page: int = 1, size: int = 30):
+        return await self.request(
+            "v1/user/illust-series",
+            params={"user_id": id, "offset": (page - 1) * size},
+        )
+
+    # 用户的小说系列
+    async def member_novel_series(self, *, id: int, page: int = 1, size: int = 30):
+        return await self.request(
+            "v1/user/novel-series", params={"user_id": id, "offset": (page - 1) * size}
+        )

hibiapi/api/pixiv/constants.py

@@ -0,0 +1,19 @@
+from typing import Any
+
+from hibiapi.utils.config import APIConfig
+
+
+class PixivConstants:
+    DEFAULT_HEADERS: dict[str, Any] = {
+        "App-OS": "ios",
+        "App-OS-Version": "14.6",
+        "User-Agent": "PixivIOSApp/7.13.3 (iOS 14.6; iPhone13,2)",
+    }
+    CLIENT_ID: str = "MOBrBDS8blbauoSck0ZfDbtuzpyT"
+    CLIENT_SECRET: str = "lsACyCD94FhDUtGTXi3QzcFE2uU1hqtDaKeqrdwj"
+    HASH_SECRET: bytes = (
+        b"28c1fdd170a5204386cb1313c7077b34f83e4aaf4aa829ce78c231e05b0bae2c"
+    )
+    CONFIG: APIConfig = APIConfig("pixiv")
+    APP_HOST: str = "https://app-api.pixiv.net"
+    AUTH_HOST: str = "https://oauth.secure.pixiv.net"

hibiapi/api/pixiv/net.py

@@ -0,0 +1,85 @@
+import asyncio
+import hashlib
+from datetime import datetime, timedelta, timezone
+from itertools import cycle
+
+from httpx import URL
+from pydantic import BaseModel, Extra, Field
+
+from hibiapi.utils.log import logger
+from hibiapi.utils.net import BaseNetClient
+
+from .constants import PixivConstants
+
+
+class AccountDataModel(BaseModel):
+    class Config:
+        extra = Extra.allow
+
+
+class PixivUserData(AccountDataModel):
+    account: str
+    id: int
+    is_premium: bool
+    mail_address: str
+    name: str
+
+
+class PixivAuthData(AccountDataModel):
+    time: datetime = Field(default_factory=datetime.now)
+    expires_in: int
+    access_token: str
+    refresh_token: str
+    user: PixivUserData
+
+
+class NetRequest(BaseNetClient):
+    def __init__(self, tokens: list[str]):
+        super().__init__(
+            headers=PixivConstants.DEFAULT_HEADERS.copy(),
+            proxies=PixivConstants.CONFIG["proxy"].as_dict(),
+        )
+        self.user_tokens = cycle(tokens)
+        self.auth_lock = asyncio.Lock()
+        self.user_tokens_dict: dict[str, PixivAuthData] = {}
+        self.headers["accept-language"] = PixivConstants.CONFIG["language"].as_str()
+
+    def get_available_user(self):
+        token = next(self.user_tokens)
+        if (auth_data := self.user_tokens_dict.get(token)) and (
+            auth_data.time + timedelta(minutes=1, seconds=auth_data.expires_in)
+            > datetime.now()
+        ):
+            return auth_data, token
+        return None, token
+
+    async def auth(self, refresh_token: str):
+        url = URL(PixivConstants.AUTH_HOST).join("/auth/token")
+        time = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S+00:00")
+        headers = {
+            **self.headers,
+            "X-Client-Time": time,
+            "X-Client-Hash": hashlib.md5(
+                time.encode() + PixivConstants.HASH_SECRET
+            ).hexdigest(),
+        }
+        payload = {
+            "get_secure_url": 1,
+            "client_id": PixivConstants.CLIENT_ID,
+            "client_secret": PixivConstants.CLIENT_SECRET,
+            "grant_type": "refresh_token",
+            "refresh_token": refresh_token,
+        }
+
+        async with self as client:
+            response = await client.post(url, data=payload, headers=headers)
+            response.raise_for_status()
+
+        self.user_tokens_dict[refresh_token] = PixivAuthData.parse_obj(response.json())
+        user_data = self.user_tokens_dict[refresh_token].user
+        logger.opt(colors=True).info(
+            f"Pixiv account <m>{user_data.id}</m> info <b>Updated</b>: "
+            f"<b><e>{user_data.name}</e>({user_data.account})</b>."
+        )
+
+        return self.user_tokens_dict[refresh_token]

hibiapi/api/qrcode.py

@@ -0,0 +1,160 @@
+from datetime import datetime
+from enum import Enum
+from io import BytesIO
+from os import fdopen
+from pathlib import Path
+from typing import Literal, Optional, cast
+
+from PIL import Image
+from pydantic import AnyHttpUrl, BaseModel, Field, validate_arguments
+from pydantic.color import Color
+from qrcode import constants
+from qrcode.image.pil import PilImage
+from qrcode.main import QRCode
+
+from hibiapi.utils.config import APIConfig
+from hibiapi.utils.decorators import ToAsync, enum_auto_doc
+from hibiapi.utils.exceptions import ClientSideException
+from hibiapi.utils.net import BaseNetClient
+from hibiapi.utils.routing import BaseHostUrl
+from hibiapi.utils.temp import TempFile
+
+Config = APIConfig("qrcode")
+
+
+class HostUrl(BaseHostUrl):
+    allowed_hosts = Config["qrcode"]["icon-site"].get(list[str])
+
+
+@enum_auto_doc
+class QRCodeLevel(str, Enum):
+    """二维码容错率"""
+
+    LOW = "L"
+    """最低容错率"""
+    MEDIUM = "M"
+    """中等容错率"""
+    QUARTILE = "Q"
+    """高容错率"""
+    HIGH = "H"
+    """最高容错率"""
+
+
+@enum_auto_doc
+class ReturnEncode(str, Enum):
+    """二维码返回的编码方式"""
+
+    raw = "raw"
+    """直接重定向到二维码图片"""
+    json = "json"
+    """返回JSON格式的二维码信息"""
+    js = "js"
+    jsc = "jsc"
+
+
+COLOR_WHITE = Color("FFFFFF")
+COLOR_BLACK = Color("000000")
+
+
+class QRInfo(BaseModel):
+    url: Optional[AnyHttpUrl] = None
+    path: Path
+    time: datetime = Field(default_factory=datetime.now)
+    data: str
+    logo: Optional[HostUrl] = None
+    level: QRCodeLevel = QRCodeLevel.MEDIUM
+    size: int = 200
+    code: Literal[0] = 0
+    status: Literal["success"] = "success"
+
+    @classmethod
+    @validate_arguments
+    async def new(
+        cls,
+        text: str,
+        *,
+        size: int = Field(
+            200,
+            gt=Config["qrcode"]["min-size"].as_number(),
+            lt=Config["qrcode"]["max-size"].as_number(),
+        ),
+        logo: Optional[HostUrl] = None,
+        level: QRCodeLevel = QRCodeLevel.MEDIUM,
+        bgcolor: Color = COLOR_WHITE,
+        fgcolor: Color = COLOR_BLACK,
+    ):
+        icon_stream = None
+        if logo is not None:
+            async with BaseNetClient() as client:
+                response = await client.get(
+                    logo, headers={"user-agent": "HibiAPI@GitHub"}, timeout=6
+                )
+                response.raise_for_status()
+            icon_stream = BytesIO(response.content)
+        return cls(
+            data=text,
+            logo=logo,
+            level=level,
+            size=size,
+            path=await cls._generate(
+                text,
+                size=size,
+                level=level,
+                icon_stream=icon_stream,
+                bgcolor=bgcolor.as_hex(),
+                fgcolor=fgcolor.as_hex(),
+            ),
+        )
+
+    @classmethod
+    @ToAsync
+    def _generate(
+        cls,
+        text: str,
+        *,
+        size: int = 200,
+        level: QRCodeLevel = QRCodeLevel.MEDIUM,
+        icon_stream: Optional[BytesIO] = None,
+        bgcolor: str = "#FFFFFF",
+        fgcolor: str = "#000000",
+    ) -> Path:
+        qr = QRCode(
+            error_correction={
+                QRCodeLevel.LOW: constants.ERROR_CORRECT_L,
+                QRCodeLevel.MEDIUM: constants.ERROR_CORRECT_M,
+                QRCodeLevel.QUARTILE: constants.ERROR_CORRECT_Q,
+                QRCodeLevel.HIGH: constants.ERROR_CORRECT_H,
+            }[level],
+            border=2,
+            box_size=8,
+        )
+        qr.add_data(text)
+        image = cast(
+            Image.Image,
+            qr.make_image(
+                PilImage,
+                back_color=bgcolor,
+                fill_color=fgcolor,
+            ).get_image(),
+        )
+        image = image.resize((size, size))
+        if icon_stream is not None:
+            try:
+                icon = Image.open(icon_stream)
+            except ValueError as e:
+                raise ClientSideException("Invalid image format.") from e
+            icon_width, icon_height = icon.size
+            image.paste(
+                icon,
+                box=(
+                    int(size / 2 - icon_width / 2),
+                    int(size / 2 - icon_height / 2),
+                    int(size / 2 + icon_width / 2),
+                    int(size / 2 + icon_height / 2),
+                ),
+                mask=icon if icon.mode == "RGBA" else None,
+            )
+        descriptor, path = TempFile.create(".png")
+        with fdopen(descriptor, "wb") as f:
+            image.save(f, format="PNG")
+        return path

hibiapi/api/sauce/__init__.py

@@ -0,0 +1,4 @@
+# flake8:noqa:F401
+from .api import DeduplicateType, HostUrl, SauceEndpoint, UploadFileIO
+from .constants import SauceConstants
+from .net import NetRequest

hibiapi/api/sauce/api.py

@@ -0,0 +1,140 @@
+import random
+from enum import IntEnum
+from io import BytesIO
+from typing import Any, Optional, overload
+
+from httpx import HTTPError
+
+from hibiapi.api.sauce.constants import SauceConstants
+from hibiapi.utils.decorators import enum_auto_doc
+from hibiapi.utils.exceptions import ClientSideException
+from hibiapi.utils.net import catch_network_error
+from hibiapi.utils.routing import BaseEndpoint, BaseHostUrl
+
+
+class UnavailableSourceException(ClientSideException):
+    code = 422
+    detail = "given image is not avaliable to fetch"
+
+
+class ImageSourceOversizedException(UnavailableSourceException):
+    code = 413
+    detail = (
+        "given image size is rather than maximum limit "
+        f"{SauceConstants.IMAGE_MAXIMUM_SIZE} bytes"
+    )
+
+
+class HostUrl(BaseHostUrl):
+    allowed_hosts = SauceConstants.IMAGE_ALLOWED_HOST
+
+
+class UploadFileIO(BytesIO):
+    @classmethod
+    def __get_validators__(cls):
+        yield cls.validate
+
+    @classmethod
+    def validate(cls, v: Any) -> BytesIO:
+        if not isinstance(v, BytesIO):
+            raise ValueError(f"Expected UploadFile, received: {type(v)}")
+        return v
+
+
+@enum_auto_doc
+class DeduplicateType(IntEnum):
+    DISABLED = 0
+    """no result deduplicating"""
+    IDENTIFIER = 1
+    """consolidate search results and deduplicate by item identifier"""
+    ALL = 2
+    """all implemented deduplicate methods such as by series name"""
+
+
+class SauceEndpoint(BaseEndpoint, cache_endpoints=False):
+    base = "https://saucenao.com"
+
+    async def fetch(self, host: HostUrl) -> UploadFileIO:
+        try:
+            response = await self.client.get(
+                url=host,
+                headers=SauceConstants.IMAGE_HEADERS,
+                timeout=SauceConstants.IMAGE_TIMEOUT,
+            )
+            response.raise_for_status()
+            if len(response.content) > SauceConstants.IMAGE_MAXIMUM_SIZE:
+                raise ImageSourceOversizedException
+            return UploadFileIO(response.content)
+        except HTTPError as e:
+            raise UnavailableSourceException(detail=str(e)) from e
+
+    @catch_network_error
+    async def request(
+        self, *, file: UploadFileIO, params: dict[str, Any]
+    ) -> dict[str, Any]:
+        response = await self.client.post(
+            url=self._join(
+                self.base,
+                "search.php",
+                params={
+                    **params,
+                    "api_key": random.choice(SauceConstants.API_KEY),
+                    "output_type": 2,
+                },
+            ),
+            files={"file": file},
+        )
+        if response.status_code >= 500:
+            response.raise_for_status()
+        return response.json()
+
+    @overload
+    async def search(
+        self,
+        *,
+        url: HostUrl,
+        size: int = 30,
+        deduplicate: DeduplicateType = DeduplicateType.ALL,
+        database: Optional[int] = None,
+        enabled_mask: Optional[int] = None,
+        disabled_mask: Optional[int] = None,
+    ) -> dict[str, Any]:
+        ...
+
+    @overload
+    async def search(
+        self,
+        *,
+        file: UploadFileIO,
+        size: int = 30,
+        deduplicate: DeduplicateType = DeduplicateType.ALL,
+        database: Optional[int] = None,
+        enabled_mask: Optional[int] = None,
+        disabled_mask: Optional[int] = None,
+    ) -> dict[str, Any]:
+        ...
+
+    async def search(
+        self,
+        *,
+        url: Optional[HostUrl] = None,
+        file: Optional[UploadFileIO] = None,
+        size: int = 30,
+        deduplicate: DeduplicateType = DeduplicateType.ALL,
+        database: Optional[int] = None,
+        enabled_mask: Optional[int] = None,
+        disabled_mask: Optional[int] = None,
+    ):
+        if url is not None:
+            file = await self.fetch(url)
+        assert file is not None
+        return await self.request(
+            file=file,
+            params={
+                "dbmask": enabled_mask,
+                "dbmaski": disabled_mask,
+                "db": database,
+                "numres": size,
+                "dedupe": deduplicate,
+            },
+        )

hibiapi/api/sauce/constants.py

@@ -0,0 +1,16 @@
+from typing import Any
+
+from hibiapi.utils.config import APIConfig
+
+_Config = APIConfig("sauce")
+
+
+class SauceConstants:
+    CONFIG: APIConfig = _Config
+    API_KEY: list[str] = _Config["net"]["api-key"].as_str_seq()
+    USER_AGENT: str = _Config["net"]["user-agent"].as_str()
+    PROXIES: dict[str, str] = _Config["proxy"].as_dict()
+    IMAGE_HEADERS: dict[str, Any] = _Config["image"]["headers"].as_dict()
+    IMAGE_ALLOWED_HOST: list[str] = _Config["image"]["allowed"].get(list[str])
+    IMAGE_MAXIMUM_SIZE: int = _Config["image"]["max-size"].as_number() * 1024
+    IMAGE_TIMEOUT: int = _Config["image"]["timeout"].as_number()

hibiapi/api/sauce/net.py

@@ -0,0 +1,11 @@
+from hibiapi.utils.net import BaseNetClient
+
+from .constants import SauceConstants
+
+
+class NetRequest(BaseNetClient):
+    def __init__(self):
+        super().__init__(
+            headers={"user-agent": SauceConstants.USER_AGENT},
+            proxies=SauceConstants.PROXIES,
+        )

hibiapi/api/tieba/__init__.py

@@ -0,0 +1,3 @@
+# flake8:noqa:F401
+from .api import Config, TiebaEndpoint
+from .net import NetRequest

hibiapi/api/tieba/api.py

@@ -0,0 +1,142 @@
+import hashlib
+from enum import Enum
+from random import randint
+from typing import Any, Optional
+
+from hibiapi.utils.config import APIConfig
+from hibiapi.utils.net import catch_network_error
+from hibiapi.utils.routing import BaseEndpoint, dont_route
+
+Config = APIConfig("tieba")
+
+
+class TiebaSignUtils:
+    salt = b"tiebaclient!!!"
+
+    @staticmethod
+    def random_digit(length: int) -> str:
+        return "".join(map(str, [randint(0, 9) for _ in range(length)]))
+
+    @staticmethod
+    def construct_content(params: dict[str, Any]) -> bytes:
+        # NOTE: this function used to construct form content WITHOUT urlencode
+        # Don't ask me why this is necessary, ask Tieba's programmers instead
+        return b"&".join(
+            map(
+                lambda k, v: (
+                    k.encode()
+                    + b"="
+                    + str(v.value if isinstance(v, Enum) else v).encode()
+                ),
+                params.keys(),
+                params.values(),
+            )
+        )
+
+    @classmethod
+    def sign(cls, params: dict[str, Any]) -> bytes:
+        params.update(
+            {
+                "_client_id": (
+                    "wappc_" + cls.random_digit(13) + "_" + cls.random_digit(3)
+                ),
+                "_client_type": 2,
+                "_client_version": "9.9.8.32",
+                **{
+                    k.upper(): str(v).strip()
+                    for k, v in Config["net"]["params"].as_dict().items()
+                    if v
+                },
+            }
+        )
+        params = {k: params[k] for k in sorted(params.keys())}
+        params["sign"] = (
+            hashlib.md5(cls.construct_content(params).replace(b"&", b"") + cls.salt)
+            .hexdigest()
+            .upper()
+        )
+        return cls.construct_content(params)
+
+
+class TiebaEndpoint(BaseEndpoint):
+    base = "http://c.tieba.baidu.com"
+
+    @dont_route
+    @catch_network_error
+    async def request(
+        self, endpoint: str, *, params: Optional[dict[str, Any]] = None
+    ) -> dict[str, Any]:
+        response = await self.client.post(
+            url=self._join(self.base, endpoint, {}),
+            content=TiebaSignUtils.sign(params or {}),
+        )
+        response.raise_for_status()
+        return response.json()
+
+    async def post_list(self, *, name: str, page: int = 1, size: int = 50):
+        return await self.request(
+            "c/f/frs/page",
+            params={
+                "kw": name,
+                "pn": page,
+                "rn": size,
+            },
+        )
+
+    async def post_detail(
+        self,
+        *,
+        tid: int,
+        page: int = 1,
+        size: int = 50,
+        reversed: bool = False,
+    ):
+        return await self.request(
+            "c/f/pb/page",
+            params={
+                **({"last": 1, "r": 1} if reversed else {}),
+                "kz": tid,
+                "pn": page,
+                "rn": size,
+            },
+        )
+
+    async def subpost_detail(
+        self,
+        *,
+        tid: int,
+        pid: int,
+        page: int = 1,
+        size: int = 50,
+    ):
+        return await self.request(
+            "c/f/pb/floor",
+            params={
+                "kz": tid,
+                "pid": pid,
+                "pn": page,
+                "rn": size,
+            },
+        )
+
+    async def user_profile(self, *, uid: int):
+        return await self.request(
+            "c/u/user/profile",
+            params={
+                "uid": uid,
+                "need_post_count": 1,
+                "has_plist": 1,
+            },
+        )
+
+    async def user_subscribed(
+        self, *, uid: int, page: int = 1
+    ):  # XXX This API required user login!
+        return await self.request(
+            "c/f/forum/like",
+            params={
+                "is_guest": 0,
+                "uid": uid,
+                "page_no": page,
+            },
+        )

hibiapi/api/tieba/net.py

@@ -0,0 +1,5 @@
+from hibiapi.utils.net import BaseNetClient
+
+
+class NetRequest(BaseNetClient):
+    pass

hibiapi/api/wallpaper/__init__.py

@@ -0,0 +1,3 @@
+# flake8:noqa:F401
+from .api import Config, WallpaperCategoryType, WallpaperEndpoint, WallpaperOrderType
+from .net import NetRequest

hibiapi/api/wallpaper/api.py

@@ -0,0 +1,125 @@
+from datetime import timedelta
+from enum import Enum
+from typing import Any, Optional
+
+from hibiapi.utils.cache import cache_config
+from hibiapi.utils.config import APIConfig
+from hibiapi.utils.decorators import enum_auto_doc
+from hibiapi.utils.net import catch_network_error
+from hibiapi.utils.routing import BaseEndpoint, dont_route
+
+Config = APIConfig("wallpaper")
+
+
+@enum_auto_doc
+class WallpaperCategoryType(str, Enum):
+    """壁纸分类"""
+
+    girl = "girl"
+    """女生"""
+    animal = "animal"
+    """动物"""
+    landscape = "landscape"
+    """自然"""
+    anime = "anime"
+    """二次元"""
+    drawn = "drawn"
+    """手绘"""
+    mechanics = "mechanics"
+    """机械"""
+    boy = "boy"
+    """男生"""
+    game = "game"
+    """游戏"""
+    text = "text"
+    """文字"""
+
+
+CATEGORY: dict[WallpaperCategoryType, str] = {
+    WallpaperCategoryType.girl: "4e4d610cdf714d2966000000",
+    WallpaperCategoryType.animal: "4e4d610cdf714d2966000001",
+    WallpaperCategoryType.landscape: "4e4d610cdf714d2966000002",
+    WallpaperCategoryType.anime: "4e4d610cdf714d2966000003",
+    WallpaperCategoryType.drawn: "4e4d610cdf714d2966000004",
+    WallpaperCategoryType.mechanics: "4e4d610cdf714d2966000005",
+    WallpaperCategoryType.boy: "4e4d610cdf714d2966000006",
+    WallpaperCategoryType.game: "4e4d610cdf714d2966000007",
+    WallpaperCategoryType.text: "5109e04e48d5b9364ae9ac45",
+}
+
+
+@enum_auto_doc
+class WallpaperOrderType(str, Enum):
+    """壁纸排序方式"""
+
+    hot = "hot"
+    """热门"""
+    new = "new"
+    """最新"""
+
+
+class WallpaperEndpoint(BaseEndpoint):
+    base = "http://service.aibizhi.adesk.com"
+
+    @dont_route
+    @catch_network_error
+    async def request(
+        self, endpoint: str, *, params: Optional[dict[str, Any]] = None
+    ) -> dict[str, Any]:
+
+        response = await self.client.get(
+            self._join(
+                base=WallpaperEndpoint.base,
+                endpoint=endpoint,
+                params=params or {},
+            )
+        )
+        return response.json()
+
+    # 壁纸有防盗链token, 不建议长时间缓存
+    @cache_config(ttl=timedelta(hours=2))
+    async def wallpaper(
+        self,
+        *,
+        category: WallpaperCategoryType,
+        limit: int = 20,
+        skip: int = 0,
+        adult: bool = True,
+        order: WallpaperOrderType = WallpaperOrderType.hot,
+    ):
+
+        return await self.request(
+            "v1/wallpaper/category/{category}/wallpaper",
+            params={
+                "limit": limit,
+                "skip": skip,
+                "adult": adult,
+                "order": order,
+                "first": 0,
+                "category": CATEGORY[category],
+            },
+        )
+
+    # 壁纸有防盗链token, 不建议长时间缓存
+    @cache_config(ttl=timedelta(hours=2))
+    async def vertical(
+        self,
+        *,
+        category: WallpaperCategoryType,
+        limit: int = 20,
+        skip: int = 0,
+        adult: bool = True,
+        order: WallpaperOrderType = WallpaperOrderType.hot,
+    ):
+
+        return await self.request(
+            "v1/vertical/category/{category}/vertical",
+            params={
+                "limit": limit,
+                "skip": skip,
+                "adult": adult,
+                "order": order,
+                "first": 0,
+                "category": CATEGORY[category],
+            },
+        )

hibiapi/api/wallpaper/constants.py

@@ -0,0 +1,8 @@
+from hibiapi.utils.config import APIConfig
+
+_CONFIG = APIConfig("wallpaper")
+
+
+class WallpaperConstants:
+    CONFIG: APIConfig = _CONFIG
+    USER_AGENT: str = _CONFIG["net"]["user-agent"].as_str()

hibiapi/api/wallpaper/net.py

@@ -0,0 +1,8 @@
+from hibiapi.utils.net import BaseNetClient
+
+from .constants import WallpaperConstants
+
+
+class NetRequest(BaseNetClient):
+    def __init__(self):
+        super().__init__(headers={"user-agent": WallpaperConstants.USER_AGENT})

hibiapi/app/__init__.py

@@ -0,0 +1,4 @@
+# flake8:noqa:F401
+from . import application, handlers, middlewares
+
+app = application.app

hibiapi/app/application.py

@@ -0,0 +1,170 @@
+import asyncio
+import re
+from contextlib import asynccontextmanager
+from ipaddress import ip_address
+from secrets import compare_digest
+from typing import Annotated
+
+import sentry_sdk
+from fastapi import Depends, FastAPI, Request, Response
+from fastapi.responses import RedirectResponse
+from fastapi.security import HTTPBasic, HTTPBasicCredentials
+from fastapi.staticfiles import StaticFiles
+from pydantic import BaseModel
+from sentry_sdk.integrations.logging import LoggingIntegration
+
+from hibiapi import __version__
+from hibiapi.app.routes import router as ImplRouter
+from hibiapi.utils.cache import cache
+from hibiapi.utils.config import Config
+from hibiapi.utils.exceptions import ClientSideException, RateLimitReachedException
+from hibiapi.utils.log import logger
+from hibiapi.utils.net import BaseNetClient
+from hibiapi.utils.temp import TempFile
+
+DESCRIPTION = (
+    """
+**A program that implements easy-to-use APIs for a variety of commonly used sites**
+
+- *Documents*:
+    - [Redoc](/docs) (Easier to read and more beautiful)
+    - [Swagger UI](/docs/test) (Integrated interactive testing function)
+
+Project: [mixmoe/HibiAPI](https://github.com/mixmoe/HibiAPI)
+
+"""
+    + Config["content"]["slogan"].as_str().strip()
+).strip()
+
+
+if Config["log"]["sentry"]["enabled"].as_bool():
+    sentry_sdk.init(
+        dsn=Config["log"]["sentry"]["dsn"].as_str(),
+        send_default_pii=Config["log"]["sentry"]["pii"].as_bool(),
+        integrations=[LoggingIntegration(level=None, event_level=None)],
+        traces_sample_rate=Config["log"]["sentry"]["sample"].get(float),
+    )
+else:
+    sentry_sdk.init()
+
+
+class AuthorizationModel(BaseModel):
+    username: str
+    password: str
+
+
+AUTHORIZATION_ENABLED = Config["authorization"]["enabled"].as_bool()
+AUTHORIZATION_ALLOWED = Config["authorization"]["allowed"].get(list[AuthorizationModel])
+
+security = HTTPBasic()
+
+
+async def basic_authorization_depend(
+    credentials: Annotated[HTTPBasicCredentials, Depends(security)],
+):
+    # NOTE: We use `compare_digest` to avoid timing attacks.
+    # Ref: https://fastapi.tiangolo.com/advanced/security/http-basic-auth/
+    for allowed in AUTHORIZATION_ALLOWED:
+        if compare_digest(credentials.username, allowed.username) and compare_digest(
+            credentials.password, allowed.password
+        ):
+            return credentials.username, credentials.password
+    raise ClientSideException(
+        f"Invalid credentials for user {credentials.username!r}",
+        status_code=401,
+        headers={"WWW-Authenticate": "Basic"},
+    )
+
+
+RATE_LIMIT_ENABLED = Config["limit"]["enabled"].as_bool()
+RATE_LIMIT_MAX = Config["limit"]["max"].as_number()
+RATE_LIMIT_INTERVAL = Config["limit"]["interval"].as_number()
+
+
+async def rate_limit_depend(request: Request):
+    if not request.client:
+        return
+
+    try:
+        client_ip = ip_address(request.client.host)
+        client_ip_hex = client_ip.packed.hex()
+        limit_key = f"rate_limit:IPv{client_ip.version}-{client_ip_hex:x}"
+    except ValueError:
+        limit_key = f"rate_limit:fallback-{request.client.host}"
+
+    request_count = await cache.incr(limit_key)
+    if request_count <= 1:
+        await cache.expire(limit_key, timeout=RATE_LIMIT_INTERVAL)
+    elif request_count > RATE_LIMIT_MAX:
+        limit_remain: int = await cache.get_expire(limit_key)
+        raise RateLimitReachedException(headers={"Retry-After": limit_remain})
+
+    return
+
+
+async def flush_sentry():
+    client = sentry_sdk.Hub.current.client
+    if client is not None:
+        client.close()
+    sentry_sdk.flush()
+    logger.debug("Sentry client has been closed")
+
+
+async def cleanup_clients():
+    opened_clients = [
+        client for client in BaseNetClient.clients if not client.is_closed
+    ]
+    if opened_clients:
+        await asyncio.gather(
+            *map(lambda client: client.aclose(), opened_clients),
+            return_exceptions=True,
+        )
+    logger.debug(f"Cleaned <r>{len(opened_clients)}</r> unclosed HTTP clients")
+
+
+@asynccontextmanager
+async def fastapi_lifespan(app: FastAPI):
+    yield
+    await asyncio.gather(cleanup_clients(), flush_sentry())
+
+
+app = FastAPI(
+    title="HibiAPI",
+    version=__version__,
+    description=DESCRIPTION,
+    docs_url="/docs/test",
+    redoc_url="/docs",
+    lifespan=fastapi_lifespan,
+)
+app.include_router(
+    ImplRouter,
+    prefix="/api",
+    dependencies=(
+        ([Depends(basic_authorization_depend)] if AUTHORIZATION_ENABLED else [])
+        + ([Depends(rate_limit_depend)] if RATE_LIMIT_ENABLED else [])
+    ),
+)
+app.mount("/temp", StaticFiles(directory=TempFile.path, check_dir=False))
+
+
+@app.get("/", include_in_schema=False)
+async def redirect():
+    return Response(status_code=302, headers={"Location": "/docs"})
+
+
+@app.get("/robots.txt", include_in_schema=False)
+async def robots():
+    content = Config["content"]["robots"].as_str().strip()
+    return Response(content, status_code=200)
+
+
+@app.middleware("http")
+async def redirect_workaround_middleware(request: Request, call_next):
+    """Temporary redirection workaround for #12"""
+    if matched := re.match(
+        r"^/(qrcode|pixiv|netease|bilibili)/(\w*)$", request.url.path
+    ):
+        service, path = matched.groups()
+        redirect_url = request.url.replace(path=f"/api/{service}/{path}")
+        return RedirectResponse(redirect_url, status_code=301)
+    return await call_next(request)

hibiapi/app/handlers.py

@@ -0,0 +1,59 @@
+from fastapi import Request, Response
+from fastapi.exceptions import HTTPException as FastAPIHTTPException
+from fastapi.exceptions import RequestValidationError as FastAPIValidationError
+from pydantic.error_wrappers import ValidationError as PydanticValidationError
+from starlette.exceptions import HTTPException as StarletteHTTPException
+
+from hibiapi.utils import exceptions
+from hibiapi.utils.log import logger
+
+from .application import app
+
+
+@app.exception_handler(exceptions.BaseServerException)
+async def exception_handler(
+    request: Request,
+    exc: exceptions.BaseServerException,
+) -> Response:
+    if isinstance(exc, exceptions.UncaughtException):
+        logger.opt(exception=exc).exception(f"Uncaught exception raised {exc.data=}:")
+
+    exc.data.url = str(request.url)  # type:ignore
+    return Response(
+        content=exc.data.json(),
+        status_code=exc.data.code,
+        headers=exc.data.headers,
+        media_type="application/json",
+    )
+
+
+@app.exception_handler(StarletteHTTPException)
+async def override_handler(
+    request: Request,
+    exc: StarletteHTTPException,
+):
+    return await exception_handler(
+        request,
+        exceptions.BaseHTTPException(
+            exc.detail,
+            code=exc.status_code,
+            headers={} if not isinstance(exc, FastAPIHTTPException) else exc.headers,
+        ),
+    )
+
+
+@app.exception_handler(AssertionError)
+async def assertion_handler(request: Request, exc: AssertionError):
+    return await exception_handler(
+        request,
+        exceptions.ClientSideException(detail=f"Assertion: {exc}"),
+    )
+
+
+@app.exception_handler(FastAPIValidationError)
+@app.exception_handler(PydanticValidationError)
+async def validation_handler(request: Request, exc: PydanticValidationError):
+    return await exception_handler(
+        request,
+        exceptions.ValidationException(detail=str(exc), validation=exc.errors()),
+    )

hibiapi/app/middlewares.py

@@ -0,0 +1,97 @@
+from collections.abc import Awaitable
+from datetime import datetime
+from typing import Callable
+
+from fastapi import Request, Response
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.middleware.gzip import GZipMiddleware
+from fastapi.middleware.trustedhost import TrustedHostMiddleware
+from sentry_sdk.integrations.asgi import SentryAsgiMiddleware
+from sentry_sdk.integrations.httpx import HttpxIntegration
+from starlette.datastructures import MutableHeaders
+
+from hibiapi.utils.config import Config
+from hibiapi.utils.exceptions import BaseServerException, UncaughtException
+from hibiapi.utils.log import LoguruHandler, logger
+from hibiapi.utils.routing import request_headers, response_headers
+
+from .application import app
+from .handlers import exception_handler
+
+RequestHandler = Callable[[Request], Awaitable[Response]]
+
+
+if Config["server"]["gzip"].as_bool():
+    app.add_middleware(GZipMiddleware)
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=Config["server"]["cors"]["origins"].get(list[str]),
+    allow_credentials=Config["server"]["cors"]["credentials"].as_bool(),
+    allow_methods=Config["server"]["cors"]["methods"].get(list[str]),
+    allow_headers=Config["server"]["cors"]["headers"].get(list[str]),
+)
+app.add_middleware(
+    TrustedHostMiddleware,
+    allowed_hosts=Config["server"]["allowed"].get(list[str]),
+)
+app.add_middleware(SentryAsgiMiddleware)
+
+HttpxIntegration.setup_once()
+
+
+@app.middleware("http")
+async def request_logger(request: Request, call_next: RequestHandler) -> Response:
+    start_time = datetime.now()
+    host, port = request.client or (None, None)
+    response = await call_next(request)
+    process_time = (datetime.now() - start_time).total_seconds() * 1000
+    response_headers.get().setdefault("X-Process-Time", f"{process_time:.3f}")
+    bg, fg = (
+        ("green", "red")
+        if response.status_code < 400
+        else ("yellow", "blue")
+        if response.status_code < 500
+        else ("red", "green")
+    )
+    status_code, method = response.status_code, request.method.upper()
+    user_agent = (
+        LoguruHandler.escape_tag(request.headers["user-agent"])
+        if "user-agent" in request.headers
+        else "<d>Unknown</d>"
+    )
+    logger.info(
+        f"<m><b>{host}</b>:{port}</m>"
+        f" | <{bg.upper()}><b><{fg}>{method}</{fg}></b></{bg.upper()}>"
+        f" | <n><b>{str(request.url)!r}</b></n>"
+        f" | <c>{process_time:.3f}ms</c>"
+        f" | <e>{user_agent}</e>"
+        f" | <b><{bg}>{status_code}</{bg}></b>"
+    )
+    return response
+
+
+@app.middleware("http")
+async def contextvar_setter(request: Request, call_next: RequestHandler):
+    request_headers.set(request.headers)
+    response_headers.set(MutableHeaders())
+    response = await call_next(request)
+    response.headers.update({**response_headers.get()})
+    return response
+
+
+@app.middleware("http")
+async def uncaught_exception_handler(
+    request: Request, call_next: RequestHandler
+) -> Response:
+    try:
+        response = await call_next(request)
+    except Exception as error:
+        response = await exception_handler(
+            request,
+            exc=(
+                error
+                if isinstance(error, BaseServerException)
+                else UncaughtException.with_exception(error)
+            ),
+        )
+    return response

hibiapi/app/routes/__init__.py

@@ -0,0 +1,52 @@
+from typing import Protocol, cast
+
+from hibiapi.app.routes import (
+    bika,
+    bilibili,
+    netease,
+    pixiv,
+    qrcode,
+    sauce,
+    tieba,
+    wallpaper,
+)
+from hibiapi.utils.config import APIConfig
+from hibiapi.utils.exceptions import ExceptionReturn
+from hibiapi.utils.log import logger
+from hibiapi.utils.routing import SlashRouter
+
+router = SlashRouter(
+    responses={
+        code: {
+            "model": ExceptionReturn,
+        }
+        for code in (400, 422, 500, 502)
+    }
+)
+
+
+class RouteInterface(Protocol):
+    router: SlashRouter
+    __mount__: str
+    __config__: APIConfig
+
+
+modules = cast(
+    list[RouteInterface],
+    [bilibili, netease, pixiv, qrcode, sauce, tieba, wallpaper, bika],
+)
+
+for module in modules:
+    mount = (
+        mount_point
+        if (mount_point := module.__mount__).startswith("/")
+        else f"/{mount_point}"
+    )
+
+    if not module.__config__["enabled"].as_bool():
+        logger.warning(
+            f"API Route <y><b>{mount}</b></y> has been "
+            "<r><b>disabled</b></r> in config."
+        )
+        continue
+    router.include_router(module.router, prefix=mount)

hibiapi/app/routes/bika.py

@@ -0,0 +1,36 @@
+from typing import Annotated
+
+from fastapi import Depends, Header
+
+from hibiapi.api.bika import (
+    BikaConstants,
+    BikaEndpoints,
+    BikaLogin,
+    ImageQuality,
+    NetRequest,
+)
+from hibiapi.utils.log import logger
+from hibiapi.utils.routing import EndpointRouter
+
+try:
+    BikaConstants.CONFIG["account"].get(BikaLogin)
+except Exception as e:
+    logger.warning(f"Bika account misconfigured: {e}")
+    BikaConstants.CONFIG["enabled"].set(False)
+
+
+async def x_image_quality(
+    x_image_quality: Annotated[ImageQuality, Header()] = ImageQuality.medium,
+):
+    if x_image_quality is None:
+        return BikaConstants.CONFIG["image_quality"].get(ImageQuality)
+    return x_image_quality
+
+
+__mount__, __config__ = "bika", BikaConstants.CONFIG
+router = EndpointRouter(tags=["Bika"], dependencies=[Depends(x_image_quality)])
+
+BikaAPIRoot = NetRequest()
+
+
+router.include_endpoint(BikaEndpoints, BikaAPIRoot)

hibiapi/app/routes/bilibili/__init__.py

@@ -0,0 +1,10 @@
+from hibiapi.api.bilibili import BilibiliConstants
+from hibiapi.app.routes.bilibili.v2 import router as RouterV2
+from hibiapi.app.routes.bilibili.v3 import router as RouterV3
+from hibiapi.utils.routing import SlashRouter
+
+__mount__, __config__ = "bilibili", BilibiliConstants.CONFIG
+
+router = SlashRouter()
+router.include_router(RouterV2, prefix="/v2")
+router.include_router(RouterV3, prefix="/v3")

hibiapi/app/routes/bilibili/v2.py

@@ -0,0 +1,6 @@
+from hibiapi.api.bilibili.api import BilibiliEndpointV2
+from hibiapi.api.bilibili.net import NetRequest
+from hibiapi.utils.routing import EndpointRouter
+
+router = EndpointRouter(tags=["Bilibili V2"])
+router.include_endpoint(BilibiliEndpointV2, NetRequest())

hibiapi/app/routes/bilibili/v3.py

@@ -0,0 +1,5 @@
+from hibiapi.api.bilibili import BilibiliEndpointV3, NetRequest
+from hibiapi.utils.routing import EndpointRouter
+
+router = EndpointRouter(tags=["Bilibili V3"])
+router.include_endpoint(BilibiliEndpointV3, NetRequest())

hibiapi/app/routes/netease.py

@@ -0,0 +1,7 @@
+from hibiapi.api.netease import NeteaseConstants, NeteaseEndpoint, NetRequest
+from hibiapi.utils.routing import EndpointRouter
+
+__mount__, __config__ = "netease", NeteaseConstants.CONFIG
+
+router = EndpointRouter(tags=["Netease"])
+router.include_endpoint(NeteaseEndpoint, NetRequest())

hibiapi/app/routes/pixiv.py

@@ -0,0 +1,26 @@
+from typing import Optional
+
+from fastapi import Depends, Header
+
+from hibiapi.api.pixiv import NetRequest, PixivConstants, PixivEndpoints
+from hibiapi.utils.log import logger
+from hibiapi.utils.routing import EndpointRouter
+
+if not (refresh_tokens := PixivConstants.CONFIG["account"]["token"].as_str_seq()):
+    logger.warning("Pixiv API token is not set, pixiv endpoint will be unavailable.")
+    PixivConstants.CONFIG["enabled"].set(False)
+
+
+async def accept_language(
+    accept_language: Optional[str] = Header(
+        None,
+        description="Accepted tag translation language",
+    )
+):
+    return accept_language
+
+
+__mount__, __config__ = "pixiv", PixivConstants.CONFIG
+
+router = EndpointRouter(tags=["Pixiv"], dependencies=[Depends(accept_language)])
+router.include_endpoint(PixivEndpoints, api_root := NetRequest(refresh_tokens))

hibiapi/app/routes/qrcode.py

@@ -0,0 +1,76 @@
+from typing import Optional
+
+from fastapi import Request, Response
+from pydantic.color import Color
+
+from hibiapi.api.qrcode import (
+    COLOR_BLACK,
+    COLOR_WHITE,
+    Config,
+    HostUrl,
+    QRCodeLevel,
+    QRInfo,
+    ReturnEncode,
+)
+from hibiapi.utils.routing import SlashRouter
+from hibiapi.utils.temp import TempFile
+
+QR_CALLBACK_TEMPLATE = (
+    r"""function {fun}(){document.write('<img class="qrcode" src="{url}"/>');}"""
+)
+
+__mount__, __config__ = "qrcode", Config
+router = SlashRouter(tags=["QRCode"])
+
+
+@router.get(
+    "/",
+    responses={
+        200: {
+            "content": {"image/png": {}, "text/javascript": {}, "application/json": {}},
+            "description": "Avaliable to return an javascript, image or json.",
+        }
+    },
+    response_model=QRInfo,
+)
+async def qrcode_api(
+    request: Request,
+    *,
+    text: str,
+    size: int = 200,
+    logo: Optional[HostUrl] = None,
+    encode: ReturnEncode = ReturnEncode.raw,
+    level: QRCodeLevel = QRCodeLevel.MEDIUM,
+    bgcolor: Color = COLOR_BLACK,
+    fgcolor: Color = COLOR_WHITE,
+    fun: str = "qrcode",
+):
+    qr = await QRInfo.new(
+        text, size=size, logo=logo, level=level, bgcolor=bgcolor, fgcolor=fgcolor
+    )
+    qr.url = TempFile.to_url(request, qr.path)  # type:ignore
+    """function {fun}(){document.write('<img class="qrcode" src="{url}"/>');}"""
+    return (
+        qr
+        if encode == ReturnEncode.json
+        else Response(
+            content=qr.json(),
+            media_type="application/json",
+            headers={"Location": qr.url},
+            status_code=302,
+        )
+        if encode == ReturnEncode.raw
+        else Response(
+            content=f"{fun}({qr.json()})",
+            media_type="text/javascript",
+        )
+        if encode == ReturnEncode.jsc
+        else Response(
+            content="function "
+            + fun
+            + '''(){document.write('<img class="qrcode" src="'''
+            + qr.url
+            + """"/>');}""",
+            media_type="text/javascript",
+        )
+    )

hibiapi/app/routes/sauce.py

@@ -0,0 +1,120 @@
+from typing import Annotated, Optional
+
+from fastapi import Depends, File, Form
+from loguru import logger
+
+from hibiapi.api.sauce import (
+    DeduplicateType,
+    HostUrl,
+    NetRequest,
+    SauceConstants,
+    SauceEndpoint,
+    UploadFileIO,
+)
+from hibiapi.utils.routing import SlashRouter
+
+if (not SauceConstants.API_KEY) or (not all(map(str.strip, SauceConstants.API_KEY))):
+    logger.warning("Sauce API key not set, SauceNAO endpoint will be unavailable")
+    SauceConstants.CONFIG["enabled"].set(False)
+
+__mount__, __config__ = "sauce", SauceConstants.CONFIG
+router = SlashRouter(tags=["SauceNAO"])
+
+SauceAPIRoot = NetRequest()
+
+
+async def request_client():
+    async with SauceAPIRoot as client:
+        yield SauceEndpoint(client)
+
+
+@router.get("/")
+async def sauce_url(
+    endpoint: Annotated[SauceEndpoint, Depends(request_client)],
+    url: HostUrl,
+    size: int = 30,
+    deduplicate: DeduplicateType = DeduplicateType.ALL,
+    database: Optional[int] = None,
+    enabled_mask: Optional[int] = None,
+    disabled_mask: Optional[int] = None,
+):
+    """
+    ## Name: `sauce_url`
+
+    > 使用SauceNAO检索网络图片
+
+    ---
+
+    ### Required:
+
+    - ***HostUrl*** **`url`**
+        - Description: 图片URL
+
+    ---
+
+    ### Optional:
+    - ***int*** `size` = `30`
+        - Description: 搜索结果数目
+    - ***DeduplicateType*** `deduplicate` = `DeduplicateType.ALL`
+        - Description: 结果去重模式
+    - ***Optional[int]*** `database` = `None`
+        - Description: 检索的数据库ID, 999为全部检索
+    - ***Optional[int]*** `enabled_mask` = `None`
+        - Description: 启用的检索数据库
+    - ***Optional[int]*** `disabled_mask` = `None`
+        - Description: 禁用的检索数据库
+    """
+    return await endpoint.search(
+        url=url,
+        size=size,
+        deduplicate=deduplicate,
+        database=database,
+        enabled_mask=enabled_mask,
+        disabled_mask=disabled_mask,
+    )
+
+
+@router.post("/")
+async def sauce_form(
+    endpoint: Annotated[SauceEndpoint, Depends(request_client)],
+    file: bytes = File(..., max_length=SauceConstants.IMAGE_MAXIMUM_SIZE),
+    size: int = Form(30),
+    deduplicate: Annotated[DeduplicateType, Form()] = DeduplicateType.ALL,
+    database: Optional[int] = Form(None),
+    enabled_mask: Optional[int] = Form(None),
+    disabled_mask: Optional[int] = Form(None),
+):
+    """
+    ## Name: `sauce_form`
+
+    > 使用SauceNAO检索表单上传图片
+
+    ---
+
+    ### Required:
+    - ***bytes*** `file`
+        - Description: 上传的图片
+
+    ---
+
+    ### Optional:
+    - ***int*** `size` = `30`
+        - Description: 搜索结果数目
+    - ***DeduplicateType*** `deduplicate` = `DeduplicateType.ALL`
+        - Description: 结果去重模式
+    - ***Optional[int]*** `database` = `None`
+        - Description: 检索的数据库ID, 999为全部检索
+    - ***Optional[int]*** `enabled_mask` = `None`
+        - Description: 启用的检索数据库
+    - ***Optional[int]*** `disabled_mask` = `None`
+        - Description: 禁用的检索数据库
+
+    """
+    return await endpoint.search(
+        file=UploadFileIO(file),
+        size=size,
+        deduplicate=deduplicate,
+        database=database,
+        disabled_mask=disabled_mask,
+        enabled_mask=enabled_mask,
+    )