#!/usr/bin/env python3 """ SolarPower / EyBond endpoint exerciser. Logs in once, then calls EVERY method on SolarPowerClient exactly once and writes each response as a JSON file under ./api_responses/. Built for one-off recon: do not commit api_responses/ or paste it anywhere public. """ import os import sys import json import time import traceback from datetime import datetime from pathlib import Path # Reuse the saved client verbatim sys.path.insert(0, "/root/solar") from solarpower_client import SolarPowerClient, BASE_URL, API_SERVERS EMAIL = os.environ.get("SOLAR_EMAIL", "alphanon") PASSWORD = os.environ.get("SOLAR_PASSWORD") if not PASSWORD: raise SystemExit("SOLAR_PASSWORD env var is required; refusing to default.") OUT_DIR = Path("/root/solar/api_responses") OUT_DIR.mkdir(parents=True, exist_ok=True) # Suppress the client's own print spam during the run. import builtins _real_print = builtins.print def _quiet_print(*a, **kw): # still print errors/warnings, just drop the URL-echo lines msg = " ".join(str(x) for x in a) if "-->" in msg or "Banner" in msg or msg.strip().startswith("╔"): return _real_print(*a, **kw) builtins.print = _quiet_print def _safe_filename(name: str) -> str: return name.replace("/", "_").replace("?", "_").replace("&", "_").replace("=", "_") def save(name: str, payload): """Write one response file. payload may be None on connection failure.""" out = OUT_DIR / f"{_safe_filename(name)}.json" if payload is None: out.write_text(json.dumps({"_error": "no response (None)"}, indent=2)) else: out.write_text(json.dumps(payload, indent=2, ensure_ascii=False, default=str)) return out def main(): started = datetime.now().isoformat(timespec="seconds") print(f"[runner] started {started}") print(f"[runner] output dir: {OUT_DIR}") print(f"[runner] base URL: {BASE_URL}") client = SolarPowerClient(BASE_URL) # Prefer a previously-cached token so we don't burn a fresh login (the # Shinemonitor gateway rate-limits logins aggressively — err=16 # ERR_PASSWORD_VERIF_FAIL is often signature replay, not bad creds). token_cache = OUT_DIR / "_tokens.json" reused = False if token_cache.exists(): try: cached = json.loads(token_cache.read_text()) ut = cached.get("utoken") zt = cached.get("ztoken") if ut and zt: client.utoken = ut client.ztoken = zt reused = True print(f"[runner] reused cached tokens (utoken={ut[:8]}…)") except Exception: pass if not reused: print("[runner] logging in as", EMAIL) ok = client.login_by_email(EMAIL, PASSWORD) save("00_login", {"success": ok, "utoken": client.utoken, "ztoken": client.ztoken}) if not ok or not client.utoken: print("[runner] FATAL: login did not return a utoken; aborting") sys.exit(2) # Cache for next time token_cache.write_text(json.dumps({"utoken": client.utoken, "ztoken": client.ztoken})) print(f"[runner] login OK, utoken={client.utoken[:8]}…") else: # Save a marker file so we know tokens were reused, not fresh. save("00_login", {"success": True, "utoken": client.utoken, "ztoken": client.ztoken, "reused": True}) # Static calls (no parameters needed beyond auth). static_calls = [ ("queryPlantCount", lambda: client.get_plant_count()), ("webQueryPlants", lambda: client.get_plants(page=0, pagesize=50)), ("queryPlantsEnergyDay", lambda: client.get_plants_energy_day()), ("queryPlantsActiveOuputPowerCurrent", lambda: client.get_plants_active_power()), ("queryDeviceCount", lambda: client.get_device_count()), ("webQueryDeviceEs", lambda: client.web_query_devices(page=0, pagesize=50)), ("webQueryCollectorsEs", lambda: client.web_query_collectors(page=0, pagesize=50)), ("webQueryDeviceStatusViewEs", lambda: client.web_query_device_status()), ("queryAccountInfo", lambda: client.get_account_info()), ] results = [] def run(name, fn): t0 = time.time() try: data = fn() dt = time.time() - t0 save(name, data) ok = data is not None size = len(json.dumps(data, default=str)) if data else 0 results.append((name, "ok" if ok else "fail", dt, size)) print(f" ✓ {name:<45s} {dt:5.2f}s {size:>6d} bytes") except Exception as e: dt = time.time() - t0 save(name, {"_exception": str(e), "_trace": traceback.format_exc()}) results.append((name, "exception", dt, 0)) print(f" ✗ {name:<45s} {dt:5.2f}s EXC {e}") print("\n[runner] === static calls ===") for name, fn in static_calls: run(name, fn) # Need a real plant_id for the plant-scoped calls. Pull from webQueryPlants. # Real response shape: {"dat": {"plant": [{...}], "total": N, "page": N, ...}} plants_resp = client.get_plants(page=0, pagesize=50) plant_id = None first_plant = None if isinstance(plants_resp, dict): inner = plants_resp.get("dat", plants_resp.get("data", plants_resp)) candidates = None if isinstance(inner, list): candidates = inner elif isinstance(inner, dict): # Singular "plant" is the actual key; keep fallbacks for safety. for key in ("plant", "plants", "rows", "list", "items"): v = inner.get(key) if isinstance(v, list) and v: candidates = v break if candidates: first_plant = candidates[0] plant_id = (first_plant.get("pid") or first_plant.get("plantid") or first_plant.get("plantId") or first_plant.get("id")) print(f"\n[runner] first plant_id resolved to: {plant_id!r}") # Plant-scoped calls if plant_id is not None: print("\n[runner] === plant-scoped calls ===") for name, fn in [ ("queryPlantInfo", lambda: client.get_plant_info(plant_id)), ("queryDevices", lambda: client.get_devices(plant_id, page=0, pagesize=50)), ("queryCollectors", lambda: client.get_collectors(plant_id, page=0, pagesize=50)), ("queryPlantsActiveOuputPowerOneDay", lambda: client.get_plants_active_power_one_day( datetime.now().strftime("%Y-%m-%d"))), ]: run(name, fn) # Collector-scoped: need a pn. Try webQueryCollectorsEs first (may be empty for # plant-scoped accounts), then queryCollectors(plant_id). pn = None for source_name in ("webQueryCollectorsEs", "queryCollectors"): src_path = OUT_DIR / f"{_safe_filename(source_name)}.json" if not src_path.exists(): continue try: src = json.loads(src_path.read_text()) except Exception: continue if not isinstance(src, dict): continue # Skip if this call itself errored out. if src.get("err", 0) not in (0, "0"): continue inner = src.get("dat", src.get("data", src)) items = None if isinstance(inner, list): items = inner elif isinstance(inner, dict): # Real key is singular "collector"; keep fallbacks. for key in ("collector", "collectors", "rows", "list", "items"): v = inner.get(key) if isinstance(v, list) and v: items = v break if items: pn = items[0].get("pn") or items[0].get("collectorPn") if pn: print(f"[runner] first pn resolved from {source_name}: {pn!r}") break # Fallback: pull pn from the device record (we know it has a pn field). if not pn: dev_path = OUT_DIR / "webQueryDeviceEs.json" if dev_path.exists(): try: src = json.loads(dev_path.read_text()) inner = src.get("dat", src.get("data", src)) items = (inner.get("device") if isinstance(inner, dict) else None) or [] if items: pn = items[0].get("pn") if pn: print(f"[runner] pn recovered from webQueryDeviceEs[0]: {pn!r}") except Exception: pass if pn: print("\n[runner] === collector-scoped calls ===") for name, fn in [ ("queryCollectorDevices", lambda: client.get_collector_devices(pn)), ("queryCollectorDevicesStatus", lambda: client.get_collector_devices_status(pn)), ]: run(name, fn) # For device-scoped calls, need (pn, devcode, devaddr, sn). Pull from # queryCollectorDevices or webQueryDeviceEs. sn = devcode = devaddr = None for src_name in ("queryCollectorDevices", "webQueryDeviceEs"): src_path = OUT_DIR / f"{_safe_filename(src_name)}.json" if not src_path.exists(): continue try: src = json.loads(src_path.read_text()) except Exception: continue if not isinstance(src, dict): continue if src.get("err", 0) not in (0, "0"): continue inner = src.get("dat", src.get("data", src)) items = None if isinstance(inner, list): items = inner elif isinstance(inner, dict): # Real key is singular "device"; keep fallbacks. for key in ("device", "devices", "rows", "list", "items"): v = inner.get(key) if isinstance(v, list) and v: items = v break if items: d = items[0] sn = d.get("sn") or d.get("deviceSn") devcode= d.get("devcode")or d.get("deviceCode") devaddr= d.get("devaddr")or d.get("deviceAddr") if sn and devcode is not None and devaddr is not None: print(f"[runner] first device resolved from {src_name}: " f"sn={sn!r} devcode={devcode!r} devaddr={devaddr!r}") break if sn is not None: print("\n[runner] === device-scoped calls ===") today = datetime.now().strftime("%Y-%m-%d") for name, fn in [ ("querySPDeviceLastData", lambda: client.get_device_last_data(pn, devcode, devaddr, sn)), ("webQueryDeviceEnergyFlowEs", lambda: client.get_device_energy_flow(pn, sn, devaddr, devcode)), ("webQueryDeviceCtrlField", lambda: client.get_device_ctrl_fields(pn, devcode, devaddr, sn)), ("queryDeviceDataOneDay", lambda: client.get_device_data_one_day(pn, devcode, sn, devaddr, today)), ]: run(name, fn) # Summary print("\n[runner] === summary ===") for name, status, dt, size in results: marker = "✓" if status == "ok" else ("✗" if status == "exception" else "·") print(f" {marker} {name:<45s} {status:<10s} {dt:5.2f}s {size:>6d}B") summary = { "started_at": started, "finished_at": datetime.now().isoformat(timespec="seconds"), "base_url": BASE_URL, "calls": [{"name": n, "status": s, "duration_s": round(d, 3), "bytes": b} for n, s, d, b in results], } (OUT_DIR / "_summary.json").write_text(json.dumps(summary, indent=2)) print(f"\n[runner] wrote {OUT_DIR/'_summary.json'}") if __name__ == "__main__": main()