oss.py
993 Bytes
import requests
import oss2
from urllib.parse import urlparse
ARCHIVE_DEV_HOST = 'archive-dev.oss-cn-beijing.aliyuncs.com'
def transfer_url(url: str | None, oss_key: str, bucket: oss2.Bucket, base_url: str) -> str:
"""
将 url 指向的文件转移到 archive-dev OSS 的 oss_key 路径。
若 url 为空或已在 archive-dev,直接返回原 url(不上传)。
返回新的 archive-dev URL。
"""
if not url:
return ''
if ARCHIVE_DEV_HOST in url:
return url
resp = requests.get(url, timeout=30)
resp.raise_for_status()
bucket.put_object(oss_key, resp.content)
return f"{base_url.rstrip('/')}/{oss_key}"
def build_oss_key(platform: str, category: str, filename: str) -> str:
"""
构造 archive-dev 内的存储路径。
platform: 'qq' | 'kugou' | 'netease'
category: 'audio' | 'cover' | 'singer' | 'album'
filename: 带扩展名的文件名
"""
return f"crawler/{platform}/{category}/{filename}"