274 lines
10 KiB
Python
274 lines
10 KiB
Python
# ==========================================
|
||
# daily_summary.py - 일일 발전 통계 집계
|
||
# ==========================================
|
||
# solar_logs 데이터를 집계하여 daily_stats 테이블에 저장
|
||
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
try:
|
||
from dotenv import load_dotenv
|
||
load_dotenv()
|
||
except ImportError:
|
||
pass
|
||
|
||
import pandas as pd
|
||
from database import get_supabase_client
|
||
from config import get_all_plants
|
||
|
||
def get_history_crawler(plant_type: str):
|
||
"""발전소 타입별 과거 데이터 크롤러 함수 반환"""
|
||
try:
|
||
if plant_type == 'nrems':
|
||
from crawlers.nrems import fetch_history_daily
|
||
return fetch_history_daily
|
||
elif plant_type == 'kremc':
|
||
from crawlers.kremc import fetch_history_daily
|
||
return fetch_history_daily
|
||
elif plant_type == 'hyundai':
|
||
from crawlers.hyundai import fetch_history_daily
|
||
return fetch_history_daily
|
||
elif plant_type == 'sun_wms':
|
||
from crawlers.sun_wms import fetch_history_daily
|
||
return fetch_history_daily
|
||
elif plant_type == 'cmsolar':
|
||
from crawlers.cmsolar import fetch_history_daily
|
||
return fetch_history_daily
|
||
except Exception as e:
|
||
print(f" ⚠️ 크롤러 모듈 임포트 실패 ({plant_type}): {e}")
|
||
return None
|
||
|
||
|
||
def get_plant_capacities(client) -> dict:
|
||
"""plants 테이블에서 용량 정보 조회"""
|
||
try:
|
||
result = client.table("plants").select("id, capacity").execute()
|
||
return {row['id']: row.get('capacity', 99.0) for row in result.data}
|
||
except Exception as e:
|
||
print(f" ⚠️ 용량 조회 실패: {e}")
|
||
return {}
|
||
|
||
|
||
def calculate_daily_stats(date_str: str = None):
|
||
"""
|
||
특정 날짜의 발전 통계 집계
|
||
|
||
Args:
|
||
date_str: 집계 대상 날짜 (YYYY-MM-DD). 미지정 시 오늘.
|
||
"""
|
||
if date_str is None:
|
||
kst = timezone(timedelta(hours=9))
|
||
date_str = datetime.now(kst).strftime('%Y-%m-%d')
|
||
|
||
print(f"\n📊 [일일 통계 집계] {date_str}")
|
||
print("-" * 60)
|
||
|
||
client = get_supabase_client()
|
||
if not client:
|
||
print("❌ Supabase 연결 실패")
|
||
return False
|
||
|
||
# 1. 용량 정보 조회
|
||
capacities = get_plant_capacities(client)
|
||
|
||
# 발전소 정보 빌드 (하이브리드 크롤링용)
|
||
all_plants = get_all_plants()
|
||
plant_info_map = {}
|
||
for p in all_plants:
|
||
is_split = p.get('options', {}).get('is_split', False)
|
||
if is_split:
|
||
plant_info_map['nrems-01'] = p
|
||
plant_info_map['nrems-02'] = p
|
||
else:
|
||
plant_info_map[p.get('id', '')] = p
|
||
|
||
# 2. 해당일 로그 조회 (KST 날짜 범위를 UTC로 변환하여 쿼리)
|
||
kst = timezone(timedelta(hours=9))
|
||
start_kst = datetime.strptime(f"{date_str} 00:00:00", "%Y-%m-%d %H:%M:%S").replace(tzinfo=kst)
|
||
end_kst = datetime.strptime(f"{date_str} 23:59:59", "%Y-%m-%d %H:%M:%S").replace(tzinfo=kst)
|
||
start_utc = start_kst.astimezone(timezone.utc).isoformat()
|
||
end_utc = end_kst.astimezone(timezone.utc).isoformat()
|
||
|
||
try:
|
||
result = client.table("solar_logs") \
|
||
.select("plant_id, current_kw, today_kwh, created_at") \
|
||
.gte("created_at", start_utc) \
|
||
.lte("created_at", end_utc) \
|
||
.order("created_at", desc=False) \
|
||
.execute()
|
||
|
||
if not result.data:
|
||
print(" ⚠️ 해당 날짜의 로그가 없습니다.")
|
||
return False
|
||
|
||
df = pd.DataFrame(result.data)
|
||
|
||
except Exception as e:
|
||
print(f" ❌ 로그 조회 실패: {e}")
|
||
return False
|
||
|
||
# 3. 발전소별 통계 계산
|
||
stats_list = []
|
||
|
||
for plant_id, group in df.groupby('plant_id'):
|
||
# 당일 최댓값 today_kwh 사용 (마지막 로그가 아닌 최댓값으로 중간 리셋 보호)
|
||
total_generation = group['today_kwh'].max() if len(group) > 0 else 0
|
||
|
||
# 최대 출력
|
||
peak_kw = group['current_kw'].max() if len(group) > 0 else 0
|
||
|
||
# 하이브리드 보정 로직 적용 (원 모니터링 사이트 통계 크롤링)
|
||
plant_info = plant_info_map.get(plant_id)
|
||
if plant_info:
|
||
plant_type = plant_info.get('type')
|
||
is_split = plant_info.get('options', {}).get('is_split', False)
|
||
history_crawler = get_history_crawler(plant_type)
|
||
if history_crawler:
|
||
try:
|
||
print(f" 📡 [{plant_id}] 원본 사이트 통계 크롤링 시도 ({date_str}) ...")
|
||
# fetch_history_daily는 리스트 반환: [{'plant_id': '...', 'date': '...', 'generation_kwh': ...}]
|
||
history_data = history_crawler(plant_info, date_str, date_str)
|
||
if history_data:
|
||
# nrems-01, nrems-02 분할 및 단일 매칭 처리
|
||
matched = [item for item in history_data if item.get('plant_id') == plant_id]
|
||
if not matched and len(history_data) == 1 and not is_split:
|
||
matched = history_data
|
||
|
||
if matched:
|
||
original_val = matched[0].get('generation_kwh', 0)
|
||
if original_val > 0:
|
||
print(f" ➔ 🟢 보정 성공: 기존 집계 {total_generation:.1f} kWh ➔ 원본 {original_val:.1f} kWh")
|
||
total_generation = original_val
|
||
else:
|
||
print(f" ➔ ⚠️ 원본 값이 0이므로 보정을 건너뜁니다.")
|
||
else:
|
||
print(f" ➔ ⚠️ 원본 데이터 중 {plant_id}와 일치하는 항목을 찾지 못했습니다.")
|
||
else:
|
||
print(f" ➔ ⚠️ 원본 데이터가 비어 있습니다.")
|
||
except Exception as ex:
|
||
print(f" ➔ ❌ 원본 사이트 크롤링 실패 (Fallback 기존 집계 적용): {ex}")
|
||
|
||
# 이용률 시간 = 발전량 / 용량
|
||
capacity = capacities.get(plant_id, 99.0)
|
||
generation_hours = round(total_generation / capacity, 2) if capacity > 0 else 0
|
||
|
||
stats = {
|
||
'plant_id': plant_id,
|
||
'date': date_str,
|
||
'total_generation': round(total_generation, 2),
|
||
'peak_kw': round(peak_kw, 2),
|
||
'generation_hours': generation_hours
|
||
}
|
||
stats_list.append(stats)
|
||
|
||
# 출력
|
||
print(f" {plant_id}: {total_generation:.1f}kWh ({generation_hours:.1f}시간, 최대 {peak_kw:.1f}kW)")
|
||
|
||
# 4. daily_stats 테이블에 Upsert
|
||
if stats_list:
|
||
try:
|
||
result = client.table("daily_stats").upsert(
|
||
stats_list,
|
||
on_conflict="plant_id,date"
|
||
).execute()
|
||
|
||
print("-" * 60)
|
||
print(f"✅ {len(stats_list)}개 발전소 통계 저장 완료")
|
||
|
||
except Exception as e:
|
||
print(f" ❌ 저장 실패: {e}")
|
||
return False
|
||
|
||
return True
|
||
|
||
|
||
def calculate_monthly_stats(target_month: str):
|
||
"""
|
||
특정 월의 발전 통계 집계 (일간 데이터 합산)
|
||
|
||
Args:
|
||
target_month: YYYY-MM
|
||
"""
|
||
print(f"\n📅 [월간 통계 집계] {target_month}")
|
||
print("-" * 60)
|
||
|
||
client = get_supabase_client()
|
||
if not client:
|
||
return False
|
||
|
||
try:
|
||
# 1. 모든 발전소 ID 조회
|
||
plants_res = client.table("plants").select("id").execute()
|
||
plant_ids = [p['id'] for p in plants_res.data]
|
||
|
||
updated_count = 0
|
||
|
||
for pid in plant_ids:
|
||
# 2. 해당 월의 Daily 합계 조회
|
||
import calendar
|
||
year_str, month_str = target_month.split("-")
|
||
last_day = calendar.monthrange(int(year_str), int(month_str))[1]
|
||
|
||
d_res = client.table("daily_stats").select("total_generation") \
|
||
.eq("plant_id", pid) \
|
||
.gte("date", f"{target_month}-01") \
|
||
.lte("date", f"{target_month}-{last_day:02d}") \
|
||
.execute()
|
||
|
||
if not d_res.data:
|
||
continue
|
||
|
||
total_gen = sum(r.get('total_generation', 0) or 0 for r in d_res.data)
|
||
|
||
# 3. Monthly Upsert
|
||
client.table("monthly_stats").upsert({
|
||
"plant_id": pid,
|
||
"month": target_month,
|
||
"total_generation": round(total_gen, 2),
|
||
"updated_at": datetime.now().isoformat()
|
||
}, on_conflict="plant_id, month").execute()
|
||
|
||
print(f" {pid}: {total_gen:.1f}kWh (Month Total)")
|
||
updated_count += 1
|
||
|
||
print("-" * 60)
|
||
print(f"✅ {updated_count}개 발전소 월간 통계 갱신 완료")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f" ❌ 월간 집계 실패: {e}")
|
||
return False
|
||
|
||
|
||
if __name__ == "__main__":
|
||
import sys
|
||
from datetime import timedelta
|
||
|
||
# 인자로 날짜 지정 가능: python daily_summary.py 2026-01-22
|
||
if len(sys.argv) > 1:
|
||
target_date = sys.argv[1]
|
||
else:
|
||
# 인자 없으면 '어제' 날짜를 기본값으로 사용
|
||
# (새벽에 실행하여 전날 데이터를 마감하는 시나리오)
|
||
yesterday = datetime.now() - timedelta(days=1)
|
||
target_date = yesterday.strftime('%Y-%m-%d')
|
||
print(f"ℹ️ 날짜 미지정 -> 어제({target_date}) 기준으로 집계합니다.")
|
||
|
||
# 1. 일간 통계 집계
|
||
success = calculate_daily_stats(target_date)
|
||
|
||
# 2. 월말 체크 및 월간 집계 트리거
|
||
# target_date가 해당 월의 마지막 날이면 월간 집계 실행
|
||
if success:
|
||
try:
|
||
current_dt = datetime.strptime(target_date, '%Y-%m-%d')
|
||
import calendar
|
||
last_day = calendar.monthrange(current_dt.year, current_dt.month)[1]
|
||
|
||
if current_dt.day == last_day:
|
||
target_month = current_dt.strftime('%Y-%m')
|
||
print(f"\n🔔 월말({target_date}) 감지 -> {target_month} 월간 집계 실행")
|
||
calculate_monthly_stats(target_month)
|
||
except Exception as e:
|
||
print(f"⚠️ 월간 집계 트리거 오류: {e}")
|
||
|