#!/usr/bin/env python3 """ rog_team テーブル専用移行スクリプト (構造変換対応) old_rogdb の rog_team から rogdb の rog_team へ構造変換を行いながらデータ移行 """ import os import sys import psycopg2 from datetime import datetime, timezone import logging # ログ設定 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) # データベース設定 OLD_ROGDB_CONFIG = { 'host': os.getenv('OLD_ROGDB_HOST', 'postgres-db'), 'database': os.getenv('OLD_ROGDB_NAME', 'old_rogdb'), 'user': os.getenv('OLD_ROGDB_USER', 'admin'), 'password': os.getenv('OLD_ROGDB_PASSWORD', 'admin123456'), 'port': int(os.getenv('OLD_ROGDB_PORT', 5432)) } ROGDB_CONFIG = { 'host': os.getenv('ROGDB_HOST', 'postgres-db'), 'database': os.getenv('ROGDB_NAME', 'rogdb'), 'user': os.getenv('ROGDB_USER', 'admin'), 'password': os.getenv('ROGDB_PASSWORD', 'admin123456'), 'port': int(os.getenv('ROGDB_PORT', 5432)) } class RogTeamMigrator: """rog_team テーブル専用移行クラス""" def __init__(self): self.old_conn = None self.new_conn = None self.old_cursor = None self.new_cursor = None self.default_event_id = None def connect_databases(self): """データベース接続""" try: logger.info("データベースに接続中...") self.old_conn = psycopg2.connect(**OLD_ROGDB_CONFIG) self.new_conn = psycopg2.connect(**ROGDB_CONFIG) self.old_conn.autocommit = True self.new_conn.autocommit = False self.old_cursor = self.old_conn.cursor() self.new_cursor = self.new_conn.cursor() logger.info("✅ データベース接続成功") return True except Exception as e: logger.error(f"❌ データベース接続エラー: {e}") return False def close_connections(self): """データベース接続クローズ""" try: if self.old_cursor: self.old_cursor.close() if self.new_cursor: self.new_cursor.close() if self.old_conn: self.old_conn.close() if self.new_conn: self.new_conn.close() logger.info("データベース接続をクローズしました") except Exception as e: logger.warning(f"接続クローズ時の警告: {e}") def get_team_zekken_and_event(self, team_id): """team_idから最新のrog_entryのzekken_numberとevent_idを取得""" try: # 旧DBのrog_entryから該当team_idの最新レコードを取得 self.old_cursor.execute(""" SELECT zekken_number, event_id, date FROM rog_entry WHERE team_id = %s AND zekken_number IS NOT NULL AND event_id IS NOT NULL ORDER BY date DESC, id DESC LIMIT 1 """, (team_id,)) result = self.old_cursor.fetchone() if result: zekken_number, event_id, entry_date = result logger.debug(f"team_id {team_id}: zekken_number={zekken_number}, event_id={event_id}, date={entry_date}") return str(zekken_number), event_id else: # rog_entryにレコードがない場合はデフォルト値を返す logger.warning(f"team_id {team_id}: rog_entryにレコードが見つかりません") return '', self.default_event_id except Exception as e: logger.error(f"team_id {team_id} のzekken_number/event_id取得エラー: {e}") return '', self.default_event_id def get_default_event_id(self): """デフォルトのevent_idを取得または作成""" try: # 既存のイベントを探す self.new_cursor.execute(""" SELECT id FROM rog_newevent2 WHERE event_name LIKE '%移行%' OR event_name LIKE '%default%' ORDER BY id LIMIT 1 """) result = self.new_cursor.fetchone() if result: event_id = result[0] logger.info(f"既存のデフォルトイベントを使用: event_id = {event_id}") return event_id # なければ最初のイベントを使用 self.new_cursor.execute(""" SELECT id FROM rog_newevent2 ORDER BY id LIMIT 1 """) result = self.new_cursor.fetchone() if result: event_id = result[0] logger.info(f"最初のイベントをデフォルトとして使用: event_id = {event_id}") return event_id # イベントがない場合はエラー logger.error("❌ rog_newevent2 テーブルにイベントが存在しません") return None except Exception as e: logger.error(f"❌ デフォルトevent_id取得エラー: {e}") return None def get_category_mapping(self): """カテゴリIDのマッピングを確認""" try: # 旧DBのカテゴリ self.old_cursor.execute("SELECT id, category_name FROM rog_newcategory ORDER BY id") old_categories = dict(self.old_cursor.fetchall()) # 新DBのカテゴリ self.new_cursor.execute("SELECT id, category_name FROM rog_newcategory ORDER BY id") new_categories = dict(self.new_cursor.fetchall()) logger.info(f"旧DB カテゴリ: {old_categories}") logger.info(f"新DB カテゴリ: {new_categories}") # 名前ベースでマッピング作成 category_mapping = {} for old_id, old_name in old_categories.items(): for new_id, new_name in new_categories.items(): if old_name == new_name: category_mapping[old_id] = new_id break else: # マッチしない場合は最初のカテゴリを使用 if new_categories: category_mapping[old_id] = min(new_categories.keys()) logger.warning(f"カテゴリマッピング失敗 - デフォルト使用: {old_id} -> {category_mapping[old_id]}") return category_mapping except Exception as e: logger.error(f"❌ カテゴリマッピング取得エラー: {e}") return {} def convert_team_record(self, old_record, category_mapping): """旧レコードを新レコード形式に変換""" old_id, old_team_name, old_category_id, old_owner_id = old_record # team_idから最新のzekken_numberとevent_idを取得 zekken_number, event_id = self.get_team_zekken_and_event(old_id) # 新しいレコード作成 new_record = { 'id': old_id, 'team_name': old_team_name, 'category_id': category_mapping.get(old_category_id, old_category_id), 'owner_id': old_owner_id, # 新しいフィールドにデフォルト値を設定 'class_name': '', # 空文字列 'event_id': event_id, # rog_entryから取得したevent_id 'location': None, # PostGIS座標は後で設定可能 'password': '', # パスワードなし 'trial': False, # 本番チーム 'zekken_number': zekken_number, # rog_entryから取得したzekken_number 'created_at': datetime.now(timezone.utc), 'updated_at': datetime.now(timezone.utc) } return new_record def migrate_rog_team(self): """rog_team テーブルのデータ移行""" logger.info("=" * 60) logger.info("rog_team テーブル構造変換移行開始") logger.info("=" * 60) try: # デフォルトevent_id取得 self.default_event_id = self.get_default_event_id() if not self.default_event_id: return False # カテゴリマッピング取得 category_mapping = self.get_category_mapping() # 旧データ取得 logger.info("旧rog_teamデータを取得中...") self.old_cursor.execute(""" SELECT id, team_name, category_id, owner_id FROM rog_team ORDER BY id """) old_records = self.old_cursor.fetchall() if not old_records: logger.info("✅ 移行対象データがありません") return True logger.info(f"移行対象レコード数: {len(old_records)}件") # 統計情報 inserted_count = 0 updated_count = 0 error_count = 0 zekken_resolved_count = 0 constraint_avoided_count = 0 # レコード別処理 for i, old_record in enumerate(old_records): try: # レコード変換 new_record = self.convert_team_record(old_record, category_mapping) team_id = new_record['id'] # zekken_number解決統計 if new_record['zekken_number']: zekken_resolved_count += 1 # 既存レコード確認（IDベース） self.new_cursor.execute( "SELECT COUNT(*) FROM rog_team WHERE id = %s", (team_id,) ) exists_by_id = self.new_cursor.fetchone()[0] > 0 # 重複制約確認（zekken_number + event_id の組み合わせ） if new_record['zekken_number']: # zekken_numberが空でない場合のみチェック self.new_cursor.execute( "SELECT COUNT(*) FROM rog_team WHERE zekken_number = %s AND event_id = %s", (new_record['zekken_number'], new_record['event_id']) ) exists_by_constraint = self.new_cursor.fetchone()[0] > 0 if exists_by_constraint and not exists_by_id: # 制約違反が発生する場合は、zekken_numberを空にしてデフォルトevent_idを使用 logger.warning(f"Team ID {team_id}: zekken_number制約回避のため空文字に変更") new_record['zekken_number'] = '' new_record['event_id'] = self.default_event_id constraint_avoided_count += 1 if exists_by_id: # UPDATE処理 update_query = """ UPDATE rog_team SET team_name = %s, category_id = %s, owner_id = %s, class_name = %s, event_id = %s, location = %s, password = %s, trial = %s, zekken_number = %s, updated_at = %s WHERE id = %s """ self.new_cursor.execute(update_query, ( new_record['team_name'], new_record['category_id'], new_record['owner_id'], new_record['class_name'], new_record['event_id'], new_record['location'], new_record['password'], new_record['trial'], new_record['zekken_number'], new_record['updated_at'], team_id )) updated_count += 1 else: # INSERT処理 insert_query = """ INSERT INTO rog_team ( id, team_name, category_id, owner_id, class_name, event_id, location, password, trial, zekken_number, created_at, updated_at ) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s ) """ self.new_cursor.execute(insert_query, ( new_record['id'], new_record['team_name'], new_record['category_id'], new_record['owner_id'], new_record['class_name'], new_record['event_id'], new_record['location'], new_record['password'], new_record['trial'], new_record['zekken_number'], new_record['created_at'], new_record['updated_at'] )) inserted_count += 1 # 進捗表示 if (i + 1) % 50 == 0: self.new_conn.commit() logger.info(f" 進捗: {i + 1}/{len(old_records)} 件処理完了") except Exception as e: error_count += 1 logger.error(f" レコード処理エラー (ID: {old_record[0]}): {e}") # トランザクションロールバック try: self.new_conn.rollback() except: pass if error_count > 10: logger.error("❌ エラー数が上限を超えました") break # 最終コミット self.new_conn.commit() # 結果サマリー logger.info("=" * 60) logger.info("rog_team 移行完了") logger.info("=" * 60) logger.info(f"挿入: {inserted_count}件") logger.info(f"更新: {updated_count}件") logger.info(f"エラー: {error_count}件") logger.info(f"zekken_number解決: {zekken_resolved_count}件") logger.info(f"制約回避: {constraint_avoided_count}件") logger.info(f"総処理: {len(old_records)}件") if error_count == 0: logger.info("✅ rog_team移行が正常に完了しました！") return True else: logger.warning(f"⚠️ {error_count}件のエラーがありました") return False except Exception as e: logger.error(f"❌ rog_team移行エラー: {e}") try: self.new_conn.rollback() except: pass return False def run(self): """移行実行""" try: if not self.connect_databases(): return False return self.migrate_rog_team() finally: self.close_connections() def main(): """メイン処理""" logger.info("rog_team テーブル構造変換移行スクリプト") migrator = RogTeamMigrator() success = migrator.run() if success: logger.info("🎉 移行が正常に完了しました！") else: logger.error("💥 移行中にエラーが発生しました") sys.exit(0 if success else 1) if __name__ == "__main__": main()