import logging from django.core.management.base import BaseCommand from django.db import connections, transaction, connection from django.db.models import Q from django.contrib.gis.geos import Point from django.utils import timezone from rog.models import Team, NewEvent2, Checkpoint, GpsCheckin, GpsLog, Entry from datetime import datetime import psycopg2 from psycopg2.extras import execute_values logger = logging.getLogger(__name__) class Command(BaseCommand): help = 'MobServerデータベースからDjangoモデルにデータを移行します' def add_arguments(self, parser): parser.add_argument( '--dry-run', action='store_true', dest='dry_run', help='実際の移行を行わず、処理内容のみを表示します', ) parser.add_argument( '--batch-size', type=int, default=100, help='バッチサイズ（デフォルト: 100）' ) def handle(self, *args, **options): dry_run = options['dry_run'] batch_size = options['batch_size'] if dry_run: self.stdout.write(self.style.WARNING('ドライランモードで実行中...')) # MobServerデータベース接続を取得 mobserver_conn = connections['mobserver'] try: with transaction.atomic(): self.migrate_events(mobserver_conn, dry_run, batch_size) self.migrate_teams(mobserver_conn, dry_run, batch_size) self.migrate_checkpoints(mobserver_conn, dry_run, batch_size) self.migrate_gps_logs(mobserver_conn, dry_run, batch_size) if dry_run: raise transaction.TransactionManagementError("ドライランのためロールバックします") except transaction.TransactionManagementError: if dry_run: self.stdout.write(self.style.SUCCESS('ドライランが完了しました（変更は保存されていません）')) else: raise except Exception as e: logger.error(f'データ移行エラー: {e}') self.stdout.write(self.style.ERROR(f'エラーが発生しました: {e}')) raise def migrate_events(self, conn, dry_run, batch_size): """イベント情報を移行""" self.stdout.write('イベント情報を移行中...') with conn.cursor() as cursor: cursor.execute("SELECT * FROM event_table") rows = cursor.fetchall() events_migrated = 0 batch_data = [] for row in rows: (event_code, event_name, start_time, event_day) = row # start_timeのデータクリーニング cleaned_start_time = start_time if start_time and isinstance(start_time, str): # セミコロンをコロンに置換 cleaned_start_time = start_time.replace(';', ':') # タイムゾーン情報を含む場合は時間部分のみ抽出 if '+' in cleaned_start_time or 'T' in cleaned_start_time: try: from datetime import datetime dt = datetime.fromisoformat(cleaned_start_time.replace('Z', '+00:00')) cleaned_start_time = dt.strftime('%H:%M:%S') except: cleaned_start_time = None if not dry_run: batch_data.append(NewEvent2( event_code=event_code, event_name=event_name, event_day=event_day, start_time=cleaned_start_time, )) events_migrated += 1 # バッチ処理 if len(batch_data) >= batch_size: if not dry_run: NewEvent2.objects.bulk_create(batch_data, ignore_conflicts=True) batch_data = [] # 残りのデータを処理 if batch_data and not dry_run: NewEvent2.objects.bulk_create(batch_data, ignore_conflicts=True) self.stdout.write(f'{events_migrated}件のイベントを移行しました') def migrate_teams(self, conn, dry_run, batch_size): """チーム情報を移行""" self.stdout.write('チーム情報を移行中...') with conn.cursor() as cursor: cursor.execute("SELECT * FROM team_table") rows = cursor.fetchall() teams_migrated = 0 batch_data = [] for row in rows: (zekken_number, event_code, team_name, class_name, password, trial) = row # 対応するイベントを取得 try: event = NewEvent2.objects.get(event_code=event_code) except NewEvent2.DoesNotExist: self.stdout.write(self.style.WARNING(f' 警告: イベント {event_code} が見つかりません。スキップします。')) continue if not dry_run: batch_data.append(Team( zekken_number=zekken_number, team_name=team_name, event=event, class_name=class_name, password=password, trial=trial, )) teams_migrated += 1 # バッチ処理 if len(batch_data) >= batch_size: if not dry_run: Team.objects.bulk_create(batch_data, ignore_conflicts=True) batch_data = [] # 残りのデータを処理 if batch_data and not dry_run: Team.objects.bulk_create(batch_data, ignore_conflicts=True) self.stdout.write(f'{teams_migrated}件のチームを移行しました') def migrate_checkpoints(self, conn, dry_run, batch_size): """チェックポイント情報を移行""" self.stdout.write('チェックポイント情報を移行中...') with conn.cursor() as cursor: cursor.execute("SELECT * FROM checkpoint_table") rows = cursor.fetchall() checkpoints_migrated = 0 batch_data = [] for row in rows: (cp_number, event_code, cp_name, latitude, longitude, photo_point, buy_point, sample_photo, colabo_company_memo) = row # 対応するイベントを取得 try: event = NewEvent2.objects.get(event_code=event_code) except NewEvent2.DoesNotExist: continue # 位置情報の処理 location = None if latitude is not None and longitude is not None: try: location = Point(longitude, latitude) # Pointは(longitude, latitude)の順序 except (ValueError, AttributeError): pass if not dry_run: batch_data.append(Checkpoint( cp_number=cp_number, event=event, cp_name=cp_name, location=location, photo_point=photo_point or 0, buy_point=buy_point or 0, sample_photo=sample_photo, colabo_company_memo=colabo_company_memo, )) checkpoints_migrated += 1 # バッチ処理 if len(batch_data) >= batch_size: if not dry_run: Checkpoint.objects.bulk_create(batch_data, ignore_conflicts=True) batch_data = [] # 残りのデータを処理 if batch_data and not dry_run: Checkpoint.objects.bulk_create(batch_data, ignore_conflicts=True) self.stdout.write(f'{checkpoints_migrated}件のチェックポイントを移行しました') def migrate_gps_logs(self, conn, dry_run, batch_size): """GPS位置情報を移行""" print('GPS位置情報を移行中...') # チームとイベントのマッピングを作成 team_to_event_map = {} for team in Team.objects.select_related('event'): if team.event: # eventがNoneでないことを確認 team_to_event_map[team.zekken_number] = team.event.id # チェックポイントのマッピングを作成 checkpoint_id_map = {} for checkpoint in Checkpoint.objects.select_related('event'): if checkpoint.event: # eventがNoneでないことを確認 key = (checkpoint.event.event_code, checkpoint.cp_number) checkpoint_id_map[key] = checkpoint.id with conn.cursor() as cursor: cursor.execute("SELECT * FROM gps_information") rows = cursor.fetchall() logs_migrated = 0 batch_data = [] for row in rows: (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) = row # 対応するチームを取得 try: event = NewEvent2.objects.get(event_code=event_code) team = Team.objects.get(zekken_number=zekken_number, event=event) # teamが存在し、eventも存在することを確認 if not team or not team.event: continue except (NewEvent2.DoesNotExist, Team.DoesNotExist): continue # 対応するチェックポイントを取得（存在する場合） checkpoint = None if cp_number is not None and cp_number != -1: try: checkpoint = Checkpoint.objects.get(cp_number=cp_number, event=event) except Checkpoint.DoesNotExist: pass # checkin_timeの設定（必須フィールド） checkin_time = timezone.now() # デフォルト値 if goal_time: try: # goal_timeはHH:MM形式と仮定 from datetime import datetime, time parsed_time = datetime.strptime(goal_time, '%H:%M').time() if create_at: checkin_time = timezone.make_aware(datetime.combine(create_at.date(), parsed_time)) else: checkin_time = timezone.make_aware(datetime.combine(datetime.now().date(), parsed_time)) except: checkin_time = timezone.make_aware(create_at) if create_at else timezone.now() elif create_at: checkin_time = timezone.make_aware(create_at) if timezone.is_naive(create_at) else create_at if not dry_run: # GpsCheckinテーブル用のデータ batch_data.append({ 'event_code': event_code, 'zekken': zekken_number, 'serial_number': serial_number, 'cp_number': cp_number or 0, 'lat': None, # 実際のMobServerデータベースから取得 'lng': None, # 実際のMobServerデータベースから取得 'checkin_time': checkin_time, 'record_time': timezone.make_aware(create_at) if create_at and timezone.is_naive(create_at) else (create_at or timezone.now()), 'location': "", # PostGISポイントは後で設定 'mobserver_id': serial_number, 'event_id': team_to_event_map.get(zekken_number), 'team_id': team.id, 'checkpoint_id': checkpoint.id if checkpoint else None }) logs_migrated += 1 # バッチ処理 if len(batch_data) >= batch_size: if not dry_run: self.bulk_insert_gps_logs(batch_data) batch_data = [] # 残りのデータを処理 if batch_data and not dry_run: self.bulk_insert_gps_logs(batch_data) print(f'{logs_migrated}件のGPS位置情報を移行しました') def bulk_insert_gps_logs(self, batch_data): """ GpsCheckinテーブルに直接SQLを使って挿入 """ if not batch_data: return with connection.cursor() as cursor: # DjangoのGpsCheckinテーブルに挿入 insert_sql = """ INSERT INTO rog_gpscheckin ( event_code, zekken, serial_number, cp_number, lat, lng, checkin_time, record_time, location, mobserver_id, event_id, team_id, checkpoint_id ) VALUES %s ON CONFLICT DO NOTHING """ # locationフィールドを除外してバリューを準備 clean_values = [] for data in batch_data: # lat/lngがある場合はPostGISポイントを作成、ない場合はNULL if data['lat'] is not None and data['lng'] is not None: location_point = f"ST_GeomFromText('POINT({data['lng']} {data['lat']})', 4326)" else: location_point = None clean_values.append(( data['event_code'], data['zekken'], data['serial_number'], data['cp_number'], data['lat'], data['lng'], data['checkin_time'], data['record_time'], location_point, data['mobserver_id'], data['event_id'], data['team_id'], data['checkpoint_id'] )) try: execute_values(cursor, insert_sql, clean_values, template=None, page_size=100) except Exception as e: logger.error(f"データ移行エラー: {e}") raise