299 lines
12 KiB
Python
299 lines
12 KiB
Python
#!/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()
|