#!/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()