feat: 阿里云 OSS 迁移至腾讯云 COS + 完善部署配置
COS 迁移: - 后端签名服务改为 COS HMAC-SHA1 表单直传签名 - config.py: OSS_* 配置项替换为 COS_SECRET_ID/KEY/REGION/BUCKET_NAME/CDN_DOMAIN - upload.py: UploadPolicyResponse 改为 COS 字段 - 前端 useOSSUpload hook: FormData 字段改为 COS 格式 - 前端 api.ts: UploadPolicyResponse 类型对齐 部署配置: - docker-compose.yml: 新增 Nginx + 前端容器，数据卷宿主机持久化 - Nginx: HTTPS + HTTP/2 + SSE 长连接 + API/前端反向代理 - backup.sh: PostgreSQL 每日备份 → 本地 + COS - .env.example: 更新为 COS 配置模板 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Your Name
co-authored by
Claude Opus 4.6
Claude Opus 4.6
parent
f02b3f4098
commit
8ab2d869fc
+75
-69
@@ -1,5 +1,5 @@
|
||||
"""
|
||||
阿里云 OSS 服务
|
||||
腾讯云 COS 服务 — 表单直传签名
|
||||
"""
|
||||
import time
|
||||
import hmac
|
||||
@@ -17,99 +17,109 @@ def generate_upload_policy(
|
||||
upload_dir: Optional[str] = None,
|
||||
) -> dict:
|
||||
"""
|
||||
生成前端直传 OSS 所需的 Policy 和签名
|
||||
生成前端直传 COS 所需的 Policy 和签名
|
||||
|
||||
COS 表单直传签名流程:
|
||||
1. key_time = "{start_time};{end_time}"
|
||||
2. sign_key = HMAC-SHA1(secret_key, key_time)
|
||||
3. policy JSON → Base64 编码
|
||||
4. string_to_sign = SHA1(policy_base64)
|
||||
5. signature = HMAC-SHA1(sign_key, string_to_sign)
|
||||
|
||||
Returns:
|
||||
{
|
||||
"accessKeyId": "...",
|
||||
"q_sign_algorithm": "sha1",
|
||||
"q_ak": "SecretId",
|
||||
"q_key_time": "{start};{end}",
|
||||
"q_signature": "...",
|
||||
"policy": "base64 encoded policy",
|
||||
"signature": "...",
|
||||
"host": "https://bucket.oss-cn-hangzhou.aliyuncs.com",
|
||||
"host": "https://bucket.cos.region.myqcloud.com",
|
||||
"dir": "uploads/2026/02/",
|
||||
"expire": 1234567890
|
||||
"expire": 1234567890,
|
||||
}
|
||||
"""
|
||||
if not settings.OSS_ACCESS_KEY_ID or not settings.OSS_ACCESS_KEY_SECRET:
|
||||
raise ValueError("OSS 配置未设置")
|
||||
if not settings.COS_SECRET_ID or not settings.COS_SECRET_KEY:
|
||||
raise ValueError("COS 配置未设置")
|
||||
|
||||
# 计算过期时间
|
||||
expire_time = int(time.time()) + expire_seconds
|
||||
expire_date = datetime.utcfromtimestamp(expire_time).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||||
# 计算时间范围
|
||||
start_time = int(time.time())
|
||||
end_time = start_time + expire_seconds
|
||||
|
||||
# key_time: "{start};{end}"
|
||||
key_time = f"{start_time};{end_time}"
|
||||
|
||||
# 默认上传目录：uploads/年/月/
|
||||
if upload_dir is None:
|
||||
now = datetime.now()
|
||||
upload_dir = f"uploads/{now.year}/{now.month:02d}/"
|
||||
|
||||
# 构建 Policy
|
||||
# 1. sign_key = HMAC-SHA1(secret_key, key_time)
|
||||
sign_key = hmac.new(
|
||||
settings.COS_SECRET_KEY.encode(),
|
||||
key_time.encode(),
|
||||
hashlib.sha1,
|
||||
).hexdigest()
|
||||
|
||||
# 2. 构建 Policy（COS 表单上传 Policy 格式）
|
||||
policy_dict = {
|
||||
"expiration": expire_date,
|
||||
"expiration": datetime.utcfromtimestamp(end_time).strftime(
|
||||
"%Y-%m-%dT%H:%M:%S.000Z"
|
||||
),
|
||||
"conditions": [
|
||||
{"bucket": settings.OSS_BUCKET_NAME},
|
||||
{"bucket": settings.COS_BUCKET_NAME},
|
||||
["starts-with", "$key", upload_dir],
|
||||
{"q-sign-algorithm": "sha1"},
|
||||
{"q-ak": settings.COS_SECRET_ID},
|
||||
{"q-sign-time": key_time},
|
||||
["content-length-range", 0, max_size_mb * 1024 * 1024],
|
||||
]
|
||||
],
|
||||
}
|
||||
|
||||
# Base64 编码 Policy
|
||||
# 3. Base64 编码 Policy
|
||||
policy_json = json.dumps(policy_dict)
|
||||
policy_base64 = base64.b64encode(policy_json.encode()).decode()
|
||||
|
||||
# 计算签名
|
||||
signature = base64.b64encode(
|
||||
hmac.new(
|
||||
settings.OSS_ACCESS_KEY_SECRET.encode(),
|
||||
policy_base64.encode(),
|
||||
hashlib.sha1
|
||||
).digest()
|
||||
).decode()
|
||||
# 4. string_to_sign = SHA1(policy_base64)
|
||||
string_to_sign = hashlib.sha1(policy_base64.encode()).hexdigest()
|
||||
|
||||
# 5. signature = HMAC-SHA1(sign_key, string_to_sign)
|
||||
signature = hmac.new(
|
||||
sign_key.encode(),
|
||||
string_to_sign.encode(),
|
||||
hashlib.sha1,
|
||||
).hexdigest()
|
||||
|
||||
# 构建 Host
|
||||
host = settings.OSS_BUCKET_DOMAIN
|
||||
if not host:
|
||||
host = f"https://{settings.OSS_BUCKET_NAME}.{settings.OSS_ENDPOINT}"
|
||||
host = f"https://{settings.COS_BUCKET_NAME}.cos.{settings.COS_REGION}.myqcloud.com"
|
||||
|
||||
return {
|
||||
"accessKeyId": settings.OSS_ACCESS_KEY_ID,
|
||||
"q_sign_algorithm": "sha1",
|
||||
"q_ak": settings.COS_SECRET_ID,
|
||||
"q_key_time": key_time,
|
||||
"q_signature": signature,
|
||||
"policy": policy_base64,
|
||||
"signature": signature,
|
||||
"host": host,
|
||||
"dir": upload_dir,
|
||||
"expire": expire_time,
|
||||
"expire": end_time,
|
||||
}
|
||||
|
||||
|
||||
def generate_sts_token(
|
||||
role_arn: str,
|
||||
session_name: str = "miaosi-upload",
|
||||
duration_seconds: int = 3600,
|
||||
) -> dict:
|
||||
"""
|
||||
生成 STS 临时凭证（需要配置 RAM 角色）
|
||||
|
||||
当前使用 Policy 签名方式，STS 方式为可选增强。
|
||||
如需启用 STS，请安装 aliyun-python-sdk-sts 并配置 RAM 角色。
|
||||
"""
|
||||
# 回退到 Policy 签名方式
|
||||
return generate_upload_policy(
|
||||
max_size_mb=settings.MAX_FILE_SIZE_MB,
|
||||
expire_seconds=duration_seconds,
|
||||
)
|
||||
|
||||
|
||||
def get_file_url(file_key: str) -> str:
|
||||
"""
|
||||
获取文件的公开访问 URL
|
||||
获取文件的访问 URL
|
||||
|
||||
优先使用 CDN 域名，否则用 COS 源站域名。
|
||||
|
||||
Args:
|
||||
file_key: 文件在 OSS 中的 key，如 "uploads/2026/02/video.mp4"
|
||||
file_key: 文件在 COS 中的 key，如 "uploads/2026/02/video.mp4"
|
||||
|
||||
Returns:
|
||||
完整的访问 URL
|
||||
"""
|
||||
host = settings.OSS_BUCKET_DOMAIN
|
||||
if not host:
|
||||
host = f"https://{settings.OSS_BUCKET_NAME}.{settings.OSS_ENDPOINT}"
|
||||
if settings.COS_CDN_DOMAIN:
|
||||
host = settings.COS_CDN_DOMAIN
|
||||
else:
|
||||
host = f"https://{settings.COS_BUCKET_NAME}.cos.{settings.COS_REGION}.myqcloud.com"
|
||||
|
||||
# 确保 host 以 https:// 开头
|
||||
if not host.startswith("http"):
|
||||
@@ -129,26 +139,22 @@ def parse_file_key_from_url(url: str) -> str:
|
||||
从完整 URL 解析出文件 key
|
||||
|
||||
Args:
|
||||
url: 完整的 OSS URL
|
||||
url: 完整的 COS URL
|
||||
|
||||
Returns:
|
||||
文件 key
|
||||
"""
|
||||
host = settings.OSS_BUCKET_DOMAIN
|
||||
if not host:
|
||||
host = f"https://{settings.OSS_BUCKET_NAME}.{settings.OSS_ENDPOINT}"
|
||||
# 尝试移除 CDN 域名
|
||||
if settings.COS_CDN_DOMAIN:
|
||||
cdn = settings.COS_CDN_DOMAIN.rstrip("/")
|
||||
if not cdn.startswith("http"):
|
||||
cdn = f"https://{cdn}"
|
||||
if url.startswith(cdn):
|
||||
return url[len(cdn):].lstrip("/")
|
||||
|
||||
# 移除 host 前缀
|
||||
if url.startswith(host):
|
||||
return url[len(host):].lstrip("/")
|
||||
|
||||
# 尝试其他格式
|
||||
if settings.OSS_BUCKET_NAME in url:
|
||||
# 格式: https://bucket.endpoint/key
|
||||
parts = url.split(settings.OSS_BUCKET_NAME + ".")
|
||||
if len(parts) > 1:
|
||||
key_part = parts[1].split("/", 1)
|
||||
if len(key_part) > 1:
|
||||
return key_part[1]
|
||||
# 尝试移除 COS 源站域名
|
||||
cos_host = f"https://{settings.COS_BUCKET_NAME}.cos.{settings.COS_REGION}.myqcloud.com"
|
||||
if url.startswith(cos_host):
|
||||
return url[len(cos_host):].lstrip("/")
|
||||
|
||||
return url
|
||||
|
||||
Reference in New Issue
Block a user