AI를 활용한 데이터 수집

AI 도구들은 데이터 수집 과정을 혁신적으로 개선할 수 있다. ChatGPT, Claude 같은 AI 어시스턴트를 활용하여 데이터 수집 코드를 생성하고, 데이터 수집 전략을 수립하며, 수집 과정에서 발생하는 문제들을 해결하는 방법을 알아보자.

ChatGPT/Claude로 수집 코드 생성

효과적인 프롬프트 작성법

AI에게 데이터 수집 코드를 요청할 때는 구체적이고 명확한 프롬프트를 작성하는 것이 중요하다.

좋은 프롬프트 예시

프롬프트: "한국 공공데이터포털에서 인구통계 데이터를 가져오는 Python 코드를 작성해주세요. 
다음 요구사항을 포함해주세요:
1. requests 라이브러리 사용
2. API 키 인증 처리
3. 에러 핸들링 포함
4. pandas DataFrame으로 결과 반환
5. 지역코드와 연도를 매개변수로 받는 함수 형태
6. 상세한 주석 포함"

AI가 생성한 코드 예시

import requests
import pandas as pd
import json
from urllib.parse import urlencode

def get_population_data(api_key, region_code, year):
    """
    공공데이터포털에서 인구통계 데이터를 조회하는 함수
    
    Parameters:
    api_key (str): 공공데이터포털 API 키
    region_code (str): 지역코드 (예: '11000' - 서울특별시)
    year (str): 조회 연도 (예: '2023')
    
    Returns:
    pandas.DataFrame: 인구통계 데이터
    """
    
    # API 엔드포인트 설정
    base_url = "http://apis.data.go.kr/1160100/service/GetPopulationKoreanService/getPopulationKorean"
    
    # 요청 매개변수 설정
    params = {
        'serviceKey': api_key,
        'numOfRows': 100,
        'pageNo': 1,
        'resultType': 'json',
        'admCode': region_code,
        'year': year
    }
    
    try:
        # API 요청 실행
        response = requests.get(base_url, params=params, timeout=30)
        response.raise_for_status()  # HTTP 에러 체크
        
        # JSON 응답 파싱
        data = response.json()
        
        # 응답 데이터 구조 확인
        if 'response' in data and 'body' in data['response']:
            body = data['response']['body']
            
            # 데이터가 있는지 확인
            if 'items' in body and body['items']:
                items = body['items']['item']
                
                # 단일 항목인 경우 리스트로 변환
                if isinstance(items, dict):
                    items = [items]
                
                # DataFrame 생성
                df = pd.DataFrame(items)
                
                # 숫자 컬럼 변환
                numeric_columns = ['totPopulation', 'malePopulation', 'femalePopulation']
                for col in numeric_columns:
                    if col in df.columns:
                        df[col] = pd.to_numeric(df[col], errors='coerce')
                
                print(f"데이터 조회 성공: {len(df)}개 행")
                return df
            else:
                print("조회된 데이터가 없습니다.")
                return pd.DataFrame()
        else:
            print(f"API 응답 오류: {data.get('response', {}).get('header', {}).get('resultMsg', '알 수 없는 오류')}")
            return pd.DataFrame()
    
    except requests.exceptions.Timeout:
        print("API 요청 시간 초과")
        return pd.DataFrame()
    except requests.exceptions.ConnectionError:
        print("API 서버 연결 실패")
        return pd.DataFrame()
    except requests.exceptions.HTTPError as e:
        print(f"HTTP 오류: {e}")
        return pd.DataFrame()
    except json.JSONDecodeError:
        print("JSON 파싱 오류")
        return pd.DataFrame()
    except Exception as e:
        print(f"예상치 못한 오류: {e}")
        return pd.DataFrame()

# 사용 예시
if __name__ == "__main__":
    # API 키와 매개변수 설정
    API_KEY = "your_api_key_here"
    REGION_CODE = "11000"  # 서울특별시
    YEAR = "2023"
    
    # 데이터 조회
    population_df = get_population_data(API_KEY, REGION_CODE, YEAR)
    
    if not population_df.empty:
        print(population_df.head())
        print(f"\n데이터 형태: {population_df.shape}")

데이터 수집 코드 리뷰와 개선

AI가 생성한 코드를 리뷰하고 개선하는 과정도 AI의 도움을 받을 수 있다.

코드 리뷰 프롬프트

프롬프트: "다음 데이터 수집 코드를 리뷰하고 개선점을 제안해주세요:
[코드 첨부]

다음 관점에서 검토해주세요:
1. 보안성 (API 키 관리 등)
2. 성능 최적화
3. 에러 핸들링 강화
4. 코드 가독성
5. 확장성"

AI의 개선 제안 예시

import requests
import pandas as pd
import json
import os
from urllib.parse import urlencode
from typing import Optional
import logging
from functools import wraps
import time

# 로깅 설정
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def retry_on_failure(max_retries=3, delay=1):
    """재시도 데코레이터"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise e
                    logger.warning(f"시도 {attempt + 1} 실패: {e}. {delay}초 후 재시도...")
                    time.sleep(delay)
            return None
        return wrapper
    return decorator

class PublicDataCollector:
    """공공데이터 수집을 위한 클래스"""
    
    def __init__(self, api_key: Optional[str] = None):
        # 환경변수에서 API 키 읽기 (보안 강화)
        self.api_key = api_key or os.getenv('PUBLIC_DATA_API_KEY')
        if not self.api_key:
            raise ValueError("API 키가 설정되지 않았습니다. 환경변수 PUBLIC_DATA_API_KEY를 설정하거나 매개변수로 전달하세요.")
        
        self.base_url = "http://apis.data.go.kr"
        self.session = requests.Session()  # 세션 재사용으로 성능 향상
        
        # 요청 헤더 설정
        self.session.headers.update({
            'User-Agent': 'PublicDataCollector/1.0',
            'Accept': 'application/json'
        })
    
    @retry_on_failure(max_retries=3, delay=2)
    def get_population_data(self, region_code: str, year: str, 
                          num_rows: int = 100) -> pd.DataFrame:
        """
        인구통계 데이터 조회 (개선된 버전)
        
        Args:
            region_code: 지역코드
            year: 조회 연도
            num_rows: 조회할 행 수
            
        Returns:
            pd.DataFrame: 인구통계 데이터
        """
        
        endpoint = "/1160100/service/GetPopulationKoreanService/getPopulationKorean"
        url = self.base_url + endpoint
        
        params = {
            'serviceKey': self.api_key,
            'numOfRows': num_rows,
            'pageNo': 1,
            'resultType': 'json',
            'admCode': region_code,
            'year': year
        }
        
        logger.info(f"인구 데이터 조회 시작: 지역코드={region_code}, 연도={year}")
        
        response = self.session.get(url, params=params, timeout=30)
        response.raise_for_status()
        
        data = response.json()
        
        # 응답 검증 및 데이터 추출
        df = self._extract_population_data(data)
        
        logger.info(f"데이터 조회 완료: {len(df)}개 행")
        return df
    
    def _extract_population_data(self, data: dict) -> pd.DataFrame:
        """응답 데이터에서 인구 데이터 추출"""
        
        if not self._validate_response(data):
            return pd.DataFrame()
        
        items = data['response']['body']['items']['item']
        
        # 단일 항목 처리
        if isinstance(items, dict):
            items = [items]
        
        df = pd.DataFrame(items)
        
        # 데이터 타입 최적화
        df = self._optimize_data_types(df)
        
        return df
    
    def _validate_response(self, data: dict) -> bool:
        """API 응답 유효성 검증"""
        
        if 'response' not in data:
            logger.error("응답에 'response' 키가 없습니다.")
            return False
        
        header = data['response'].get('header', {})
        result_code = header.get('resultCode', '')
        
        if result_code != '00':
            error_msg = header.get('resultMsg', '알 수 없는 오류')
            logger.error(f"API 오류: {result_code} - {error_msg}")
            return False
        
        body = data['response'].get('body', {})
        if 'items' not in body or not body['items']:
            logger.warning("조회된 데이터가 없습니다.")
            return False
        
        return True
    
    def _optimize_data_types(self, df: pd.DataFrame) -> pd.DataFrame:
        """데이터 타입 최적화"""
        
        # 숫자 컬럼 변환
        numeric_columns = ['totPopulation', 'malePopulation', 'femalePopulation']
        for col in numeric_columns:
            if col in df.columns:
                df[col] = pd.to_numeric(df[col], errors='coerce')
        
        # 문자열 컬럼 최적화
        for col in df.select_dtypes(include=['object']).columns:
            df[col] = df[col].astype('string')
        
        return df
    
    def __del__(self):
        """세션 정리"""
        if hasattr(self, 'session'):
            self.session.close()

# 사용 예시
if __name__ == "__main__":
    try:
        collector = PublicDataCollector()
        df = collector.get_population_data("11000", "2023")
        print(df.head())
    except Exception as e:
        logger.error(f"데이터 수집 실패: {e}")

AI와 함께하는 디버깅

오류 해결 프롬프트

프롬프트: "다음 데이터 수집 코드에서 오류가 발생했습니다:

[오류 메시지]
KeyError: 'items'

[관련 코드]
data = response.json()
items = data['response']['body']['items']['item']

이 오류의 원인과 해결 방법을 알려주세요. 그리고 이런 오류를 방지하는 방법도 제안해주세요."

AI의 디버깅 도움

def safe_extract_api_data(response_data, path):
    """
    API 응답에서 안전하게 데이터를 추출하는 함수
    
    Args:
        response_data (dict): API 응답 데이터
        path (list): 추출할 데이터의 경로 (예: ['response', 'body', 'items', 'item'])
    
    Returns:
        any: 추출된 데이터 또는 None
    """
    
    current_data = response_data
    
    for key in path:
        if isinstance(current_data, dict) and key in current_data:
            current_data = current_data[key]
        else:
            print(f"경로에서 키 '{key}'를 찾을 수 없습니다.")
            print(f"현재 데이터 구조: {type(current_data)}")
            if isinstance(current_data, dict):
                print(f"사용 가능한 키: {list(current_data.keys())}")
            return None
    
    return current_data

# 사용 예시
def robust_data_extraction(response):
    """강건한 데이터 추출"""
    
    try:
        data = response.json()
        
        # 응답 구조 확인
        print("API 응답 구조:")
        print(json.dumps(data, indent=2, ensure_ascii=False)[:500])
        
        # 안전한 데이터 추출
        items = safe_extract_api_data(data, ['response', 'body', 'items', 'item'])
        
        if items is None:
            # 대안 경로 시도
            items = safe_extract_api_data(data, ['response', 'body', 'items'])
            if items is None:
                items = safe_extract_api_data(data, ['data'])
        
        return items
        
    except json.JSONDecodeError as e:
        print(f"JSON 파싱 오류: {e}")
        print(f"응답 내용: {response.text[:200]}")
        return None
    except Exception as e:
        print(f"예상치 못한 오류: {e}")
        return None

AI 기반 데이터 수집 전략

새로운 데이터 소스 발굴

AI에게 특정 분야의 데이터 소스를 추천받을 수 있다.

프롬프트: "한국의 부동산 시장 분석을 위한 데이터 소스를 추천해주세요. 
다음 정보를 포함해주세요:
1. 공공 데이터 소스
2. 민간 데이터 제공업체
3. 각 데이터 소스의 특징과 한계
4. 데이터 수집 방법 (API, 크롤링 등)
5. 법적 고려사항"

자동화된 데이터 수집 파이프라인

import schedule
import time
from datetime import datetime
import logging

class AutomatedDataPipeline:
    """AI 도움으로 설계된 자동화 데이터 파이프라인"""
    
    def __init__(self, config):
        self.config = config
        self.logger = self._setup_logging()
        self.collectors = self._initialize_collectors()
    
    def _setup_logging(self):
        """로깅 설정"""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('data_pipeline.log'),
                logging.StreamHandler()
            ]
        )
        return logging.getLogger(__name__)
    
    def _initialize_collectors(self):
        """데이터 수집기 초기화"""
        collectors = {}
        
        # 공공데이터 수집기
        if 'public_data' in self.config:
            collectors['public_data'] = PublicDataCollector(
                self.config['public_data']['api_key']
            )
        
        # 추가 수집기들...
        
        return collectors
    
    def collect_daily_data(self):
        """일일 데이터 수집"""
        self.logger.info("일일 데이터 수집 시작")
        
        try:
            # 각 데이터 소스에서 수집
            for source_name, collector in self.collectors.items():
                self.logger.info(f"{source_name} 데이터 수집 중...")
                
                data = self._collect_from_source(source_name, collector)
                
                if data is not None and not data.empty:
                    self._save_data(data, source_name)
                    self.logger.info(f"{source_name} 데이터 수집 완료: {len(data)}개 행")
                else:
                    self.logger.warning(f"{source_name} 데이터 수집 실패")
        
        except Exception as e:
            self.logger.error(f"데이터 수집 중 오류 발생: {e}")
    
    def _collect_from_source(self, source_name, collector):
        """특정 소스에서 데이터 수집"""
        
        if source_name == 'public_data':
            return collector.get_population_data("11000", "2023")
        
        # 다른 소스들...
        
        return None
    
    def _save_data(self, data, source_name):
        """데이터 저장"""
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"{source_name}_{timestamp}.csv"
        
        data.to_csv(f"data/{filename}", index=False)
        self.logger.info(f"데이터 저장 완료: {filename}")
    
    def setup_schedule(self):
        """스케줄 설정"""
        # 매일 오전 9시에 데이터 수집
        schedule.every().day.at("09:00").do(self.collect_daily_data)
        
        # 매주 월요일 오전 6시에 주간 리포트
        schedule.every().monday.at("06:00").do(self.generate_weekly_report)
        
        self.logger.info("스케줄 설정 완료")
    
    def generate_weekly_report(self):
        """주간 리포트 생성"""
        self.logger.info("주간 리포트 생성 중...")
        # 리포트 생성 로직
        pass
    
    def run(self):
        """파이프라인 실행"""
        self.setup_schedule()
        
        self.logger.info("데이터 파이프라인 시작")
        
        while True:
            schedule.run_pending()
            time.sleep(60)  # 1분마다 스케줄 확인

# 설정 파일
config = {
    'public_data': {
        'api_key': 'your_api_key_here'
    }
}

# 파이프라인 실행
# pipeline = AutomatedDataPipeline(config)
# pipeline.run()

AI를 활용한 데이터 품질 검증

def ai_assisted_data_validation(df, data_source_info):
    """
    AI 도움으로 설계된 데이터 품질 검증
    
    이 함수는 AI와의 대화를 통해 개발된 검증 로직을 포함합니다.
    """
    
    validation_results = {
        'source': data_source_info['name'],
        'validation_time': datetime.now(),
        'checks': {},
        'overall_quality': 'UNKNOWN'
    }
    
    # 1. 기본 구조 검증
    validation_results['checks']['structure'] = {
        'total_rows': len(df),
        'total_columns': len(df.columns),
        'memory_usage_mb': df.memory_usage(deep=True).sum() / 1024 / 1024
    }
    
    # 2. 데이터 완성도 검증
    completeness_score = 0
    missing_data = {}
    
    for col in df.columns:
        missing_count = df[col].isnull().sum()
        missing_percent = (missing_count / len(df)) * 100
        missing_data[col] = {
            'missing_count': missing_count,
            'missing_percent': round(missing_percent, 2)
        }
        
        # 완성도 점수 계산
        completeness_score += max(0, 100 - missing_percent)
    
    completeness_score = completeness_score / len(df.columns)
    validation_results['checks']['completeness'] = {
        'score': round(completeness_score, 2),
        'missing_data': missing_data
    }
    
    # 3. 데이터 일관성 검증
    consistency_issues = []
    
    # 날짜 컬럼 일관성 확인
    for col in df.columns:
        if 'date' in col.lower() or 'time' in col.lower():
            try:
                date_series = pd.to_datetime(df[col], errors='coerce')
                invalid_dates = date_series.isnull().sum() - df[col].isnull().sum()
                if invalid_dates > 0:
                    consistency_issues.append(f"{col}: {invalid_dates}개의 잘못된 날짜 형식")
            except:
                pass
    
    # 숫자 컬럼 범위 확인
    for col in df.select_dtypes(include=['number']).columns:
        q1 = df[col].quantile(0.25)
        q3 = df[col].quantile(0.75)
        iqr = q3 - q1
        lower_bound = q1 - 1.5 * iqr
        upper_bound = q3 + 1.5 * iqr
        
        outliers = ((df[col] < lower_bound) | (df[col] > upper_bound)).sum()
        if outliers > len(df) * 0.05:  # 5% 이상이 이상치라면
            consistency_issues.append(f"{col}: {outliers}개의 잠재적 이상치")
    
    validation_results['checks']['consistency'] = {
        'issues': consistency_issues,
        'issue_count': len(consistency_issues)
    }
    
    # 4. 전체 품질 점수 계산
    quality_score = completeness_score
    
    # 일관성 문제로 점수 차감
    consistency_penalty = min(30, len(consistency_issues) * 5)
    quality_score -= consistency_penalty
    
    # 품질 등급 결정
    if quality_score >= 90:
        validation_results['overall_quality'] = 'EXCELLENT'
    elif quality_score >= 75:
        validation_results['overall_quality'] = 'GOOD'
    elif quality_score >= 60:
        validation_results['overall_quality'] = 'FAIR'
    else:
        validation_results['overall_quality'] = 'POOR'
    
    validation_results['quality_score'] = round(quality_score, 2)
    
    return validation_results

# 사용 예시
# data_source_info = {'name': 'Public Population Data'}
# validation_result = ai_assisted_data_validation(df, data_source_info)
# print(json.dumps(validation_result, indent=2, default=str, ensure_ascii=False))

정리

AI를 활용한 데이터 수집은 효율성과 품질을 크게 향상시킬 수 있다. 코드 생성부터 디버깅, 품질 검증까지 AI의 도움을 받으면 더 빠르고 정확한 데이터 수집이 가능하다. 하지만 AI가 생성한 코드는 항상 검토하고 테스트해야 하며, 도메인 지식과 결합해서 사용하는 것이 중요하다.

다음 섹션에서는 데이터 입수 실무 고려사항에 대해 알아보겠다.