claudetools/temp/bardach_purge_notes.py

#!/usr/bin/env python3
"""Purge junk notes from 223 Bardach contacts in Microsoft 365."""

import json
import subprocess
import sys
import time

TENANT_ID = "dd4a82e8-85a3-44ac-8800-07945ab4d95f"
CLIENT_ID = "fabb3421-8b34-484b-bc17-e46de9703418"
CLIENT_SECRET = "~QJ8Q~NyQSs4OcGqHZyPrA2CVnq9KBfKiimntbMO"
SCOPE = "https://graph.microsoft.com/.default"
USER = "barbara@bardach.net"
TOKEN_URL = f"https://login.microsoftonline.com/{TENANT_ID}/oauth2/v2.0/token"

def get_token():
    """Acquire OAuth2 token using client credentials."""
    result = subprocess.run(
        ["curl", "-s", "-X", "POST", TOKEN_URL,
         "-H", "Content-Type: application/x-www-form-urlencoded",
         "-d", f"client_id={CLIENT_ID}&scope={SCOPE}&client_secret={CLIENT_SECRET}&grant_type=client_credentials"],
        capture_output=True, text=True
    )
    data = json.loads(result.stdout)
    if "access_token" not in data:
        print(f"[ERROR] Token acquisition failed: {data}")
        sys.exit(1)
    return data["access_token"]

def patch_contact(token, contact_id, display_name):
    """PATCH a contact to clear personalNotes."""
    url = f"https://graph.microsoft.com/v1.0/users/{USER}/contacts/{contact_id}"
    result = subprocess.run(
        ["curl", "-s", "-o", "/dev/null", "-w", "%{http_code}",
         "-X", "PATCH", url,
         "-H", f"Authorization: Bearer {token}",
         "-H", "Content-Type: application/json",
         "-d", '{"personalNotes": ""}'],
        capture_output=True, text=True
    )
    status_code = result.stdout.strip()
    return status_code

def main():
    # Load analysis file
    with open("D:/ClaudeTools/temp/bardach_notes_analysis.json", "r", encoding="utf-8") as f:
        data = json.load(f)

    # Collect all contact IDs to purge
    contacts_to_purge = []

    # iCloud junk notes (217)
    for c in data["A_junk_boilerplate"]["icloud_warnings"]:
        contacts_to_purge.append((c["id"], c["displayName"], "icloud_junk"))

    # Empty/whitespace notes (6)
    for c in data["A_junk_boilerplate"]["empty_whitespace"]:
        contacts_to_purge.append((c["id"], c["displayName"], "empty"))

    total = len(contacts_to_purge)
    print(f"[INFO] Total contacts to purge: {total}")
    print(f"  - iCloud junk: {len(data['A_junk_boilerplate']['icloud_warnings'])}")
    print(f"  - Empty/whitespace: {len(data['A_junk_boilerplate']['empty_whitespace'])}")
    print()

    token = get_token()
    print("[OK] Token acquired")

    successes = 0
    failures = 0
    failed_contacts = []

    for i, (cid, name, category) in enumerate(contacts_to_purge, 1):
        # Re-acquire token every 200 operations
        if i > 1 and (i - 1) % 200 == 0:
            print(f"[INFO] Re-acquiring token at operation {i}...")
            token = get_token()
            print("[OK] Token re-acquired")

        status = patch_contact(token, cid, name)
        if status == "200":
            successes += 1
        else:
            failures += 1
            failed_contacts.append({"name": name, "id": cid, "status": status, "category": category})

        # Progress every 50
        if i % 50 == 0 or i == total:
            print(f"[INFO] Progress: {i}/{total} | Successes: {successes} | Failures: {failures}")

        # Small delay to avoid throttling
        if i % 4 == 0:
            time.sleep(0.1)

    print()
    print("=" * 60)
    print(f"[DONE] Notes purge complete")
    print(f"  Total processed: {total}")
    print(f"  Successes: {successes}")
    print(f"  Failures: {failures}")

    if failed_contacts:
        print()
        print("[WARNING] Failed contacts:")
        for fc in failed_contacts:
            print(f"  - {fc['name']} (status={fc['status']}, category={fc['category']})")

if __name__ == "__main__":
    main()