315 lines
12 KiB
Python
315 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
|
||
import logging
|
||
|
||
# ロギング設定
|
||
logging.basicConfig(level=logging.INFO)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
class PathConversionPreview:
|
||
"""パス変換プレビューサービス"""
|
||
|
||
def __init__(self):
|
||
self.s3_bucket = settings.AWS_STORAGE_BUCKET_NAME
|
||
self.s3_region = settings.AWS_S3_REGION_NAME
|
||
self.s3_base_url = f"https://{self.s3_bucket}.s3.{self.s3_region}.amazonaws.com"
|
||
|
||
def convert_local_path_to_s3_url(self, local_path, event_code, team_name, image_type='checkin'):
|
||
"""ローカルパスをS3 URLに変換（プレビュー用、100文字制限対応）"""
|
||
try:
|
||
filename = os.path.basename(local_path)
|
||
|
||
if image_type == 'goal' or local_path.startswith('goals/'):
|
||
s3_path = f"s3://{self.s3_bucket}/{event_code}/goals/{team_name}/{filename}"
|
||
else:
|
||
s3_path = f"s3://{self.s3_bucket}/{event_code}/{team_name}/{filename}"
|
||
|
||
# 100文字制限チェック
|
||
if len(s3_path) > 100:
|
||
# 短縮版: ファイル名のみを使用
|
||
if image_type == 'goal' or local_path.startswith('goals/'):
|
||
s3_path = f"s3://{self.s3_bucket}/goals/{filename}"
|
||
else:
|
||
s3_path = f"s3://{self.s3_bucket}/checkin/{filename}"
|
||
|
||
# それでも長い場合はファイル名だけ
|
||
if len(s3_path) > 100:
|
||
s3_path = f"s3://{self.s3_bucket}/{filename}"
|
||
|
||
return s3_path
|
||
|
||
except Exception as e:
|
||
return f"ERROR: {str(e)}"
|
||
|
||
def is_already_s3_url(self, path):
|
||
"""既にS3 URLかどうかを判定"""
|
||
return (
|
||
path and (
|
||
path.startswith('https://') or
|
||
path.startswith('http://') or
|
||
path.startswith('s3://') or
|
||
's3' in path.lower() or
|
||
'amazonaws' in path.lower()
|
||
)
|
||
)
|
||
|
||
def preview_goal_images(self, limit=10):
|
||
"""GoalImagesのパス変換をプレビュー"""
|
||
print("\n=== GoalImages パス変換プレビュー ===")
|
||
|
||
goal_images = GoalImages.objects.filter(goalimage__isnull=False).exclude(goalimage='')
|
||
total_count = goal_images.count()
|
||
|
||
print(f"総対象件数: {total_count:,}件")
|
||
print(f"プレビュー件数: {min(limit, total_count)}件\n")
|
||
|
||
already_s3_count = 0
|
||
conversion_examples = []
|
||
|
||
for i, goal_img in enumerate(goal_images[:limit]):
|
||
original_path = str(goal_img.goalimage)
|
||
|
||
if self.is_already_s3_url(original_path):
|
||
already_s3_count += 1
|
||
status = "🔗 既にS3 URL"
|
||
converted_path = original_path
|
||
else:
|
||
status = "🔄 変換対象"
|
||
converted_path = self.convert_local_path_to_s3_url(
|
||
original_path,
|
||
goal_img.event_code,
|
||
goal_img.team_name,
|
||
'goal'
|
||
)
|
||
|
||
conversion_examples.append({
|
||
'id': goal_img.id,
|
||
'event_code': goal_img.event_code,
|
||
'team_name': goal_img.team_name,
|
||
'cp_number': goal_img.cp_number,
|
||
'original_path': original_path,
|
||
'converted_path': converted_path,
|
||
'status': status
|
||
})
|
||
|
||
print(f"{i+1:2d}. ID={goal_img.id} {status}")
|
||
print(f" イベント: {goal_img.event_code} | チーム: {goal_img.team_name}")
|
||
print(f" 元: {original_path}")
|
||
print(f" → : {converted_path}")
|
||
print()
|
||
|
||
return {
|
||
'total_count': total_count,
|
||
'already_s3_count': already_s3_count,
|
||
'conversion_examples': conversion_examples
|
||
}
|
||
|
||
def preview_checkin_images(self, limit=10):
|
||
"""CheckinImagesのパス変換をプレビュー"""
|
||
print("\n=== CheckinImages パス変換プレビュー ===")
|
||
|
||
checkin_images = CheckinImages.objects.filter(checkinimage__isnull=False).exclude(checkinimage='')
|
||
total_count = checkin_images.count()
|
||
|
||
print(f"総対象件数: {total_count:,}件")
|
||
print(f"プレビュー件数: {min(limit, total_count)}件\n")
|
||
|
||
already_s3_count = 0
|
||
conversion_examples = []
|
||
|
||
for i, checkin_img in enumerate(checkin_images[:limit]):
|
||
original_path = str(checkin_img.checkinimage)
|
||
|
||
if self.is_already_s3_url(original_path):
|
||
already_s3_count += 1
|
||
status = "🔗 既にS3 URL"
|
||
converted_path = original_path
|
||
else:
|
||
status = "🔄 変換対象"
|
||
converted_path = self.convert_local_path_to_s3_url(
|
||
original_path,
|
||
checkin_img.event_code,
|
||
checkin_img.team_name,
|
||
'checkin'
|
||
)
|
||
|
||
conversion_examples.append({
|
||
'id': checkin_img.id,
|
||
'event_code': checkin_img.event_code,
|
||
'team_name': checkin_img.team_name,
|
||
'cp_number': checkin_img.cp_number,
|
||
'original_path': original_path,
|
||
'converted_path': converted_path,
|
||
'status': status
|
||
})
|
||
|
||
print(f"{i+1:2d}. ID={checkin_img.id} {status}")
|
||
print(f" イベント: {checkin_img.event_code} | チーム: {checkin_img.team_name}")
|
||
print(f" 元: {original_path}")
|
||
print(f" → : {converted_path}")
|
||
print()
|
||
|
||
return {
|
||
'total_count': total_count,
|
||
'already_s3_count': already_s3_count,
|
||
'conversion_examples': conversion_examples
|
||
}
|
||
|
||
def analyze_path_patterns(self):
|
||
"""パスパターンを分析"""
|
||
print("\n=== パスパターン分析 ===")
|
||
|
||
# GoalImagesのパターン分析
|
||
goal_paths = GoalImages.objects.filter(goalimage__isnull=False).exclude(goalimage='').values_list('goalimage', flat=True)
|
||
goal_patterns = {}
|
||
|
||
for path in goal_paths:
|
||
path_str = str(path)
|
||
if self.is_already_s3_url(path_str):
|
||
pattern = "S3_URL"
|
||
elif path_str.startswith('goals/'):
|
||
pattern = "goals/YYMMDD/filename"
|
||
elif '/' in path_str:
|
||
parts = path_str.split('/')
|
||
pattern = f"{parts[0]}/..."
|
||
else:
|
||
pattern = "filename_only"
|
||
|
||
goal_patterns[pattern] = goal_patterns.get(pattern, 0) + 1
|
||
|
||
# CheckinImagesのパターン分析
|
||
checkin_paths = CheckinImages.objects.filter(checkinimage__isnull=False).exclude(checkinimage='').values_list('checkinimage', flat=True)
|
||
checkin_patterns = {}
|
||
|
||
for path in checkin_paths:
|
||
path_str = str(path)
|
||
if self.is_already_s3_url(path_str):
|
||
pattern = "S3_URL"
|
||
elif path_str.startswith('checkin/'):
|
||
pattern = "checkin/YYMMDD/filename"
|
||
elif '/' in path_str:
|
||
parts = path_str.split('/')
|
||
pattern = f"{parts[0]}/..."
|
||
else:
|
||
pattern = "filename_only"
|
||
|
||
checkin_patterns[pattern] = checkin_patterns.get(pattern, 0) + 1
|
||
|
||
print("GoalImages パスパターン:")
|
||
for pattern, count in sorted(goal_patterns.items(), key=lambda x: x[1], reverse=True):
|
||
print(f" {pattern}: {count:,}件")
|
||
|
||
print("\nCheckinImages パスパターン:")
|
||
for pattern, count in sorted(checkin_patterns.items(), key=lambda x: x[1], reverse=True):
|
||
print(f" {pattern}: {count:,}件")
|
||
|
||
return {
|
||
'goal_patterns': goal_patterns,
|
||
'checkin_patterns': checkin_patterns
|
||
}
|
||
|
||
def generate_preview_report(self, goal_preview, checkin_preview, patterns):
|
||
"""プレビューレポートを生成"""
|
||
report = {
|
||
'preview_timestamp': datetime.now().isoformat(),
|
||
's3_configuration': {
|
||
'bucket': self.s3_bucket,
|
||
'region': self.s3_region,
|
||
'base_url': self.s3_base_url
|
||
},
|
||
'goal_images': goal_preview,
|
||
'checkin_images': checkin_preview,
|
||
'path_patterns': patterns,
|
||
'summary': {
|
||
'total_goal_images': goal_preview['total_count'],
|
||
'total_checkin_images': checkin_preview['total_count'],
|
||
'total_images': goal_preview['total_count'] + checkin_preview['total_count'],
|
||
'already_s3_urls': goal_preview['already_s3_count'] + checkin_preview['already_s3_count'],
|
||
'requires_conversion': (goal_preview['total_count'] - goal_preview['already_s3_count']) +
|
||
(checkin_preview['total_count'] - checkin_preview['already_s3_count'])
|
||
}
|
||
}
|
||
|
||
# レポートファイルの保存
|
||
report_file = f'path_conversion_preview_{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"🎯 総画像数: {report['summary']['total_images']:,}件")
|
||
print(f" - ゴール画像: {report['summary']['total_goal_images']:,}件")
|
||
print(f" - チェックイン画像: {report['summary']['total_checkin_images']:,}件")
|
||
print(f"🔗 既にS3 URL: {report['summary']['already_s3_urls']:,}件")
|
||
print(f"🔄 変換必要: {report['summary']['requires_conversion']:,}件")
|
||
print(f"📄 詳細レポート: {report_file}")
|
||
|
||
return report
|
||
|
||
def main():
|
||
"""メイン関数"""
|
||
print("="*60)
|
||
print("👀 データベースパス変換プレビューツール")
|
||
print("="*60)
|
||
print("このツールは以下を実行します:")
|
||
print("1. パス変換のプレビュー表示")
|
||
print("2. パスパターンの分析")
|
||
print("3. 変換レポートの生成")
|
||
print()
|
||
print("⚠️ 注意: データベースの実際の更新は行いません")
|
||
print()
|
||
|
||
# プレビュー件数の設定
|
||
try:
|
||
limit_input = input("プレビュー表示件数を入力してください [デフォルト: 10]: ").strip()
|
||
limit = int(limit_input) if limit_input else 10
|
||
if limit <= 0:
|
||
limit = 10
|
||
except ValueError:
|
||
limit = 10
|
||
|
||
print(f"\n🔍 パス変換をプレビューします（各タイプ{limit}件まで表示）...\n")
|
||
|
||
# プレビュー実行
|
||
preview_service = PathConversionPreview()
|
||
|
||
# 1. パスパターン分析
|
||
patterns = preview_service.analyze_path_patterns()
|
||
|
||
# 2. GoalImagesプレビュー
|
||
goal_preview = preview_service.preview_goal_images(limit)
|
||
|
||
# 3. CheckinImagesプレビュー
|
||
checkin_preview = preview_service.preview_checkin_images(limit)
|
||
|
||
# 4. レポート生成
|
||
report = preview_service.generate_preview_report(goal_preview, checkin_preview, patterns)
|
||
|
||
print("\n✅ プレビューが完了しました！")
|
||
print("実際の変換を実行する場合は update_image_paths_to_s3.py を使用してください。")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|