360 lines
12 KiB
Python
360 lines
12 KiB
Python
# ==========================================
|
|
# crawlers/sun_wms.py - Sun-WMS 크롤러 (6호기)
|
|
# ==========================================
|
|
|
|
import requests
|
|
import re
|
|
import time
|
|
from .base import create_session, safe_float
|
|
|
|
def fetch_data(plant_info):
|
|
"""
|
|
Sun-WMS 발전소 데이터 수집
|
|
"""
|
|
plant_id = plant_info.get('id', 'sunwms-06')
|
|
auth = plant_info.get('auth', {})
|
|
system = plant_info.get('system', {})
|
|
company_name = plant_info.get('company_name', '태양과바람')
|
|
plant_name = plant_info.get('name', '6호기')
|
|
|
|
payload_id = auth.get('payload_id', '')
|
|
payload_pw = auth.get('payload_pw', '')
|
|
login_url = system.get('login_url', '')
|
|
data_url = system.get('data_url', '')
|
|
|
|
session = create_session()
|
|
|
|
headers = {
|
|
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120.0.0.0 Safari/537.36',
|
|
'Referer': 'http://tb6.sun-wms.com/public/main/login.php',
|
|
'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8'
|
|
}
|
|
|
|
# 1. 로그인
|
|
login_data = {
|
|
'act': 'loginChk',
|
|
'user_id': payload_id,
|
|
'user_pass': payload_pw
|
|
}
|
|
|
|
try:
|
|
res = session.post(login_url, data=login_data, headers=headers)
|
|
if res.status_code != 200:
|
|
return []
|
|
except Exception as e:
|
|
print(f"❌ {plant_name} 접속 에러: {e}")
|
|
return []
|
|
|
|
# 2. 데이터 요청
|
|
try:
|
|
timestamp = int(time.time() * 1000)
|
|
res = session.get(f"{data_url}?time={timestamp}", headers=headers)
|
|
res.encoding = 'euc-kr'
|
|
|
|
content = res.text
|
|
|
|
match_kw = re.search(r"id=['\"]cur_power['\"].*?value=['\"]([^'\"]+)['\"]", content)
|
|
curr_kw = float(match_kw.group(1)) if match_kw else 0.0
|
|
|
|
match_today = re.search(r"id=['\"]today_power['\"].*?value=['\"]([^'\"]+)['\"]", content)
|
|
today_kwh = float(match_today.group(1)) if match_today else 0.0
|
|
|
|
status = "🟢 정상" if curr_kw > 0 else "💤 대기"
|
|
|
|
return [{
|
|
'id': plant_id,
|
|
'name': f'{company_name} {plant_name}',
|
|
'kw': curr_kw,
|
|
'today': today_kwh,
|
|
'status': status
|
|
}]
|
|
|
|
except Exception as e:
|
|
print(f"❌ {plant_name} 에러: {e}")
|
|
return []
|
|
|
|
|
|
def fetch_history_hourly(plant_info, start_date, end_date):
|
|
"""
|
|
Sun-WMS 발전소의 시간대별 과거 데이터 수집
|
|
|
|
실제 엔드포인트: /public/statics/statics.php
|
|
파라미터: tab01=0&tab02=1&tab03=2&tord=1&s_day=YYYY-MM-DD
|
|
"""
|
|
from datetime import datetime, timedelta
|
|
|
|
results = []
|
|
plant_id = plant_info.get('id', 'sunwms-06')
|
|
auth = plant_info.get('auth', {})
|
|
system = plant_info.get('system', {})
|
|
plant_name = plant_info.get('name', '6호기')
|
|
|
|
payload_id = auth.get('payload_id', '')
|
|
payload_pw = auth.get('payload_pw', '')
|
|
login_url = system.get('login_url', '')
|
|
|
|
# base_url 추출
|
|
base_url = system.get('base_url', '')
|
|
if not base_url and 'http' in login_url:
|
|
base_url = login_url.split('/public')[0]
|
|
|
|
statics_url = system.get('statics_url', f"{base_url}/public/statics/statics.php")
|
|
|
|
session = create_session()
|
|
|
|
print(f"\n{'='*60}")
|
|
print(f"[Sun-WMS History] {plant_name} ({start_date} ~ {end_date})")
|
|
print(f"{'='*60}")
|
|
|
|
headers = {
|
|
'User-Agent': 'Mozilla/5.0',
|
|
'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8'
|
|
}
|
|
|
|
login_data = {'act': 'loginChk', 'user_id': payload_id, 'user_pass': payload_pw}
|
|
|
|
try:
|
|
res = session.post(login_url, data=login_data, headers=headers)
|
|
if res.status_code == 200:
|
|
print(" ✓ Login successful")
|
|
else:
|
|
print(" ✗ Login failed")
|
|
return results
|
|
except Exception as e:
|
|
print(f" ✗ Login error: {e}")
|
|
return results
|
|
|
|
# 날짜 반복
|
|
current_date = datetime.strptime(start_date, '%Y-%m-%d')
|
|
end_dt = datetime.strptime(end_date, '%Y-%m-%d')
|
|
|
|
while current_date <= end_dt:
|
|
date_str = current_date.strftime('%Y-%m-%d')
|
|
|
|
# 실제 확인된 시간별 엔드포인트
|
|
params = {
|
|
'tab01': '0',
|
|
'tab02': '1',
|
|
'tab03': '2',
|
|
'tord': '1',
|
|
's_day': date_str
|
|
}
|
|
|
|
try:
|
|
res = session.get(statics_url, params=params, headers=headers, timeout=10)
|
|
res.encoding = 'euc-kr'
|
|
|
|
if res.status_code == 200:
|
|
data = res.json()
|
|
# 시간별 데이터 파싱
|
|
hourly_data = data.get('data', []) or data.get('list', [])
|
|
|
|
if isinstance(hourly_data, list) and len(hourly_data) > 0:
|
|
print(f" ✓ Found {len(hourly_data)} hourly records")
|
|
|
|
for item in hourly_data:
|
|
hour = str(item.get('hour', item.get('time', '00'))).zfill(2)
|
|
generation_kwh = safe_float(item.get('generation', item.get('kwh', 0)))
|
|
current_kw = safe_float(item.get('power', item.get('kw', 0)))
|
|
|
|
timestamp = f"{date_str} {hour}:00:00"
|
|
|
|
results.append({
|
|
'plant_id': plant_id,
|
|
'timestamp': timestamp,
|
|
'generation_kwh': generation_kwh,
|
|
'current_kw': current_kw
|
|
})
|
|
else:
|
|
print(f" ⚠ No data for {date_str}")
|
|
else:
|
|
print(f" ✗ HTTP {res.status_code}")
|
|
|
|
except Exception as e:
|
|
print(f" ✗ Error: {e}")
|
|
|
|
current_date += timedelta(days=1)
|
|
|
|
print(f"\n{'='*60}")
|
|
print(f"[Total] Collected {len(results)} hourly records")
|
|
print(f"{'='*60}\n")
|
|
|
|
return results
|
|
|
|
|
|
def fetch_history_daily(plant_info, start_date, end_date):
|
|
"""
|
|
Sun-WMS 발전소의 일별 과거 데이터 수집
|
|
|
|
실제 엔드포인트: /public/statics/statics.php
|
|
파라미터: tab01=0&tab02=2&tab03=2&tord=2&s_day=YYYY-MM-DD&e_day=YYYY-MM-DD
|
|
"""
|
|
from datetime import datetime
|
|
|
|
results = []
|
|
plant_id = plant_info.get('id', 'sunwms-06')
|
|
auth = plant_info.get('auth', {})
|
|
system = plant_info.get('system', {})
|
|
plant_name = plant_info.get('name', '6호기')
|
|
|
|
payload_id = auth.get('payload_id', '')
|
|
payload_pw = auth.get('payload_pw', '')
|
|
login_url = system.get('login_url', '')
|
|
|
|
# base_url 추출
|
|
base_url = system.get('base_url', '')
|
|
if not base_url and 'http' in login_url:
|
|
base_url = login_url.split('/public')[0]
|
|
|
|
statics_url = system.get('statics_url', f"{base_url}/public/statics/statics.php")
|
|
|
|
session = create_session()
|
|
|
|
print(f"\n{'='*60}")
|
|
print(f"[Sun-WMS Daily] {plant_name} ({start_date} ~ {end_date})")
|
|
print(f"{'='*60}")
|
|
|
|
headers = {
|
|
'User-Agent': 'Mozilla/5.0',
|
|
'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8'
|
|
}
|
|
|
|
login_data = {'act': 'loginChk', 'user_id': payload_id, 'user_pass': payload_pw}
|
|
|
|
try:
|
|
res = session.post(login_url, data=login_data, headers=headers)
|
|
if res.status_code == 200:
|
|
print(" ✓ Login successful")
|
|
else:
|
|
print(" ✗ Login failed")
|
|
return results
|
|
except Exception as e:
|
|
print(f" ✗ Login error: {e}")
|
|
return results
|
|
|
|
# 실제 확인된 일별 엔드포인트
|
|
params = {
|
|
'tab01': '0',
|
|
'tab02': '2',
|
|
'tab03': '2',
|
|
'tord': '2',
|
|
's_day': start_date,
|
|
'e_day': end_date
|
|
}
|
|
|
|
try:
|
|
res = session.get(statics_url, params=params, headers=headers, timeout=10)
|
|
res.encoding = 'euc-kr'
|
|
|
|
if res.status_code == 200:
|
|
data = res.json()
|
|
# 일별 데이터 파싱
|
|
daily_data = data.get('data', []) or data.get('list', [])
|
|
|
|
if isinstance(daily_data, list) and len(daily_data) > 0:
|
|
for item in daily_data:
|
|
date_str = item.get('date', item.get('day', ''))
|
|
generation_kwh = safe_float(item.get('generation', item.get('kwh', 0)))
|
|
current_kw = safe_float(item.get('power', item.get('kw', 0)))
|
|
|
|
results.append({
|
|
'plant_id': plant_id,
|
|
'date': date_str,
|
|
'generation_kwh': generation_kwh,
|
|
'current_kw': current_kw
|
|
})
|
|
print(f" ✓ {date_str}: {generation_kwh:.2f}kWh")
|
|
except Exception as e:
|
|
print(f" ✗ Error: {e}")
|
|
|
|
print(f"[Total] Collected {len(results)} daily records\n")
|
|
return results
|
|
|
|
|
|
def fetch_history_monthly(plant_info, start_month, end_month):
|
|
"""
|
|
Sun-WMS 발전소의 월별 과거 데이터 수집
|
|
|
|
실제 엔드포인트: /public/statics/statics.php
|
|
파라미터: tab01=0&tab02=3&tab03=2&tord=3&s_day=YYYY-MM&e_day=YYYY-MM
|
|
"""
|
|
from datetime import datetime
|
|
from dateutil.relativedelta import relativedelta
|
|
|
|
results = []
|
|
plant_id = plant_info.get('id', 'sunwms-06')
|
|
auth = plant_info.get('auth', {})
|
|
system = plant_info.get('system', {})
|
|
plant_name = plant_info.get('name', '6호기')
|
|
|
|
payload_id = auth.get('payload_id', '')
|
|
payload_pw = auth.get('payload_pw', '')
|
|
login_url = system.get('login_url', '')
|
|
|
|
# base_url 추출
|
|
base_url = system.get('base_url', '')
|
|
if not base_url and 'http' in login_url:
|
|
base_url = login_url.split('/public')[0]
|
|
|
|
statics_url = system.get('statics_url', f"{base_url}/public/statics/statics.php")
|
|
|
|
session = create_session()
|
|
|
|
print(f"\n{'='*60}")
|
|
print(f"[Sun-WMS Monthly] {plant_name} ({start_month} ~ {end_month})")
|
|
print(f"{'='*60}")
|
|
|
|
headers = {
|
|
'User-Agent': 'Mozilla/5.0',
|
|
'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8'
|
|
}
|
|
|
|
login_data = {'act': 'loginChk', 'user_id': payload_id, 'user_pass': payload_pw}
|
|
|
|
try:
|
|
res = session.post(login_url, data=login_data, headers=headers)
|
|
if res.status_code == 200:
|
|
print(" ✓ Login successful")
|
|
else:
|
|
print(" ✗ Login failed")
|
|
return results
|
|
except Exception as e:
|
|
print(f" ✗ Login error: {e}")
|
|
return results
|
|
|
|
# 실제 확인된 월별 엔드포인트
|
|
params = {
|
|
'tab01': '0',
|
|
'tab02': '3',
|
|
'tab03': '2',
|
|
'tord': '3',
|
|
's_day': start_month,
|
|
'e_day': end_month
|
|
}
|
|
|
|
try:
|
|
res = session.get(statics_url, params=params, headers=headers, timeout=10)
|
|
res.encoding = 'euc-kr'
|
|
|
|
if res.status_code == 200:
|
|
data = res.json()
|
|
# 월별 데이터 파싱
|
|
monthly_data = data.get('data', []) or data.get('list', [])
|
|
|
|
if isinstance(monthly_data, list) and len(monthly_data) > 0:
|
|
for item in monthly_data:
|
|
month_str = item.get('month', item.get('date', ''))
|
|
generation_kwh = safe_float(item.get('generation', item.get('kwh', item.get('monthTotal', 0))))
|
|
|
|
results.append({
|
|
'plant_id': plant_id,
|
|
'month': month_str[:7] if len(month_str) >= 7 else month_str,
|
|
'generation_kwh': generation_kwh
|
|
})
|
|
print(f" ✓ {month_str[:7]}: {generation_kwh:.1f}kWh")
|
|
except Exception as e:
|
|
print(f" ✗ Error: {e}")
|
|
|
|
print(f"[Total] Collected {len(results)} monthly records\n")
|
|
return results
|