solorpower/crawler/daily_summary.py

274 lines
10 KiB
Python
Raw Permalink Blame History

This file contains invisible Unicode characters

This file contains invisible Unicode characters that are indistinguishable to humans but may be processed differently by a computer. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

# ==========================================
# 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}")