278 lines
9.1 KiB
Python
278 lines
9.1 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
图片迁移到新 OSS 存储桶
|
||
集成配置和上传功能
|
||
"""
|
||
|
||
import os
|
||
import re
|
||
import oss2
|
||
from pathlib import Path
|
||
|
||
# 本地图片目录
|
||
GIFT_DIR = 'public/images/gifts'
|
||
OUTFIT_DIR = 'public/images/outfits'
|
||
|
||
# OSS 上传路径（保持与数据库一致）
|
||
OSS_GIFT_PREFIX = 'uploads/gifts/'
|
||
OSS_OUTFIT_PREFIX = 'uploads/outfit/'
|
||
|
||
class OSSMigrator:
|
||
def __init__(self):
|
||
self.config = {}
|
||
self.bucket = None
|
||
|
||
def get_input(self, prompt, default=None):
|
||
"""获取用户输入"""
|
||
if default:
|
||
prompt = f"{prompt} [{default}]: "
|
||
else:
|
||
prompt = f"{prompt}: "
|
||
|
||
value = input(prompt).strip()
|
||
return value if value else default
|
||
|
||
def validate_endpoint(self, endpoint):
|
||
"""验证 Endpoint 格式"""
|
||
endpoint = endpoint.replace('https://', '').replace('http://', '')
|
||
return endpoint
|
||
|
||
def configure(self):
|
||
"""配置新 OSS 信息"""
|
||
print("=" * 60)
|
||
print("配置新的 OSS 存储桶")
|
||
print("=" * 60)
|
||
print("\n请输入新的 OSS 配置信息：\n")
|
||
|
||
# 获取配置信息
|
||
self.config['access_key_id'] = self.get_input("Access Key ID")
|
||
if not self.config['access_key_id']:
|
||
print("✗ Access Key ID 不能为空")
|
||
return False
|
||
|
||
self.config['access_key_secret'] = self.get_input("Access Key Secret")
|
||
if not self.config['access_key_secret']:
|
||
print("✗ Access Key Secret 不能为空")
|
||
return False
|
||
|
||
self.config['bucket_name'] = self.get_input("Bucket 名称")
|
||
if not self.config['bucket_name']:
|
||
print("✗ Bucket 名称不能为空")
|
||
return False
|
||
|
||
self.config['endpoint'] = self.get_input("Endpoint", "oss-cn-hangzhou.aliyuncs.com")
|
||
self.config['endpoint'] = self.validate_endpoint(self.config['endpoint'])
|
||
|
||
# 显示配置摘要
|
||
print("\n" + "=" * 60)
|
||
print("配置摘要")
|
||
print("=" * 60)
|
||
print(f"Access Key ID: {self.config['access_key_id'][:8]}...")
|
||
print(f"Access Key Secret: {self.config['access_key_secret'][:8]}...")
|
||
print(f"Bucket 名称: {self.config['bucket_name']}")
|
||
print(f"Endpoint: {self.config['endpoint']}")
|
||
print(f"CDN 域名: https://{self.config['bucket_name']}.{self.config['endpoint']}")
|
||
|
||
# 确认
|
||
confirm = self.get_input("\n确认配置？(y/n)", "y")
|
||
if confirm.lower() != 'y':
|
||
print("已取消")
|
||
return False
|
||
|
||
return True
|
||
|
||
def init_oss_bucket(self):
|
||
"""初始化 OSS 连接"""
|
||
try:
|
||
auth = oss2.Auth(self.config['access_key_id'], self.config['access_key_secret'])
|
||
self.bucket = oss2.Bucket(auth, self.config['endpoint'], self.config['bucket_name'])
|
||
return True
|
||
except Exception as e:
|
||
print(f"✗ 初始化 OSS 失败: {e}")
|
||
return False
|
||
|
||
def test_oss_connection(self):
|
||
"""测试 OSS 连接"""
|
||
try:
|
||
for obj in oss2.ObjectIterator(self.bucket, max_keys=1):
|
||
pass
|
||
return True
|
||
except Exception as e:
|
||
print(f"✗ OSS 连接失败: {e}")
|
||
return False
|
||
|
||
def upload_file(self, local_path, oss_path):
|
||
"""上传单个文件到 OSS"""
|
||
try:
|
||
result = self.bucket.put_object_from_file(oss_path, local_path)
|
||
if result.status == 200:
|
||
return True
|
||
else:
|
||
print(f" ✗ 上传失败 (状态码: {result.status})")
|
||
return False
|
||
except Exception as e:
|
||
print(f" ✗ 上传错误: {e}")
|
||
return False
|
||
|
||
def upload_gifts(self):
|
||
"""上传礼物图片"""
|
||
print(f"\n📦 上传礼物图片")
|
||
print(f"本地目录: {GIFT_DIR}")
|
||
print(f"OSS 路径: {OSS_GIFT_PREFIX}")
|
||
|
||
if not os.path.exists(GIFT_DIR):
|
||
print(f"✗ 目录不存在: {GIFT_DIR}")
|
||
return
|
||
|
||
files = [f for f in os.listdir(GIFT_DIR) if f.endswith(('.png', '.jpg', '.jpeg', '.gif'))]
|
||
print(f"找到 {len(files)} 个文件")
|
||
|
||
success = 0
|
||
failed = 0
|
||
skipped = 0
|
||
|
||
for filename in files:
|
||
local_path = os.path.join(GIFT_DIR, filename)
|
||
oss_path = OSS_GIFT_PREFIX + filename
|
||
|
||
# 检查文件是否已存在
|
||
try:
|
||
self.bucket.head_object(oss_path)
|
||
print(f" ○ 已存在: {filename}")
|
||
skipped += 1
|
||
continue
|
||
except oss2.exceptions.NoSuchKey:
|
||
pass
|
||
|
||
print(f" ↑ 上传: {filename}")
|
||
if self.upload_file(local_path, oss_path):
|
||
success += 1
|
||
print(f" ✓ 成功")
|
||
else:
|
||
failed += 1
|
||
|
||
print(f"\n礼物上传完成: 成功 {success}, 失败 {failed}, 跳过 {skipped}")
|
||
|
||
def upload_outfits(self):
|
||
"""上传服装图片"""
|
||
print(f"\n👔 上传服装图片")
|
||
print(f"本地目录: {OUTFIT_DIR}")
|
||
print(f"OSS 路径: {OSS_OUTFIT_PREFIX}")
|
||
|
||
if not os.path.exists(OUTFIT_DIR):
|
||
print(f"✗ 目录不存在: {OUTFIT_DIR}")
|
||
return
|
||
|
||
files = [f for f in os.listdir(OUTFIT_DIR) if f.endswith(('.png', '.jpg', '.jpeg', '.gif'))]
|
||
print(f"找到 {len(files)} 个文件")
|
||
|
||
success = 0
|
||
failed = 0
|
||
skipped = 0
|
||
|
||
# 服装图片需要按类型分类上传
|
||
outfit_types = {
|
||
'top': ['tshirt', 'shirt', 'sleeve', 'strap', 'shoulder', 'crop', 'cardigan', 'jacket', 'sweatshirt'],
|
||
'bottom': ['jeans', 'skirt', 'shorts', 'pants', 'sweatpants'],
|
||
'dress': ['dress', 'uniform', 'hanfu', 'lolita']
|
||
}
|
||
|
||
for filename in files:
|
||
local_path = os.path.join(OUTFIT_DIR, filename)
|
||
|
||
# 判断服装类型
|
||
outfit_type = 'dress'
|
||
for type_name, keywords in outfit_types.items():
|
||
if any(keyword in filename.lower() for keyword in keywords):
|
||
outfit_type = type_name
|
||
break
|
||
|
||
oss_path = f"{OSS_OUTFIT_PREFIX}{outfit_type}/{filename}"
|
||
|
||
# 检查文件是否已存在
|
||
try:
|
||
self.bucket.head_object(oss_path)
|
||
print(f" ○ 已存在: {filename} ({outfit_type})")
|
||
skipped += 1
|
||
continue
|
||
except oss2.exceptions.NoSuchKey:
|
||
pass
|
||
|
||
print(f" ↑ 上传: {filename} -> {outfit_type}/")
|
||
if self.upload_file(local_path, oss_path):
|
||
success += 1
|
||
print(f" ✓ 成功")
|
||
else:
|
||
failed += 1
|
||
|
||
print(f"\n服装上传完成: 成功 {success}, 失败 {failed}, 跳过 {skipped}")
|
||
|
||
def generate_env_config(self):
|
||
"""生成 .env 配置内容"""
|
||
endpoint_with_https = f"https://{self.config['endpoint']}"
|
||
cdn_domain = f"https://{self.config['bucket_name']}.{self.config['endpoint']}"
|
||
|
||
env_content = f"""
|
||
# ===== 新的 OSS 配置 =====
|
||
ALIYUN_OSS_ACCESS_KEY_ID={self.config['access_key_id']}
|
||
ALIYUN_OSS_ACCESS_KEY_SECRET={self.config['access_key_secret']}
|
||
ALIYUN_OSS_BUCKET_NAME={self.config['bucket_name']}
|
||
ALIYUN_OSS_ENDPOINT={endpoint_with_https}
|
||
ALIYUN_OSS_CDN_DOMAIN={cdn_domain}
|
||
"""
|
||
return env_content
|
||
|
||
def run(self):
|
||
"""运行迁移流程"""
|
||
print("=" * 60)
|
||
print("图片迁移到新 OSS 存储桶")
|
||
print("=" * 60)
|
||
|
||
# 步骤1: 配置
|
||
if not self.configure():
|
||
return
|
||
|
||
# 步骤2: 初始化 OSS
|
||
print("\n正在连接 OSS...")
|
||
if not self.init_oss_bucket():
|
||
return
|
||
|
||
# 步骤3: 测试连接
|
||
if not self.test_oss_connection():
|
||
print("\n✗ 无法连接到 OSS，请检查配置")
|
||
return
|
||
|
||
print("✓ OSS 连接成功")
|
||
|
||
# 步骤4: 上传礼物图片
|
||
self.upload_gifts()
|
||
|
||
# 步骤5: 上传服装图片
|
||
self.upload_outfits()
|
||
|
||
# 步骤6: 显示后续操作
|
||
print("\n" + "=" * 60)
|
||
print("✓ 上传完成！")
|
||
print("=" * 60)
|
||
print("\n下一步操作：")
|
||
print("\n1. 将以下内容添加到 .env 文件：")
|
||
print(self.generate_env_config())
|
||
print("\n2. 重启服务：")
|
||
print(" 开发\\1-启动项目.bat")
|
||
print("\n数据库无需修改，图片路径保持不变！")
|
||
|
||
def main():
|
||
try:
|
||
migrator = OSSMigrator()
|
||
migrator.run()
|
||
except KeyboardInterrupt:
|
||
print("\n\n用户中断操作")
|
||
except Exception as e:
|
||
print(f"\n✗ 发生错误: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
if __name__ == '__main__':
|
||
main()
|