#!/usr/bin/env python3 """ GPS情報（通過データ）移行スクリプト gifurogeのgps_informationテーブルから新しいrogdbシステムに通過データを移行 """ import os import sys import django from datetime import datetime import psycopg2 from django.utils import timezone from django.db import transaction # Django設定 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings') django.setup() from rog.models import ( GpsLog, GpsCheckin, CheckinExtended, Entry, NewEvent2, CustomUser, Team, Waypoint, Location2025 ) class GpsInformationMigrator: def __init__(self): # 環境変数から接続情報を取得 self.gifuroge_conn_params = { 'host': os.environ.get('PG_HOST', 'postgres-db'), 'database': 'gifuroge', 'user': os.environ.get('POSTGRES_USER', 'postgres'), 'password': os.environ.get('POSTGRES_PASS', 'password'), 'port': os.environ.get('PG_PORT', 5432), } # 統計情報 self.stats = { 'total_gps_info': 0, 'migrated_gps_logs': 0, 'migrated_checkins': 0, 'skipped_records': 0, 'errors': 0, 'error_details': [] } def connect_to_gifuroge(self): """gifurogeデータベースに接続""" try: conn = psycopg2.connect(**self.gifuroge_conn_params) return conn except Exception as e: print(f"❌ gifurogeデータベース接続エラー: {e}") return None def get_gps_information_data(self): """gifurogeのgps_informationデータを取得""" conn = self.connect_to_gifuroge() if not conn: return [] try: cursor = conn.cursor() # まずテーブル構造を確認 cursor.execute(""" SELECT column_name, data_type FROM information_schema.columns WHERE table_name = 'gps_information' AND table_schema = 'public' ORDER BY ordinal_position; """) columns = cursor.fetchall() print("=== gps_information テーブル構造 ===") for col in columns: print(f"- {col[0]}: {col[1]}") # データ数確認 cursor.execute("SELECT COUNT(*) FROM gps_information;") total_count = cursor.fetchone()[0] self.stats['total_gps_info'] = total_count print(f"\n📊 gps_information 総レコード数: {total_count}") if total_count == 0: print("⚠️ gps_informationテーブルにデータがありません") return [] # 全データを取得（テーブル構造に合わせて修正） cursor.execute(""" SELECT serial_number, zekken_number, event_code, cp_number, image_address, goal_time, late_point, create_at, create_user, update_at, update_user, buy_flag, minus_photo_flag, colabo_company_memo FROM gps_information ORDER BY create_at, serial_number; """) data = cursor.fetchall() print(f"✅ {len(data)}件のgps_informationデータを取得しました") return data except Exception as e: print(f"❌ データ取得エラー: {e}") self.stats['errors'] += 1 self.stats['error_details'].append(f"データ取得エラー: {e}") return [] finally: if conn: conn.close() def find_matching_entry(self, zekken_number, event_code): """ゼッケン番号とイベントコードからEntryを検索""" try: # NewEvent2でイベントを検索 events = NewEvent2.objects.filter(event_name__icontains=event_code) if not events.exists(): # イベントコードの部分一致で検索 events = NewEvent2.objects.filter( event_name__icontains=event_code.replace('_', ' ') ) for event in events: # ゼッケン番号でEntryを検索 entries = Entry.objects.filter( event=event, zekken_number=zekken_number ) if entries.exists(): return entries.first() # 見つからない場合はNone return None except Exception as e: print(f"⚠️ Entry検索エラー (ゼッケン: {zekken_number}, イベント: {event_code}): {e}") return None def find_matching_location(self, cp_number): """CP番号からLocationを検索""" try: if not cp_number: return None # Location2025から検索 locations = Location2025.objects.filter(cp_number=cp_number) if locations.exists(): return locations.first() # 部分一致で検索 locations = Location2025.objects.filter(cp_number__icontains=str(cp_number)) if locations.exists(): return locations.first() return None except Exception as e: print(f"⚠️ Location検索エラー (CP: {cp_number}): {e}") return None def migrate_gps_record(self, record): """個別のGPS記録を移行""" try: (serial_number, zekken_number, event_code, cp_number, image_address, goal_time, late_point, create_at, create_user, update_at, update_user, buy_flag, minus_photo_flag, colabo_company_memo) = record # checkin_timeはcreate_atを使用 checkin_time = create_at or timezone.now() # Entryを検索 entry = self.find_matching_entry(zekken_number, event_code) if not entry: print(f"⚠️ Entry未発見: ゼッケン{zekken_number}, イベント{event_code}") self.stats['skipped_records'] += 1 return False # Locationを検索（オプション） location = self.find_matching_location(cp_number) if cp_number else None # 既存のGpsLogをチェック existing_log = GpsLog.objects.filter( zekken_number=str(zekken_number), event_code=event_code, checkin_time=checkin_time ).first() if existing_log: print(f"⚠️ 既存記録をスキップ: ゼッケン{zekken_number}, {checkin_time}") self.stats['skipped_records'] += 1 return False # GpsLogを作成 gps_log = GpsLog.objects.create( serial_number=serial_number or 0, zekken_number=str(zekken_number), event_code=event_code, cp_number=str(cp_number) if cp_number else '', image_address=image_address or '', checkin_time=checkin_time, goal_time=goal_time or '', late_point=late_point or 0, create_at=create_at or timezone.now(), create_user=create_user or '', update_at=update_at or timezone.now(), update_user=update_user or '', buy_flag=buy_flag or False, minus_photo_flag=minus_photo_flag or False, colabo_company_memo=colabo_company_memo or '', is_service_checked=False, # デフォルト値 score=0, # デフォルト値 scoreboard_url='' # デフォルト値 ) self.stats['migrated_gps_logs'] += 1 # CheckinExtendedも作成（通過記録として） if cp_number and location: try: checkin_extended = CheckinExtended.objects.create( entry=entry, location=location, checkin_time=checkin_time, image_url=image_address or '', score_override=0, # デフォルト値 notes=f"移行データ: {colabo_company_memo}", is_verified=False # デフォルト値 ) self.stats['migrated_checkins'] += 1 print(f"✅ チェックイン記録作成: ゼッケン{zekken_number}, CP{cp_number}") except Exception as e: print(f"⚠️ CheckinExtended作成エラー: {e}") print(f"✅ GPS記録移行完了: ゼッケン{zekken_number}, {checkin_time}") return True except Exception as e: print(f"❌ GPS記録移行エラー: {e}") self.stats['errors'] += 1 self.stats['error_details'].append(f"GPS記録移行エラー: {e}") return False def run_migration(self): """メイン移行処理""" print("🚀 GPS情報移行スクリプト開始") print("=" * 50) # gifurogeからデータを取得 gps_data = self.get_gps_information_data() if not gps_data: print("❌ 移行するデータがありません") return print(f"\n📋 {len(gps_data)}件のGPS記録の移行を開始...") # バッチ処理で移行 batch_size = 100 total_batches = (len(gps_data) + batch_size - 1) // batch_size for batch_num in range(total_batches): start_idx = batch_num * batch_size end_idx = min(start_idx + batch_size, len(gps_data)) batch_data = gps_data[start_idx:end_idx] print(f"\n📦 バッチ {batch_num + 1}/{total_batches} ({len(batch_data)}件) 処理中...") with transaction.atomic(): for record in batch_data: self.migrate_gps_record(record) # 統計レポート self.print_migration_report() def print_migration_report(self): """移行結果レポート""" print("\n" + "=" * 50) print("📊 GPS情報移行完了レポート") print("=" * 50) print(f"📋 総GPS記録数: {self.stats['total_gps_info']}") print(f"✅ 移行済みGpsLog: {self.stats['migrated_gps_logs']}") print(f"✅ 移行済みCheckin: {self.stats['migrated_checkins']}") print(f"⚠️ スキップ記録: {self.stats['skipped_records']}") print(f"❌ エラー数: {self.stats['errors']}") if self.stats['error_details']: print("\n❌ エラー詳細:") for error in self.stats['error_details'][:10]: # 最初の10個だけ表示 print(f" - {error}") if len(self.stats['error_details']) > 10: print(f" ... 他 {len(self.stats['error_details']) - 10} 件") success_rate = (self.stats['migrated_gps_logs'] / max(self.stats['total_gps_info'], 1)) * 100 print(f"\n📈 移行成功率: {success_rate:.1f}%") print("=" * 50) def main(): """メイン実行関数""" migrator = GpsInformationMigrator() migrator.run_migration() if __name__ == '__main__': main()