Files
yingqing/core/backend/apps/assets/storage.py
T
zycandClaude Opus 4.8 8ba76e5bb7 perf(core): 资产预览改 TOS 公读直链 + 通知只取首页,减刷新开销与缓存失效
- 资产 preview_url 从「逐图签发预签名 URL」改为虚拟主机式公读直链
  (https://{bucket}.{host}/{key})。桶公读已验证免签可访问;直链稳定可被浏览器/CDN
  缓存,而签名链每次都变、1h 过期反而打不到缓存。boto3 也移出序列化热路径。
- allNotifications 不再逐页翻全部(原 O(N) 随消息增长拖慢每次刷新),只取首页 100 条
  (侧边栏徽标 unread_count + 团队动态仅展示最近 6 条已足够);「共 N」改用后端真实 count。
- perf-probe:page_size 由 API 级硬闸确定性守住后,浏览器端「资产翻页」降为提示
  (翻 1~2 页是资产真超 200 的合法分页,仅 ≥3 页才疑似 page_size 失效)。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-17 16:58:48 +08:00

61 lines
2.0 KiB
Python

from dataclasses import dataclass
from typing import BinaryIO
from urllib.parse import quote, urlparse
import boto3
from botocore.config import Config
from django.conf import settings
@dataclass(frozen=True)
class StoredObject:
bucket: str
object_key: str
content_type: str
size_bytes: int
class TosStorage:
def __init__(self) -> None:
tos = settings.TOS
self.bucket = tos["bucket"]
self.client = boto3.client(
"s3",
endpoint_url=tos["endpoint"],
aws_access_key_id=tos["access_key_id"],
aws_secret_access_key=tos["secret_access_key"],
region_name="cn-shanghai",
config=Config(s3={"addressing_style": "virtual"}),
)
def upload_fileobj(self, *, fileobj: BinaryIO, object_key: str, content_type: str) -> StoredObject:
fileobj.seek(0, 2)
size = fileobj.tell()
fileobj.seek(0)
self.client.upload_fileobj(
fileobj,
self.bucket,
object_key,
ExtraArgs={"ContentType": content_type},
)
return StoredObject(
bucket=self.bucket,
object_key=object_key,
content_type=content_type,
size_bytes=size,
)
def presigned_get_url(self, *, object_key: str, expires_in: int = 3600) -> str:
return self.client.generate_presigned_url(
"get_object",
Params={"Bucket": self.bucket, "Key": object_key},
ExpiresIn=expires_in,
)
def public_url(self, *, object_key: str, bucket: str | None = None) -> str:
"""桶公读时的虚拟主机式直链:https://{bucket}.{host}/{key}。
相比预签名 URL:不签名(省 boto3 客户端/签名开销)、且 URL 稳定可被浏览器/CDN 缓存
(签名链每次都变、1h 过期,反而打不到缓存)。仅用于公开可读的预览图。"""
host = urlparse(settings.TOS["endpoint"]).netloc
return f"https://{bucket or self.bucket}.{host}/{quote(object_key, safe='/')}"