#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright (C) 2026 Christian Wirtz <doc@snowheaven.de>
# This file is part of the Anker Solix Checkmk extension.
# License: GPL-2.0-only, see LICENSE in the repository root.
"""Checkmk special agent for Anker Solix systems (Solarbank 2 E1600 Pro and compatible models).

Logs into the Anker Power/Solix cloud API (the same undocumented API used by the Anker app),
fetches the site list, "scene info" (home screen data), power cutoff (min/max state-of-
charge limits) and bound-device info (firmware version, WiFi signal) for every accessible
site/device, and emits the relevant Solarbank fields as Checkmk agent sections.

Protocol notes
--------------
Anker does not publish this API. The login handshake (ECDH key exchange on the P-256 curve,
AES-256-CBC password encryption) and the endpoint paths used below were derived from the
community-maintained, MIT-licensed reference implementation at
https://github.com/thomluther/anker-solix-api (not vendored here; this script only
re-implements the small, read-only subset needed for monitoring). As with any unofficial
API, Anker can change or break it at any time - and has already been observed to return
inconsistent shapes for the same endpoint (see extract_bind_device_info()), so parsing
throughout this file is deliberately defensive rather than trusting a fixed schema.

This agent is intentionally READ-ONLY: it never calls any of the control/settings endpoints
(no schedule changes, no power limits, no firmware updates) - it only *reads* the firmware
version that is already reported alongside other device info.

Session caching
----------------
Anker's login token is valid for about 7 days. Logging in fresh on every single agent run
(i.e. every check cycle, typically every 1-2 minutes) looks like many rapid logins from a
"new device" to Anker's servers and can get the account rate-limited/locked out. To avoid
that, the auth token is cached to disk (per email+country, under $OMD_ROOT/var/check_mk/cache
when running inside a Checkmk site) and reused until shortly before it expires; a fresh
login only happens when there is no usable cached session, or the cached one is rejected by
the API (e.g. because it was invalidated server-side).
"""

from __future__ import annotations

import argparse
import base64
import hashlib
import json
import logging
import os
import sys
import tempfile
import time
from collections.abc import Iterable
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any
from urllib import error as urlerror
from urllib import request as urlrequest

try:
    from cryptography.hazmat.backends import default_backend
    from cryptography.hazmat.primitives import padding, serialization
    from cryptography.hazmat.primitives.asymmetric import ec
    from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
except ImportError as exc:  # pragma: no cover - guarded at runtime, not by tests
    sys.stderr.write(
        "agent_anker_solix requires the 'cryptography' Python package, which is missing "
        f"from this Checkmk site's Python environment: {exc}\n"
    )
    sys.exit(2)

LOGGER = logging.getLogger("agent_anker_solix")

# --------------------------------------------------------------------------------------
# Protocol constants
# --------------------------------------------------------------------------------------

# Uncompressed EC public key (0x04 + 32 byte X + 32 byte Y) used by both Anker Api server
# clusters for the ECDH key exchange during login. This is a fixed, published protocol
# constant of the Anker cloud API, not a secret.
ANKER_SERVER_PUBLIC_KEY_HEX = (
    "04c5c00c4f8d1197cc7c3167c52bf7acb054d722f0ef08dcd7e0883236e0d72a3868d975"
    "0cb47fa4619248f3d83f0f662671dadc6e2d31c2f41db0161651c7c076"
)

API_SERVERS = {
    "eu": "https://ankerpower-api-eu.anker.com",
    "com": "https://ankerpower-api.anker.com",
}

# Country -> server cluster assignment as used by the Anker app. Countries missing from
# this table fall back to the EU cluster (see resolve_api_base()).
API_COUNTRIES = {
    "com": [
        "DZ", "LB", "SY", "EG", "LY", "TN", "MA", "JO", "PS", "AR", "AU", "BR",
        "HK", "IN", "MX", "NG", "NZ", "RU", "SG", "ZA", "KR", "TW", "US", "CA",
    ],
    "eu": [
        "DE", "BE", "EL", "LT", "PT", "BG", "ES", "LU", "CZ", "FR", "HU", "SI",
        "DK", "HR", "MT", "SK", "IT", "NL", "FI", "EE", "CY", "AT", "SE", "IE",
        "LV", "PL", "UK", "IS", "NO", "LI", "CH", "BA", "ME", "MD", "MK", "GE",
        "AL", "RS", "TR", "UA", "XK", "AM", "BY", "AZ", "IL", "RO", "JP",
    ],
}

API_LOGIN = "passport/login"
API_SITE_LIST = "power_service/v1/site/get_site_list"
API_SCENE_INFO = "power_service/v1/site/get_scen_info"
API_POWER_CUTOFF = "power_service/v1/app/compatible/get_power_cutoff"
API_BIND_DEVICES = "power_service/v1/app/get_relate_and_bind_devices"

API_HEADERS = {
    "content-type": "application/json",
    "model-type": "DESKTOP",
    "app-name": "anker_power",
    "os-type": "android",
}

# device-level "status" (cloud connectivity)
DEVICE_STATUS_DESC = {"0": "offline", "1": "online"}

# device-level "charging_status" (Solarbank operating mode)
CHARGING_STATUS_DESC = {
    "0": "detection",
    "1": "bypass",
    "2": "discharge",
    "3": "charge",
    "4": "wakeup",
    "5": "fully_charged",
    "6": "full_bypass",
    "7": "standby",
    "116": "cold_wakeup",
}

DEFAULT_TIMEOUT = 20
REQUEST_RETRIES = 2

# Re-login proactively if the cached token has less than this much validity left, so a
# long-running batch of requests never races an expiry mid-way through.
TOKEN_REFRESH_MARGIN_SECONDS = 3600


class AnkerSolixApiError(Exception):
    """Raised for any error response from the Anker Solix cloud API."""


class AnkerSolixAuthError(AnkerSolixApiError):
    """Raised specifically when the API rejects the current session (HTTP 401/403).

    Distinguished from AnkerSolixApiError so callers can tell "this session token is no
    longer valid, try logging in again" apart from other errors (wrong credentials,
    account locked, network issues, ...) that must NOT be retried automatically -
    retrying those would just send more failed requests at an already-struggling account.
    """


# --------------------------------------------------------------------------------------
# Login crypto
# --------------------------------------------------------------------------------------


def _generate_keypair() -> tuple[ec.EllipticCurvePrivateKey, str]:
    """Create an ephemeral P-256 keypair and return (private_key, raw public key hex)."""
    private_key = ec.generate_private_key(ec.SECP256R1(), default_backend())
    public_hex = private_key.public_key().public_bytes(
        encoding=serialization.Encoding.X962,
        format=serialization.PublicFormat.UncompressedPoint,
    ).hex()
    return private_key, public_hex


def _derive_shared_key(private_key: ec.EllipticCurvePrivateKey) -> bytes:
    """Derive the 32 byte ECDH shared secret with Anker's fixed server public key."""
    server_public_key = ec.EllipticCurvePublicKey.from_encoded_point(
        ec.SECP256R1(), bytes.fromhex(ANKER_SERVER_PUBLIC_KEY_HEX)
    )
    return private_key.exchange(ec.ECDH(), server_public_key)


def _encrypt_password(password: str, shared_key: bytes) -> str:
    """AES-256-CBC encrypt the password with the ECDH shared key (key = secret, IV = secret[:16])."""
    cipher = Cipher(
        algorithms.AES(shared_key), modes.CBC(shared_key[:16]), backend=default_backend()
    )
    encryptor = cipher.encryptor()
    padder = padding.PKCS7(128).padder()
    padded = padder.update(password.encode("utf-8")) + padder.finalize()
    return base64.b64encode(encryptor.update(padded) + encryptor.finalize()).decode("ascii")


def resolve_api_base(country: str, base_override: str | None = None) -> str:
    """Return the Anker API base URL for the given ISO country code."""
    if base_override:
        return base_override.rstrip("/")
    country = (country or "").upper()
    for region, countries in API_COUNTRIES.items():
        if country in countries:
            return API_SERVERS[region]
    LOGGER.warning("Unknown country %r, defaulting to EU Anker API server", country)
    return API_SERVERS["eu"]


# --------------------------------------------------------------------------------------
# HTTP client
# --------------------------------------------------------------------------------------


class AnkerSolixClient:
    """Minimal, read-only client for the Anker Power/Solix cloud API."""

    def __init__(
        self,
        email: str,
        password: str,
        country: str,
        timeout: int = DEFAULT_TIMEOUT,
        base_override: str | None = None,
        cache_dir: Path | None = None,
    ) -> None:
        self.email = email
        self.password = password
        self.country = (country or "").upper()
        self.timeout = timeout
        self.api_base = resolve_api_base(self.country, base_override)
        self.token: str | None = None
        self.gtoken: str | None = None
        self._session_from_cache = False
        self._cache_dir = cache_dir or self._default_cache_dir()

    @staticmethod
    def _default_cache_dir() -> Path:
        # $OMD_ROOT/var/check_mk/cache is a persistent (non-tmpfs) per-site directory that
        # already exists on every Checkmk site; fall back to a temp dir for standalone use
        # (e.g. running this script outside a site, or in tests).
        omd_root = os.environ.get("OMD_ROOT")
        if omd_root:
            return Path(omd_root) / "var" / "check_mk" / "cache" / "anker_solix"
        return Path(tempfile.gettempdir()) / "anker_solix"

    def _cache_file(self) -> Path:
        digest = hashlib.sha256(f"{self.email.lower()}:{self.country}".encode("utf-8")).hexdigest()
        return self._cache_dir / f"{digest}.json"

    def _load_cached_session(self) -> bool:
        """Try to reuse a still-valid cached session. Returns True if one was loaded."""
        try:
            raw = self._cache_file().read_text(encoding="utf-8")
            data = json.loads(raw)
            token = data["token"]
            gtoken = data["gtoken"]
            expires_at = float(data["expires_at"])
        except (OSError, json.JSONDecodeError, KeyError, TypeError, ValueError):
            return False
        if not token or not gtoken:
            return False
        if expires_at - time.time() < TOKEN_REFRESH_MARGIN_SECONDS:
            LOGGER.debug("Cached Anker session for %s has expired or is expiring soon", self.email)
            return False
        self.token = token
        self.gtoken = gtoken
        self._session_from_cache = True
        LOGGER.debug(
            "Reusing cached Anker session for %s (valid until %s)",
            self.email,
            datetime.fromtimestamp(expires_at).astimezone().isoformat(),
        )
        return True

    def _save_cached_session(self, expires_at: float) -> None:
        path = self._cache_file()
        try:
            path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
            tmp_path = path.with_suffix(".tmp")
            tmp_path.write_text(
                json.dumps({"token": self.token, "gtoken": self.gtoken, "expires_at": expires_at}),
                encoding="utf-8",
            )
            os.chmod(tmp_path, 0o600)
            tmp_path.replace(path)
        except OSError as exc:
            # Caching is an optimization, not a correctness requirement: if it fails (e.g.
            # read-only filesystem) we just fall back to logging in fresh every run.
            LOGGER.warning("Could not persist the Anker session cache at %s: %s", path, exc)

    def _clear_cached_session(self) -> None:
        try:
            self._cache_file().unlink(missing_ok=True)
        except OSError:
            pass

    def _post(self, endpoint: str, payload: dict | None, headers: dict) -> dict:
        url = f"{self.api_base}/{endpoint}"
        body = json.dumps(payload or {}).encode("utf-8")
        req = urlrequest.Request(url, data=body, headers=headers, method="POST")
        last_err: Exception | None = None
        http_status: int | None = None
        for attempt in range(1, REQUEST_RETRIES + 2):
            try:
                with urlrequest.urlopen(req, timeout=self.timeout) as resp:
                    raw = resp.read()
                break
            except urlerror.HTTPError as exc:
                raw = exc.read()
                # Anker returns structured JSON error bodies even for 4xx/5xx, so fall through
                # to the shared decode/error handling below instead of raising immediately.
                last_err = exc
                http_status = exc.code
                if exc.code not in (429, 500, 502, 503, 504) or attempt > REQUEST_RETRIES:
                    break
                time.sleep(1.5 * attempt)
                continue
            except urlerror.URLError as exc:
                last_err = exc
                if attempt > REQUEST_RETRIES:
                    raise AnkerSolixApiError(
                        f"Could not reach Anker API at {url}: {exc}"
                    ) from exc
                time.sleep(1.5 * attempt)
                continue
        else:
            raise AnkerSolixApiError(f"Could not reach Anker API at {url}: {last_err}")

        try:
            data = json.loads(raw.decode("utf-8"))
        except (json.JSONDecodeError, UnicodeDecodeError) as exc:
            if http_status in (401, 403):
                raise AnkerSolixAuthError(
                    f"Anker API rejected the current session for {endpoint} "
                    f"(HTTP {http_status})"
                ) from exc
            raise AnkerSolixApiError(
                f"Anker API returned a non-JSON response for {endpoint}: {exc}"
            ) from exc

        code = data.get("code")
        if http_status in (401, 403):
            raise AnkerSolixAuthError(
                f"Anker API rejected the current session for {endpoint} "
                f"(HTTP {http_status}, code {code}): {data.get('msg', 'unknown error')}"
            )
        if code not in (0, None):
            raise AnkerSolixApiError(
                f"Anker API error {code} for {endpoint}: {data.get('msg', 'unknown error')}"
            )
        return data

    def login(self, force: bool = False) -> None:
        """Authenticate and store the auth token / gtoken used for subsequent requests.

        Reuses a cached session (see module docstring) unless `force` is set or no valid
        cached session is available, in which case a fresh login is performed against the
        API and the result is cached again for next time.
        """
        self._session_from_cache = False
        if not force and self._load_cached_session():
            return

        private_key, public_hex = _generate_keypair()
        shared_key = _derive_shared_key(private_key)
        now = datetime.now().astimezone()
        tz_offset_ms = round((now.utcoffset() or timedelta(0)).total_seconds() * 1000)
        payload = {
            "ab": self.country,
            "client_secret_info": {"public_key": public_hex},
            "enc": 0,
            "email": self.email,
            "password": _encrypt_password(self.password, shared_key),
            "time_zone": tz_offset_ms,
            "transaction": str(round(time.time() * 1000)),
        }
        try:
            resp = self._post(API_LOGIN, payload, dict(API_HEADERS))
        except AnkerSolixApiError as exc:
            raise AnkerSolixApiError(f"Login failed for {self.email}: {exc}") from exc

        data = resp.get("data") or {}
        self.token = data.get("auth_token")
        user_id = data.get("user_id")
        if not self.token or not user_id:
            raise AnkerSolixApiError(
                f"Login for {self.email} did not return a usable auth token "
                "(check email/password/country)"
            )
        self.gtoken = hashlib.md5(user_id.encode("utf-8")).hexdigest()
        expires_at = data.get("token_expires_at")
        if expires_at:
            self._save_cached_session(float(expires_at))
        else:
            LOGGER.debug(
                "Login response for %s had no token_expires_at, not caching the session",
                self.email,
            )
        LOGGER.info("Logged in to Anker Solix as %s", data.get("nick_name") or self.email)

    def _auth_headers(self) -> dict:
        if not self.token or not self.gtoken:
            raise AnkerSolixApiError("Not logged in")
        headers = dict(API_HEADERS)
        headers["country"] = self.country
        headers["gtoken"] = self.gtoken
        headers["x-auth-token"] = self.token
        return headers

    def _request_authenticated(self, endpoint: str, payload: dict | None) -> dict:
        """POST to an authenticated endpoint, transparently logging in first if needed.

        If the (cached) session is rejected by the API, this logs in fresh exactly once
        and retries - but only when the rejected session came from the cache; a session
        that was JUST freshly obtained being immediately rejected means something else is
        wrong (e.g. account locked), and retrying that would only make it worse.
        """
        if self.token is None:
            self.login()
        try:
            return self._post(endpoint, payload, self._auth_headers())
        except AnkerSolixAuthError:
            if not self._session_from_cache:
                raise
            LOGGER.info("Cached Anker session was rejected by the API, logging in again")
            self._clear_cached_session()
            self.login(force=True)
            return self._post(endpoint, payload, self._auth_headers())

    def get_site_list(self) -> list[dict]:
        resp = self._request_authenticated(API_SITE_LIST, {})
        return (resp.get("data") or {}).get("site_list") or []

    def get_scene_info(self, site_id: str) -> dict:
        resp = self._request_authenticated(API_SCENE_INFO, {"site_id": site_id})
        return resp.get("data") or {}

    def get_power_cutoff(self, device_sn: str, site_id: str) -> dict:
        resp = self._request_authenticated(
            API_POWER_CUTOFF, {"site_id": site_id, "device_sn": device_sn}
        )
        return resp.get("data") or {}

    def get_bind_devices(self) -> list[dict]:
        """Account-wide list of bound devices (firmware version, WiFi signal, ...)."""
        resp = self._request_authenticated(API_BIND_DEVICES, {})
        return resp.get("data") or []


# --------------------------------------------------------------------------------------
# Data extraction (pure functions, unit-testable without any network access)
# --------------------------------------------------------------------------------------


def _to_number(value: Any) -> float | None:
    """Best-effort conversion of Anker's string-encoded numeric fields to float."""
    if value is None or value == "":
        return None
    try:
        return float(value)
    except (TypeError, ValueError):
        return None


def _extract_battery_power(device: dict, charging_status_desc: str) -> tuple[float | None, float | None]:
    """Return (charge_power_w, discharge_power_w) for a Solarbank device.

    Newer firmware reports "bat_charge_power"/"bat_discharge_power" directly - unambiguous,
    one of them is simply 0 depending on current direction. Older firmware/API responses
    (and this is a real bug this extension had: confirmed against a live Solarbank 2 Pro
    that was discharging) only have a single, direction-less "charging_power" field, whose
    meaning is inferred here from charging_status as a best-effort fallback; if that status
    is itself ambiguous (bypass, standby, detection, ...), neither value is reported rather
    than guessing wrong.
    """
    if "bat_charge_power" in device or "bat_discharge_power" in device:
        return (
            _to_number(device.get("bat_charge_power")) or 0.0,
            _to_number(device.get("bat_discharge_power")) or 0.0,
        )
    legacy = _to_number(device.get("charging_power"))
    if legacy is None:
        return None, None
    if charging_status_desc == "charge":
        return legacy, 0.0
    if charging_status_desc == "discharge":
        return 0.0, legacy
    return None, None


def extract_solarbank_rows(scene_info: dict, site_id: str, site_name: str) -> list[dict]:
    """Build one flat dict per Solarbank device out of a raw get_scen_info() response."""
    rows: list[dict] = []
    solarbank_info = scene_info.get("solarbank_info") or {}
    device_list = solarbank_info.get("solarbank_list") or []
    for device in device_list:
        status_code = str(device.get("status", ""))
        charging_status_code = str(device.get("charging_status", ""))
        charging_status_desc = CHARGING_STATUS_DESC.get(charging_status_code, "unknown")
        charge_power, discharge_power = _extract_battery_power(device, charging_status_desc)
        row = {
            "site_id": site_id,
            "site_name": site_name,
            "device_sn": device.get("device_sn", ""),
            "device_pn": device.get("device_pn", ""),
            "device_name": device.get("device_name") or device.get("device_sn", ""),
            # NOTE: Anker's API reuses the "battery_power" field name to mean the
            # battery state of charge in percent (0-100) for Solarbank 2/3 generations,
            # not a power value in Watt. Confirmed against reference fixture data.
            "battery_soc_percent": _to_number(device.get("battery_power")),
            "battery_charge_power_w": charge_power,
            "battery_discharge_power_w": discharge_power,
            "solar_input_power_w": _to_number(device.get("photovoltaic_power")),
            "output_power_w": _to_number(device.get("output_power")),
            "status_code": status_code,
            "status_desc": DEVICE_STATUS_DESC.get(status_code, "unknown"),
            "charging_status_code": charging_status_code,
            "charging_status_desc": charging_status_desc,
            "sub_package_num": device.get("sub_package_num"),
        }
        if len(device_list) == 1:
            # Per-PV-string input power, AC-coupled power and battery heating power are
            # physically properties of a specific Solarbank unit (its own MC4/AC inputs,
            # its own battery), but Anker's API only ever reports them once, at the site's
            # "solarbank_info" level, not per device in solarbank_list - there is no way to
            # tell which bank they belong to when a site has more than one. With exactly
            # one Solarbank at this site, though, there is no ambiguity: attribute them to
            # it (see extract_site_row(), which does the opposite for multi-bank sites).
            row["solar_power_1_w"] = _to_number(solarbank_info.get("solar_power_1"))
            row["solar_power_2_w"] = _to_number(solarbank_info.get("solar_power_2"))
            row["solar_power_3_w"] = _to_number(solarbank_info.get("solar_power_3"))
            row["solar_power_4_w"] = _to_number(solarbank_info.get("solar_power_4"))
            row["ac_power_w"] = _to_number(solarbank_info.get("ac_power"))
            row["heating_power_w"] = _to_number(solarbank_info.get("pei_heating_power"))
        rows.append(row)
    return rows


def extract_power_cutoff(cutoff_data: dict) -> dict:
    """Pull the effective minimum/maximum state-of-charge limits out of a
    get_power_cutoff() response.

    Newer firmware reports "discharge_lower_limit"/"charge_upper_limit" directly (this is
    what the Anker app itself shows as the battery's discharge/charge limit). Older
    firmware only exposes a list of selectable presets ("power_cutoff_data"); in that case
    the preset with "is_selected" set is the active minimum SoC, and there is no separate
    configurable maximum.
    """
    soc_min = _to_number(cutoff_data.get("discharge_lower_limit"))
    if soc_min is None:
        for preset in cutoff_data.get("power_cutoff_data") or []:
            if not isinstance(preset, dict):
                continue
            if int(preset.get("is_selected", 0) or 0) > 0:
                soc_min = _to_number(preset.get("output_cutoff_data"))
                break
    return {
        "soc_min_percent": soc_min,
        "soc_max_percent": _to_number(cutoff_data.get("charge_upper_limit")),
    }


def extract_bind_device_info(devices: list) -> dict[str, dict]:
    """Turn a get_relate_and_bind_devices() response into a dict keyed by device serial.

    Anker's own account-wide "bound devices" list has been observed to mix in entries
    that are not device objects at all (e.g. plain strings) - this endpoint is not used
    for anything essential, so any entry that doesn't look like a device dict is simply
    skipped rather than raising and losing the whole agent run over it.
    """
    result: dict[str, dict] = {}
    for device in devices:
        if not isinstance(device, dict):
            continue
        device_sn = device.get("device_sn")
        if not device_sn:
            continue
        result[device_sn] = {
            "firmware_version": device.get("device_sw_version") or None,
            "wifi_signal_dbm": _to_number(device.get("rssi")),
            "wifi_name": device.get("wifi_name") or None,
            "wifi_online": device.get("wifi_online"),
        }
    return result


def extract_site_row(scene_info: dict, site_id: str, site_name: str) -> dict | None:
    """Build one flat dict describing the site-wide Solarbank system, if any bank is present."""
    solarbank_info = scene_info.get("solarbank_info") or {}
    if not solarbank_info.get("solarbank_list"):
        return None
    grid_info = scene_info.get("grid_info") or {}
    retain_load_raw = str(scene_info.get("retain_load", "") or "")
    total_soc = _to_number(solarbank_info.get("total_battery_power"))
    statistics_by_type = {
        str(entry.get("type")): entry for entry in scene_info.get("statistics") or []
    }
    device_list = solarbank_info.get("solarbank_list") or []
    row = {
        "site_id": site_id,
        "site_name": site_name,
        "solarbank_count": len(device_list),
        "total_battery_soc_percent": total_soc * 100 if total_soc is not None else None,
        "total_solar_input_power_w": _to_number(solarbank_info.get("total_photovoltaic_power")),
        "total_charge_power_w": _to_number(solarbank_info.get("total_charging_power")),
        "battery_discharge_power_w": _to_number(solarbank_info.get("battery_discharge_power")),
        "to_home_load_power_w": _to_number(solarbank_info.get("to_home_load")),
        "home_load_power_w": _to_number(scene_info.get("home_load_power")),
        "retain_load_w": _to_number(retain_load_raw.rstrip("Ww")),
        "grid_to_home_power_w": _to_number(grid_info.get("grid_to_home_power")),
        "solar_to_grid_power_w": _to_number(grid_info.get("photovoltaic_to_grid_power")),
        "has_smart_meter": bool(grid_info.get("grid_list")),
        # cumulative, lifetime counters as reported by the Anker app's site overview
        "lifetime_output_kwh": _to_number((statistics_by_type.get("1") or {}).get("total")),
        "lifetime_co2_saved_kg": _to_number((statistics_by_type.get("2") or {}).get("total")),
        "lifetime_savings": _to_number((statistics_by_type.get("3") or {}).get("total")),
        "lifetime_savings_unit": (statistics_by_type.get("3") or {}).get("unit") or "",
    }
    # Per-PV-string power, AC-coupled power and battery heating power are NEVER put on
    # the site row - they belong to a specific Solarbank:
    #
    # - With exactly one Solarbank at this site, that's unambiguous, so
    #   extract_solarbank_rows() puts them on that device's row instead.
    # - With zero or more than one Solarbank, they are dropped entirely rather than
    #   guessed at: Anker's API only reports "solar_power_1".."_4" once per site, and
    #   there is no confirmed real-world data for what that means for a multi-Solarbank
    #   site - e.g. whether it's a meaningful sum ("all of string position 1 across every
    #   bank", which would be physically meaningless, since different banks' same-numbered
    #   strings have no relation to each other) or something else entirely.
    #
    # "total_solar_input_power_w" above, by contrast, is explicitly named as a total and
    # has been confirmed (against reference data) to equal the sum of the four strings for
    # a single-bank site, so it stays reliable regardless of how many Solarbanks are at
    # the site - it's the right metric to look at here for any site with more than one.
    return row


# --------------------------------------------------------------------------------------
# Agent output
# --------------------------------------------------------------------------------------


def build_agent_output(
    sites: Iterable[dict],
    scene_info_by_site: dict[str, dict],
    extra_by_device: dict[str, dict] | None = None,
) -> str:
    """Render the Checkmk agent section output for the given sites.

    extra_by_device holds additional per-device fields (power cutoff limits, firmware
    version, WiFi signal, ...) fetched from endpoints beyond get_scen_info, keyed by
    device serial number, merged into each Solarbank device's row.
    """
    extra_by_device = extra_by_device or {}
    solarbank_lines: list[str] = []
    site_lines: list[str] = []

    for site in sites:
        site_id = site.get("site_id", "")
        site_name = site.get("site_name") or site_id
        scene_info = scene_info_by_site.get(site_id)
        if scene_info is None:
            continue
        for row in extract_solarbank_rows(scene_info, site_id, site_name):
            row.update(extra_by_device.get(row["device_sn"], {}))
            solarbank_lines.append(json.dumps(row, sort_keys=True))
        site_row = extract_site_row(scene_info, site_id, site_name)
        if site_row is not None:
            site_lines.append(json.dumps(site_row, sort_keys=True))

    out = ["<<<anker_solix_solarbank:sep(0)>>>"]
    out.extend(solarbank_lines)
    out.append("<<<anker_solix_site:sep(0)>>>")
    out.extend(site_lines)
    return "\n".join(out) + "\n"


# --------------------------------------------------------------------------------------
# CLI
# --------------------------------------------------------------------------------------


def parse_arguments(argv: list[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--email", required=True, help="Anker account email address")
    password_group = parser.add_mutually_exclusive_group(required=True)
    password_group.add_argument(
        "--password",
        help="Anker account password, given directly in plain text. Only meant for manual "
        "CLI use (e.g. testing) - this puts the password in the process's command line, "
        "which Checkmk itself never does; it always uses --password-id instead.",
    )
    password_group.add_argument(
        "--password-id",
        dest="password_id",
        help="Anker account password as a Checkmk password store reference (\"<id>:<path>\"), "
        "resolved via cmk.password_store.v1_unstable.dereference_secret(). This is what the "
        "special agent's server_side_calls plugin actually passes; not meant to be set by "
        "hand.",
    )
    parser.add_argument(
        "--country",
        required=True,
        help="Two letter ISO country code of the Anker account (e.g. DE)",
    )
    parser.add_argument(
        "--site-id",
        action="append",
        dest="site_ids",
        default=None,
        help="Restrict monitoring to this site ID; may be given multiple times. "
        "Default: monitor all sites the account can see.",
    )
    parser.add_argument(
        "--timeout",
        type=int,
        default=DEFAULT_TIMEOUT,
        help="HTTP request timeout in seconds (default: %(default)s)",
    )
    parser.add_argument(
        "--debug", action="store_true", help="Enable debug logging to stderr"
    )
    return parser.parse_args(argv)


def resolve_password(args: argparse.Namespace) -> str:
    """Get the real Anker account password out of parsed CLI arguments.

    --password carries it directly (CLI/debug use). --password-id carries a Checkmk
    password store reference ("<id>:<path>") instead - this is what Checkmk itself always
    uses, so that the real password never ends up on the process's command line (visible
    via e.g. `ps`). Resolving it requires cmk.password_store, which is only importable when
    running inside a Checkmk >= 2.5.0 site's Python - which every real invocation, via the
    special agent mechanism, always is.
    """
    if args.password is not None:
        return args.password

    try:
        from cmk.password_store.v1_unstable import dereference_secret
    except ImportError as exc:
        raise AnkerSolixApiError(
            "Got --password-id, but the 'cmk.password_store' module (Checkmk >= 2.5.0) is "
            f"not available in this Python environment: {exc}"
        ) from exc
    try:
        return dereference_secret(args.password_id).reveal()
    except Exception as exc:  # noqa: BLE001 - covers PasswordStoreError and friends
        raise AnkerSolixApiError(f"Could not resolve --password-id: {exc}") from exc


def main(argv: list[str] | None = None) -> int:
    args = parse_arguments(argv if argv is not None else sys.argv[1:])
    logging.basicConfig(
        level=logging.DEBUG if args.debug else logging.WARNING,
        format="%(levelname)s %(name)s: %(message)s",
        stream=sys.stderr,
    )

    # Test-only escape hatches, e.g. to point at a local mock server and an isolated cache
    # directory. Not exposed as CLI flags so they can never end up in a WATO rule.
    base_override = os.environ.get("ANKER_SOLIX_API_BASE_OVERRIDE")
    cache_dir_override = os.environ.get("ANKER_SOLIX_CACHE_DIR_OVERRIDE")

    try:
        password = resolve_password(args)
    except AnkerSolixApiError as exc:
        sys.stderr.write(f"agent_anker_solix: {exc}\n")
        return 1

    client = AnkerSolixClient(
        email=args.email,
        password=password,
        country=args.country,
        timeout=args.timeout,
        base_override=base_override,
        cache_dir=Path(cache_dir_override) if cache_dir_override else None,
    )

    try:
        client.login()
        sites = client.get_site_list()
        if args.site_ids:
            wanted = set(args.site_ids)
            sites = [s for s in sites if s.get("site_id") in wanted]
        if not sites:
            LOGGER.warning("Anker account %s has no accessible sites", args.email)

        # From here on, every step fetches supplementary data for one site/device at a
        # time. Anker's API is undocumented and has been observed to return unexpected
        # shapes (e.g. a plain string where a device object was expected) - each step is
        # therefore deliberately isolated with a broad `except Exception`, not just
        # AnkerSolixApiError: a bug or surprise in any one optional piece of data must
        # degrade to "that piece is missing", never crash the whole run and lose
        # everything that WAS fetched successfully (this happened for real: a bad entry
        # in get_bind_devices() used to take down the entire agent).
        scene_info_by_site: dict[str, dict] = {}
        for site in sites:
            site_id = site.get("site_id")
            if not site_id:
                continue
            try:
                scene_info_by_site[site_id] = client.get_scene_info(site_id)
            except Exception as exc:  # noqa: BLE001 - see comment above
                LOGGER.error("Skipping site %s: %s", site.get("site_name", site_id), exc)

        bind_device_info: dict[str, dict] = {}
        try:
            bind_device_info = extract_bind_device_info(client.get_bind_devices())
        except Exception as exc:  # noqa: BLE001 - see comment above
            LOGGER.warning("Could not fetch bound-device info (firmware/WiFi signal): %s", exc)

        extra_by_device: dict[str, dict] = {}
        for site_id, scene_info in scene_info_by_site.items():
            for device in (scene_info.get("solarbank_info") or {}).get("solarbank_list") or []:
                if not isinstance(device, dict):
                    continue
                device_sn = device.get("device_sn")
                if not device_sn:
                    continue
                extra = dict(bind_device_info.get(device_sn, {}))
                try:
                    extra.update(extract_power_cutoff(client.get_power_cutoff(device_sn, site_id)))
                except Exception as exc:  # noqa: BLE001 - see comment above
                    LOGGER.warning(
                        "Could not fetch power cutoff settings for %s: %s", device_sn, exc
                    )
                extra_by_device[device_sn] = extra

        sys.stdout.write(build_agent_output(sites, scene_info_by_site, extra_by_device))
        return 0
    except AnkerSolixApiError as exc:
        sys.stderr.write(f"agent_anker_solix: {exc}\n")
        return 1
    except Exception as exc:  # noqa: BLE001 - last-resort safety net
        # Login and the core site-list fetch above are not individually guarded (unlike
        # the optional supplementary steps further down), since a failure there genuinely
        # means there is no useful data for this run. Still: report it as a clean,
        # non-zero agent failure with a full traceback for debugging, rather than an
        # unhandled crash - this exists as a defense-in-depth backstop for bugs neither of
        # us has found yet, not as a substitute for fixing the specific ones we do find.
        LOGGER.exception("agent_anker_solix: unexpected error: %s", exc)
        return 1


if __name__ == "__main__":
    sys.exit(main())
