프로젝트

불법주정차 단속 위치정보 적재

import csv
import mysql.connector
import pandas as pd

# MySQL 연결 설정
mysql_host = 'localhost'
mysql_user = 'root'
mysql_password = '****'  # MySQL 비밀번호를 입력하세요
mysql_database = 'programmers_da'

# CSV 파일 경로
csv_file_path = 'C:/Users/user/OneDrive/바탕 화면/데이터 분석 데브코스/프로젝트/불법주정차/불법주정차 단속 위치정보(22.11월_23.10월).csv'

# MySQL 연결
conn = mysql.connector.connect(
    host=mysql_host,
    user=mysql_user,
    password=mysql_password,
    database=mysql_database
)

# CSV 파일을 pandas DataFrame으로 읽어 데이터 형식 확인
df = pd.read_csv(csv_file_path, delimiter=',', encoding='ANSI')

# 커서 생성
cursor = conn.cursor()

# 데이터프레임의 데이터 형식을 MySQL 데이터 형식으로 매핑
mysql_type_mapping = {
    'int64': 'INT',
    'float64': 'DOUBLE',
    'object': 'TEXT'  
}

# 데이터프레임의 컬럼과 데이터 형식을 기반으로 MySQL 쿼리 생성
columns = [
    "단속일 TEXT",
    "시간 TEXT",
    "구주소 TEXT",
    "도로명 TEXT",
    "경도 DOUBLE",
    "위도 DOUBLE"
]

# 테이블 삭제
drop_table_query = "DROP TABLE IF EXISTS Crackdown_Location"
cursor.execute(drop_table_query)

create_table_query = f"CREATE TABLE IF NOT EXISTS Crackdown_Location ({', '.join(columns)})"
cursor.execute(create_table_query)

# 레코드를 일부씩 MySQL에 적재
insert_query = f"INSERT INTO Crackdown_Location ({', '.join(df.columns)}) VALUES ({', '.join(['%s' for _ in df.columns])})"
batch_size = 1000  # 원하는 배치 크기로 조정
records = []

with open(csv_file_path, 'r', encoding='ANSI') as csvfile:
    csv_reader = csv.reader(csvfile, delimiter=',')  
    header = next(csv_reader)  

    for i, row in enumerate(csv_reader, start=1):
        try:
            # '경도'와 '위도' 컬럼이 빈 문자열인 경우 None으로 처리
            row[4] = round(float(row[4]), 10) if row[4] else None  # '경도'
            row[5] = round(float(row[5]), 10) if row[5] else None  # '위도'
            records.append(tuple(row))
        except ValueError as e:
            print(f"Error converting values to float at row {i}: {e}")

        if len(records) >= batch_size:
            try:
                cursor.executemany(insert_query, records)
                conn.commit()
            except mysql.connector.errors.DataError as e:
                print(f"Error inserting records at row {i}: {e}")
                # Handle the error (skip the row, modify the data, etc.)
            records = []

    # 마지막으로 남은 레코드들을 적재
    if records:
        try:
            cursor.executemany(insert_query, records)
            conn.commit()
        except mysql.connector.errors.DataError as e:
            print(f"Error inserting final records: {e}")
            # Handle the error (skip the row, modify the data, etc.)

# 연결 종료
cursor.close()
conn.close()