#!/usr/bin/env python3 """ Old RogDB データ移行スクリプト (エラー修正版) old_rogdb の rog_* テーブルから rogdb の rog_* テーブルへデータを更新・挿入 """ import os import sys import psycopg2 from datetime import datetime, timezone from typing import Optional, Dict, List, Tuple 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)) } # PostgreSQL予約語をクォートで囲む必要があるカラム RESERVED_KEYWORDS = { 'like', 'order', 'group', 'user', 'table', 'where', 'select', 'insert', 'update', 'delete', 'create', 'drop', 'alter', 'index', 'constraint' } class RogTableMigrator: """Rogテーブル移行クラス (エラー修正版)""" def __init__(self): self.old_conn = None self.new_conn = None self.old_cursor = None self.new_cursor = None self.migration_stats = {} def quote_column_if_needed(self, column_name): """予約語やキャメルケースの場合はダブルクォートで囲む""" # 予約語チェック if column_name.lower() in RESERVED_KEYWORDS: return f'"{column_name}"' # キャメルケースや大文字が含まれる場合もクォート if any(c.isupper() for c in column_name) or column_name != column_name.lower(): return f'"{column_name}"' return column_name def connect_databases(self): """データベース接続""" try: logger.info("データベースに接続中...") self.old_conn = psycopg2.connect(**OLD_ROGDB_CONFIG) self.new_conn = psycopg2.connect(**ROGDB_CONFIG) # autocommitを有効にしてトランザクション問題を回避 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_rog_tables(self): """rog_で始まるテーブル一覧を取得""" try: # old_rogdbのrog_テーブル一覧 self.old_cursor.execute(""" SELECT table_name FROM information_schema.tables WHERE table_schema = 'public' AND table_name LIKE 'rog_%' ORDER BY table_name """) old_tables = [row[0] for row in self.old_cursor.fetchall()] # rogdbのrog_テーブル一覧 self.new_cursor.execute(""" SELECT table_name FROM information_schema.tables WHERE table_schema = 'public' AND table_name LIKE 'rog_%' ORDER BY table_name """) new_tables = [row[0] for row in self.new_cursor.fetchall()] # 共通テーブル common_tables = list(set(old_tables) & set(new_tables)) logger.info(f"old_rogdb rog_テーブル: {len(old_tables)}個") logger.info(f"rogdb rog_テーブル: {len(new_tables)}個") logger.info(f"共通 rog_テーブル: {len(common_tables)}個") return old_tables, new_tables, common_tables except Exception as e: logger.error(f"テーブル一覧取得エラー: {e}") return [], [], [] def get_table_structure(self, table_name, cursor): """テーブル構造を取得 (エラーハンドリング強化)""" try: cursor.execute(""" SELECT column_name, data_type, is_nullable, column_default FROM information_schema.columns WHERE table_name = %s AND table_schema = 'public' ORDER BY ordinal_position """, (table_name,)) columns = cursor.fetchall() return { 'columns': [col[0] for col in columns], 'details': columns } except Exception as e: logger.error(f"テーブル構造取得エラー ({table_name}): {e}") # トランザクションエラーを回避 try: cursor.connection.rollback() except: pass return {'columns': [], 'details': []} def get_primary_key(self, table_name, cursor): """主キーカラムを取得""" try: cursor.execute(""" SELECT a.attname FROM pg_index i JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey) WHERE i.indrelid = %s::regclass AND i.indisprimary """, (table_name,)) pk_columns = [row[0] for row in cursor.fetchall()] return pk_columns if pk_columns else ['id'] except Exception as e: logger.warning(f"主キー取得エラー ({table_name}): {e}") return ['id'] # デフォルトでidを使用 def migrate_table_data(self, table_name): """個別テーブルのデータ移行 (エラー修正版)""" logger.info(f"\n=== {table_name} データ移行開始 ===") try: # テーブル構造比較 old_structure = self.get_table_structure(table_name, self.old_cursor) new_structure = self.get_table_structure(table_name, self.new_cursor) old_columns = set(old_structure['columns']) new_columns = set(new_structure['columns']) common_columns = old_columns & new_columns if not common_columns: logger.warning(f"⚠️ {table_name}: 共通カラムがありません") return {'inserted': 0, 'updated': 0, 'errors': 0} logger.info(f"共通カラム ({len(common_columns)}個): {sorted(common_columns)}") # 主キー取得 pk_columns = self.get_primary_key(table_name, self.new_cursor) logger.info(f"主キー: {pk_columns}") # カラム名を予約語対応でクォート quoted_columns = [self.quote_column_if_needed(col) for col in common_columns] columns_str = ', '.join(quoted_columns) # old_rogdbからデータ取得 try: self.old_cursor.execute(f"SELECT {columns_str} FROM {table_name}") old_records = self.old_cursor.fetchall() except Exception as e: logger.error(f"❌ {table_name} データ取得エラー: {e}") return {'inserted': 0, 'updated': 0, 'errors': 1} if not old_records: logger.info(f"✅ {table_name}: 移行対象データなし") return {'inserted': 0, 'updated': 0, 'errors': 0} logger.info(f"移行対象レコード数: {len(old_records)}件") # 統計情報 inserted_count = 0 updated_count = 0 error_count = 0 # レコード移行処理 for i, record in enumerate(old_records): try: # レコードデータを辞書形式に変換 record_dict = dict(zip(common_columns, record)) # 主キー値を取得 pk_values = [] pk_conditions = [] for pk_col in pk_columns: if pk_col in record_dict: pk_values.append(record_dict[pk_col]) quoted_pk_col = self.quote_column_if_needed(pk_col) pk_conditions.append(f"{quoted_pk_col} = %s") if not pk_values: error_count += 1 continue # 既存レコード確認 check_query = f"SELECT COUNT(*) FROM {table_name} WHERE {' AND '.join(pk_conditions)}" self.new_cursor.execute(check_query, pk_values) exists = self.new_cursor.fetchone()[0] > 0 if exists: # UPDATE処理 set_clauses = [] update_values = [] for col in common_columns: if col not in pk_columns: # 主キー以外を更新 quoted_col = self.quote_column_if_needed(col) set_clauses.append(f"{quoted_col} = %s") update_values.append(record_dict[col]) if set_clauses: update_query = f""" UPDATE {table_name} SET {', '.join(set_clauses)} WHERE {' AND '.join(pk_conditions)} """ self.new_cursor.execute(update_query, update_values + pk_values) updated_count += 1 else: # INSERT処理 insert_columns = list(common_columns) insert_values = [record_dict[col] for col in insert_columns] quoted_insert_columns = [self.quote_column_if_needed(col) for col in insert_columns] placeholders = ', '.join(['%s'] * len(insert_columns)) insert_query = f""" INSERT INTO {table_name} ({', '.join(quoted_insert_columns)}) VALUES ({placeholders}) """ self.new_cursor.execute(insert_query, insert_values) inserted_count += 1 # 定期的なコミット if (i + 1) % 100 == 0: self.new_conn.commit() logger.info(f" 進捗: {i + 1}/{len(old_records)} 件処理完了") except Exception as e: error_count += 1 logger.error(f" レコード処理エラー (行{i+1}): {e}") # トランザクションロールバック try: self.new_conn.rollback() except: pass if error_count > 10: # エラー上限 logger.error(f"❌ {table_name}: エラー数が上限を超えました") break # 最終コミット try: self.new_conn.commit() except Exception as e: logger.error(f"コミットエラー: {e}") # 結果サマリー stats = { 'inserted': inserted_count, 'updated': updated_count, 'errors': error_count, 'total': len(old_records) } logger.info(f"✅ {table_name} 移行完了:") logger.info(f" 挿入: {inserted_count}件") logger.info(f" 更新: {updated_count}件") logger.info(f" エラー: {error_count}件") self.migration_stats[table_name] = stats return stats except Exception as e: logger.error(f"❌ {table_name} 移行エラー: {e}") try: self.new_conn.rollback() except: pass return {'inserted': 0, 'updated': 0, 'errors': 1} def run_migration(self, exclude_tables=None): """全体移行実行""" if exclude_tables is None: exclude_tables = [] logger.info("=" * 80) logger.info("Old RogDB → RogDB データ移行開始") logger.info("=" * 80) try: # データベース接続 if not self.connect_databases(): return False # テーブル一覧取得 old_tables, new_tables, common_tables = self.get_rog_tables() if not common_tables: logger.error("❌ 移行対象テーブルがありません") return False # 除外テーブルをフィルタリング migration_tables = [t for t in common_tables if t not in exclude_tables] logger.info(f"移行対象テーブル ({len(migration_tables)}個): {migration_tables}") # 各テーブルを移行 total_inserted = 0 total_updated = 0 total_errors = 0 for table_name in migration_tables: stats = self.migrate_table_data(table_name) total_inserted += stats['inserted'] total_updated += stats['updated'] total_errors += stats['errors'] # 全体サマリー出力 logger.info("=" * 80) logger.info("移行完了サマリー") logger.info("=" * 80) logger.info(f"処理対象テーブル: {len(migration_tables)}個") logger.info(f"総挿入件数: {total_inserted}件") logger.info(f"総更新件数: {total_updated}件") logger.info(f"総エラー件数: {total_errors}件") logger.info("\n--- テーブル別詳細 ---") for table_name, stats in self.migration_stats.items(): logger.info(f"{table_name}: 挿入{stats['inserted']}, 更新{stats['updated']}, エラー{stats['errors']}") if total_errors > 0: logger.warning(f"⚠️ {total_errors}件のエラーがありました") else: logger.info("✅ 全ての移行が正常に完了しました！") return True except Exception as e: logger.error(f"❌ 移行処理エラー: {e}") return False finally: self.close_connections() def main(): """メイン処理""" logger.info("Old RogDB → RogDB データ移行スクリプト") # 除外テーブル設定 exclude_tables = [] if os.getenv('EXCLUDE_TABLES'): exclude_tables = [t.strip() for t in os.getenv('EXCLUDE_TABLES').split(',')] logger.info(f"除外テーブル: {exclude_tables}") # 移行実行 migrator = RogTableMigrator() success = migrator.run_migration(exclude_tables=exclude_tables) if success: logger.info("移行処理が完了しました") sys.exit(0) else: logger.error("移行処理が失敗しました") sys.exit(1) if __name__ == "__main__": main()