refactor(assets_proxy): use decorators to simplify code

app/helpers/asset_proxy_helper.py

@@ -0,0 +1,108 @@
+"""资源代理辅助方法与路由装饰器。"""
+
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable
+from functools import wraps
+import re
+from typing import Any
+
+from app.config import settings
+
+from fastapi import Response
+from pydantic import BaseModel
+
+Handler = Callable[..., Awaitable[Any]]
+
+
+def _replace_asset_urls_in_string(value: str) -> str:
+    result = value
+    custom_domain = settings.custom_asset_domain
+    asset_prefix = settings.asset_proxy_prefix
+    avatar_prefix = settings.avatar_proxy_prefix
+    beatmap_prefix = settings.beatmap_proxy_prefix
+    audio_proxy_base_url = f"{settings.server_url}api/private/audio/beatmapset"
+
+    result = re.sub(
+        r"^https://assets\.ppy\.sh/",
+        f"https://{asset_prefix}.{custom_domain}/",
+        result,
+    )
+
+    result = re.sub(
+        r"^https://b\.ppy\.sh/preview/(\d+)\\.mp3",
+        rf"{audio_proxy_base_url}/\1",
+        result,
+    )
+
+    result = re.sub(
+        r"^//b\.ppy\.sh/preview/(\d+)\\.mp3",
+        rf"{audio_proxy_base_url}/\1",
+        result,
+    )
+
+    result = re.sub(
+        r"^https://a\.ppy\.sh/",
+        f"https://{avatar_prefix}.{custom_domain}/",
+        result,
+    )
+
+    result = re.sub(
+        r"https://b\.ppy\.sh/",
+        f"https://{beatmap_prefix}.{custom_domain}/",
+        result,
+    )
+    return result
+
+
+def _replace_asset_urls_in_data(data: Any) -> Any:
+    if isinstance(data, str):
+        return _replace_asset_urls_in_string(data)
+    if isinstance(data, list):
+        return [_replace_asset_urls_in_data(item) for item in data]
+    if isinstance(data, tuple):
+        return tuple(_replace_asset_urls_in_data(item) for item in data)
+    if isinstance(data, dict):
+        return {key: _replace_asset_urls_in_data(value) for key, value in data.items()}
+    return data
+
+
+async def replace_asset_urls(data: Any) -> Any:
+    """替换数据中的 osu! 资源 URL。"""
+
+    if not settings.enable_asset_proxy:
+        return data
+
+    if hasattr(data, "model_dump"):
+        raw = data.model_dump()
+        processed = _replace_asset_urls_in_data(raw)
+        try:
+            return data.__class__(**processed)
+        except Exception:
+            return processed
+
+    if isinstance(data, (dict, list, tuple, str)):
+        return _replace_asset_urls_in_data(data)
+
+    return data
+
+
+def asset_proxy_response(func: Handler) -> Handler:
+    """装饰器：在返回响应前替换资源 URL。"""
+
+    @wraps(func)
+    async def wrapper(*args, **kwargs):
+        result = await func(*args, **kwargs)
+
+        if not settings.enable_asset_proxy:
+            return result
+
+        if isinstance(result, Response):
+            return result
+
+        if isinstance(result, BaseModel):
+            result = result.model_dump()
+
+        return _replace_asset_urls_in_data(result)
+
+    return wrapper  # type: ignore[return-value]