298 lines
12 KiB
Python
298 lines
12 KiB
Python
#!/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()
|
|
|
|
# 테이블 자동 생성
|
|
cursor.execute("""
|
|
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()
|
|
|
|
# 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()
|