feat: 新增后台复制店铺商品目录接口

支持按用户ID复制商品类型、专区与商品,OSS图片独立拷贝,并提供覆盖/补充两种模式。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
XingQue
2026-07-09 22:58:56 +08:00
parent 581295443a
commit 214b52badd
6 changed files with 388 additions and 1 deletions

View File

@@ -184,6 +184,88 @@ def _upload_to_aliyun_oss(file_obj, file_path):
return None
def _url_to_oss_key(file_url):
"""从完整 URL 或相对路径提取 OSS 对象 Key"""
if not file_url:
return None
file_url = str(file_url).strip()
if not file_url:
return None
oss_type = getattr(settings, 'OSS_TYPE', 'tencent')
if oss_type == 'tencent':
domain = getattr(settings, 'COS_DOMAIN', '')
elif oss_type == 'aliyun':
domain = getattr(settings, 'ALIYUN_OSS_DOMAIN', '')
else:
return None
if file_url.startswith('http'):
if domain and domain not in file_url:
return None
prefix = f"{domain.rstrip('/')}/"
if file_url.startswith(prefix):
return file_url[len(prefix):]
return file_url.split('://', 1)[-1].split('/', 1)[-1]
return file_url.lstrip('/')
def copy_in_oss(src_key_or_url, dest_key):
"""
在 OSS 内复制对象(服务端复制,不经过本地)
src_key_or_url: 源对象 Key 或完整 URL
dest_key: 目标对象 Key相对路径
返回: 成功 True失败 False
"""
try:
src_key = _url_to_oss_key(src_key_or_url)
if not src_key or not dest_key:
return False
dest_key = dest_key.lstrip('/')
oss_type = getattr(settings, 'OSS_TYPE', 'tencent')
if oss_type == 'tencent':
return _copy_in_tencent_cos(src_key, dest_key)
elif oss_type == 'aliyun':
return _copy_in_aliyun_oss(src_key, dest_key)
return False
except Exception as e:
print(f"OSS复制失败: {e}")
return False
def _copy_in_tencent_cos(src_key, dest_key):
client = get_oss_client()
bucket = getattr(settings, 'COS_BUCKET', '')
region = getattr(settings, 'COS_REGION', 'ap-shanghai')
try:
client.copy_object(
Bucket=bucket,
Key=dest_key,
CopySource={
'Bucket': bucket,
'Region': region,
'Key': src_key,
},
)
return True
except Exception as e:
print(f"腾讯云COS复制失败: {e}")
return False
def _copy_in_aliyun_oss(src_key, dest_key):
bucket = get_oss_client()
try:
result = bucket.copy_object(src_key, dest_key)
return result.status == 200
except Exception as e:
print(f"阿里云OSS复制失败: {e}")
return False
def delete_from_oss(file_url):
"""
从OSS删除文件