#!/usr/bin/env python3 """ Solar Production Data Scraper for e-SenZ Scrapes solar production data from https://esenz.co.in/esenzCustomerLoginV3.aspx """ import time import json import logging import os import re from datetime import datetime from selenium import webdriver from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC from selenium.webdriver.chrome.service import Service from selenium.webdriver.chrome.options import Options from bs4 import BeautifulSoup # Configure logging - ONLY TO FILE, NOT STDOUT logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('solar_scraper.log') ] ) logger = logging.getLogger(__name__) class SolarScraper: def __init__(self, username, password, headless=True): self.username = username self.password = password self.base_url = "https://esenz.co.in/esenzCustomerLoginV3.aspx" self.driver = None self.headless = headless def setup_driver(self): """Setup Chrome WebDriver with appropriate options""" try: chrome_options = Options() if self.headless: chrome_options.add_argument("--headless") chrome_options.add_argument("--no-sandbox") chrome_options.add_argument("--disable-dev-shm-usage") chrome_options.add_argument("--disable-gpu") chrome_options.add_argument("--window-size=1920,1080") chrome_options.add_argument("--disable-web-security") chrome_options.add_argument("--allow-running-insecure-content") # Set browser path for different systems (RPi first, then Mac) chromium_path = "/usr/bin/chromium-browser" brave_path = "/Applications/Brave Browser.app/Contents/MacOS/Brave Browser" if os.path.exists(chromium_path): chrome_options.binary_location = chromium_path logger.info("Using Chromium Browser (RPi)") elif os.path.exists(brave_path): chrome_options.binary_location = brave_path logger.info("Using Brave Browser (Mac)") else: logger.warning("No compatible browser found, using default Chrome") # Try to use system ChromeDriver first (RPi), then local (Mac) system_chromedriver = "/usr/bin/chromedriver" local_chromedriver = os.path.join(os.getcwd(), "chromedriver-mac-x64", "chromedriver") if os.path.exists(system_chromedriver): service = Service(system_chromedriver) logger.info(f"Using system ChromeDriver at: {system_chromedriver} (RPi)") elif os.path.exists(local_chromedriver): service = Service(local_chromedriver) logger.info(f"Using local ChromeDriver at: {local_chromedriver} (Mac)") else: # Try to use ChromeDriver without specifying path (let Selenium find it) logger.info("Using default ChromeDriver (system PATH)") service = Service() self.driver = webdriver.Chrome(service=service, options=chrome_options) self.driver.implicitly_wait(10) logger.info("WebDriver setup successful") except Exception as e: logger.error(f"WebDriver setup failed: {e}") raise def login(self): """Login to the e-SenZ website""" try: logger.info("Navigating to login page...") self.driver.get(self.base_url) # Wait for page to load time.sleep(3) # Find and fill username field logger.info("Entering username...") username_field = WebDriverWait(self.driver, 10).until( EC.presence_of_element_located((By.NAME, "idTextBoxCustomerSYSID")) ) username_field.clear() username_field.send_keys(self.username) # Find and fill password field logger.info("Entering password...") password_field = self.driver.find_element(By.NAME, "idTextBoxCustomerPassword") password_field.clear() password_field.send_keys(self.password) # Click login button logger.info("Clicking login button...") login_button = self.driver.find_element(By.NAME, "idButtonLogin") login_button.click() # Wait for redirect to dashboard time.sleep(5) # Check if login was successful if "dashboard" in self.driver.current_url.lower() or "default" in self.driver.current_url.lower(): logger.info("Login successful!") return True else: logger.error("Login failed - still on login page") return False except Exception as e: logger.error(f"Login failed: {str(e)}") return False def extract_solar_data(self): """Extract solar production data from the dashboard""" try: logger.info("Extracting solar data...") # Wait for the page to fully load and for AJAX data to populate time.sleep(15) # Try to wait for specific data elements to appear try: WebDriverWait(self.driver, 20).until( lambda driver: driver.execute_script("return document.body.innerHTML.includes('kWh')") ) logger.info("Page content loaded") # Additional wait for dynamic content time.sleep(5) except: logger.warning("Timeout waiting for page content") # Get page source and parse with BeautifulSoup soup = BeautifulSoup(self.driver.page_source, 'html.parser') # Initialize data dictionary solar_data = { 'timestamp': datetime.now().isoformat(), 'current_power': None, 'today_energy': None, 'yesterday_energy': None, 'monthly_energy': None, 'total_energy': None, 'generation_factor_today': None, 'generation_factor_yesterday': None, 'cuf_yesterday': None, 'performance_ratio_today': None, 'irradiation_today': None } # Try to extract data using various selectors # This will need to be adjusted based on the actual HTML structure # Try to extract data using JavaScript first try: # Execute JavaScript to get data from the page js_data = self.driver.execute_script(""" var data = {}; // Look for elements with actual values var elements = document.querySelectorAll('*'); for (var i = 0; i < elements.length; i++) { var el = elements[i]; var text = el.textContent || el.innerText; // Look for power values (extract just the number and unit) if (text && text.match(/\\d+(\\.\\d+)?\\s*(W|kW)/)) { var match = text.match(/(\\d+(\\.\\d+)?)\\s*(W|kW)/); if (match) { data.current_power = match[1] + ' ' + match[3]; } } // Look for energy values with numbers (extract just the number and unit) if (text && text.match(/\\d+(\\.\\d+)?\\s*kWh/)) { var match = text.match(/(\\d+(\\.\\d+)?)\\s*kWh/); if (match) { if (text.includes('Today') || text.includes('today')) data.today_energy = match[1] + ' kWh'; else if (text.includes('Yesterday') || text.includes('Yest')) data.yesterday_energy = match[1] + ' kWh'; else if (text.includes('Month') || text.includes('month')) data.monthly_energy = match[1] + ' kWh'; else if (text.includes('Total') || text.includes('total')) data.total_energy = match[1] + ' kWh'; } } // Look for patterns like "0.00" followed by "This Month" or similar if (text && text.match(/^\\d+(\\.\\d+)?$/)) { var value = text.trim(); // Check if this number is followed by a month-related text in nearby elements var nextElement = el.nextElementSibling; if (nextElement && (nextElement.textContent.includes('Month') || nextElement.textContent.includes('month'))) { data.monthly_energy = value + ' kWh'; } } // Also look for patterns like "1.30 Today (kWh)" or "27.86 Yest. (kWh)" if (text && text.match(/\\d+(\\.\\d+)?\\s+(Today|Yest|Month|Total)/)) { var match = text.match(/(\\d+(\\.\\d+)?)\\s+(Today|Yest|Month|Total)/); if (match) { var value = match[1]; var type = match[3]; if (type === 'Today') data.today_energy = value + ' kWh'; else if (type === 'Yest') data.yesterday_energy = value + ' kWh'; else if (type === 'Month') data.monthly_energy = value + ' kWh'; else if (type === 'Total') data.total_energy = value + ' kWh'; } } // Look for GF values (extract just the number) if (text && text.match(/\\d+(\\.\\d+)?\\s*GF/)) { var match = text.match(/(\\d+(\\.\\d+)?)/); if (match) { if (text.includes('Today')) data.gf_today = match[1]; else if (text.includes('Yesterday')) data.gf_yesterday = match[1]; } } // Look for CUF values (extract just the number) if (text && text.match(/\\d+(\\.\\d+)?\\s*CUF/)) { var match = text.match(/(\\d+(\\.\\d+)?)/); if (match) { data.cuf_yesterday = match[1]; } } // Look for PR values (extract just the number or dash) if (text && text.match(/(\\d+(\\.\\d+)?|-)\\s*PR/)) { var match = text.match(/(\\d+(\\.\\d+)?|-)/); if (match) { data.pr_today = match[1]; } } } return data; """) # Update solar_data with JavaScript results for key, value in js_data.items(): if value: solar_data[key] = value except Exception as e: logger.warning(f"JavaScript extraction failed: {e}") # Fallback: Look for current power (usually in a gauge or large display) power_elements = soup.find_all(string=lambda text: text and 'W' in text and any(char.isdigit() for char in text)) if power_elements and not solar_data.get('current_power'): for element in power_elements: if 'W' in element and any(char.isdigit() for char in element): solar_data['current_power'] = element.strip() break # Fallback: Look for energy values (kWh) - only if JavaScript didn't find them if not solar_data.get('today_energy') or not solar_data.get('yesterday_energy') or not solar_data.get('monthly_energy'): energy_elements = soup.find_all(string=lambda text: text and 'kWh' in text) for element in energy_elements: text = element.strip() # Only extract if we have a number followed by kWh if re.search(r'\d+\.?\d*\s*kWh', text): if 'Today' in text and not solar_data.get('today_energy'): match = re.search(r'(\d+\.?\d*)\s*kWh', text) if match: solar_data['today_energy'] = match.group(1) + ' kWh' elif 'Yesterday' in text and not solar_data.get('yesterday_energy'): match = re.search(r'(\d+\.?\d*)\s*kWh', text) if match: solar_data['yesterday_energy'] = match.group(1) + ' kWh' elif 'month' in text.lower() and not solar_data.get('monthly_energy'): match = re.search(r'(\d+\.?\d*)\s*kWh', text) if match: solar_data['monthly_energy'] = match.group(1) + ' kWh' elif 'Total' in text and not solar_data.get('total_energy'): match = re.search(r'(\d+\.?\d*)\s*kWh', text) if match: solar_data['total_energy'] = match.group(1) + ' kWh' # Fallback: Look for GF (Generation Factor) values gf_elements = soup.find_all(string=lambda text: text and 'GF' in text) for element in gf_elements: text = element.strip() if 'Today' in text and not solar_data.get('gf_today'): solar_data['generation_factor_today'] = text elif 'Yesterday' in text and not solar_data.get('gf_yesterday'): solar_data['generation_factor_yesterday'] = text # Fallback: Look for CUF (Capacity Utilization Factor) cuf_elements = soup.find_all(string=lambda text: text and 'CUF' in text) if cuf_elements and not solar_data.get('cuf_yesterday'): solar_data['cuf_yesterday'] = cuf_elements[0].strip() # Fallback: Look for PR (Performance Ratio) pr_elements = soup.find_all(string=lambda text: text and 'PR' in text) if pr_elements and not solar_data.get('pr_today'): solar_data['performance_ratio_today'] = pr_elements[0].strip() # Fallback: Look for Irradiation irr_elements = soup.find_all(string=lambda text: text and 'Irr' in text) if irr_elements and not solar_data.get('irradiation_today'): solar_data['irradiation_today'] = irr_elements[0].strip() logger.info(f"Extracted data: {solar_data}") # Validate that we got real data (not just labels) if (solar_data.get('today_energy') and not solar_data['today_energy'].startswith('E Today') and not solar_data['today_energy'].startswith('Today')): logger.info("✅ Valid data extracted") return solar_data else: logger.warning("❌ Extracted data appears to be labels, not values") return solar_data # Return anyway to avoid infinite recursion except Exception as e: logger.error(f"Data extraction failed: {str(e)}") return None def save_data(self, data): """Save extracted data to JSON file""" try: filename = f"solar_data_{datetime.now().strftime('%Y%m%d')}.json" # Load existing data if file exists existing_data = [] try: with open(filename, 'r') as f: existing_data = json.load(f) except FileNotFoundError: pass # Add new data existing_data.append(data) # Save updated data with open(filename, 'w') as f: json.dump(existing_data, f, indent=2) logger.info(f"Data saved to {filename}") except Exception as e: logger.error(f"Failed to save data: {str(e)}") def scrape_once(self): """Perform one complete scraping cycle""" try: self.setup_driver() if self.login(): data = self.extract_solar_data() if data: self.save_data(data) return data else: logger.error("Failed to extract data") return None else: logger.error("Failed to login") return None except Exception as e: logger.error(f"Scraping failed: {str(e)}") return None finally: if self.driver: self.driver.quit() def close(self): """Close the browser""" if self.driver: self.driver.quit() def main(): """Main function to test the scraper""" username = "314" password = "solgen314" scraper = SolarScraper(username, password, headless=False) # Set headless=False for testing try: data = scraper.scrape_once() if data: print("Scraping successful!") print(json.dumps(data, indent=2)) else: print("Scraping failed!") except Exception as e: print(f"Error: {str(e)}") finally: scraper.close() if __name__ == "__main__": main()