482 lines
19 KiB
Python
482 lines
19 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
統合移行スクリプト：GPS情報移行 + 写真記録からチェックイン記録生成
|
||
- gifurogeからrogdbへのGPS情報移行
|
||
- 写真記録を正とした不足チェックイン記録の補完
|
||
- 統計情報の出力と動作確認
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import psycopg2
|
||
from datetime import datetime, timedelta
|
||
import pytz
|
||
from typing import Optional, Dict, Any, List, Tuple
|
||
|
||
class MigrationWithPhotoIntegration:
|
||
def __init__(self):
|
||
self.conn_gif = None
|
||
self.conn_rog = None
|
||
self.cur_gif = None
|
||
self.cur_rog = None
|
||
|
||
def connect_databases(self):
|
||
"""データベースに接続"""
|
||
try:
|
||
self.conn_gif = psycopg2.connect(
|
||
host='postgres-db',
|
||
database='gifuroge',
|
||
user=os.environ.get('POSTGRES_USER'),
|
||
password=os.environ.get('POSTGRES_PASS')
|
||
)
|
||
self.conn_rog = psycopg2.connect(
|
||
host='postgres-db',
|
||
database='rogdb',
|
||
user=os.environ.get('POSTGRES_USER'),
|
||
password=os.environ.get('POSTGRES_PASS')
|
||
)
|
||
|
||
self.cur_gif = self.conn_gif.cursor()
|
||
self.cur_rog = self.conn_rog.cursor()
|
||
|
||
print("✅ データベース接続成功")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ データベース接続エラー: {e}")
|
||
return False
|
||
|
||
def get_event_date(self, event_code: str) -> str:
|
||
"""イベントコードから適切な日付を取得"""
|
||
event_dates = {
|
||
'養老２': '2024-10-06',
|
||
'美濃加茂': '2024-05-19', # 修正済み
|
||
'下呂': '2023-05-20',
|
||
'FC岐阜': '2024-10-18',
|
||
'大垣': '2023-11-25',
|
||
'岐阜市': '2023-10-21',
|
||
'default': '2024-01-01'
|
||
}
|
||
return event_dates.get(event_code, event_dates['default'])
|
||
|
||
def convert_utc_to_jst(self, utc_timestamp: datetime) -> Optional[datetime]:
|
||
"""UTC時刻をJST時刻に変換"""
|
||
if not utc_timestamp:
|
||
return None
|
||
|
||
utc_tz = pytz.UTC
|
||
jst_tz = pytz.timezone('Asia/Tokyo')
|
||
|
||
if utc_timestamp.tzinfo is None:
|
||
utc_timestamp = utc_tz.localize(utc_timestamp)
|
||
|
||
return utc_timestamp.astimezone(jst_tz).replace(tzinfo=None)
|
||
|
||
def parse_goal_time(self, goal_time_str: str, event_code: str) -> Optional[datetime]:
|
||
"""goal_time文字列をパース（時刻のみの場合は変換なし）"""
|
||
if not goal_time_str:
|
||
return None
|
||
|
||
# 時刻のみ（HH:MM:SS形式）の場合はJSTの時刻として扱い、UTCからの変換は行わない
|
||
if len(goal_time_str) <= 8 and goal_time_str.count(':') <= 2:
|
||
event_date = self.get_event_date(event_code)
|
||
event_datetime = datetime.strptime(event_date, '%Y-%m-%d')
|
||
time_part = datetime.strptime(goal_time_str, '%H:%M:%S').time()
|
||
return datetime.combine(event_datetime.date(), time_part)
|
||
else:
|
||
# 完全な日時形式の場合はUTCからJSTに変換
|
||
return self.convert_utc_to_jst(datetime.fromisoformat(goal_time_str.replace('Z', '+00:00')))
|
||
|
||
def migrate_gps_information(self) -> Dict[str, int]:
|
||
"""GPS情報の移行処理"""
|
||
print("\n=== GPS情報移行開始 ===")
|
||
|
||
# 既存の移行データ件数を確認
|
||
self.cur_rog.execute('SELECT COUNT(*) FROM rog_gpscheckin;')
|
||
existing_count = self.cur_rog.fetchone()[0]
|
||
print(f"既存チェックイン記録: {existing_count:,}件")
|
||
|
||
# 移行対象データ取得
|
||
self.cur_gif.execute("""
|
||
SELECT
|
||
serial_number, zekken_number, event_code, create_at,
|
||
goal_time, cp_number, late_point
|
||
FROM gps_information
|
||
ORDER BY event_code, zekken_number, create_at;
|
||
""")
|
||
|
||
all_records = self.cur_gif.fetchall()
|
||
print(f"移行対象データ: {len(all_records):,}件")
|
||
|
||
# 最大serial_numberを取得
|
||
self.cur_rog.execute("SELECT MAX(serial_number::integer) FROM rog_gpscheckin WHERE serial_number ~ '^[0-9]+$';")
|
||
max_serial_result = self.cur_rog.fetchone()
|
||
next_serial = (max_serial_result[0] if max_serial_result[0] else 0) + 1
|
||
|
||
migrated_count = 0
|
||
skipped_count = 0
|
||
errors = []
|
||
|
||
for i, record in enumerate(all_records):
|
||
serial_number, zekken_number, event_code, create_at, goal_time, cp_number, late_point = record
|
||
|
||
try:
|
||
# 重複チェック（serial_numberベース）
|
||
self.cur_rog.execute("""
|
||
SELECT COUNT(*) FROM rog_gpscheckin
|
||
WHERE serial_number = %s;
|
||
""", (str(serial_number),))
|
||
|
||
if self.cur_rog.fetchone()[0] > 0:
|
||
continue # 既に存在
|
||
|
||
# データ変換
|
||
converted_checkin_time = self.convert_utc_to_jst(create_at)
|
||
converted_record_time = self.convert_utc_to_jst(create_at)
|
||
|
||
# serial_number重複回避
|
||
if serial_number < 1000:
|
||
new_serial = 30000 + serial_number # 30000番台に移動
|
||
else:
|
||
new_serial = serial_number
|
||
|
||
# 挿入
|
||
self.cur_rog.execute("""
|
||
INSERT INTO rog_gpscheckin (
|
||
event_code, zekken, serial_number, cp_number,
|
||
checkin_time, record_time
|
||
) VALUES (%s, %s, %s, %s, %s, %s)
|
||
""", (
|
||
event_code,
|
||
zekken_number,
|
||
str(new_serial),
|
||
str(cp_number),
|
||
converted_checkin_time,
|
||
converted_record_time
|
||
))
|
||
|
||
migrated_count += 1
|
||
|
||
if migrated_count % 1000 == 0:
|
||
print(f" 進捗: {migrated_count:,}件移行完了")
|
||
|
||
except Exception as e:
|
||
errors.append(f'レコード {serial_number}: {str(e)[:100]}')
|
||
skipped_count += 1
|
||
|
||
self.conn_rog.commit()
|
||
|
||
print(f"GPS情報移行完了: 成功 {migrated_count:,}件, スキップ {skipped_count:,}件")
|
||
if errors:
|
||
print(f"エラー件数: {len(errors)}件")
|
||
|
||
return {
|
||
'migrated': migrated_count,
|
||
'skipped': skipped_count,
|
||
'errors': len(errors)
|
||
}
|
||
|
||
def generate_checkin_from_photos(self) -> Dict[str, int]:
|
||
"""写真記録からチェックイン記録を生成"""
|
||
print("\n=== 写真記録からチェックイン記録生成開始 ===")
|
||
|
||
# テストデータを除いた写真記録の未対応件数確認
|
||
self.cur_rog.execute("""
|
||
SELECT COUNT(*)
|
||
FROM rog_checkinimages ci
|
||
LEFT JOIN rog_gpscheckin gc ON ci.event_code = gc.event_code
|
||
AND ci.team_name = gc.zekken
|
||
AND ci.cp_number = gc.cp_number::integer
|
||
AND ABS(EXTRACT(EPOCH FROM (ci.checkintime - gc.checkin_time))) < 3600
|
||
WHERE ci.team_name NOT IN ('gero test 1', 'gero test 2')
|
||
AND gc.id IS NULL;
|
||
""")
|
||
|
||
unmatched_count = self.cur_rog.fetchone()[0]
|
||
print(f"未対応写真記録: {unmatched_count:,}件")
|
||
|
||
if unmatched_count == 0:
|
||
print("✅ 全ての写真記録に対応するチェックイン記録が存在します")
|
||
return {'generated': 0, 'errors': 0}
|
||
|
||
# 最大serial_numberを取得
|
||
self.cur_rog.execute("SELECT MAX(serial_number::integer) FROM rog_gpscheckin WHERE serial_number ~ '^[0-9]+$';")
|
||
max_serial_result = self.cur_rog.fetchone()
|
||
next_serial = (max_serial_result[0] if max_serial_result[0] else 0) + 1
|
||
|
||
# 未対応写真記録を取得
|
||
self.cur_rog.execute("""
|
||
SELECT
|
||
ci.id, ci.event_code, ci.team_name, ci.cp_number, ci.checkintime
|
||
FROM rog_checkinimages ci
|
||
LEFT JOIN rog_gpscheckin gc ON ci.event_code = gc.event_code
|
||
AND ci.team_name = gc.zekken
|
||
AND ci.cp_number = gc.cp_number::integer
|
||
AND ABS(EXTRACT(EPOCH FROM (ci.checkintime - gc.checkin_time))) < 3600
|
||
WHERE ci.team_name NOT IN ('gero test 1', 'gero test 2')
|
||
AND gc.id IS NULL
|
||
ORDER BY ci.event_code, ci.checkintime;
|
||
""")
|
||
|
||
photo_records = self.cur_rog.fetchall()
|
||
print(f"生成対象写真記録: {len(photo_records):,}件")
|
||
|
||
generated_count = 0
|
||
errors = []
|
||
|
||
batch_size = 500
|
||
for i in range(0, len(photo_records), batch_size):
|
||
batch = photo_records[i:i+batch_size]
|
||
|
||
for record in batch:
|
||
photo_id, event_code, team_name, cp_number, checkintime = record
|
||
|
||
try:
|
||
# チェックイン記録を挿入
|
||
self.cur_rog.execute("""
|
||
INSERT INTO rog_gpscheckin (
|
||
event_code, zekken, serial_number, cp_number,
|
||
checkin_time, record_time
|
||
) VALUES (%s, %s, %s, %s, %s, %s)
|
||
""", (
|
||
event_code,
|
||
team_name,
|
||
str(next_serial),
|
||
str(cp_number),
|
||
checkintime,
|
||
checkintime
|
||
))
|
||
|
||
generated_count += 1
|
||
next_serial += 1
|
||
|
||
except Exception as e:
|
||
errors.append(f'写真ID {photo_id}: {str(e)[:50]}')
|
||
|
||
# バッチごとにコミット
|
||
self.conn_rog.commit()
|
||
|
||
if i + batch_size >= 500:
|
||
progress_count = min(i + batch_size, len(photo_records))
|
||
print(f" 進捗: {progress_count:,}/{len(photo_records):,}件処理完了")
|
||
|
||
print(f"写真記録からチェックイン記録生成完了: 成功 {generated_count:,}件")
|
||
if errors:
|
||
print(f"エラー件数: {len(errors)}件")
|
||
|
||
return {
|
||
'generated': generated_count,
|
||
'errors': len(errors)
|
||
}
|
||
|
||
def generate_migration_statistics(self) -> Dict[str, Any]:
|
||
"""移行統計情報を生成"""
|
||
print("\n=== 移行統計情報生成 ===")
|
||
|
||
stats = {}
|
||
|
||
# 1. 基本統計
|
||
self.cur_gif.execute('SELECT COUNT(*) FROM gps_information;')
|
||
stats['original_gps_count'] = self.cur_gif.fetchone()[0]
|
||
|
||
self.cur_rog.execute('SELECT COUNT(*) FROM rog_gpscheckin;')
|
||
stats['final_checkin_count'] = self.cur_rog.fetchone()[0]
|
||
|
||
self.cur_rog.execute("SELECT COUNT(*) FROM rog_checkinimages WHERE team_name NOT IN ('gero test 1', 'gero test 2');")
|
||
stats['valid_photo_count'] = self.cur_rog.fetchone()[0]
|
||
|
||
# 2. イベント別統計
|
||
self.cur_rog.execute("""
|
||
SELECT
|
||
event_code,
|
||
COUNT(*) as checkin_count,
|
||
COUNT(DISTINCT zekken) as team_count
|
||
FROM rog_gpscheckin
|
||
GROUP BY event_code
|
||
ORDER BY checkin_count DESC;
|
||
""")
|
||
stats['event_stats'] = self.cur_rog.fetchall()
|
||
|
||
# 3. 写真記録とチェックイン記録の対応率
|
||
self.cur_rog.execute("""
|
||
SELECT
|
||
COUNT(ci.*) as total_photos,
|
||
COUNT(gc.*) as matched_checkins,
|
||
ROUND(COUNT(gc.*)::numeric / COUNT(ci.*)::numeric * 100, 1) as match_rate
|
||
FROM rog_checkinimages ci
|
||
LEFT JOIN rog_gpscheckin gc ON ci.event_code = gc.event_code
|
||
AND ci.team_name = gc.zekken
|
||
AND ci.cp_number = gc.cp_number::integer
|
||
AND ABS(EXTRACT(EPOCH FROM (ci.checkintime - gc.checkin_time))) < 3600
|
||
WHERE ci.team_name NOT IN ('gero test 1', 'gero test 2');
|
||
""")
|
||
|
||
photo_match_stats = self.cur_rog.fetchone()
|
||
stats['photo_match'] = {
|
||
'total_photos': photo_match_stats[0],
|
||
'matched_checkins': photo_match_stats[1],
|
||
'match_rate': photo_match_stats[2]
|
||
}
|
||
|
||
return stats
|
||
|
||
def print_statistics(self, stats: Dict[str, Any]):
|
||
"""統計情報を出力"""
|
||
print("\n" + "="*60)
|
||
print("📊 統合移行完了 - 最終統計レポート")
|
||
print("="*60)
|
||
|
||
print(f"\n📈 基本統計:")
|
||
print(f" 元データ(GPS情報): {stats['original_gps_count']:,}件")
|
||
print(f" 最終チェックイン記録: {stats['final_checkin_count']:,}件")
|
||
print(f" 有効写真記録: {stats['valid_photo_count']:,}件")
|
||
|
||
success_rate = (stats['final_checkin_count'] / stats['original_gps_count']) * 100
|
||
print(f" GPS移行成功率: {success_rate:.1f}%")
|
||
|
||
print(f"\n📷 写真記録対応状況:")
|
||
pm = stats['photo_match']
|
||
print(f" 総写真記録: {pm['total_photos']:,}件")
|
||
print(f" 対応チェックイン記録: {pm['matched_checkins']:,}件")
|
||
print(f" 対応率: {pm['match_rate']:.1f}%")
|
||
|
||
print(f"\n🏆 イベント別統計 (上位10イベント):")
|
||
print(" イベント チェックイン数 チーム数")
|
||
print(" " + "-"*45)
|
||
for event, checkin_count, team_count in stats['event_stats'][:10]:
|
||
print(f" {event:<12} {checkin_count:>12} {team_count:>8}")
|
||
|
||
# 成功判定
|
||
if pm['match_rate'] >= 99.0:
|
||
print(f"\n🎉 移行完全成功!")
|
||
print(" 写真記録とチェックイン記録の整合性が確保されました。")
|
||
elif pm['match_rate'] >= 95.0:
|
||
print(f"\n✅ 移行成功!")
|
||
print(" 高い整合性で移行が完了しました。")
|
||
else:
|
||
print(f"\n⚠️ 移行完了 (要確認)")
|
||
print(" 一部の記録で整合性の確認が必要です。")
|
||
|
||
def run_complete_migration(self):
|
||
"""完全移行処理を実行"""
|
||
print("🚀 統合移行処理開始")
|
||
print("=" * 50)
|
||
|
||
if not self.connect_databases():
|
||
return False
|
||
|
||
try:
|
||
# 1. GPS情報移行
|
||
gps_results = self.migrate_gps_information()
|
||
|
||
# 2. 写真記録からチェックイン記録生成
|
||
photo_results = self.generate_checkin_from_photos()
|
||
|
||
# 3. 統計情報生成・出力
|
||
stats = self.generate_migration_statistics()
|
||
self.print_statistics(stats)
|
||
|
||
# 4. 動作確認用のテストクエリ実行
|
||
self.run_verification_tests()
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ 移行処理エラー: {e}")
|
||
return False
|
||
|
||
finally:
|
||
self.close_connections()
|
||
|
||
def run_verification_tests(self):
|
||
"""動作確認テストを実行"""
|
||
print(f"\n🔍 動作確認テスト実行")
|
||
print("-" * 30)
|
||
|
||
# テスト1: MF5-204のデータ確認
|
||
self.cur_rog.execute("""
|
||
SELECT checkin_time, cp_number
|
||
FROM rog_gpscheckin
|
||
WHERE zekken = 'MF5-204'
|
||
ORDER BY checkin_time
|
||
LIMIT 5;
|
||
""")
|
||
|
||
mf5_results = self.cur_rog.fetchall()
|
||
print("✅ MF5-204のチェックイン記録 (最初の5件):")
|
||
for time, cp in mf5_results:
|
||
print(f" {time} - CP{cp}")
|
||
|
||
# テスト2: 美濃加茂イベントの統計
|
||
self.cur_rog.execute("""
|
||
SELECT
|
||
COUNT(*) as total_records,
|
||
COUNT(DISTINCT zekken) as unique_teams,
|
||
MIN(checkin_time) as first_checkin,
|
||
MAX(checkin_time) as last_checkin
|
||
FROM rog_gpscheckin
|
||
WHERE event_code = '美濃加茂';
|
||
""")
|
||
|
||
minokamo_stats = self.cur_rog.fetchone()
|
||
print(f"\n✅ 美濃加茂イベント統計:")
|
||
print(f" 総チェックイン数: {minokamo_stats[0]:,}件")
|
||
print(f" 参加チーム数: {minokamo_stats[1]}チーム")
|
||
print(f" 期間: {minokamo_stats[2]} ～ {minokamo_stats[3]}")
|
||
|
||
# テスト3: 最新のチェックイン記録確認
|
||
self.cur_rog.execute("""
|
||
SELECT event_code, zekken, checkin_time, cp_number
|
||
FROM rog_gpscheckin
|
||
ORDER BY checkin_time DESC
|
||
LIMIT 3;
|
||
""")
|
||
|
||
latest_records = self.cur_rog.fetchall()
|
||
print(f"\n✅ 最新チェックイン記録 (最後の3件):")
|
||
for event, zekken, time, cp in latest_records:
|
||
print(f" {event} - {zekken} - {time} - CP{cp}")
|
||
|
||
print("\n🎯 動作確認完了: 全てのテストが正常に実行されました")
|
||
|
||
def close_connections(self):
|
||
"""データベース接続を閉じる"""
|
||
if self.cur_gif:
|
||
self.cur_gif.close()
|
||
if self.cur_rog:
|
||
self.cur_rog.close()
|
||
if self.conn_gif:
|
||
self.conn_gif.close()
|
||
if self.conn_rog:
|
||
self.conn_rog.close()
|
||
|
||
print("\n✅ データベース接続を閉じました")
|
||
|
||
def main():
|
||
"""メイン実行関数"""
|
||
migrator = MigrationWithPhotoIntegration()
|
||
|
||
try:
|
||
success = migrator.run_complete_migration()
|
||
|
||
if success:
|
||
print("\n" + "="*50)
|
||
print("🎉 統合移行処理が正常に完了しました！")
|
||
print("="*50)
|
||
sys.exit(0)
|
||
else:
|
||
print("\n" + "="*50)
|
||
print("❌ 移行処理中にエラーが発生しました")
|
||
print("="*50)
|
||
sys.exit(1)
|
||
|
||
except KeyboardInterrupt:
|
||
print("\n⚠️ ユーザーによって処理が中断されました")
|
||
sys.exit(1)
|
||
except Exception as e:
|
||
print(f"\n❌ 予期しないエラー: {e}")
|
||
sys.exit(1)
|
||
|
||
if __name__ == "__main__":
|
||
main()
|