diff --git a/crawler/backward_backfill.py b/crawler/backward_backfill.py new file mode 100644 index 0000000..a7e077d --- /dev/null +++ b/crawler/backward_backfill.py @@ -0,0 +1,286 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +""" +backward_backfill.py - 과거 발전 데이터 역추적 백필 스케줄러 + +어제자부터 과거로 하루씩 거슬러 올라가며 발전소별 데이터를 수집하고 Supabase에 반영합니다. +호출 속도 조절(Rate limiting) 및 자동 종료 조건을 지원합니다. +""" + +import os +import sys +import time +import sqlite3 +import argparse +import importlib +from datetime import datetime, timedelta +from dotenv import load_dotenv + +# .env 로드 +load_dotenv() + +# Windows 인코딩 설정 +if sys.platform.startswith('win'): + sys.stdout.reconfigure(encoding='utf-8') + sys.stderr.reconfigure(encoding='utf-8') + +# 프로젝트 루트 경로 추가 +current_dir = os.path.dirname(os.path.abspath(__file__)) +sys.path.append(current_dir) + +from config import get_all_plants +from database import save_history, get_supabase_client + +# DB 경로 설정 +DB_PATH = os.path.join(current_dir, "crawler_manager.db") + +def get_db_connection(): + """SQLite 연결 반환""" + return sqlite3.connect(DB_PATH, timeout=10.0) + +def init_backfill_states(yesterday_str: str): + """ + 각 발전소의 백필 상태를 SQLite에 초기화합니다. + 이미 존재하면 건너뛰고 없으면 신규 등록합니다. + """ + plants = get_all_plants() + with get_db_connection() as conn: + cursor = conn.cursor() + + # NREMS 분할 호기도 식별 가능하게 리스트 빌드 + site_ids = [] + for p in plants: + is_split = p.get('options', {}).get('is_split', False) + if is_split: + site_ids.extend(['nrems-01', 'nrems-02']) + else: + site_ids.append(p.get('id')) + + for site_id in site_ids: + if not site_id: + continue + cursor.execute("SELECT 1 FROM backfill_state WHERE site_id = ?", (site_id,)) + if not cursor.fetchone(): + cursor.execute(""" + INSERT INTO backfill_state (site_id, last_backfilled_date, consecutive_zero_count, status) + VALUES (?, ?, 0, 'RUNNING') + """, (site_id, yesterday_str)) + print(f" 📝 [Backfill Init] {site_id} 상태 등록 완료 (시작일: {yesterday_str})") + conn.commit() + +def update_backfill_state(site_id: str, last_date: str, zero_count: int, status: str): + """발전소의 백필 진행 상태를 업데이트합니다.""" + with get_db_connection() as conn: + cursor = conn.cursor() + cursor.execute(""" + UPDATE backfill_state + SET last_backfilled_date = ?, + consecutive_zero_count = ?, + status = ? + WHERE site_id = ? + """, (last_date, zero_count, status, site_id)) + conn.commit() + +def get_running_states(): + """RUNNING 상태인 발전소들의 정보를 가져옵니다.""" + with get_db_connection() as conn: + cursor = conn.cursor() + cursor.execute(""" + SELECT site_id, last_backfilled_date, consecutive_zero_count, status + FROM backfill_state + WHERE status = 'RUNNING' + """) + rows = cursor.fetchall() + return {row[0]: {"last_date": row[1], "zero_count": row[2], "status": row[3]} for row in rows} + +def get_history_crawler(plant_type: str): + """발전소 타입별 과거 데이터 크롤러 함수 동적 로드""" + try: + crawler_module = importlib.import_module(f"crawlers.{plant_type}") + if hasattr(crawler_module, 'fetch_history_daily'): + return crawler_module.fetch_history_daily + except Exception as e: + print(f" ⚠️ 크롤러 모듈 로드 실패 ({plant_type}): {e}") + return None + +def process_backfill(max_days_per_run: int = 7, delay_sec: float = 2.0, dry_run: bool = False): + """ + 백필 작업을 수행합니다. + """ + print(f"\n⚡ [과거 데이터 역추적 백필 시작] Dry-Run: {dry_run}, 1회당 최대 일수: {max_days_per_run}일, 대기시간: {delay_sec}초") + print("=" * 80) + + # 1. 상태 데이터 로드 + running_states = get_running_states() + if not running_states: + print("🎉 모든 발전소의 백필 작업이 완료되었거나 실행 중인 작업이 없습니다.") + return + + # 2. 발전소 설정 매핑 + all_plants = get_all_plants() + plant_configs = {} + for p in all_plants: + is_split = p.get('options', {}).get('is_split', False) + if is_split: + # 1호기, 2호기 분할 세팅 복사 + p1 = p.copy() + p1['id'] = 'nrems-01' + p1['options'] = p['options'].copy() + p1['options']['split_index'] = 1 + + p2 = p.copy() + p2['id'] = 'nrems-02' + p2['options'] = p['options'].copy() + p2['options']['split_index'] = 2 + + plant_configs['nrems-01'] = p1 + plant_configs['nrems-02'] = p2 + else: + plant_configs[p.get('id', '')] = p + + # 3. 각 발전소별 백필 진행 + for site_id, state in running_states.items(): + plant_config = plant_configs.get(site_id) + if not plant_config: + print(f"⚠️ [{site_id}] 설정을 config.py에서 찾을 수 없습니다. 스킵합니다.") + continue + + plant_type = plant_config.get('type') + plant_name = plant_config.get('name') + start_date_str = plant_config.get('start_date', '2020-01-01') + + last_date_str = state["last_date"] + zero_count = state["zero_count"] + + print(f"\n▶️ [{site_id} - {plant_name}] 백필 진행 ({last_date_str}부터 과거로 역추적)") + print(f" 가동개시일: {start_date_str}, 현재 연속 무발전 일수: {zero_count}") + + crawler_fn = get_history_crawler(plant_type) + if not crawler_fn: + print(f" ❌ [{site_id}] 해당 타입({plant_type})의 fetch_history_daily 크롤러가 없습니다. 스킵.") + continue + + last_dt = datetime.strptime(last_date_str, "%Y-%m-%d") + start_dt = datetime.strptime(start_date_str, "%Y-%m-%d") + + days_processed = 0 + current_dt = last_dt + + while days_processed < max_days_per_run: + # 하루 이전 날짜 계산 + current_dt = current_dt - timedelta(days=1) + target_date_str = current_dt.strftime("%Y-%m-%d") + + # 가동개시일 이전이면 종료 + if current_dt < start_dt: + print(f" 🏁 가동개시일({start_date_str}) 이전 날짜에 도달했습니다. 백필을 마감합니다.") + if not dry_run: + update_backfill_state(site_id, target_date_str, zero_count, 'COMPLETED') + break + + print(f" 📅 [{target_date_str}] 수집 시도 중...") + + if dry_run: + # Dry run 시뮬레이션 + print(f" [Dry-Run] 크롤링 호출 시뮬레이션: {site_id} @ {target_date_str}") + days_processed += 1 + last_date_str = target_date_str + time.sleep(0.1) + continue + + # 실 크롤링 실행 + try: + # fetch_history_daily는 리스트 반환: [{'plant_id': '...', 'date': '...', 'generation_kwh': ...}] + history_data = crawler_fn(plant_config, target_date_str, target_date_str) + time.sleep(delay_sec) # Rate Limit 딜레이 + + matched_val = 0.0 + has_data = False + + if history_data: + # site_id 매칭 처리 + matched = [item for item in history_data if item.get('plant_id') == site_id] + # NREMS 분할이 아니거나 단일 결과 리스트인 경우 fallback + is_split = plant_config.get('options', {}).get('is_split', False) + if not matched and len(history_data) == 1 and not is_split: + matched = history_data + + if matched: + matched_val = matched[0].get('generation_kwh', 0.0) + has_data = True + + if has_data: + print(f" ➔ 🟢 수집 완료: {matched_val:.2f} kWh") + if matched_val <= 0.0: + zero_count += 1 + print(f" ➔ ⚠️ 발전량이 0입니다. (연속 {zero_count}일)") + else: + zero_count = 0 # 발전 확인 시 카운트 리셋 + + # DB 저장 (Supabase) + # matched list 형태로 save_history에 전달 + save_history([{ + 'plant_id': site_id, + 'date': target_date_str, + 'generation_kwh': matched_val + }], 'daily') + else: + zero_count += 1 + print(f" ➔ ⚠️ 원본 데이터가 존재하지 않습니다. (연속 {zero_count}일)") + + except Exception as e: + zero_count += 1 + print(f" ➔ ❌ 크롤링 오류 발생: {e} (연속 {zero_count}일)") + time.sleep(delay_sec) + + days_processed += 1 + last_date_str = target_date_str + + # 상태 업데이트 (매 날짜 진행 마다 유실 방지 위해 업데이트) + update_backfill_state(site_id, last_date_str, zero_count, 'RUNNING') + + # 연속 무발전/데이터 누락 30일 조건 확인 + if zero_count >= 30: + print(f" 🏁 연속 30일 동안 데이터가 없거나 발전량이 0입니다. 백필을 마감합니다.") + update_backfill_state(site_id, last_date_str, zero_count, 'COMPLETED') + break + + print("=" * 80) + print("⚡ [과거 데이터 백필 실행 완료]") + +def main(): + parser = argparse.ArgumentParser(description="SolarPower 과거 발전 데이터 역추적 백필 스케줄러") + parser.add_argument("--dry-run", action="store_true", help="실제 웹 요청 및 DB 저장을 하지 않고 시뮬레이션만 수행") + parser.add_argument("--days", type=int, default=7, help="1회 실행 시 각 발전소당 수집할 최대 과거 일수 (기본: 7일)") + parser.add_argument("--delay", type=float, default=2.0, help="크롤링 요청 간의 시간 대기 초 (기본: 2.0초)") + parser.add_argument("--yesterday", type=str, help="최초 백필 시작 날짜 YYYY-MM-DD (미지정 시 어제)") + parser.add_argument("--reset-site", type=str, help="특정 발전소의 백필 상태를 재설정하여 처음부터 다시 동작하게 함") + + args = parser.parse_args() + + # 1. 초기 시작일 설정 (어제 기준) + if args.yesterday: + yesterday_str = args.yesterday + else: + yesterday_str = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d") + + # 2. 백필 상태 초기화 + init_backfill_states(yesterday_str) + + # 3. 특정 사이트 리셋 옵션 + if args.reset_site: + with get_db_connection() as conn: + cursor = conn.cursor() + cursor.execute(""" + UPDATE backfill_state + SET last_backfilled_date = ?, consecutive_zero_count = 0, status = 'RUNNING' + WHERE site_id = ? + """, (yesterday_str, args.reset_site)) + conn.commit() + print(f"🔄 [{args.reset_site}] 백필 상태가 {yesterday_str} 기준 'RUNNING'으로 재설정되었습니다.") + + # 4. 백필 작업 진행 + process_backfill(max_days_per_run=args.days, delay_sec=args.delay, dry_run=args.dry_run) + +if __name__ == "__main__": + main() diff --git a/crawler/crawler_manager.py b/crawler/crawler_manager.py index 9f22b30..595a842 100644 --- a/crawler/crawler_manager.py +++ b/crawler/crawler_manager.py @@ -64,6 +64,13 @@ class CrawlerManager: detected_minute INTEGER, detected_at TEXT ); + + CREATE TABLE IF NOT EXISTS backfill_state ( + site_id TEXT PRIMARY KEY, + last_backfilled_date TEXT, + consecutive_zero_count INTEGER DEFAULT 0, + status TEXT DEFAULT 'RUNNING' + ); """) conn.commit()