update scalable_sync.py to enhance activity recording by validating drafts and committing them to Wealthfolio

scalable_sync.py

@@ -329,9 +329,9 @@ class WealthfolioMCP:
             return False
 
     def record_activities(self, activities):
-        payload = [
+        activity_payloads = [
             {
-                "accountId": activity.account_id,
+                "account": activity.account_id,
                 "activityDate": activity.date,
                 "symbol": activity.symbol,
                 "activityType": activity.activity_type,
@@ -341,18 +341,73 @@ class WealthfolioMCP:
                 "fee": activity.fee,
                 "tax": activity.tax,
                 "currency": activity.currency,
-                "source": activity.source,
+                "notes": f"scalable:{activity.external_id}",
-                "externalId": activity.external_id,
             }
             for activity in activities
         ]
-        result = asyncio.run(
+
-            self._connect_and_call("record_activities", {"activities": payload})
+        draft_result = asyncio.run(
+            self._connect_and_call(
+                "record_activities",
+                {"activities": activity_payloads},
+            )
         )
-        if getattr(result, "is_error", False):
+
-            raise RuntimeError(f"record_activities error: {result.content}")
+        if getattr(draft_result, "is_error", False):
-        log.info("record_activities OK: %s activities", len(activities))
+            raise RuntimeError(
-        return result
+                f"record_activities error: {draft_result.content}"
+            )
+
+        draft_data = draft_result.structured_content or {}
+        drafts = draft_data.get("drafts", [])
+
+        if len(drafts) != len(activities):
+            raise RuntimeError(
+                "Wealthfolio returned an incomplete draft batch: "
+                f"expected {len(activities)}, got {len(drafts)}"
+            )
+
+        invalid = [
+            draft for draft in drafts
+            if not draft.get("accountId")
+               or not draft.get("activityType")
+               or not draft.get("activityDate")
+        ]
+        if invalid:
+            raise RuntimeError(
+                "Wealthfolio returned invalid activity drafts: "
+                + json.dumps(invalid, default=str)
+            )
+
+        commit_result = asyncio.run(
+            self._connect_and_call(
+                "commit_activity_drafts",
+                {"drafts": drafts},
+            )
+        )
+
+        if getattr(commit_result, "is_error", False):
+            raise RuntimeError(
+                f"commit_activity_drafts error: {commit_result.content}"
+            )
+
+        commit_data = commit_result.structured_content or {}
+        if commit_data.get("errors"):
+            raise RuntimeError(
+                "Wealthfolio failed to persist activities: "
+                + json.dumps(commit_data["errors"], default=str)
+            )
+
+        created = commit_data.get("created", [])
+        if len(created) != len(activities):
+            raise RuntimeError(
+                "Wealthfolio commit was incomplete: "
+                f"expected {len(activities)}, got {len(created)}; "
+                + json.dumps(commit_data, default=str)
+            )
+
+        log.info("Wealthfolio committed %s activities", len(created))
+        return created
 
 
 # ---------------------------------------------------------------------------