diff --git a/backend/src/main.py b/backend/src/main.py index fb0d721..1c04cb1 100644 --- a/backend/src/main.py +++ b/backend/src/main.py @@ -1,13 +1,20 @@ from fastapi import FastAPI +from pydantic import BaseModel app = FastAPI(title="HanchuESS Solar Backend API") + +class DecryptRequest(BaseModel): + encrypted_payload: str + + @app.get("/", tags=["Root"]) def root(): return {"message": "Welcome to the HanchuESS Solar Backend API!"} @app.get("/get_access_token", tags=["HanchuESS"]) def get_access_token(): + """Get access token by logging into HanchuESS""" from service.hanchu_service import HanchuESSService hanchu_service = HanchuESSService() @@ -17,6 +24,61 @@ def get_access_token(): except Exception as e: return {"error": str(e)} +@app.post("/decrypt_payload", tags=["Payload"]) +def decrypt_payload(request: DecryptRequest): + """Decrypt an AES-encrypted HanchuESS payload""" + from service.hanchu_service import HanchuESSService + + try: + hanchu_service = HanchuESSService() + decrypted_data = hanchu_service.decrypt_payload(request.encrypted_payload) + + return { + "decrypted_data": decrypted_data, + "data_type": type(decrypted_data).__name__ + } + except Exception as e: + import traceback + return {"error": str(e), "traceback": traceback.format_exc()} + +@app.get("/get_power_chart", tags=["HanchuESS"]) +def get_power_chart(): + """Get 65-second power chart data from HanchuESS""" + from service.hanchu_service import HanchuESSService + + try: + hanchu_service = HanchuESSService() + + # Get power chart data (will automatically handle authentication) + power_data = hanchu_service.get_power_chart() + + return power_data + except Exception as e: + import traceback + return {"error": str(e), "traceback": traceback.format_exc()} + +@app.get("/get_power_minute_chart", tags=["HanchuESS"]) +def get_power_minute_chart(start_ts: int = None, end_ts: int = None): + """Get minute-by-minute power chart data from HanchuESS + + Args: + start_ts: Optional start timestamp in milliseconds + end_ts: Optional end timestamp in milliseconds + """ + from service.hanchu_service import HanchuESSService + + try: + hanchu_service = HanchuESSService() + + # Get minute chart data (will automatically handle authentication) + chart_data = hanchu_service.get_power_minute_chart(start_ts=start_ts, end_ts=end_ts) + + return chart_data + except Exception as e: + import traceback + return {"error": str(e), "traceback": traceback.format_exc()} + + if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8050) \ No newline at end of file diff --git a/backend/src/service/hanchu_service.py b/backend/src/service/hanchu_service.py index 9272dda..dca04eb 100644 --- a/backend/src/service/hanchu_service.py +++ b/backend/src/service/hanchu_service.py @@ -3,8 +3,9 @@ import json import os import requests -from Crypto.Cipher import AES -from Crypto.Util.Padding import pad +from Crypto.Cipher import AES, PKCS1_v1_5 +from Crypto.PublicKey import RSA +from Crypto.Util.Padding import pad, unpad from dotenv import load_dotenv @@ -12,6 +13,13 @@ load_dotenv() class HanchuESSService: + # RSA public key from the JavaScript code + RSA_PUBLIC_KEY = """-----BEGIN PUBLIC KEY----- +MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQCVg7RFDLMGM4O98d1zWKI5RQan +jci3iY4qlpgsH76fUn3GnZtqjbRk37lCQDv6AhgPNXRPpty81+g909/c4yzySKaP +CcDZv7KdCRB1mVxkq+0z4EtKx9EoTXKnFSDBaYi2srdal1tM3gGOsNTDN58CzYPX +nDGPX7+EHS1Mm4aVDQIDAQAB +-----END PUBLIC KEY-----""" def __init__(self): self.name = "HanchuESS Service" @@ -24,6 +32,10 @@ class HanchuESSService: self.verify_ssl = os.getenv("HANCHU_VERIFY_SSL", "true").lower() == "true" self.hanchu_username = os.getenv("HANCHU_USERNAME", "") self.hanchu_password = os.getenv("HANCHU_PASSWORD", "") + self.base_serial_number = os.getenv("BASE_SERIAL_NUMBER", "") + + # Cache for access token + self._access_token = None # Safety checks if len(self.aes_key) not in (16, 24, 32): @@ -42,28 +54,72 @@ class HanchuESSService: ciphertext = cipher.encrypt(pad(data.encode("utf-8"), AES.block_size)) return base64.b64encode(ciphertext).decode("utf-8") + def decrypt_payload(self, encrypted_data: str) -> dict | str: + """ + Decrypt base64-encoded AES-CBC payload and return the original data. + """ + ciphertext = base64.b64decode(encrypted_data) + cipher = AES.new(self.aes_key, AES.MODE_CBC, self.aes_iv) + decrypted = unpad(cipher.decrypt(ciphertext), AES.block_size) + decrypted_str = decrypted.decode("utf-8") + + # Try to parse as JSON, return string if it fails + try: + return json.loads(decrypted_str) + except json.JSONDecodeError: + return decrypted_str + + def encrypt_password_rsa(self, password: str) -> str: + """ + Encrypt password using RSA public key (matches JavaScript GO function). + Returns base64-encoded encrypted password. + """ + public_key = RSA.import_key(self.RSA_PUBLIC_KEY) + cipher = PKCS1_v1_5.new(public_key) + encrypted = cipher.encrypt(password.encode('utf-8')) + return base64.b64encode(encrypted).decode('utf-8') + def get_access_token(self) -> str: """ Authenticate with Hanchu ESS and return access token. + Uses double encryption: RSA for password, then AES for entire payload. + Caches the token to avoid unnecessary logins. """ + # Return cached token if available + if self._access_token: + return self._access_token + + # Step 1: RSA encrypt the password + encrypted_password = self.encrypt_password_rsa(self.hanchu_password) + # Step 2: Build payload with encrypted password payload = { "account": self.hanchu_username, - "password": self.hanchu_password, + "pwd": encrypted_password, "loginType": self.login_type, } + # Step 3: AES encrypt the entire payload encrypted_payload = self.encrypt_payload(payload) + # Step 4: Send to API with correct headers headers = { - "Content-Type": "application/json", - "Accept": "application/json", - "User-Agent": "Mozilla/5.0", + "Content-Type": "text/plain", + "Accept": "application/json, text/plain, */*", + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", + "appplat": "iess", + "locale": "en", + "timezone": "Africa/Accra", + "timeselected": "GMT", + "version": "1.0", + "crypto-version": "1.0.0", + "Origin": "https://iess3.hanchuess.com", + "Referer": "https://iess3.hanchuess.com/login", } response = requests.post( self.login_url, - json={"data": encrypted_payload}, + data=encrypted_payload, headers=headers, timeout=self.timeout, verify=self.verify_ssl, @@ -73,8 +129,138 @@ class HanchuESSService: result = response.json() try: - return result["data"]["accessToken"] + # The token is directly in the 'data' field as a JWT string + if result.get("success") and result.get("data"): + self._access_token = result["data"] + return self._access_token + else: + raise RuntimeError( + f"Hanchu login failed: {json.dumps(result, ensure_ascii=False)}" + ) except (KeyError, TypeError): raise RuntimeError( f"Hanchu login failed: {json.dumps(result, ensure_ascii=False)}" ) + + def get_power_chart(self, access_token: str = None) -> dict: + """ + Get 65-second power chart data from HanchuESS. + + Args: + access_token: Optional JWT token from login. If not provided, will get one automatically. + + Returns: + Power chart data from the API + """ + # Get access token if not provided + if not access_token: + # Check if we have a cached token first + if self._access_token: + access_token = self._access_token + else: + access_token = self.get_access_token() + + # Build payload with serial number + payload = {"sn": self.base_serial_number} + + # AES encrypt the payload + encrypted_payload = self.encrypt_payload(payload) + + # Send to API with access token + url = "https://iess3.hanchuess.com/gateway/platform/pcs/powerChart" + headers = { + "Content-Type": "text/plain", + "Accept": "application/json, text/plain, */*", + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", + "appplat": "iess", + "locale": "en", + "timezone": "Africa/Accra", + "timeselected": "GMT", + "version": "1.0", + "crypto-version": "1.0.0", + "access-token": access_token, + "Origin": "https://iess3.hanchuess.com", + "Referer": "https://iess3.hanchuess.com/", + } + + response = requests.post( + url, + data=encrypted_payload, + headers=headers, + timeout=self.timeout, + verify=self.verify_ssl, + ) + + response.raise_for_status() + return response.json() + + def get_power_minute_chart(self, access_token: str = None, start_ts: int = None, end_ts: int = None) -> dict: + """ + Get minute-by-minute power chart data from HanchuESS. + + Args: + access_token: Optional JWT token from login. If not provided, will get one automatically. + start_ts: Start timestamp in milliseconds. If not provided, defaults to start of today. + end_ts: End timestamp in milliseconds. If not provided, defaults to end of today. + + Returns: + Power minute chart data from the API + """ + # Get access token if not provided + if not access_token: + # Check if we have a cached token first + if self._access_token: + access_token = self._access_token + else: + access_token = self.get_access_token() + + # Set default timestamps if not provided (today's data) + if not start_ts or not end_ts: + from datetime import datetime, timezone, timedelta + now = datetime.now(timezone.utc) + # 30 minutes ago + start_time = now - timedelta(minutes=30) + start_ts = int(start_time.timestamp() * 1000) + # Current time + end_ts = int(now.timestamp() * 1000) + + # Build payload + payload = { + "sn": self.base_serial_number, + "devType": "2", + "maxCount": 1440, # 24 hours * 60 minutes + "dataTimeTsEnd": end_ts, + "dataTimeTsStart": start_ts, + "masterSum": True + } + + # AES encrypt the payload + encrypted_payload = self.encrypt_payload(payload) + + # Send to API with access token + url = "https://iess3.hanchuess.com/gateway/platform/pcs/powerMinuteChart" + headers = { + "Content-Type": "text/plain", + "Accept": "application/json, text/plain, */*", + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", + "appplat": "iess", + "locale": "en", + "timezone": "Africa/Accra", + "timeselected": "GMT", + "version": "1.0", + "crypto-version": "1.0.0", + "access-token": access_token, + "Origin": "https://iess3.hanchuess.com", + "Referer": "https://iess3.hanchuess.com/", + } + + response = requests.post( + url, + data=encrypted_payload, + headers=headers, + timeout=self.timeout, + verify=self.verify_ssl, + ) + + response.raise_for_status() + return response.json()