# ========================================== # crawlers/base.py - 크롤러 공통 유틸리티 # ========================================== import requests def safe_float(value): """ 안전한 float 변환 None, 빈 문자열, 콤마 포함 숫자 등을 처리 """ if value is None: return 0.0 try: return float(str(value).replace(',', '')) except (ValueError, TypeError): return 0.0 def create_session(): """기본 설정된 requests 세션 생성""" session = requests.Session() return session def get_default_headers(): """기본 HTTP 헤더 반환""" return { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120.0.0.0 Safari/537.36', 'Accept': 'application/json, text/plain, */*' } def determine_status(current_kw): """발전량 기반 상태 결정""" if current_kw > 0: return "🟢 정상" else: return "💤 대기" def format_result(name, kw, today, plant_id, status=None): """결과 딕셔너리 포맷 통일""" if status is None: status = determine_status(kw) return { 'name': name, 'kw': kw, 'today': today, 'id': plant_id, 'status': status } def validate_data_quality(data_list, value_key='generation_kwh'): """ 데이터 품질 검증 Returns: dict: { 'is_valid': bool, 'warnings': list, 'all_zero': bool, 'duplicate_ratio': float } """ if not data_list or len(data_list) == 0: return { 'is_valid': False, 'warnings': ['데이터 없음'], 'all_zero': True, 'duplicate_ratio': 0.0 } warnings = [] values = [safe_float(item.get(value_key, 0)) for item in data_list] # 모두 0인 경우 체크 all_zero = all(v == 0 for v in values) if all_zero: warnings.append('모든 값이 0 - 실제 데이터가 아닐 가능성') # 연속 중복 체크 if len(values) > 1: duplicates = 0 for i in range(len(values) - 1): if values[i] == values[i+1]: duplicates += 1 duplicate_ratio = duplicates / (len(values) - 1) if duplicate_ratio > 0.8: warnings.append(f'연속 중복 비율 {duplicate_ratio*100:.1f}% - 실제 데이터가 아닐 가능성') else: duplicate_ratio = 0.0 is_valid = not all_zero and duplicate_ratio < 0.8 return { 'is_valid': is_valid, 'warnings': warnings, 'all_zero': all_zero, 'duplicate_ratio': duplicate_ratio }