codeSearch/test/clean_database_admin.py

82 lines
2.8 KiB
Python

import sqlite3
def clean_fault_code_list():
"""
fault_code_list 테이블에서 외래키(manufacturer_id)가 설정되지 않은 레코드를 삭제합니다.
이 스크립트는 관리자용으로, 데이터베이스 정리 전용입니다.
"""
conn = sqlite3.connect('fault_codes.db')
cursor = conn.cursor()
try:
# 현재 상태 확인
cursor.execute("SELECT COUNT(*) FROM fault_code_list")
total_records = cursor.fetchone()[0]
cursor.execute("SELECT COUNT(*) FROM fault_code_list WHERE manufacturer_id IS NULL")
null_records = cursor.fetchone()[0]
print(f"현재 총 레코드 수: {total_records}")
print(f"manufacturer_id가 NULL인 레코드 수: {null_records}")
# 외래키 제약 활성화
cursor.execute("PRAGMA foreign_keys = ON")
# 외래키가 설정되지 않은 레코드 삭제
cursor.execute("DELETE FROM fault_code_list WHERE manufacturer_id IS NULL")
deleted = cursor.rowcount
print(f"삭제된 레코드 수: {deleted}")
# 제작사 테이블 중복 제거
cursor.execute("""
DELETE FROM manufacturers
WHERE id NOT IN (
SELECT MIN(id)
FROM manufacturers
GROUP BY name
)
""")
deleted_manufacturers = cursor.rowcount
print(f"제거된 중복 제작사 수: {deleted_manufacturers}")
# 외래키 설정 업데이트
cursor.execute("""
UPDATE fault_code_list
SET manufacturer_id = (
SELECT id FROM manufacturers
WHERE name = '우진'
LIMIT 1
)
WHERE manufacturer_id IS NULL
""")
updated = cursor.rowcount
print(f"업데이트된 레코드 수: {updated}")
# 최종 상태 확인
cursor.execute("SELECT COUNT(*) FROM fault_code_list")
final_records = cursor.fetchone()[0]
cursor.execute("SELECT COUNT(*) FROM manufacturers")
manufacturer_count = cursor.fetchone()[0]
print(f"최종 레코드 수: {final_records}")
print(f"제작사 수: {manufacturer_count}")
conn.commit()
print("데이터베이스 정리가 완료되었습니다.")
except sqlite3.Error as e:
conn.rollback()
print(f"오류 발생: {e}")
finally:
conn.close()
if __name__ == "__main__":
user_input = input("이 스크립트는 데이터베이스에서 외래키가 설정되지 않은 레코드를 삭제합니다. 계속하시겠습니까? (y/n): ")
if user_input.lower() == 'y':
clean_fault_code_list()
else:
print("스크립트 실행이 취소되었습니다.")