ivooskamp/backupchecks
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix Cove importer: correct API payload format and response parsing

Browse Source 
- Login: use lowercase username/password params and id="jsonrpc"
- Login: visa is at top-level of response (not inside result)
- EnumerateAccountStatistics: use lowercase query param, RecordsCount
  instead of RecordCount, remove DisplayColumns (not needed)
- _flatten_settings: Settings items are single-key dicts like
  {"D09F00": "5"}, not {Key: ..., Value: ...} - use dict.update()
- _cove_enumerate: unwrap nested result and handle Accounts key
- _process_account: AccountId is top-level field, not from Settings

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Ivo Oskamp 2026-02-23 10:21:01 +01:00
parent 467f350184
commit 3bd8178464
1 changed files with 53 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

85
containers/backupchecks/src/backend/app/cove_importer.py
View File

@ -76,15 +76,20 @@ def _cove_login(url: str, username: str, password: str) -> tuple[str, int]:
"""
payload = {
"jsonrpc": "2.0",
"id": "jsonrpc",
"method": "Login",
"id": 1,
"params": {
"Username": username,
"Password": password,
"username": username,
"password": password,
},
}
try:
resp = requests.post(url, json=payload, timeout=30)
resp = requests.post(
url,
json=payload,
headers={"Content-Type": "application/json"},
timeout=30,
)
resp.raise_for_status()
data = resp.json()
except requests.RequestException as exc:
@ -92,21 +97,22 @@ def _cove_login(url: str, username: str, password: str) -> tuple[str, int]:
except ValueError as exc:
raise CoveImportError(f"Cove login response is not valid JSON: {exc}") from exc
result = data.get("result") or {}
if not result:
error = data.get("error") or {}
raise CoveImportError(f"Cove login failed: {error.get('message', 'unknown error')}")
if "error" in data and data["error"]:
error = data["error"]
msg = error.get("message") or str(error) if isinstance(error, dict) else str(error)
raise CoveImportError(f"Cove login failed: {msg}")
visa = result.get("Visa") or ""
# Visa is returned at the top level of the response (not inside result)
visa = data.get("visa") or ""
if not visa:
raise CoveImportError("Cove login succeeded but no Visa token returned")
raise CoveImportError("Cove login succeeded but no visa token returned")
# Partner/account info may be nested
account_info = result.get("Accounts", [{}])[0] if result.get("Accounts") else {}
# PartnerId is inside result
result = data.get("result") or {}
partner_id = (
account_info.get("PartnerId")
result.get("PartnerId")
or result.get("PartnerID")
or result.get("PartnerId")
or result.get("result", {}).get("PartnerId")
or 0
)
@ -126,21 +132,25 @@ def _cove_enumerate(
"""
payload = {
"jsonrpc": "2.0",
"method": "EnumerateAccountStatistics",
"id": 2,
"visa": visa,
"id": "jsonrpc",
"method": "EnumerateAccountStatistics",
"params": {
"Query": {
"query": {
"PartnerId": partner_id,
"DisplayColumns": COVE_COLUMNS,
"StartRecordNumber": start,
"RecordCount": count,
"RecordsCount": count,
"Columns": COVE_COLUMNS,
}
},
}
try:
resp = requests.post(url, json=payload, timeout=60)
resp = requests.post(
url,
json=payload,
headers={"Content-Type": "application/json"},
timeout=60,
)
resp.raise_for_status()
data = resp.json()
except requests.RequestException as exc:
@ -148,18 +158,32 @@ def _cove_enumerate(
except ValueError as exc:
raise CoveImportError(f"Cove EnumerateAccountStatistics response is not valid JSON: {exc}") from exc
result = data.get("result") or {}
if not result:
error = data.get("error") or {}
raise CoveImportError(f"Cove EnumerateAccountStatistics failed: {error.get('message', 'unknown error')}")
if "error" in data and data["error"]:
error = data["error"]
msg = error.get("message") or str(error) if isinstance(error, dict) else str(error)
raise CoveImportError(f"Cove EnumerateAccountStatistics failed: {msg}")
return result.get("result", []) or []
result = data.get("result")
if result is None:
return []
# Unwrap possible nested result (same as test script)
if isinstance(result, dict) and "result" in result:
result = result["result"]
# Accounts can be a list directly or wrapped in an "Accounts" key
if isinstance(result, list):
return result
if isinstance(result, dict):
return result.get("Accounts", []) or []
return []
def _flatten_settings(account: dict) -> dict:
"""Convert the Settings array in an account dict to a flat key→value dict.
Cove returns settings as a list of {Key, Value} objects.
Cove returns settings as a list of single-key dicts, e.g.:
[{"D09F00": "5"}, {"I1": "device name"}, ...]
This flattens them so we can do `flat['D09F00']` instead of iterating.
"""
flat: dict[str, Any] = {}
@ -167,10 +191,7 @@ def _flatten_settings(account: dict) -> dict:
if isinstance(settings_list, list):
for item in settings_list:
if isinstance(item, dict):
key = item.get("Key") or item.get("key")
value = item.get("Value") if "Value" in item else item.get("value")
if key is not None:
flat[str(key)] = value
flat.update(item)
return flat
@ -274,8 +295,8 @@ def _process_account(account: dict, settings) -> bool:
flat = _flatten_settings(account)
# AccountId – try both the top-level field and Settings
account_id = account.get("AccountId") or account.get("AccountID") or flat.get("I18")
# AccountId is a top-level field (not in Settings)
account_id = account.get("AccountId") or account.get("AccountID")
if not account_id:
return False
try: