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()
|