69 lines
2.1 KiB
Python
69 lines
2.1 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
Supabase 동기화 테스트 스크립트
|
|
"""
|
|
|
|
import logging
|
|
logging.basicConfig(level=logging.DEBUG, format='%(levelname)s: %(message)s')
|
|
|
|
from core.supabase_client import SupabaseClient, SupabaseSyncManager
|
|
|
|
def main():
|
|
# 연결 테스트
|
|
client = SupabaseClient()
|
|
print('=== Supabase 연결 테스트 ===')
|
|
if client.test_connection():
|
|
print('연결 성공!')
|
|
else:
|
|
print('연결 실패')
|
|
return
|
|
|
|
# 편성 정보 조회
|
|
print('\n=== 편성 정보 조회 ===')
|
|
trains = client.get_trains()
|
|
print(f'조회된 편성 수: {len(trains)}')
|
|
if trains:
|
|
for t in trains[:5]:
|
|
print(f' - 편성 {t.train_set}: {t.train_id}, 제조사={t.manufacturer}, 신형={t.is_new}')
|
|
|
|
# 역 정보 조회
|
|
print('\n=== 역 정보 조회 (1호선) ===')
|
|
stations = client.get_stations(line_number=1)
|
|
print(f'조회된 역 수: {len(stations)}')
|
|
if stations:
|
|
for s in stations[:5]:
|
|
print(f' - {s.station_id}: {s.station_name}')
|
|
|
|
# 제조사 정보 조회
|
|
print('\n=== 제조사 정보 조회 ===')
|
|
manufacturers = client.get_manufacturers()
|
|
print(f'조회된 제조사 수: {len(manufacturers)}')
|
|
for m in manufacturers:
|
|
print(f' - {m.id}: {m.manufact}')
|
|
|
|
# 동기화 테스트
|
|
print('\n=== 마스터 데이터 동기화 ===')
|
|
sync_manager = SupabaseSyncManager()
|
|
results = sync_manager.sync_all()
|
|
print(f'동기화 결과: {results}')
|
|
|
|
# 로컬 DB에서 확인
|
|
print('\n=== 로컬 DB 확인 ===')
|
|
from core.settings_manager import get_settings_manager
|
|
settings = get_settings_manager()
|
|
|
|
local_trains = settings.get_all_trains()
|
|
print(f'로컬 편성 수: {len(local_trains)}')
|
|
|
|
local_stations = settings.get_all_stations()
|
|
print(f'로컬 역 수: {len(local_stations)}')
|
|
|
|
local_manufacturers = settings.get_all_manufacturers()
|
|
print(f'로컬 제조사 수: {len(local_manufacturers)}')
|
|
|
|
print('\n=== 테스트 완료 ===')
|
|
|
|
if __name__ == '__main__':
|
|
main()
|
|
|