My-Weekly-ToDo-List/scripts/test-cloudkit-reminders.py

#!/usr/bin/env python3
"""
Test script to probe the CloudKit API for modern iOS 13+ Reminders.
Uses pyicloud's authenticated session to make direct CloudKit requests.

Usage:
  python3 scripts/test-cloudkit-reminders.py <email> <password> [--session-dir <dir>]
"""

import sys
import json
import os
from urllib.parse import urlencode

SESSION_DIR = os.environ.get(
    'ICLOUD_SESSION_DIR',
    os.path.join(os.path.dirname(__file__), '..', 'data', 'icloud-sessions')
)


def main():
    if len(sys.argv) < 3:
        print("Usage: python3 test-cloudkit-reminders.py <email> <password>")
        sys.exit(1)

    email = sys.argv[1]
    password = sys.argv[2]

    from pyicloud import PyiCloudService

    os.makedirs(SESSION_DIR, exist_ok=True)

    print("Authenticating with iCloud...")
    api = PyiCloudService(
        apple_id=email,
        password=password,
        cookie_directory=SESSION_DIR,
    )

    if api.requires_2fa or api.requires_2sa:
        print("ERROR: 2FA required. Please authenticate first.")
        sys.exit(1)

    print("Authenticated successfully!\n")

    # Get the ckdatabasews URL
    ck_url = api.data["webservices"]["ckdatabasews"]["url"]
    print(f"CloudKit URL: {ck_url}")

    params = dict(api.params)
    container = "com.apple.reminders"
    base = f"{ck_url}/database/1/{container}/production/private"

    print(f"Reminders base: {base}\n")

    # Step 1: List zones
    print("=" * 60)
    print("STEP 1: Listing zones")
    print("=" * 60)

    zones_url = f"{base}/zones/list?{urlencode(params)}"
    resp = api.session.post(zones_url, data="{}", headers={"Content-type": "text/plain"})
    print(f"Status: {resp.status_code}")
    zones_data = resp.json()

    if "zones" in zones_data:
        zones = zones_data["zones"]
        print(f"Found {len(zones)} zone(s):")
        for z in zones:
            zone_id = z.get("zoneID", {})
            print(f"  - {zone_id.get('zoneName', '?')} (owner: {zone_id.get('ownerRecordName', '?')[:30]}...)")
    else:
        print(f"Response: {json.dumps(zones_data, indent=2)[:1000]}")
        print("\nNo zones found. Exiting.")
        sys.exit(1)

    # Step 2: For each zone, discover record types via zone changes
    print("\n" + "=" * 60)
    print("STEP 2: Discovering record types via zone changes")
    print("=" * 60)

    for z in zones:
        zone_id = z.get("zoneID", {})
        zone_name = zone_id.get("zoneName", "unknown")
        print(f"\n--- Zone: {zone_name} ---")

        changes_url = f"{base}/changes/zone?{urlencode(params)}"
        changes_body = json.dumps({"zoneID": zone_id})
        resp = api.session.post(changes_url, data=changes_body, headers={"Content-type": "text/plain"})
        print(f"Status: {resp.status_code}")
        data = resp.json()

        if "records" in data:
            records = data["records"]
            record_types = {}
            for r in records:
                rt = r.get("recordType", "unknown")
                if rt not in record_types:
                    record_types[rt] = {"count": 0, "sample": r}
                record_types[rt]["count"] += 1

            print(f"Total records: {len(records)}")
            print(f"Record types: {list(record_types.keys())}")

            for rt, info in record_types.items():
                print(f"\n  === {rt} ({info['count']} records) ===")
                sample = info["sample"]
                fields = sample.get("fields", {})
                print(f"  Fields: {list(fields.keys())}")

                # Show field values for the first record
                for fname, fval in fields.items():
                    val = fval.get("value") if isinstance(fval, dict) else fval
                    val_str = str(val)[:80]
                    ftype = fval.get("type", "?") if isinstance(fval, dict) else "?"
                    print(f"    {fname} ({ftype}): {val_str}")

                # Print titles of all records of this type (up to 20)
                if "title" in fields or "name" in fields:
                    print(f"\n  All {rt} titles:")
                    count = 0
                    for r in records:
                        if r.get("recordType") == rt:
                            f = r.get("fields", {})
                            title = None
                            for tfield in ["title", "name", "Title", "Name"]:
                                if tfield in f:
                                    tv = f[tfield]
                                    title = tv.get("value") if isinstance(tv, dict) else tv
                                    break
                            if title:
                                print(f"    - {title}")
                            count += 1
                            if count >= 40:
                                print(f"    ... (showing first 40 of {info['count']})")
                                break
        else:
            print(f"Response: {json.dumps(data, indent=2)[:1000]}")

        # Also check moreComing
        if data.get("moreComing"):
            print(f"\n  NOTE: moreComing=true, there are more records.")
            print(f"  syncToken: {data.get('syncToken', '')[:50]}...")

    # Step 3: Try direct queries for common record types
    print("\n" + "=" * 60)
    print("STEP 3: Trying direct queries for common record types")
    print("=" * 60)

    for z in zones:
        zone_id = z.get("zoneID", {})
        zone_name = zone_id.get("zoneName", "unknown")
        print(f"\n--- Zone: {zone_name} ---")

        for rt in ["Checklist", "List", "Collection", "ReminderList", "REMCDList",
                    "Task", "Reminder", "Item", "REMCDReminder"]:
            query_url = f"{base}/records/query?{urlencode(params)}"
            query_body = json.dumps({
                "query": {"recordType": rt},
                "zoneID": zone_id,
                "resultsLimit": 5,
            })
            try:
                resp = api.session.post(query_url, data=query_body, headers={"Content-type": "text/plain"})
                result = resp.json()
                records = result.get("records", [])
                error = result.get("serverErrorCode", "")

                if records:
                    print(f"  {rt}: {len(records)} records (may be more)")
                    # Show first record fields
                    r = records[0]
                    fields = r.get("fields", {})
                    print(f"    Fields: {list(fields.keys())}")
                elif error and error != "NOT_FOUND":
                    print(f"  {rt}: error={error}")
                # Skip NOT_FOUND silently
            except Exception as e:
                print(f"  {rt}: Exception: {e}")

    print("\n" + "=" * 60)
    print("DONE")
    print("=" * 60)


if __name__ == "__main__":
    main()