Added duplicate field check and Google Sheet protections check. Creat…

check_sheet_protections.py

@@ -0,0 +1,90 @@
+#!/usr/bin/python3.12
+"""
+Standalone diagnostic: list protected ranges/sheets on a Google Sheet,
+and check whether a given service account email is allowed to bypass them.
+
+Usage:
+    python3 check_sheet_protections.py \
+        --spreadsheet-id 1_nnBcCt4MPaAzaXqvAVD2pMv8AX9wPj8BBfgI-05Ri8 \
+        --creds /path/to/configuration.json
+"""
+
+import argparse
+import json
+from google.oauth2 import service_account
+from googleapiclient.discovery import build
+
+
+def connect(credentials_file):
+    scopes = ['https://www.googleapis.com/auth/spreadsheets.readonly']
+    creds = service_account.Credentials.from_service_account_file(
+        credentials_file, scopes=scopes
+    )
+    return build('sheets', 'v4', credentials=creds)
+
+
+def get_service_account_email(credentials_file):
+    with open(credentials_file) as f:
+        data = json.load(f)
+    return data.get('client_email', 'UNKNOWN')
+
+
+def describe_range(rng):
+    if not rng:
+        return "ENTIRE SHEET (no bounds set)"
+    parts = []
+    if 'startRowIndex' in rng or 'endRowIndex' in rng:
+        parts.append(f"rows {rng.get('startRowIndex', '0')}-{rng.get('endRowIndex', 'end')}")
+    if 'startColumnIndex' in rng or 'endColumnIndex' in rng:
+        parts.append(f"cols {rng.get('startColumnIndex', '0')}-{rng.get('endColumnIndex', 'end')}")
+    return ", ".join(parts) if parts else "ENTIRE SHEET (no bounds set)"
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Check Google Sheet protected ranges.")
+    parser.add_argument('--spreadsheet-id', required=True)
+    parser.add_argument('--creds', required=True, help="Path to service account credentials JSON.")
+    args = parser.parse_args()
+
+    sa_email = get_service_account_email(args.creds)
+    print(f"Service account: {sa_email}\n")
+
+    service = connect(args.creds)
+    result = service.spreadsheets().get(
+        spreadsheetId=args.spreadsheet_id,
+        fields="sheets(properties(sheetId,title),protectedRanges)"
+    ).execute()
+
+    found_any = False
+    for sheet in result.get('sheets', []):
+        title = sheet['properties']['title']
+        for pr in sheet.get('protectedRanges', []):
+            found_any = True
+            editors = pr.get('editors', {})
+            editor_users = editors.get('users', [])
+            domain_can_edit = editors.get('domainUsersCanEdit', False)
+
+            print(f"Sheet/Tab:     {title}")
+            print(f"Description:   {pr.get('description', '(none)')}")
+            print(f"Scope:         {describe_range(pr.get('range'))}")
+            print(f"Warning only:  {pr.get('warningOnly', False)}")
+            print(f"Editor list:   {editor_users if editor_users else '(empty/not visible)'}")
+            print(f"Domain can edit: {domain_can_edit}")
+
+            if pr.get('warningOnly', False):
+                print("  -> Warning-only: this protection would NOT block the service account's writes.")
+            elif sa_email in editor_users:
+                print(f"  -> Service account IS listed as an allowed editor for this protection.")
+            else:
+                print(f"  -> Service account is NOT listed. THIS PROTECTION IS LIKELY BLOCKING WRITES.")
+            print()
+
+    if not found_any:
+        print("No protected ranges found on any sheet/tab in this spreadsheet.")
+        print("(If writes are still failing with a 400 protection error, double check")
+        print(" you're pointing at the correct spreadsheet ID / tab name, or that the")
+        print(" service account even has enough access to see the protection metadata.)")
+
+
+if __name__ == "__main__":
+    main()

scan-batch-dir

@@ -119,6 +119,117 @@ def connect_to_google_sheet(credentials_file: str):
     except Exception as e:
         raise Exception(f"Failed to create Google Sheets service: {str(e)}")
 
+def get_service_account_email(credentials_file: str) -> str:
+    """
+    Reads the 'client_email' field out of a Google service account
+    credentials JSON file.
+
+    Args:
+        credentials_file (str): Path to the Google service account credentials file.
+
+    Returns:
+        str or None: The service account email if found, else None.
+    """
+    try:
+        with open(credentials_file, "r") as f:
+            data = json.load(f)
+        return data.get("client_email")
+    except Exception as e:
+        logger.error(f"get_service_account_email - Failed to read credentials file: {str(e)}")
+        return None
+
+
+def check_sheet_protections(spreadsheet_id: str, credentials_file: str) -> tuple:
+    """
+    Checks all protected ranges/sheets in a Google Spreadsheet and determines
+    whether the service account associated with credentials_file is allowed
+    to edit within each protected range.
+
+    Args:
+        spreadsheet_id (str): The ID of the Google Sheet.
+        credentials_file (str): Path to the Google service account credentials file.
+
+    Returns:
+        tuple: (can_edit_all, blocking_protections, message)
+            can_edit_all (bool): True if no protections block the service account,
+                False if one or more protections would block writes.
+            blocking_protections (list of dict): Details of each protection that
+                the service account is NOT permitted to bypass. Each dict has
+                keys: sheet, description, range, warning_only.
+            message (str): Human-readable summary.
+    """
+    try:
+        sa_email = get_service_account_email(credentials_file)
+        if not sa_email:
+            msg = "check_sheet_protections - Could not determine service account email from credentials file."
+            logger.error(msg)
+            return False, [], msg
+
+        service = connect_to_google_sheet(credentials_file)
+
+        result = service.spreadsheets().get(
+            spreadsheetId=spreadsheet_id,
+            fields="sheets(properties(sheetId,title),protectedRanges)"
+        ).execute()
+
+        blocking_protections = []
+
+        for sheet in result.get('sheets', []):
+            title = sheet['properties']['title']
+            for pr in sheet.get('protectedRanges', []):
+                warning_only = pr.get('warningOnly', False)
+
+                # Warning-only protections do not block API writes.
+                if warning_only:
+                    continue
+
+                editors = pr.get('editors', {})
+                editor_users = editors.get('users', [])
+                domain_can_edit = editors.get('domainUsersCanEdit', False)
+
+                # If the service account is explicitly listed, it can edit.
+                if sa_email in editor_users:
+                    continue
+
+                # If domain-wide editing is allowed, and the service account
+                # belongs to that domain, it can edit. We can't reliably
+                # verify domain membership here, so we flag it as a caveat
+                # rather than assuming it's safe.
+                if domain_can_edit:
+                    logger.warning(
+                        f"check_sheet_protections - Sheet '{title}' protection allows "
+                        f"domain editors; unable to verify if '{sa_email}' is in-domain."
+                    )
+
+                # Otherwise, this protection blocks the service account.
+                blocking_protections.append({
+                    'sheet': title,
+                    'description': pr.get('description', ''),
+                    'range': pr.get('range', {}),
+                    'warning_only': warning_only,
+                })
+
+        if blocking_protections:
+            msg = (
+                f"check_sheet_protections - Service account '{sa_email}' is blocked by "
+                f"{len(blocking_protections)} protected range(s): "
+                + ", ".join(
+                    f"[{p['sheet']}: {p['description'] or 'unnamed'}]"
+                    for p in blocking_protections
+                )
+            )
+            logger.error(msg)
+            return False, blocking_protections, msg
+        else:
+            msg = f"check_sheet_protections - Service account '{sa_email}' can edit all ranges checked."
+            logger.info(msg)
+            return True, [], msg
+
+    except Exception as e:
+        msg = f"check_sheet_protections - An error occurred: {str(e)}"
+        logger.error(msg)
+        return False, [], msg
+
 def pad_rows(data: list) -> tuple:
     """
     Pads each row in the data to ensure all rows have the same length.
@@ -1534,6 +1645,17 @@ def main():
     print(f"Reading Google Sheet: {google_sheet_id},{google_sheet_name}")
     df = read_google_sheet(google_sheet_id, google_sheet_name, google_credentials)
 
+    # Check that the service account can actually write to this sheet before
+    # doing all the work of scanning the directory.
+    can_edit, blocking, msg = check_sheet_protections(google_sheet_id, google_credentials)
+    if not can_edit:
+        logger.error(f"Aborting: {msg}")
+        print(f"Error: The service account cannot edit this sheet due to protected ranges:")
+        for p in blocking:
+            print(f"  - Sheet '{p['sheet']}': {p['description'] or '(no description)'} — range: {p['range']}")
+        print("Contact the spreadsheet owner to remove protection or add the service account as an editor.")
+        sys.exit(1)
+
     # Fail fast if the sheet could not be read, rather than crashing later
     # with an unhelpful AttributeError on a None DataFrame.
     if df is None:
@@ -1544,7 +1666,13 @@ def main():
     # Ensure all required_columns exist.
     for col in required_columns:
         df = add_column(df,col)
-
+
+    dupe_cols = df.columns[df.columns.duplicated()].tolist()
+    if dupe_cols:
+        logger.error(f"Duplicate columns detected in sheet: {dupe_cols}")
+        print(f"Duplicate columns detected in sheet: {dupe_cols}")
+        sys.exit(1)
+
     # Fetch taxonomy terms once and cache as a name->tid dict.
     logger.info(f"Fetching Islandora Models taxonomy terms.")
     print(f"Fetching Islandora Models taxonomy terms.")