Pine
985b85cb4e
- infrastructure/oss.py 重写：服务端上传/预签名上下行/CDN鉴权直链(方式A)/resolve_url·to_object_path 出入口规范 - config.py 新增 OSS_REGION/OSS_CDN_BASE_URL/OSS_CDN_AUTH_KEY - media_upload 统一走 oss 单例，返回 CDN 直链；dispatcher /oss 前缀路由 - 管理端用户列表头像出口 CDN 化
222 lines
9.4 KiB
Python
222 lines
9.4 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""基础设施层 · OSS 对象存储统一封装（aioboto3，兼容阿里云 OSS / MinIO / S3）。
|
||
|
||
全项目所有文件上传/下载的唯一入口，供各端口（培训/园区/运营端/平台）调用：
|
||
- upload() 服务端中转上传（表单 multipart 走后端）
|
||
- presigned_put() 生成临时上传链接（前端直传，绕过后端带宽）
|
||
- presigned_get() 生成临时下载链接
|
||
- object_url() 对象的稳定相对路径 /oss/<key>（入库用；经下载路由跳转预签名 URL）
|
||
|
||
未配置 OSS_ENDPOINT 时降级本地 serverdata/uploads，保证开发环境无 OSS 也可用。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from typing import Optional
|
||
|
||
from .. import config
|
||
|
||
|
||
class OSS:
|
||
"""对象存储：上传 / 预签名上下行 URL。配置了 OSS_ENDPOINT 才启用，否则落本地上传目录。"""
|
||
|
||
def __init__(self) -> None:
|
||
self._client: Optional[object] = None
|
||
self.enabled = bool(config.OSS_ENDPOINT)
|
||
self.bucket = config.OSS_BUCKET
|
||
self.local_dir = config.SERVERDATA_DIR / "uploads"
|
||
self.local_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
async def _get_client(self):
|
||
if self._client is None:
|
||
import aioboto3
|
||
from botocore.config import Config
|
||
|
||
session = aioboto3.Session(
|
||
aws_access_key_id=config.OSS_ACCESS_KEY,
|
||
aws_secret_access_key=config.OSS_SECRET_KEY,
|
||
)
|
||
# aioboto3 的 client 是异步上下文管理器，需 enter 后长期复用；
|
||
# 阿里云 OSS 要求 virtual-hosted style；不支持 PutObject CRC32 trailer（botocore 新版默认）
|
||
self._client = await session.client(
|
||
"s3",
|
||
endpoint_url=config.OSS_ENDPOINT,
|
||
region_name=config.OSS_REGION or "cn-heyuan",
|
||
config=Config(
|
||
s3={
|
||
"addressing_style": "virtual",
|
||
"payload_signing_enabled": True,
|
||
},
|
||
signature_version="s3v4",
|
||
request_checksum_calculation="when_required",
|
||
response_checksum_validation="when_required",
|
||
),
|
||
).__aenter__()
|
||
return self._client
|
||
|
||
def clean_key(self, key: str) -> str:
|
||
"""规整对象 key：去首尾斜杠、拒目录穿越。"""
|
||
key = (key or "").strip().strip("/")
|
||
if not key or ".." in key.split("/"):
|
||
raise ValueError(f"非法对象 key：{key!r}")
|
||
return key
|
||
|
||
# ---- 上传 ----
|
||
|
||
async def upload(self, key: str, data: bytes, content_type: str = "application/octet-stream") -> str:
|
||
"""服务端上传对象，成功返回稳定相对路径 /oss/<key>（入库/回传客户端统一用这个）。"""
|
||
key = self.clean_key(key)
|
||
if self.enabled:
|
||
client = await self._get_client()
|
||
await client.put_object(Bucket=self.bucket, Key=key, Body=data, ContentType=content_type)
|
||
return f"/oss/{key}"
|
||
dest = self.local_dir / key
|
||
dest.parent.mkdir(parents=True, exist_ok=True)
|
||
dest.write_bytes(data)
|
||
return f"/uploads/{key}"
|
||
|
||
# ---- 下载 / 访问 ----
|
||
|
||
def object_url(self, key: str) -> str:
|
||
"""对象的稳定相对路径（下载经 GET /oss/<key> 跳转预签名 URL，桶私有也可读）。"""
|
||
return f"/oss/{self.clean_key(key)}"
|
||
|
||
async def presigned_get(self, key: str, expires: int = 3600) -> str:
|
||
"""临时下载链接（预签名 GET，走 OSS 源站）。"""
|
||
key = self.clean_key(key)
|
||
if not self.enabled:
|
||
return f"/uploads/{key}"
|
||
client = await self._get_client()
|
||
return await client.generate_presigned_url(
|
||
"get_object", Params={"Bucket": self.bucket, "Key": key}, ExpiresIn=expires
|
||
)
|
||
|
||
async def download_url(self, key: str, expires: int = 3600) -> str:
|
||
"""统一下载链接：配置了 CDN 加速域名则走 CDN（带鉴权），否则退回预签名直链。"""
|
||
if config.OSS_CDN_BASE_URL:
|
||
return self.cdn_url(key)
|
||
return await self.presigned_get(key, expires)
|
||
|
||
# ---- CDN URL 鉴权（阿里云「鉴权方式A」）----
|
||
|
||
# 「永久有效」实现：鉴权时间戳取 32 位最大值（2038-01），叠加控制台有效期内长期可访问
|
||
_CDN_TS_MAX = "7FFFFFFF" # 方式A 要求大写 hex
|
||
|
||
def _cdn_sign(self, key: str, ts_hex: str) -> str:
|
||
"""鉴权方式A：md5(密钥 + URI + timestamp)，URI 以 / 开头，时间戳大写 hex。"""
|
||
import hashlib
|
||
|
||
uri = f"/{key}"
|
||
return hashlib.md5(f"{config.OSS_CDN_AUTH_KEY}{uri}{ts_hex}".encode()).hexdigest()
|
||
|
||
def cdn_url(self, key: str) -> str:
|
||
"""CDN 鉴权直链（方式A）：{CDN}/{md5签名}/{时间戳}/{key}；未配密钥则裸链接。"""
|
||
key = self.clean_key(key)
|
||
base = config.OSS_CDN_BASE_URL.rstrip("/")
|
||
if not config.OSS_CDN_AUTH_KEY:
|
||
return f"{base}/{key}"
|
||
ts_hex = self._CDN_TS_MAX
|
||
return f"{base}/{self._cdn_sign(key, ts_hex)}/{ts_hex}/{key}"
|
||
|
||
def direct_url(self, key: str) -> str:
|
||
"""直链（不经服务端代理，直接返回给前端）：
|
||
CDN 加速域名（带鉴权）> OSS 公网直链 > 本地 /uploads。"""
|
||
key = self.clean_key(key)
|
||
if config.OSS_CDN_BASE_URL:
|
||
return self.cdn_url(key)
|
||
if self.enabled:
|
||
return self.public_url(key)
|
||
return f"/uploads/{key}"
|
||
|
||
async def presigned_put(self, key: str, expires: int = 600, content_type: str = "application/octet-stream") -> str:
|
||
"""临时上传链接（预签名 PUT，供前端直传）。"""
|
||
key = self.clean_key(key)
|
||
if not self.enabled:
|
||
raise RuntimeError("OSS 未配置，无法生成直传链接")
|
||
client = await self._get_client()
|
||
return await client.generate_presigned_url(
|
||
"put_object",
|
||
Params={"Bucket": self.bucket, "Key": key, "ContentType": content_type},
|
||
ExpiresIn=expires,
|
||
)
|
||
|
||
def public_url(self, key: str) -> str:
|
||
"""对象的公网直链（仅公共读桶可直接访问；私有桶 403）。"""
|
||
endpoint = (config.OSS_ENDPOINT or "").rstrip("/")
|
||
return f"{endpoint}/{self.bucket}/{self.clean_key(key)}"
|
||
|
||
|
||
# 模块级单例：全项目统一通过 `from ..infrastructure.oss import oss` 使用
|
||
oss = OSS()
|
||
|
||
|
||
def resolve_url(u: str | None) -> str:
|
||
"""资源 URL 统一出口（读取时把对象路径生成为可访问 URL）：
|
||
1) 对象路径（/oss/<key>、/uploads/<f>）补全为站点公开绝对 URL；
|
||
2) 本站 CDN 裸链（历史入库的无鉴权链接）自动重签为鉴权直链；
|
||
3) 其余完整 URL 原样返回。"""
|
||
import os as _os
|
||
|
||
if not u:
|
||
return ""
|
||
u = str(u).strip()
|
||
base = _os.environ.get("PUBLIC_BASE", "https://opc.pinesound.cn").rstrip("/")
|
||
if not (u.startswith("http://") or u.startswith("https://")):
|
||
if u.startswith("/oss/"):
|
||
# OSS 对象路径 → 直接生成 CDN 鉴权直链（未启用 OSS 时回退站点路径）
|
||
if oss.enabled and config.OSS_CDN_BASE_URL:
|
||
try:
|
||
return oss.cdn_url(u[len("/oss/"):])
|
||
except ValueError:
|
||
return u
|
||
return base + u
|
||
return base + ("" if u.startswith("/") else "/") + u
|
||
# 历史 CDN 裸链（无鉴权段）→ 提取 key 重签（鉴权上线前的存量数据）
|
||
cdn = config.OSS_CDN_BASE_URL
|
||
if cdn and config.OSS_CDN_AUTH_KEY and u.startswith(f"{cdn}/") and "/7FFFFFFF/" not in u:
|
||
key = u[len(cdn) + 1:].split("?")[0]
|
||
try:
|
||
return oss.cdn_url(key)
|
||
except ValueError:
|
||
return u
|
||
return u
|
||
|
||
|
||
# 本站资源 URL 前缀（用于入库归一化：完整 URL → 对象路径）
|
||
_OSS_PUBLIC_PREFIXES = (
|
||
lambda: (config.OSS_CDN_BASE_URL or "") + "/",
|
||
lambda: (config.OSS_ENDPOINT or "").rstrip("/") + "/" + config.OSS_BUCKET + "/",
|
||
)
|
||
|
||
|
||
def to_object_path(u: str | None) -> str:
|
||
"""入库归一化（写入规范）：本站资源完整 URL → 对象相对路径（/oss/<key>、/uploads/<f>）。
|
||
|
||
换 OSS/CDN 供应商时只改本层配置，数据库无需迁移。
|
||
外部 URL（如微信头像）与本站已有相对路径原样保留。
|
||
"""
|
||
import os as _os
|
||
|
||
if not u:
|
||
return ""
|
||
u = str(u).strip()
|
||
if not (u.startswith("http://") or u.startswith("https://")):
|
||
return u # 已是路径/外链原样
|
||
for prefix_fn in _OSS_PUBLIC_PREFIXES:
|
||
prefix = prefix_fn()
|
||
if prefix and u.startswith(prefix):
|
||
rest = u[len(prefix):].split("?")[0].strip("/")
|
||
parts = rest.split("/")
|
||
# CDN 鉴权方式A：{sign}/{timestamp}/{key}——剥掉前两段
|
||
if len(parts) >= 3 and parts[1].upper() == "7FFFFFFF":
|
||
rest = "/".join(parts[2:])
|
||
if ".." in rest.split("/"):
|
||
return u
|
||
return f"/oss/{rest}"
|
||
# 站点自身域名下的 /oss、/uploads 路径 → 还原为相对路径
|
||
base = _os.environ.get("PUBLIC_BASE", "https://opc.pinesound.cn").rstrip("/")
|
||
if u.startswith(base + "/"):
|
||
rest = u[len(base) + 1:].split("?")[0]
|
||
if rest.startswith(("oss/", "uploads/")):
|
||
return "/" + rest
|
||
return u # 外部 URL 原样（如微信头像）
|