NRL-1798 Add delete pointers script

Author: sandyforresternhs

scripts/delete_pointers_by_id.py

@@ -0,0 +1,317 @@
+#!/usr/bin/env python
+import json
+from datetime import datetime, timedelta, timezone
+from typing import Any, Dict, List
+
+import boto3
+import fire
+
+dynamodb = boto3.client("dynamodb")
+
+
+def _load_pointers_from_file(pointers_file: str) -> list[str]:
+    """
+    Read pointers from a file. Supports:
+     - JSON array of objects with an "id" field
+     - line-delimited plain text (one id per line)
+
+    Returns a list of pointer id strings. Prints a warning for skipped malformed JSON entries.
+    """
+    with open(pointers_file, "r") as fh:
+        content = fh.read().strip()
+
+    if not content:
+        return []
+
+    # JSON path
+    if content.startswith("[") or content.startswith("{"):
+        try:
+            data = json.loads(content)
+        except json.JSONDecodeError as e:
+            raise ValueError(f"Failed to parse JSON file {pointers_file}: {e}") from e
+
+        if not isinstance(data, list):
+            raise ValueError("JSON file must contain an array of objects")
+
+        parsed_ids: list[str] = []
+        skipped_count = 0
+        for item in data:
+            if (
+                isinstance(item, dict)
+                and "id" in item
+                and isinstance(item["id"], str)
+                and item["id"].strip()
+            ):
+                parsed_ids.append(item["id"].strip())
+            else:
+                skipped_count += 1
+
+        if skipped_count:
+            print(
+                f"Warning: skipped {skipped_count} malformed entries in JSON file {pointers_file}"
+            )
+
+        return parsed_ids
+
+    # Plain text fallback
+    return [line.strip() for line in content.splitlines() if line.strip()]
+
+
+def _check_pointers_match_ods_code(
+    ods_code: str, pointer_ids: List[str]
+) -> tuple[List[str], List[str]]:
+    """
+    Validate that pointer IDs are in line with the provided ODS code.
+    Returns (matched_ids, mismatched_ids)
+    """
+    matched = []
+    mismatched = []
+
+    for pointer_id in pointer_ids:
+        if pointer_id.startswith(f"{ods_code}-"):
+            matched.append(pointer_id)
+        else:
+            mismatched.append(pointer_id)
+
+    return matched, mismatched
+
+
+def _batch_get_existing_pointers(
+    table_name: str, pointer_ids: List[str]
+) -> tuple[List[str], List[str]]:
+    """
+    Check which pointers exist using BatchGetItem (max 100 items per request).
+    Returns (existing_ids, not_found_ids)
+    """
+    existing = []
+    not_found = []
+
+    for batch_idx in range(0, len(pointer_ids), 100):
+        batch_ids = pointer_ids[batch_idx : batch_idx + 100]
+
+        keys = [
+            {
+                "pk": {"S": f"D#{pointer_id}"},
+                "sk": {"S": f"D#{pointer_id}"},
+            }
+            for pointer_id in batch_ids
+        ]
+
+        response = dynamodb.batch_get_item(RequestItems={table_name: {"Keys": keys}})
+
+        found_ids = {
+            item["pk"]["S"][2:]
+            for item in response.get("Responses", {}).get(table_name, [])
+        }
+
+        for pointer_id in batch_ids:
+            if pointer_id in found_ids:
+                existing.append(pointer_id)
+            else:
+                not_found.append(pointer_id)
+
+    return existing, not_found
+
+
+def _batch_delete_pointers(
+    table_name: str, pointer_ids: List[str]
+) -> tuple[List[str], List[str]]:
+    """
+    Delete pointers using BatchWriteItem (max 25 items per request).
+    Returns (deleted_ids, failed_ids)
+    """
+    pointers_deleted = []
+    failed_deletes_set: set[str] = set()
+
+    for _batch_id in range(0, len(pointer_ids), 25):
+        batch_ptrs = pointer_ids[_batch_id : _batch_id + 25]
+
+        batch = [
+            {
+                "DeleteRequest": {
+                    "Key": {
+                        "pk": {"S": f"D#{pointer_id}"},
+                        "sk": {"S": f"D#{pointer_id}"},
+                    }
+                }
+            }
+            for pointer_id in batch_ptrs
+        ]
+
+        result = dynamodb.batch_write_item(RequestItems={table_name: batch})
+        unprocessed = result.get("UnprocessedItems", {}).get(table_name, [])
+
+        # Collect unprocessed IDs
+        for item in unprocessed:
+            pk_val = item["DeleteRequest"]["Key"]["pk"]["S"]
+            failed_deletes_set.add(pk_val[2:])  # Remove "D#"
+
+        # Only count successfully deleted items (batch size minus unprocessed)
+        successfully_deleted = [p for p in batch_ptrs if p not in failed_deletes_set]
+        pointers_deleted.extend(successfully_deleted)
+
+        if len(pointers_deleted) % 1000 == 0 and len(pointers_deleted) > 0:
+            print(".", end="", flush=True)
+
+    return pointers_deleted, sorted(failed_deletes_set)
+
+
+def _delete_pointers_by_id(
+    table_name: str,
+    ods_code: str,
+    pointers_to_delete: List[str] | None = None,
+    pointers_file: str | None = None,
+) -> Dict[str, Any]:
+    """
+    Delete pointers from DynamoDB table.
+
+    Can accept pointers as:
+    - list of strings: pointers_to_delete=["id1", "id2"]
+    - JSON file: pointers_file=/path/to/pointers.json (array of objects with "id" field)
+    - text file: pointers_file=/path/to/ids.txt (one id per line)
+
+    Parameters:
+    - table_name: DynamoDB table name
+    - ods_code: ODS code of the organisation that the pointers belong to
+    - pointers_to_delete: list of pointer ids to delete
+    - pointers_file: path to JSON file (array of objects with "id" field) or text file (one id per line)
+    """
+    if pointers_to_delete is None and pointers_file is None:
+        raise ValueError("Provide either pointers_to_delete or pointers_file")
+
+    if pointers_to_delete is not None and pointers_file is not None:
+        raise ValueError("Provide either pointers_to_delete or pointers_file, not both")
+
+    # Load pointers from file if provided
+    if pointers_file:
+        pointers_to_delete = _load_pointers_from_file(pointers_file)
+
+    if len(pointers_to_delete) == 0:
+        return {
+            "pointers_to_delete": 0,
+            "ods_code": ods_code,
+            "ods_code_matched": [],
+            "ods_code_mismatched": [],
+            "pointer_not_found": [],
+            "deleted_pointers": [],
+            "failed_deletes": [],
+            "deletes-took-secs": 0,
+        }
+
+    start_time = datetime.now(tz=timezone.utc)
+
+    print(
+        f"Validating {len(pointers_to_delete)} pointers against ODS code {ods_code}..."
+    )
+    matched_pointers, mismatched_pointers = _check_pointers_match_ods_code(
+        ods_code, pointers_to_delete
+    )
+
+    print(
+        f"Validate pointer's ODS code: {len(matched_pointers)} matched, {len(mismatched_pointers)} mismatched"
+    )
+
+    if len(matched_pointers) == 0:
+        print(f"None of the pointer IDs are a match for ODS code {ods_code}. Exiting.")
+        end_time = datetime.now(tz=timezone.utc)
+        return {
+            "pointers_to_delete": len(pointers_to_delete),
+            "ods_code": ods_code,
+            "ods_code_matched": {
+                "count": len(matched_pointers),
+                "ids": matched_pointers,
+            },
+            "ods_code_mismatched": {
+                "count": len(mismatched_pointers),
+                "ids": mismatched_pointers,
+            },
+            "pointer_not_found": {
+                "count": 0,
+                "ids": [],
+            },
+            "deleted_pointers": {
+                "count": 0,
+                "ids": [],
+            },
+            "failed_deletes": {
+                "count": 0,
+                "ids": [],
+            },
+            "deletes-took-secs": timedelta.total_seconds(end_time - start_time),
+        }
+
+    print(f"Checking existence of {len(matched_pointers)} pointers in {table_name}...")
+    existing_pointers, not_found_pointers = _batch_get_existing_pointers(
+        table_name, matched_pointers
+    )
+
+    print(
+        f"Found {len(existing_pointers)} existing pointers to delete, {len(not_found_pointers)} not found."
+    )
+
+    if len(existing_pointers) == 0:
+        print("No pointers found to delete. Exiting.")
+        end_time = datetime.now(tz=timezone.utc)
+        return {
+            "pointers_to_delete": len(pointers_to_delete),
+            "ods_code": ods_code,
+            "ods_code_matched": {
+                "count": len(matched_pointers),
+                "ids": matched_pointers,
+            },
+            "ods_code_mismatched": {
+                "count": len(mismatched_pointers),
+                "ids": mismatched_pointers,
+            },
+            "pointer_not_found": {
+                "count": len(not_found_pointers),
+                "ids": not_found_pointers,
+            },
+            "deleted_pointers": {
+                "count": 0,
+                "ids": [],
+            },
+            "failed_deletes": {
+                "count": 0,
+                "ids": [],
+            },
+            "deletes-took-secs": timedelta.total_seconds(end_time - start_time),
+        }
+
+    # Proceed with deletion using BatchWriteItem
+    pointers_deleted, failed_deletes = _batch_delete_pointers(
+        table_name, existing_pointers
+    )
+
+    end_time = datetime.now(tz=timezone.utc)
+
+    print(" Done")
+    return {
+        "pointers_to_delete": len(pointers_to_delete),
+        "ods_code": ods_code,
+        "ods_code_matched": {
+            "count": len(matched_pointers),
+            "ids": matched_pointers,
+        },
+        "ods_code_mismatched": {
+            "count": len(mismatched_pointers),
+            "ids": mismatched_pointers,
+        },
+        "pointer_not_found": {
+            "count": len(not_found_pointers),
+            "ids": not_found_pointers,
+        },
+        "deleted_pointers": {
+            "count": len(pointers_deleted),
+            "ids": pointers_deleted,
+        },
+        "failed_deletes": {
+            "count": len(failed_deletes),
+            "ids": failed_deletes,
+        },
+        "deletes-took-secs": timedelta.total_seconds(end_time - start_time),
+    }
+
+
+if __name__ == "__main__":
+    fire.Fire(_delete_pointers_by_id)