6bcd8c1a21
Introduces asset proxy configuration and services to enable replacement of osu! resource URLs with custom domains. Updates API endpoints and caching services to process and rewrite resource URLs when asset proxy is enabled. Adds documentation and environment variables for asset proxy setup.
93 lines
3.1 KiB
Python
93 lines
3.1 KiB
Python
"""
|
||
资源文件代理服务
|
||
提供URL替换方案：将osu!官方资源URL替换为自定义域名
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from typing import Any
|
||
from app.config import settings
|
||
from app.log import logger
|
||
|
||
|
||
class AssetProxyService:
|
||
"""资源代理服务 - 仅URL替换模式"""
|
||
|
||
def __init__(self):
|
||
# 从配置获取自定义assets域名和前缀
|
||
self.custom_asset_domain = settings.custom_asset_domain
|
||
self.asset_proxy_prefix = settings.asset_proxy_prefix
|
||
self.avatar_proxy_prefix = settings.avatar_proxy_prefix
|
||
self.beatmap_proxy_prefix = settings.beatmap_proxy_prefix
|
||
|
||
async def replace_asset_urls(self, data: Any) -> Any:
|
||
"""
|
||
递归替换数据中的osu!资源URL为自定义域名
|
||
"""
|
||
# 处理Pydantic模型
|
||
if hasattr(data, 'model_dump'):
|
||
# 转换为字典，处理后再转换回模型
|
||
data_dict = data.model_dump()
|
||
processed_dict = await self.replace_asset_urls(data_dict)
|
||
# 尝试从字典重新创建模型
|
||
try:
|
||
return data.__class__(**processed_dict)
|
||
except Exception:
|
||
# 如果重新创建失败，返回字典
|
||
return processed_dict
|
||
elif isinstance(data, dict):
|
||
result = {}
|
||
for key, value in data.items():
|
||
result[key] = await self.replace_asset_urls(value)
|
||
return result
|
||
elif isinstance(data, list):
|
||
return [await self.replace_asset_urls(item) for item in data]
|
||
elif isinstance(data, str):
|
||
# 替换各种osu!资源域名
|
||
result = data
|
||
|
||
# 替换 assets.ppy.sh (用户头像、封面、奖章等)
|
||
result = re.sub(
|
||
r"https://assets\.ppy\.sh/",
|
||
f"https://{self.asset_proxy_prefix}.{self.custom_asset_domain}/",
|
||
result
|
||
)
|
||
|
||
# 替换 b.ppy.sh 预览音频 (保持//前缀)
|
||
result = re.sub(
|
||
r"//b\.ppy\.sh/",
|
||
f"//{self.beatmap_proxy_prefix}.{self.custom_asset_domain}/",
|
||
result
|
||
)
|
||
|
||
# 替换 https://b.ppy.sh 预览音频 (转换为//前缀)
|
||
result = re.sub(
|
||
r"https://b\.ppy\.sh/",
|
||
f"//{self.beatmap_proxy_prefix}.{self.custom_asset_domain}/",
|
||
result
|
||
)
|
||
|
||
# 替换 a.ppy.sh 头像
|
||
result = re.sub(
|
||
r"https://a\.ppy\.sh/",
|
||
f"https://{self.avatar_proxy_prefix}.{self.custom_asset_domain}/",
|
||
result
|
||
)
|
||
|
||
return result
|
||
else:
|
||
return data
|
||
|
||
|
||
# 全局实例
|
||
_asset_proxy_service: AssetProxyService | None = None
|
||
|
||
|
||
def get_asset_proxy_service() -> AssetProxyService:
|
||
"""获取资源代理服务实例"""
|
||
global _asset_proxy_service
|
||
if _asset_proxy_service is None:
|
||
_asset_proxy_service = AssetProxyService()
|
||
return _asset_proxy_service
|