#!/usr/bin/env python """ データベースパス更新ロールバックスクリプト バックアップファイルからパス情報を復元します。 """ import os import sys import django from pathlib import Path import json from datetime import datetime # 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 django.db import transaction import logging # ロギング設定 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(f'rollback_log_{datetime.now().strftime("%Y%m%d_%H%M%S")}.log'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) class PathRollbackService: """パスロールバックサービス""" def __init__(self, backup_file): self.backup_file = backup_file self.rollback_stats = { 'total_goal_images': 0, 'restored_goal_images': 0, 'total_checkin_images': 0, 'restored_checkin_images': 0, 'failed_restores': [] } def load_backup_data(self): """バックアップデータを読み込み""" try: with open(self.backup_file, 'r', encoding='utf-8') as f: backup_data = json.load(f) logger.info(f"バックアップファイル読み込み成功: {self.backup_file}") logger.info(f"バックアップ日時: {backup_data.get('backup_timestamp', 'Unknown')}") logger.info(f"GoalImages: {len(backup_data.get('goal_images', []))}件") logger.info(f"CheckinImages: {len(backup_data.get('checkin_images', []))}件") return backup_data except FileNotFoundError: raise Exception(f"バックアップファイルが見つかりません: {self.backup_file}") except json.JSONDecodeError: raise Exception(f"バックアップファイルの形式が不正です: {self.backup_file}") except Exception as e: raise Exception(f"バックアップファイル読み込みエラー: {str(e)}") def rollback_goal_images(self, goal_images_backup): """GoalImagesをロールバック""" logger.info("=== GoalImagesロールバック開始 ===") self.rollback_stats['total_goal_images'] = len(goal_images_backup) restored_count = 0 with transaction.atomic(): for backup_item in goal_images_backup: try: goal_img = GoalImages.objects.get(id=backup_item['id']) original_path = backup_item['original_path'] goal_img.goalimage = original_path goal_img.save() restored_count += 1 if restored_count <= 5: # 最初の5件のみログ出力 logger.info(f"✅ GoalImage ID={goal_img.id}: パス復元完了") except GoalImages.DoesNotExist: logger.warning(f"⚠️ GoalImage ID={backup_item['id']} が存在しません（削除済み）") self.rollback_stats['failed_restores'].append({ 'type': 'goal', 'id': backup_item['id'], 'reason': 'Record not found' }) except Exception as e: logger.error(f"❌ GoalImage ID={backup_item['id']} 復元エラー: {str(e)}") self.rollback_stats['failed_restores'].append({ 'type': 'goal', 'id': backup_item['id'], 'reason': str(e) }) self.rollback_stats['restored_goal_images'] = restored_count logger.info(f"GoalImagesロールバック完了: {restored_count}件復元") def rollback_checkin_images(self, checkin_images_backup): """CheckinImagesをロールバック""" logger.info("=== CheckinImagesロールバック開始 ===") self.rollback_stats['total_checkin_images'] = len(checkin_images_backup) restored_count = 0 with transaction.atomic(): for backup_item in checkin_images_backup: try: checkin_img = CheckinImages.objects.get(id=backup_item['id']) original_path = backup_item['original_path'] checkin_img.checkinimage = original_path checkin_img.save() restored_count += 1 if restored_count <= 5: # 最初の5件のみログ出力 logger.info(f"✅ CheckinImage ID={checkin_img.id}: パス復元完了") except CheckinImages.DoesNotExist: logger.warning(f"⚠️ CheckinImage ID={backup_item['id']} が存在しません（削除済み）") self.rollback_stats['failed_restores'].append({ 'type': 'checkin', 'id': backup_item['id'], 'reason': 'Record not found' }) except Exception as e: logger.error(f"❌ CheckinImage ID={backup_item['id']} 復元エラー: {str(e)}") self.rollback_stats['failed_restores'].append({ 'type': 'checkin', 'id': backup_item['id'], 'reason': str(e) }) self.rollback_stats['restored_checkin_images'] = restored_count logger.info(f"CheckinImagesロールバック完了: {restored_count}件復元") def generate_rollback_report(self): """ロールバックレポートを生成""" logger.info("=== ロールバックレポート生成 ===") total_restored = self.rollback_stats['restored_goal_images'] + self.rollback_stats['restored_checkin_images'] total_processed = self.rollback_stats['total_goal_images'] + self.rollback_stats['total_checkin_images'] report = { 'rollback_timestamp': datetime.now().isoformat(), 'backup_file': self.backup_file, 'summary': { 'total_processed': total_processed, 'total_restored': total_restored, 'goal_images_restored': self.rollback_stats['restored_goal_images'], 'checkin_images_restored': self.rollback_stats['restored_checkin_images'], 'failed_restores': len(self.rollback_stats['failed_restores']), 'success_rate': (total_restored / max(total_processed, 1) * 100) }, 'failed_restores': self.rollback_stats['failed_restores'] } # レポートファイルの保存 report_file = f'rollback_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("🔄 ロールバックレポート") print("="*60) print(f"📄 バックアップファイル: {self.backup_file}") print(f"📊 処理総数: {total_processed:,}件") print(f"✅ 復元成功: {total_restored:,}件 ({report['summary']['success_rate']:.1f}%)") print(f" - ゴール画像: {self.rollback_stats['restored_goal_images']:,}件") print(f" - チェックイン画像: {self.rollback_stats['restored_checkin_images']:,}件") print(f"❌ 失敗: {len(self.rollback_stats['failed_restores'])}件") print(f"📄 詳細レポート: {report_file}") if len(self.rollback_stats['failed_restores']) > 0: print("\n⚠️ 失敗した復元:") for failure in self.rollback_stats['failed_restores'][:5]: print(f" - {failure['type']} ID={failure['id']}: {failure['reason']}") if len(self.rollback_stats['failed_restores']) > 5: print(f" ... 他 {len(self.rollback_stats['failed_restores']) - 5} 件") return report def run_rollback(self): """メインロールバック処理""" logger.info("🔄 ロールバック開始") print("🔄 データベースパスをロールバックします...") try: # 1. バックアップデータ読み込み backup_data = self.load_backup_data() # 2. GoalImagesロールバック if backup_data.get('goal_images'): self.rollback_goal_images(backup_data['goal_images']) # 3. CheckinImagesロールバック if backup_data.get('checkin_images'): self.rollback_checkin_images(backup_data['checkin_images']) # 4. レポート生成 report = self.generate_rollback_report() logger.info("✅ ロールバック完了") print("\n✅ ロールバックが完了しました！") return report except Exception as e: logger.error(f"💥 ロールバック中に重大なエラーが発生: {str(e)}") print(f"\n💥 ロールバックエラー: {str(e)}") raise def list_backup_files(): """利用可能なバックアップファイルをリスト表示""" backup_files = [] for file in Path('.').glob('path_update_backup_*.json'): try: with open(file, 'r', encoding='utf-8') as f: backup_data = json.load(f) timestamp = backup_data.get('backup_timestamp', 'Unknown') goal_count = len(backup_data.get('goal_images', [])) checkin_count = len(backup_data.get('checkin_images', [])) backup_files.append({ 'file': str(file), 'timestamp': timestamp, 'goal_count': goal_count, 'checkin_count': checkin_count }) except: continue return backup_files def main(): """メイン関数""" print("="*60) print("🔄 データベースパスロールバックツール") print("="*60) # バックアップファイルのリスト表示 backup_files = list_backup_files() if not backup_files: print("❌ バックアップファイルが見つかりません。") print(" path_update_backup_*.json ファイルが存在することを確認してください。") return print("利用可能なバックアップファイル:") for i, backup in enumerate(backup_files, 1): print(f"{i}. {backup['file']}") print(f" 日時: {backup['timestamp']}") print(f" GoalImages: {backup['goal_count']:,}件, CheckinImages: {backup['checkin_count']:,}件") print() # バックアップファイル選択 try: choice = input(f"ロールバックするファイルを選択してください [1-{len(backup_files)}]: ").strip() choice_idx = int(choice) - 1 if choice_idx < 0 or choice_idx >= len(backup_files): print("❌ 無効な選択です。") return selected_backup = backup_files[choice_idx]['file'] except (ValueError, KeyboardInterrupt): print("❌ ロールバックをキャンセルしました。") return # 確認プロンプト print(f"\n⚠️ 以下のファイルからロールバックします:") print(f" {selected_backup}") print() print("このロールバックにより、現在のS3パス情報は元のローカルパスに戻ります。") confirm = input("ロールバックを実行しますか？ [y/N]: ").strip().lower() if confirm not in ['y', 'yes']: print("ロールバックをキャンセルしました。") return # ロールバック実行 rollback_service = PathRollbackService(selected_backup) rollback_service.run_rollback() if __name__ == "__main__": main()