feat: implement backward historical backfill scheduler with SQLite state tracking

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()

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()