#!/usr/bin/env python """ ローカル画像をS3に移行するスクリプト 使用方法: python migrate_local_images_to_s3.py 機能: - GoalImagesのローカル画像をS3に移行 - CheckinImagesのローカル画像をS3に移行 - 標準画像（start/goal/rule/map）も移行対象（存在する場合） - 移行後にデータベースのパスを更新 - バックアップとロールバック機能付き """ import os import sys import django from pathlib import Path import json from datetime import datetime import shutil import traceback # Django settings setup BASE_DIR = Path(__file__).resolve().parent sys.path.append(str(BASE_DIR)) os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings') django.setup() from django.conf import settings from rog.models import GoalImages, CheckinImages from rog.services.s3_service import S3Service from django.core.files.storage import default_storage from django.core.files.base import ContentFile import logging # ロギング設定 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(f'migration_log_{datetime.now().strftime("%Y%m%d_%H%M%S")}.log'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) class ImageMigrationService: """画像移行サービス""" def __init__(self): self.s3_service = S3Service() self.migration_stats = { 'total_goal_images': 0, 'total_checkin_images': 0, 'successfully_migrated_goal': 0, 'successfully_migrated_checkin': 0, 'failed_migrations': [], 'migration_details': [] } self.backup_file = f'migration_backup_{datetime.now().strftime("%Y%m%d_%H%M%S")}.json' def backup_database_state(self): """移行前のデータベース状態をバックアップ""" logger.info("データベース状態をバックアップ中...") backup_data = { 'goal_images': [], 'checkin_images': [], 'migration_timestamp': datetime.now().isoformat() } # GoalImages のバックアップ for goal_img in GoalImages.objects.all(): backup_data['goal_images'].append({ 'id': goal_img.id, 'original_path': str(goal_img.goalimage) if goal_img.goalimage else None, 'user_id': goal_img.user.id, 'team_name': goal_img.team_name, 'event_code': goal_img.event_code, 'cp_number': goal_img.cp_number }) # CheckinImages のバックアップ for checkin_img in CheckinImages.objects.all(): backup_data['checkin_images'].append({ 'id': checkin_img.id, 'original_path': str(checkin_img.checkinimage) if checkin_img.checkinimage else None, 'user_id': checkin_img.user.id, 'team_name': checkin_img.team_name, 'event_code': checkin_img.event_code, 'cp_number': checkin_img.cp_number }) with open(self.backup_file, 'w', encoding='utf-8') as f: json.dump(backup_data, f, ensure_ascii=False, indent=2) logger.info(f"バックアップ完了: {self.backup_file}") return backup_data def migrate_goal_images(self): """ゴール画像をS3に移行""" logger.info("=== ゴール画像の移行開始 ===") goal_images = GoalImages.objects.filter(goalimage__isnull=False).exclude(goalimage='') self.migration_stats['total_goal_images'] = goal_images.count() logger.info(f"移行対象のゴール画像: {self.migration_stats['total_goal_images']}件") for goal_img in goal_images: try: logger.info(f"処理中: GoalImage ID={goal_img.id}, Path={goal_img.goalimage}") # ローカルファイルパスの構築 local_file_path = os.path.join(settings.MEDIA_ROOT, str(goal_img.goalimage)) if not os.path.exists(local_file_path): logger.warning(f"ファイルが見つかりません: {local_file_path}") self.migration_stats['failed_migrations'].append({ 'type': 'goal', 'id': goal_img.id, 'reason': 'File not found', 'path': local_file_path }) continue # ファイルを読み込み with open(local_file_path, 'rb') as f: file_content = f.read() # ContentFileとして準備 file_name = os.path.basename(local_file_path) content_file = ContentFile(file_content, name=file_name) # S3にアップロード（ゴール画像として扱う） s3_url = self.s3_service.upload_checkin_image( image_file=content_file, event_code=goal_img.event_code, team_code=goal_img.team_name, cp_number=goal_img.cp_number, is_goal=True # ゴール画像フラグ ) if s3_url: # データベースを更新（S3パスを保存） original_path = str(goal_img.goalimage) goal_img.goalimage = s3_url.replace(f'https://{settings.AWS_S3_CUSTOM_DOMAIN}/', '') goal_img.save() self.migration_stats['successfully_migrated_goal'] += 1 self.migration_stats['migration_details'].append({ 'type': 'goal', 'id': goal_img.id, 'original_path': original_path, 'new_s3_url': s3_url, 'local_file': local_file_path }) logger.info(f"✅ 成功: {file_name} -> {s3_url}") else: raise Exception("S3アップロードが失敗しました") except Exception as e: logger.error(f"❌ エラー: GoalImage ID={goal_img.id}, Error={str(e)}") logger.error(traceback.format_exc()) self.migration_stats['failed_migrations'].append({ 'type': 'goal', 'id': goal_img.id, 'reason': str(e), 'path': str(goal_img.goalimage) }) def migrate_checkin_images(self): """チェックイン画像をS3に移行""" logger.info("=== チェックイン画像の移行開始 ===") checkin_images = CheckinImages.objects.filter(checkinimage__isnull=False).exclude(checkinimage='') self.migration_stats['total_checkin_images'] = checkin_images.count() logger.info(f"移行対象のチェックイン画像: {self.migration_stats['total_checkin_images']}件") for checkin_img in checkin_images: try: logger.info(f"処理中: CheckinImage ID={checkin_img.id}, Path={checkin_img.checkinimage}") # ローカルファイルパスの構築 local_file_path = os.path.join(settings.MEDIA_ROOT, str(checkin_img.checkinimage)) if not os.path.exists(local_file_path): logger.warning(f"ファイルが見つかりません: {local_file_path}") self.migration_stats['failed_migrations'].append({ 'type': 'checkin', 'id': checkin_img.id, 'reason': 'File not found', 'path': local_file_path }) continue # ファイルを読み込み with open(local_file_path, 'rb') as f: file_content = f.read() # ContentFileとして準備 file_name = os.path.basename(local_file_path) content_file = ContentFile(file_content, name=file_name) # S3にアップロード s3_url = self.s3_service.upload_checkin_image( image_file=content_file, event_code=checkin_img.event_code, team_code=checkin_img.team_name, cp_number=checkin_img.cp_number ) if s3_url: # データベースを更新（S3パスを保存） original_path = str(checkin_img.checkinimage) checkin_img.checkinimage = s3_url.replace(f'https://{settings.AWS_S3_CUSTOM_DOMAIN}/', '') checkin_img.save() self.migration_stats['successfully_migrated_checkin'] += 1 self.migration_stats['migration_details'].append({ 'type': 'checkin', 'id': checkin_img.id, 'original_path': original_path, 'new_s3_url': s3_url, 'local_file': local_file_path }) logger.info(f"✅ 成功: {file_name} -> {s3_url}") else: raise Exception("S3アップロードが失敗しました") except Exception as e: logger.error(f"❌ エラー: CheckinImage ID={checkin_img.id}, Error={str(e)}") logger.error(traceback.format_exc()) self.migration_stats['failed_migrations'].append({ 'type': 'checkin', 'id': checkin_img.id, 'reason': str(e), 'path': str(checkin_img.checkinimage) }) def migrate_standard_images(self): """規定画像をS3に移行（存在する場合）""" logger.info("=== 規定画像の移行チェック開始 ===") standard_types = ['start', 'goal', 'rule', 'map'] media_root = Path(settings.MEDIA_ROOT) # 各イベントフォルダをチェック events_found = set() # GoalImagesとCheckinImagesから一意のイベントコードを取得 goal_events = set(GoalImages.objects.values_list('event_code', flat=True)) checkin_events = set(CheckinImages.objects.values_list('event_code', flat=True)) all_events = goal_events.union(checkin_events) logger.info(f"検出されたイベント: {all_events}") for event_code in all_events: # 各標準画像タイプをチェック for image_type in standard_types: # 一般的な画像拡張子をチェック for ext in ['.jpg', '.jpeg', '.png', '.JPG', '.JPEG', '.PNG']: # 複数の可能なパスパターンをチェック possible_paths = [ media_root / f'{event_code}_{image_type}{ext}', media_root / event_code / f'{image_type}{ext}', media_root / 'standards' / event_code / f'{image_type}{ext}', media_root / f'{image_type}_{event_code}{ext}', ] for possible_path in possible_paths: if possible_path.exists(): try: logger.info(f"規定画像発見: {possible_path}") # ファイルを読み込み with open(possible_path, 'rb') as f: file_content = f.read() # ContentFileとして準備 content_file = ContentFile(file_content, name=possible_path.name) # S3にアップロード s3_url = self.s3_service.upload_standard_image( image_file=content_file, event_code=event_code, image_type=image_type ) if s3_url: self.migration_stats['migration_details'].append({ 'type': 'standard', 'event_code': event_code, 'image_type': image_type, 'original_path': str(possible_path), 'new_s3_url': s3_url }) logger.info(f"✅ 規定画像移行成功: {possible_path.name} -> {s3_url}") break # 同じタイプの画像が見つかったら他のパスはスキップ except Exception as e: logger.error(f"❌ 規定画像移行エラー: {possible_path}, Error={str(e)}") self.migration_stats['failed_migrations'].append({ 'type': 'standard', 'event_code': event_code, 'image_type': image_type, 'reason': str(e), 'path': str(possible_path) }) def generate_migration_report(self): """移行レポートを生成""" logger.info("=== 移行レポート生成 ===") report = { 'migration_timestamp': datetime.now().isoformat(), 'summary': { 'total_goal_images': self.migration_stats['total_goal_images'], 'successfully_migrated_goal': self.migration_stats['successfully_migrated_goal'], 'total_checkin_images': self.migration_stats['total_checkin_images'], 'successfully_migrated_checkin': self.migration_stats['successfully_migrated_checkin'], 'total_failed': len(self.migration_stats['failed_migrations']), 'success_rate_goal': ( self.migration_stats['successfully_migrated_goal'] / max(self.migration_stats['total_goal_images'], 1) * 100 ), 'success_rate_checkin': ( self.migration_stats['successfully_migrated_checkin'] / max(self.migration_stats['total_checkin_images'], 1) * 100 ) }, 'details': self.migration_stats['migration_details'], 'failures': self.migration_stats['failed_migrations'] } # レポートファイルの保存 report_file = f'migration_report_{datetime.now().strftime("%Y%m%d_%H%M%S")}.json' with open(report_file, 'w', encoding='utf-8') as f: json.dump(report, f, ensure_ascii=False, indent=2) # コンソール出力 print("\n" + "="*60) print("🎯 画像S3移行レポート") print("="*60) print(f"📊 ゴール画像: {report['summary']['successfully_migrated_goal']}/{report['summary']['total_goal_images']} " f"({report['summary']['success_rate_goal']:.1f}%)") print(f"📊 チェックイン画像: {report['summary']['successfully_migrated_checkin']}/{report['summary']['total_checkin_images']} " f"({report['summary']['success_rate_checkin']:.1f}%)") print(f"❌ 失敗数: {report['summary']['total_failed']}") print(f"📄 詳細レポート: {report_file}") print(f"💾 バックアップファイル: {self.backup_file}") if report['summary']['total_failed'] > 0: print("\n⚠️ 失敗した移行:") for failure in self.migration_stats['failed_migrations'][:5]: # 最初の5件のみ表示 print(f" - {failure['type']} ID={failure.get('id', 'N/A')}: {failure['reason']}") if len(self.migration_stats['failed_migrations']) > 5: print(f" ... 他 {len(self.migration_stats['failed_migrations']) - 5} 件") return report def run_migration(self): """メイン移行処理""" logger.info("🚀 画像S3移行開始") print("🚀 画像S3移行を開始します...") try: # 1. バックアップ self.backup_database_state() # 2. ゴール画像移行 self.migrate_goal_images() # 3. チェックイン画像移行 self.migrate_checkin_images() # 4. 規定画像移行 self.migrate_standard_images() # 5. レポート生成 report = self.generate_migration_report() logger.info("✅ 移行完了") print("\n✅ 移行が完了しました！") return report except Exception as e: logger.error(f"💥 移行中に重大なエラーが発生: {str(e)}") logger.error(traceback.format_exc()) print(f"\n💥 移行エラー: {str(e)}") print(f"バックアップファイル: {self.backup_file}") raise def main(): """メイン関数""" print("="*60) print("🔄 ローカル画像S3移行ツール") print("="*60) print("このツールは以下を実行します:") print("1. データベースの現在の状態をバックアップ") print("2. GoalImages のローカル画像をS3に移行") print("3. CheckinImages のローカル画像をS3に移行") print("4. 標準画像（存在する場合）をS3に移行") print("5. 移行レポートの生成") print() # 確認プロンプト confirm = input("移行を開始しますか？ [y/N]: ").strip().lower() if confirm not in ['y', 'yes']: print("移行をキャンセルしました。") return # 移行実行 migration_service = ImageMigrationService() migration_service.run_migration() if __name__ == "__main__": main()