From bf14a2bed55039fbd38af273b1e3c3ed420f1719 Mon Sep 17 00:00:00 2001 From: Jordan Padams Date: Mon, 23 Mar 2026 18:45:57 -0700 Subject: [PATCH 1/3] Improve compare_s3_manifests CLI help and add progress status output - Clarify --missing-output, --unverifiable-output, --weak-output help strings to indicate they are output file paths (with defaults shown) - Add status prints throughout main() for loading, indexing, comparing (with 10% interval progress), and writing each output file Co-Authored-By: Claude Sonnet 4.6 --- .../data_integrity/compare_s3_manifests.py | 188 ++++++++++++++++++ 1 file changed, 188 insertions(+) create mode 100644 src/pds/pdc/data_integrity/compare_s3_manifests.py diff --git a/src/pds/pdc/data_integrity/compare_s3_manifests.py b/src/pds/pdc/data_integrity/compare_s3_manifests.py new file mode 100644 index 0000000..d70f672 --- /dev/null +++ b/src/pds/pdc/data_integrity/compare_s3_manifests.py @@ -0,0 +1,188 @@ +#!/usr/bin/env python3 +""" +Compare two S3 checksum manifests. + +Goal: +- Verify that every object in OLD has at least one matching object in NEW + by content signature, regardless of key name. + +Primary signature: + (size, checksum_type, checksum_value) + +Fallback/debug fields: + checksum_algorithm, etag + +Outputs: +- summary to stdout +- missing coverage CSV (confirmed missing from NEW) +- unverifiable CSV (no checksum available; cannot confirm presence in NEW) +- weak rows CSV (NEW-side rows without usable checksums) + +Exit codes: + 0 — PASS: every source object is confirmed present in destination + 1 — FAIL: one or more source objects are missing or unverifiable +""" + +from __future__ import annotations + +import argparse +import csv +import gzip +import io +from collections import Counter +from typing import Dict, List, Tuple + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument("--old", required=True, nargs="+", help="OLD manifest CSV(s), plain or gzipped") + parser.add_argument("--new", required=True, nargs="+", help="NEW manifest CSV(s), plain or gzipped") + parser.add_argument( + "--missing-output", + default="missing_in_new.csv", + help="Output CSV file path for OLD objects with no matching signature in NEW (default: %(default)s)", + ) + parser.add_argument( + "--unverifiable-output", + default="unverifiable.csv", + help="Output CSV file path for OLD objects with no checksum; cannot confirm presence in NEW (default: %(default)s)", + ) + parser.add_argument( + "--weak-output", + default="weak_checksum_rows.csv", + help="Output CSV file path for NEW-side rows without usable checksums (default: %(default)s)", + ) + return parser.parse_args() + + +def _open_csv(path: str) -> io.TextIOWrapper: + if path.endswith(".gz"): + return gzip.open(path, "rt", encoding="utf-8", newline="") # type: ignore[return-value] + return open(path, "r", newline="", encoding="utf-8") + + +def load_rows(paths: List[str]) -> List[Dict[str, str]]: + rows: List[Dict[str, str]] = [] + for path in paths: + with _open_csv(path) as f: + rows.extend(csv.DictReader(f)) + return rows + + +def usable_signature(row: Dict[str, str]) -> bool: + value = (row.get("checksum_value") or "").strip() + ctype = (row.get("checksum_type") or "").strip() + size = (row.get("size") or "").strip() + return bool(size and ctype and value and not value.startswith("ERROR:")) + + +def signature(row: Dict[str, str]) -> Tuple[str, str, str]: + return ( + (row.get("size") or "").strip(), + (row.get("checksum_type") or "").strip(), + (row.get("checksum_value") or "").strip(), + ) + + +def main() -> int: + args = parse_args() + + print(f"Loading OLD manifest(s): {args.old}") + old_rows = load_rows(args.old) + print(f" Loaded {len(old_rows):,} rows from OLD manifest(s)") + + print(f"Loading NEW manifest(s): {args.new}") + new_rows = load_rows(args.new) + print(f" Loaded {len(new_rows):,} rows from NEW manifest(s)") + + new_sig_counts: Counter = Counter() + new_weak_rows: List[Dict[str, str]] = [] + + print("Indexing NEW manifest signatures...") + for row in new_rows: + if usable_signature(row): + new_sig_counts[signature(row)] += 1 + else: + new_weak_rows.append(row) + print(f" {len(new_sig_counts):,} unique signatures indexed; {len(new_weak_rows):,} weak/unusable rows") + + missing_rows: List[Dict[str, str]] = [] + unverifiable_rows: List[Dict[str, str]] = [] + + old_bucket = (old_rows[0].get("bucket") or "OLD") if old_rows else "OLD" + new_bucket = (new_rows[0].get("bucket") or "NEW") if new_rows else "NEW" + + old_total = len(old_rows) + old_covered = 0 + + print(f"Comparing {old_total:,} OLD objects against NEW...") + report_interval = max(1, old_total // 10) + for i, row in enumerate(old_rows, 1): + if i % report_interval == 0 or i == old_total: + print(f" Progress: {i:,}/{old_total:,} ({100 * i // old_total}%)") + if not usable_signature(row): + unverifiable_rows.append(row) + continue + + sig = signature(row) + if new_sig_counts[sig] > 0: + old_covered += 1 + else: + missing_rows.append(row) + + row_fieldnames = list(old_rows[0].keys()) if old_rows else [] + + print(f"Writing missing objects report -> {args.missing_output}") + with open(args.missing_output, "w", newline="", encoding="utf-8") as f: + writer = csv.DictWriter(f, fieldnames=row_fieldnames) + writer.writeheader() + writer.writerows(missing_rows) + + print(f"Writing unverifiable objects report -> {args.unverifiable_output}") + with open(args.unverifiable_output, "w", newline="", encoding="utf-8") as f: + writer = csv.DictWriter(f, fieldnames=row_fieldnames) + writer.writeheader() + writer.writerows(unverifiable_rows) + + if new_weak_rows: + new_fieldnames = list(new_rows[0].keys()) if new_rows else [] + print(f"Writing weak checksum report -> {args.weak_output}") + with open(args.weak_output, "w", newline="", encoding="utf-8") as f: + writer = csv.DictWriter(f, fieldnames=new_fieldnames) + writer.writeheader() + writer.writerows(new_weak_rows) + + missing_count = len(missing_rows) + unverifiable_count = len(unverifiable_rows) + success = (missing_count == 0 and unverifiable_count == 0) + + print("=== S3 Manifest Comparison Summary ===") + print(f"Source bucket: {old_bucket}") + print(f"Destination bucket: {new_bucket}") + print() + print(f"{old_bucket} objects total: {old_total}") + print(f"{old_bucket} objects confirmed in {new_bucket}: {old_covered}") + print(f"{old_bucket} objects confirmed missing: {missing_count}") + print(f"{old_bucket} objects unable to verify: {unverifiable_count}") + print(f"{new_bucket} objects with weak checksum: {len(new_weak_rows)}") + print() + print(f"Confirmed missing report: {args.missing_output}") + print(f"Unverifiable report: {args.unverifiable_output}") + if new_weak_rows: + print(f"Weak checksum report: {args.weak_output}") + print() + if success: + print("RESULT: PASS") + print(f"Every object in {old_bucket} is confirmed present in {new_bucket}.") + else: + print("RESULT: FAIL") + if missing_count: + print(f" {missing_count} object(s) from {old_bucket} have no matching content signature in {new_bucket}.") + if unverifiable_count: + print(f" {unverifiable_count} object(s) from {old_bucket} have no checksum and cannot be verified.") + + return 0 if success else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) From 7254fbdda934c0d872d13bdc9b8aeb78fb8ffaf3 Mon Sep 17 00:00:00 2001 From: Jordan Padams Date: Wed, 1 Apr 2026 14:46:48 -0700 Subject: [PATCH 2/3] Add data integrity tools, update README, and modernize version loading - Add pdc-build-checksum-manifest and pdc-compare-manifests CLI entry points - Add build_s3_checksum_manifest.py for generating S3 checksum manifests - Add compare_s3_manifests.py improvements (multipart ETag handling, better docs) - Add scripts/csv-converter.py utility - Add CLAUDE.md project guidance file - Expand README with full usage docs for all tools - Replace deprecated pkg_resources with importlib.resources in __init__.py Co-Authored-By: Claude Sonnet 4.6 --- CLAUDE.md | 67 +++ README.md | 203 ++++++++- scripts/csv-converter.py | 87 ++++ setup.cfg | 2 + src/pds/pdc/__init__.py | 4 +- src/pds/pdc/data_integrity/__init__.py | 0 .../build_s3_checksum_manifest.py | 237 +++++++++++ .../data_integrity/compare_s3_manifests.py | 389 +++++++++++++----- src/pds/pdc/s3_download.py | 70 +++- 9 files changed, 935 insertions(+), 124 deletions(-) create mode 100644 CLAUDE.md create mode 100755 scripts/csv-converter.py create mode 100644 src/pds/pdc/data_integrity/__init__.py create mode 100644 src/pds/pdc/data_integrity/build_s3_checksum_manifest.py diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..58812bc --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,67 @@ +# CLAUDE.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. + +## Project Overview + +`pds.pdc-cloud-tools` provides CLI tools and utilities for accessing and managing Planetary Data Cloud (PDC) resources on AWS S3, part of the NASA Planetary Data System (PDS) ecosystem. The package uses Python namespace packages under `pds.pdc`. + +## Development Setup + +```bash +pip install --editable '.[dev]' +pre-commit install +pre-commit install -t pre-push +pre-commit install -t prepare-commit-msg +pre-commit install -t commit-msg +``` + +## Commands + +### Testing +```bash +pytest # run all tests +pytest tests/path/to/test.py # run single test file +ptw # watch mode +tox -e py39 # run tests via tox (parallel with coverage) +``` + +### Linting +```bash +tox -e lint # run all linters via pre-commit +pre-commit run --all # run pre-commit hooks directly +``` + +### Build +```bash +python setup.py sdist bdist_wheel +``` + +## Code Style + +- Line length: 120 characters (Black + Flake8) +- Docstrings: Google style (enforced by pydocstyle via flake8-docstrings) +- Import order: enforced by `reorder-python-imports` pre-commit hook +- Type annotations: mypy runs on `src/` at pre-commit time + +## Architecture + +The source lives under `src/pds/pdc/` using PDS namespace packaging. CLI entry points are registered in `setup.cfg`: + +- **`pdc-s3-download`** (`s3_download.py`) — Downloads S3 objects filtered by prefix, preserving directory structure. Accepts `--source-profile`, `--source-bucket`, `--source-prefix`, `--local-dest-dir` (also reads `AWS_PROFILE` and `AWS_BUCKET` env vars). Optional `--start-datetime` and `--end-datetime` (UTC, `YYYY-MM-DD` or `YYYY-MM-DDTHH:MM:SS`) filter by last-modified time. + +- **`pdc-inventory-summary`** (`inventory_summary.py`) — Summarizes S3 inventory files (gzipped CSV). Reads all `.gz` files in a directory tree and reports total object count and cumulative size. Assumes object size is in CSV column index 2. + +- **`data_integrity/`** — Data integrity verification for cloud migrations: + - `build_s3_checksum_manifest.py` — Generates a CSV manifest of S3 objects with checksums (prefers CRC64NVME, falls back to others). Supports `--resume-from` for incremental runs. Outputs: `bucket, key, size, checksum_algorithm, checksum_type, checksum_value, etag`. + - `compare_s3_manifests.py` — Compares two manifests by matching `(size, checksum_type, checksum_value)` tuples. Identifies objects in OLD manifest missing from NEW, and flags weak checksums. Returns exit code 1 if any OLD objects are unmatched. + +## CI/CD + +GitHub Actions workflows in `.github/workflows/`: +- `branch-cicd.yaml` — runs on feature branches +- `unstable-cicd.yaml` — runs on `main` pushes, publishes SNAPSHOT to Test PyPI via NASA-PDS Roundup Action +- `stable-cicd.yaml` — runs on `release/*` branches for stable releases +- `terraform_cicd.yaml` — infrastructure deployment + +Secrets required by CI: `ADMIN_GITHUB_TOKEN`, `TEST_PYPI_USERNAME`, `TEST_PYPI_PASSWORD`. diff --git a/README.md b/README.md index f35ea4e..4433396 100644 --- a/README.md +++ b/README.md @@ -9,11 +9,208 @@ Install with: pip install pds.cloud-tools -If possible, make it so that your program works out of the box without any additional configuration—but see the [Configuration](###configuration) section for details. -To execute, run: +## S3 Download (`pdc-s3-download`) - (put your run commands here) +Downloads objects from an S3 bucket matching a given prefix to a local directory, preserving the key structure as a directory tree. + +### Basic usage + +```bash +pdc-s3-download \ + --source-profile my-aws-profile \ + --source-bucket my-bucket \ + --source-prefix path/to/objects/ \ + --local-dest-dir ./downloads +``` + +`--source-profile` and `--source-bucket` can also be supplied via the `AWS_PROFILE` and `AWS_BUCKET` environment variables. + +### Filtering by last-modified time + +Use `--start-datetime` and/or `--end-datetime` to restrict downloads to objects last modified within a time range. All datetimes are UTC. Accepted formats: `YYYY-MM-DD` or `YYYY-MM-DDTHH:MM:SS`. + +| Arguments provided | Behavior | +|--------------------|----------| +| `--start-datetime` only | Download objects modified from start to now | +| `--end-datetime` only | Download objects modified up to end | +| Both | Download objects modified within [start, end] | +| Neither | Download all matching objects | + +```bash +# Download only objects modified in January 2025 +pdc-s3-download \ + --source-profile my-aws-profile \ + --source-bucket my-bucket \ + --source-prefix path/to/objects/ \ + --local-dest-dir ./downloads \ + --start-datetime 2025-01-01 \ + --end-datetime 2025-02-01 + +# Download everything modified since a specific timestamp +pdc-s3-download \ + --source-profile my-aws-profile \ + --source-bucket my-bucket \ + --source-prefix path/to/objects/ \ + --local-dest-dir ./downloads \ + --start-datetime 2025-06-15T12:00:00 +``` + + +## Data Integrity Verification (S3 Bucket Migration) + +Use these two tools to verify that every object in an OLD S3 bucket has a matching copy in a NEW bucket — by content (size + checksum), not just by key name. + +### Overview + +| Command | Purpose | +|---------|---------| +| `pdc-build-checksum-manifest` | Generate a CSV manifest of every object in a bucket with its checksum, size, and ETag | +| `pdc-compare-manifests` | Compare two manifests and report any OLD objects not represented in NEW | + +### AWS Credentials Setup + +The scripts use standard boto3 credential resolution. Options, in order of preference: + +**Named profile** (recommended): + +```ini +# ~/.aws/credentials +[old-account] +aws_access_key_id = AKIA... +aws_secret_access_key = ... + +[new-account] +aws_access_key_id = AKIA... +aws_secret_access_key = ... +``` + +**IAM role assumption** (for cross-account access without long-lived keys): + +```ini +# ~/.aws/config +[profile new-account] +role_arn = arn:aws:iam::123456789012:role/DataMigrationRole +source_profile = old-account +region = us-west-2 +``` + +**Environment variables** (for CI/automation): + +```bash +export AWS_ACCESS_KEY_ID=... +export AWS_SECRET_ACCESS_KEY=... +export AWS_SESSION_TOKEN=... # required when using temporary credentials +``` + +**Minimum required IAM permissions** for each bucket: + +```json +{ + "Effect": "Allow", + "Action": [ + "s3:ListBucket", + "s3:GetObject", + "s3:GetObjectAttributes" + ], + "Resource": [ + "arn:aws:s3:::your-bucket-name", + "arn:aws:s3:::your-bucket-name/*" + ] +} +``` + +### Step-by-Step Workflow + +**Step 1 — Generate the OLD bucket manifest:** + +```bash +pdc-build-checksum-manifest \ + --bucket old-bucket-name \ + --output old_manifest.csv \ + --profile old-account \ + --region us-west-2 +``` + +**Step 2 — Generate the NEW bucket manifest:** + +```bash +pdc-build-checksum-manifest \ + --bucket new-bucket-name \ + --output new_manifest.csv \ + --profile new-account \ + --region us-west-2 +``` + +**Step 3 — Compare the manifests:** + +```bash +pdc-compare-manifests \ + --old old_manifest.csv \ + --new new_manifest.csv \ + --missing-output missing_in_new.csv \ + --weak-output weak_checksums.csv +``` + +The comparison exits with code `0` (PASS) if every OLD object has a matching content signature in NEW, or `1` (FAIL) if any are missing. + +Sample output: + +``` +=== S3 Manifest Comparison Summary === +OLD objects total: 4823901 +OLD objects covered by NEW: 4823901 +OLD objects missing in NEW: 0 +OLD objects with weak checksum: 142 +Missing report: missing_in_new.csv +Weak rows report: weak_checksums.csv + +RESULT: PASS +Every OLD object has at least one matching content signature in NEW. +``` + +### Handling Buckets with Millions of Objects + +Manifest generation uses the `list_objects_v2` paginator, which pages through arbitrarily large buckets automatically — there is no per-bucket size limit. Progress is printed to stderr every 1,000 objects. + +The comparison step loads both manifests into memory simultaneously. For very large buckets (tens of millions of objects), ensure the host has sufficient RAM. + +### Resuming After a Failure + +If manifest generation is interrupted (timeout, credential expiry, network error), resume by passing the partial output file as `--resume-from`. Both `--output` and `--resume-from` must point to the same file: + +```bash +# Initial run (interrupted): +pdc-build-checksum-manifest \ + --bucket old-bucket-name \ + --output old_manifest.csv \ + --profile old-account + +# Resume run — pass the same file to both flags: +pdc-build-checksum-manifest \ + --bucket old-bucket-name \ + --output old_manifest.csv \ + --resume-from old_manifest.csv \ + --profile old-account +``` + +The resume run appends new rows to the existing CSV and skips any key already recorded in it. + +### Understanding the Output Files + +**`missing_in_new.csv`** — OLD objects not found in NEW, with a `reason` column: + +| Reason | Meaning | +|--------|---------| +| `NO_MATCHING_SIGNATURE_IN_NEW` | Object content (size + checksum) not present in NEW | +| `NO_USABLE_CHECKSUM` | Object has no stored checksum; cannot verify by content signature | + +**`weak_checksums.csv`** — rows from either manifest lacking a usable checksum. This occurs when objects were uploaded without `--checksum-algorithm`. These objects cannot be verified by content signature alone. + +### Limitations + +- Checksums are only present if they were stored at upload time. Objects copied with `aws s3 cp` without `--checksum-algorithm` will have no checksum and appear in the weak-checksum report. +- Matching is by content signature `(size, checksum_type, checksum_value)`, not by key. An object that was renamed or moved but has identical content will be counted as covered. ## Code of Conduct diff --git a/scripts/csv-converter.py b/scripts/csv-converter.py new file mode 100755 index 0000000..027e681 --- /dev/null +++ b/scripts/csv-converter.py @@ -0,0 +1,87 @@ +#!/usr/bin/env python3 + +import csv +import gzip +import glob +import os +from pathlib import Path + +# Input: one or more raw S3 Inventory CSV.gz files +INPUT_GLOBS = [ + "/Users/jpadams/proj/pds/pdsen/workspace/cloud-tools/data/inventories/new_bucket_raw/**/*.csv.gz", +] + +# Output: one normalized gzipped CSV +OUTPUT_FILE = "/Users/jpadams/proj/pds/pdsen/workspace/cloud-tools/data/inventories/new_bucket_processed/new_row_level_normalized.csv.gz" + +HEADER = [ + "bucket", + "key", + "size", + "checksum_algorithm", + "checksum_type", + "checksum_value", + "etag", +] + +def iter_input_files(patterns): + for pattern in patterns: + for path in glob.glob(pattern, recursive=True): + yield path + +def main(): + input_files = list(iter_input_files(INPUT_GLOBS)) + if not input_files: + raise SystemExit("No input files found.") + + Path(os.path.dirname(OUTPUT_FILE) or ".").mkdir(parents=True, exist_ok=True) + + rows_written = 0 + + with gzip.open(OUTPUT_FILE, "wt", newline="", encoding="utf-8") as out_f: + writer = csv.writer(out_f, quoting=csv.QUOTE_MINIMAL) + writer.writerow(HEADER) + + for input_file in input_files: + with gzip.open(input_file, "rt", newline="", encoding="utf-8") as in_f: + reader = csv.reader(in_f) + + for row in reader: + if not row: + continue + + # Skip header rows if present + if row[0] == "Bucket" or row[0] == "bucket": + continue + + # We only need these source positions from the raw inventory row: + # 0 = bucket + # 1 = key + # 2 = size + # 4 = e_tag + try: + bucket = row[0] + key = row[1] + size = row[2] + etag = row[4] + except IndexError: + raise ValueError(f"Unexpected row shape in {input_file}: {row}") + + if not etag: + continue + + writer.writerow([ + bucket, + key, + size, + "etag", + "etag", + etag, + etag, + ]) + rows_written += 1 + + print(f"Wrote {rows_written} rows to {OUTPUT_FILE}") + +if __name__ == "__main__": + main() diff --git a/setup.cfg b/setup.cfg index 22a5ae9..3274ac6 100644 --- a/setup.cfg +++ b/setup.cfg @@ -83,6 +83,8 @@ dev = console_scripts = pdc-s3-download=pds.pdc.s3_download:main pdc-inventory-summary=pds.pdc.inventory_summary:main + pdc-build-checksum-manifest=pds.pdc.data_integrity.build_s3_checksum_manifest:main + pdc-compare-manifests=pds.pdc.data_integrity.compare_s3_manifests:main [options.packages.find] # Don't change this. Needed to find packages under src/ diff --git a/src/pds/pdc/__init__.py b/src/pds/pdc/__init__.py index f89df4c..96d145f 100644 --- a/src/pds/pdc/__init__.py +++ b/src/pds/pdc/__init__.py @@ -1,9 +1,9 @@ # -*- coding: utf-8 -*- """My PDS Module.""" -import pkg_resources +from importlib.resources import files -__version__ = pkg_resources.resource_string(__name__, "VERSION.txt").decode("utf-8").strip() +__version__ = files(__name__).joinpath("VERSION.txt").read_text(encoding="utf-8").strip() # For future consideration: diff --git a/src/pds/pdc/data_integrity/__init__.py b/src/pds/pdc/data_integrity/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/pds/pdc/data_integrity/build_s3_checksum_manifest.py b/src/pds/pdc/data_integrity/build_s3_checksum_manifest.py new file mode 100644 index 0000000..0803c2b --- /dev/null +++ b/src/pds/pdc/data_integrity/build_s3_checksum_manifest.py @@ -0,0 +1,237 @@ +#!/usr/bin/env python3 +""" +Build a per-object checksum manifest for an S3 bucket. + +Checksum resolution order (first match wins): + 1. Native S3 checksum from GetObjectAttributes (CRC64NVME preferred) + 2. MD5 from x-amz-meta-s3cmd-attrs user metadata (HeadObject fallback) + 3. No checksum recorded — object will appear as unverifiable in comparison + +Output CSV columns: + bucket, key, size, checksum_algorithm, checksum_type, checksum_value, etag + +Usage: + pdc-build-checksum-manifest --bucket my-bucket --output manifest.csv + +Optional: + --prefix PDS4/ + --profile my-aws-profile + --region us-west-2 + --resume-from manifest.csv # resume an interrupted run (same file as --output) +""" + +from __future__ import annotations + +import argparse +import csv +import sys +import time +from typing import Dict, Iterable, Optional, Set, Tuple + +import boto3 +from botocore.config import Config +from botocore.exceptions import ClientError + + +CHECKSUM_FIELDS = [ + ("CRC64NVME", "ChecksumCRC64NVME"), + ("CRC32C", "ChecksumCRC32C"), + ("CRC32", "ChecksumCRC32"), + ("SHA256", "ChecksumSHA256"), + ("SHA1", "ChecksumSHA1"), +] + +CSV_HEADERS = [ + "bucket", + "key", + "size", + "checksum_algorithm", + "checksum_type", + "checksum_value", + "etag", +] + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument("--bucket", required=True, help="S3 bucket name") + parser.add_argument("--output", required=True, help="Output CSV path") + parser.add_argument("--prefix", default="", help="Optional prefix filter") + parser.add_argument("--profile", default=None, help="AWS profile") + parser.add_argument("--region", default=None, help="AWS region") + parser.add_argument( + "--resume-from", + default=None, + help="Existing manifest CSV to skip already-processed keys", + ) + return parser.parse_args() + + +def build_session(profile: Optional[str], region: Optional[str]): + session_kwargs = {} + if profile: + session_kwargs["profile_name"] = profile + if region: + session_kwargs["region_name"] = region + return boto3.Session(**session_kwargs) + + +def load_existing_keys(csv_path: str) -> Set[str]: + keys: Set[str] = set() + with open(csv_path, "r", newline="", encoding="utf-8") as f: + reader = csv.DictReader(f) + for row in reader: + keys.add(row["key"]) + return keys + + +def choose_checksum(resp: Dict) -> Tuple[str, str, str]: + """ + Prefer CRC64NVME if present. Otherwise take the first available checksum. + Returns: (algorithm, checksum_type, checksum_value) + """ + checksum = resp.get("Checksum", {}) or {} + checksum_type = checksum.get("ChecksumType", "") + + for algo_name, field_name in CHECKSUM_FIELDS: + value = checksum.get(field_name) + if value: + return algo_name, checksum_type, value + + return "", checksum_type, "" + + +def list_objects(s3_client, bucket: str, prefix: str) -> Iterable[Dict]: + paginator = s3_client.get_paginator("list_objects_v2") + kwargs = {"Bucket": bucket} + if prefix: + kwargs["Prefix"] = prefix + + for page in paginator.paginate(**kwargs): + for obj in page.get("Contents", []): + yield obj + + +def get_object_attrs_with_retry(s3_client, bucket: str, key: str, retries: int = 5) -> Dict: + delay = 1.0 + for attempt in range(retries): + try: + return s3_client.get_object_attributes( + Bucket=bucket, + Key=key, + ObjectAttributes=["Checksum", "ObjectSize", "ETag"], + ) + except ClientError as e: + code = e.response.get("Error", {}).get("Code", "") + if code in {"SlowDown", "Throttling", "RequestTimeout", "InternalError"} and attempt < retries - 1: + time.sleep(delay) + delay *= 2 + continue + raise + + +def head_object_with_retry(s3_client, bucket: str, key: str, retries: int = 5) -> Dict: + delay = 1.0 + for attempt in range(retries): + try: + return s3_client.head_object(Bucket=bucket, Key=key) + except ClientError as e: + code = e.response.get("Error", {}).get("Code", "") + if code in {"SlowDown", "Throttling", "RequestTimeout", "InternalError"} and attempt < retries - 1: + time.sleep(delay) + delay *= 2 + continue + raise + + +def parse_s3cmd_md5(attrs_value: str) -> str: + """Extract md5 from an x-amz-meta-s3cmd-attrs metadata string. + + Example value: + atime:1733437044/ctime:1733164973/gid:4000/gname:pds/md5:c3de3759.../mode:33188/... + + Returns the md5 hex string, or empty string if not found. + """ + for part in attrs_value.split("/"): + if part.startswith("md5:"): + return part[4:].strip() + return "" + + +def main() -> int: + args = parse_args() + + session = build_session(args.profile, args.region) + s3_client = session.client( + "s3", + config=Config( + retries={"max_attempts": 10, "mode": "standard"}, + ), + ) + + processed_keys: Set[str] = set() + write_header = True + + mode = "w" + if args.resume_from: + processed_keys = load_existing_keys(args.resume_from) + mode = "a" + write_header = False + + with open(args.output, mode, newline="", encoding="utf-8") as f: + writer = csv.writer(f) + if write_header: + writer.writerow(CSV_HEADERS) + + count = 0 + for obj in list_objects(s3_client, args.bucket, args.prefix): + key = obj["Key"] + if key in processed_keys: + continue + + try: + attrs = get_object_attrs_with_retry(s3_client, args.bucket, key) + algo, checksum_type, checksum_value = choose_checksum(attrs) + size = attrs.get("ObjectSize", obj.get("Size", "")) + etag = (attrs.get("ETag") or "").strip('"') + + if not checksum_value: + # No native S3 checksum — try x-amz-meta-s3cmd-attrs metadata. + head = head_object_with_retry(s3_client, args.bucket, key) + s3cmd_attrs = (head.get("Metadata") or {}).get("s3cmd-attrs", "") + if s3cmd_attrs: + md5 = parse_s3cmd_md5(s3cmd_attrs) + if md5: + algo = "S3CMD-MD5" + checksum_type = "MD5" + checksum_value = md5 + + except ClientError as e: + # Record a row with blanks so the comparison script can flag it. + err_code = e.response.get("Error", {}).get("Code", "UNKNOWN") + size = obj.get("Size", "") + etag = "" + algo = "" + checksum_type = "" + checksum_value = f"ERROR:{err_code}" + + writer.writerow([ + args.bucket, + key, + size, + algo, + checksum_type, + checksum_value, + etag, + ]) + count += 1 + + if count % 1000 == 0: + print(f"Processed {count} objects...", file=sys.stderr) + + print(f"Done. Wrote manifest to {args.output}", file=sys.stderr) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/pds/pdc/data_integrity/compare_s3_manifests.py b/src/pds/pdc/data_integrity/compare_s3_manifests.py index d70f672..157be2e 100644 --- a/src/pds/pdc/data_integrity/compare_s3_manifests.py +++ b/src/pds/pdc/data_integrity/compare_s3_manifests.py @@ -1,26 +1,54 @@ #!/usr/bin/env python3 -""" -Compare two S3 checksum manifests. +"""Compare two S3 checksum manifests to verify a bucket migration. Goal: -- Verify that every object in OLD has at least one matching object in NEW - by content signature, regardless of key name. - -Primary signature: - (size, checksum_type, checksum_value) - -Fallback/debug fields: - checksum_algorithm, etag - -Outputs: -- summary to stdout -- missing coverage CSV (confirmed missing from NEW) -- unverifiable CSV (no checksum available; cannot confirm presence in NEW) -- weak rows CSV (NEW-side rows without usable checksums) + Confirm that every object in OLD has at least one matching object in NEW + by content signature, regardless of key name or bucket name. + +Matching strategy: + Objects are matched on the tuple (size, checksum_type, checksum_value). + Key names are intentionally ignored so renamed objects still count as present. + +Multipart ETags: + AWS S3 computes ETags differently for multipart uploads: the value is a hash + of part hashes suffixed with the part count (e.g. "abc123-4"). When an + object is re-uploaded or copied with a different part size the ETag changes, + even if the object bytes are identical. These checksums are therefore NOT + stable across migrations and cannot be used to confirm presence in NEW. + Any OLD row whose checksum_value matches the pattern ``-`` is + treated as unverifiable and written to --unverifiable-output rather than + --missing-output. + +Output files: + --missing-output (default: missing_in_new.csv) + OLD-side rows whose (size, checksum_type, checksum_value) signature was + not found anywhere in NEW. These objects have stable checksums (not + multipart ETags) and are confirmed absent from the destination. + Action required: investigate why they were not migrated. + + --unverifiable-output (default: unverifiable.csv) + OLD-side rows that cannot be confirmed present in NEW for one of two + reasons: + (a) The row has no usable checksum at all (empty, ERROR:*, etc.). + (b) The row has a multipart ETag (value ends in -), which changes + when objects are re-uploaded and therefore cannot reliably match + against the destination. + The ``reason`` column appended to each output row records which case + applies: ``no_checksum`` or ``multipart_etag``. + Action required: use a different verification method (e.g. byte-level + comparison, S3 object metadata, or re-checksumming) for these objects. + + --weak-output (default: weak_checksum_rows.csv) + NEW-side rows that have no usable checksum. These entries cannot be + used to confirm migration coverage for any OLD object. + Action required: re-run checksum generation on NEW for these objects. Exit codes: - 0 — PASS: every source object is confirmed present in destination - 1 — FAIL: one or more source objects are missing or unverifiable + 0 — PASS: every OLD object is either confirmed present in NEW or is + unverifiable due to a multipart ETag (i.e. missing_count == 0) + 1 — FAIL: one or more OLD objects with stable checksums have no matching + signature in NEW, or there are unverifiable rows beyond multipart + ETags """ from __future__ import annotations @@ -29,28 +57,53 @@ import csv import gzip import io -from collections import Counter -from typing import Dict, List, Tuple +import os +import re +import sqlite3 +import tempfile +from typing import Dict, Iterator, Optional, Tuple + +_MULTIPART_ETAG_RE = re.compile(r"^[0-9a-fA-F]+-\d+$") + +MANIFEST_FIELDNAMES = ["bucket", "key", "size", "checksum_algorithm", "checksum_type", "checksum_value", "etag"] def parse_args() -> argparse.Namespace: - parser = argparse.ArgumentParser() + parser = argparse.ArgumentParser( + description="Compare two S3 checksum manifests to verify a bucket migration.", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +Output files +------------ +--missing-output + OLD rows whose (size, checksum_type, checksum_value) was not found in NEW. + These have stable checksums and are confirmed absent — action required. + +--unverifiable-output + OLD rows that cannot be verified: either no usable checksum ('no_checksum') + or a multipart ETag that changes across re-uploads ('multipart_etag'). + A 'reason' column is appended to each row. + +--weak-output + NEW rows with no usable checksum; cannot be matched against OLD objects. +""", + ) parser.add_argument("--old", required=True, nargs="+", help="OLD manifest CSV(s), plain or gzipped") parser.add_argument("--new", required=True, nargs="+", help="NEW manifest CSV(s), plain or gzipped") parser.add_argument( "--missing-output", default="missing_in_new.csv", - help="Output CSV file path for OLD objects with no matching signature in NEW (default: %(default)s)", + help="Output CSV for OLD objects with no matching signature in NEW (default: %(default)s)", ) parser.add_argument( "--unverifiable-output", default="unverifiable.csv", - help="Output CSV file path for OLD objects with no checksum; cannot confirm presence in NEW (default: %(default)s)", + help="Output CSV for OLD objects that cannot be verified (no checksum or multipart ETag) (default: %(default)s)", ) parser.add_argument( "--weak-output", default="weak_checksum_rows.csv", - help="Output CSV file path for NEW-side rows without usable checksums (default: %(default)s)", + help="Output CSV for NEW-side rows without usable checksums (default: %(default)s)", ) return parser.parse_args() @@ -61,22 +114,54 @@ def _open_csv(path: str) -> io.TextIOWrapper: return open(path, "r", newline="", encoding="utf-8") -def load_rows(paths: List[str]) -> List[Dict[str, str]]: - rows: List[Dict[str, str]] = [] +def _iter_rows(paths: list) -> Iterator[Dict[str, str]]: + """Yield rows from all manifest CSVs with consistent fieldnames. + + The first file's header determines the canonical fieldnames. + Subsequent files that lack a header (or have a different header) + are read with those canonical fieldnames so all rows are consistent. + """ + canonical: Optional[list] = None for path in paths: with _open_csv(path) as f: - rows.extend(csv.DictReader(f)) - return rows - - -def usable_signature(row: Dict[str, str]) -> bool: + # Peek at the first line to detect whether the file has a header. + first_line = f.readline() + if not first_line: + continue + first_vals = next(csv.reader([first_line.rstrip("\n")])) + + if canonical is None: + # Decide whether the first line is a header or data. + if first_vals == MANIFEST_FIELDNAMES or first_vals[0] in ("bucket", "key"): + # Looks like a header — use it as canonical fieldnames. + canonical = first_vals + else: + # No header — use the known manifest fieldnames and emit first line as data. + canonical = MANIFEST_FIELDNAMES + yield dict(zip(canonical, first_vals)) + else: + if first_vals != canonical: + # No header — first line is data; emit it then read the rest. + yield dict(zip(canonical, first_vals)) + # Either way, the header was consumed; use canonical fieldnames for + # the remainder so all rows are consistently keyed. + + yield from csv.DictReader(f, fieldnames=canonical) + + +def _usable_signature(row: Dict[str, str]) -> bool: value = (row.get("checksum_value") or "").strip() ctype = (row.get("checksum_type") or "").strip() size = (row.get("size") or "").strip() return bool(size and ctype and value and not value.startswith("ERROR:")) -def signature(row: Dict[str, str]) -> Tuple[str, str, str]: +def _is_multipart_etag(row: Dict[str, str]) -> bool: + value = (row.get("checksum_value") or "").strip() + return bool(_MULTIPART_ETAG_RE.match(value)) + + +def _signature(row: Dict[str, str]) -> Tuple[str, str, str]: return ( (row.get("size") or "").strip(), (row.get("checksum_type") or "").strip(), @@ -84,102 +169,184 @@ def signature(row: Dict[str, str]) -> Tuple[str, str, str]: ) +def _build_new_index( + db: sqlite3.Connection, + paths: list, + weak_output_path: str, +) -> Tuple[int, int, Optional[list]]: + """Stream NEW manifests into SQLite signature index; stream weak rows to file. + + Returns (unique_sig_count, weak_count, new_fieldnames). + """ + db.execute( + """ + CREATE TABLE new_sigs ( + size TEXT NOT NULL, + checksum_type TEXT NOT NULL, + checksum_value TEXT NOT NULL, + cnt INTEGER NOT NULL DEFAULT 1, + PRIMARY KEY (size, checksum_type, checksum_value) + ) + """ + ) + + unique_sigs = 0 + weak_count = 0 + new_fieldnames: Optional[list] = None + weak_writer: Optional[csv.DictWriter] = None + weak_file = None + + try: + weak_file = open(weak_output_path, "w", newline="", encoding="utf-8") + + for row in _iter_rows(paths): + if new_fieldnames is None: + new_fieldnames = list(row.keys()) + weak_writer = csv.DictWriter(weak_file, fieldnames=new_fieldnames, extrasaction="ignore") + weak_writer.writeheader() + + if _usable_signature(row): + sig = _signature(row) + db.execute( + "INSERT INTO new_sigs(size, checksum_type, checksum_value) VALUES (?,?,?)" + " ON CONFLICT(size, checksum_type, checksum_value) DO UPDATE SET cnt = cnt + 1", + sig, + ) + unique_sigs += 1 + else: + assert weak_writer is not None + weak_writer.writerow(row) + weak_count += 1 + + db.commit() + finally: + if weak_file is not None: + weak_file.close() + + return unique_sigs, weak_count, new_fieldnames + + def main() -> int: args = parse_args() - print(f"Loading OLD manifest(s): {args.old}") - old_rows = load_rows(args.old) - print(f" Loaded {len(old_rows):,} rows from OLD manifest(s)") - - print(f"Loading NEW manifest(s): {args.new}") - new_rows = load_rows(args.new) - print(f" Loaded {len(new_rows):,} rows from NEW manifest(s)") - - new_sig_counts: Counter = Counter() - new_weak_rows: List[Dict[str, str]] = [] - - print("Indexing NEW manifest signatures...") - for row in new_rows: - if usable_signature(row): - new_sig_counts[signature(row)] += 1 - else: - new_weak_rows.append(row) - print(f" {len(new_sig_counts):,} unique signatures indexed; {len(new_weak_rows):,} weak/unusable rows") - - missing_rows: List[Dict[str, str]] = [] - unverifiable_rows: List[Dict[str, str]] = [] - - old_bucket = (old_rows[0].get("bucket") or "OLD") if old_rows else "OLD" - new_bucket = (new_rows[0].get("bucket") or "NEW") if new_rows else "NEW" - - old_total = len(old_rows) - old_covered = 0 - - print(f"Comparing {old_total:,} OLD objects against NEW...") - report_interval = max(1, old_total // 10) - for i, row in enumerate(old_rows, 1): - if i % report_interval == 0 or i == old_total: - print(f" Progress: {i:,}/{old_total:,} ({100 * i // old_total}%)") - if not usable_signature(row): - unverifiable_rows.append(row) - continue - - sig = signature(row) - if new_sig_counts[sig] > 0: - old_covered += 1 - else: - missing_rows.append(row) - - row_fieldnames = list(old_rows[0].keys()) if old_rows else [] - - print(f"Writing missing objects report -> {args.missing_output}") - with open(args.missing_output, "w", newline="", encoding="utf-8") as f: - writer = csv.DictWriter(f, fieldnames=row_fieldnames) - writer.writeheader() - writer.writerows(missing_rows) - - print(f"Writing unverifiable objects report -> {args.unverifiable_output}") - with open(args.unverifiable_output, "w", newline="", encoding="utf-8") as f: - writer = csv.DictWriter(f, fieldnames=row_fieldnames) - writer.writeheader() - writer.writerows(unverifiable_rows) - - if new_weak_rows: - new_fieldnames = list(new_rows[0].keys()) if new_rows else [] - print(f"Writing weak checksum report -> {args.weak_output}") - with open(args.weak_output, "w", newline="", encoding="utf-8") as f: - writer = csv.DictWriter(f, fieldnames=new_fieldnames) - writer.writeheader() - writer.writerows(new_weak_rows) - - missing_count = len(missing_rows) - unverifiable_count = len(unverifiable_rows) - success = (missing_count == 0 and unverifiable_count == 0) + with tempfile.NamedTemporaryFile(suffix=".db", delete=True) as tmp: + db_path = tmp.name + + db = sqlite3.connect(db_path) + db.execute("PRAGMA journal_mode=WAL") + db.execute("PRAGMA synchronous=NORMAL") + db.execute("PRAGMA cache_size=-131072") # 128 MB page cache + + try: + print(f"Indexing NEW manifest(s) into SQLite: {args.new}") + unique_sigs, weak_count, new_fieldnames = _build_new_index(db, args.new, args.weak_output) + print(f" {unique_sigs:,} unique signatures indexed; {weak_count:,} weak/unusable rows") + if weak_count: + print(f" Weak checksum report -> {args.weak_output}") + + old_total = 0 + old_covered = 0 + missing_count = 0 + unverifiable_count = 0 + multipart_etag_count = 0 + old_bucket = "OLD" + new_bucket = "NEW" + old_fieldnames: Optional[list] = None + + print(f"Comparing OLD manifest(s) against NEW index: {args.old}") + with ( + open(args.missing_output, "w", newline="", encoding="utf-8") as miss_f, + open(args.unverifiable_output, "w", newline="", encoding="utf-8") as unver_f, + ): + miss_writer: Optional[csv.DictWriter] = None + unver_writer: Optional[csv.DictWriter] = None + + report_every = 100_000 + + for row in _iter_rows(args.old): + if old_fieldnames is None: + old_fieldnames = list(row.keys()) + miss_writer = csv.DictWriter(miss_f, fieldnames=old_fieldnames, extrasaction="ignore") + miss_writer.writeheader() + unver_fieldnames = old_fieldnames + ["reason"] + unver_writer = csv.DictWriter(unver_f, fieldnames=unver_fieldnames, extrasaction="ignore") + unver_writer.writeheader() + old_bucket = row.get("bucket") or "OLD" + + old_total += 1 + if old_total % report_every == 0: + print(f" Progress: {old_total:,} rows processed...") + + if not _usable_signature(row): + assert unver_writer is not None + unver_writer.writerow({**row, "reason": "no_checksum"}) + unverifiable_count += 1 + continue + + if _is_multipart_etag(row): + assert unver_writer is not None + unver_writer.writerow({**row, "reason": "multipart_etag"}) + unverifiable_count += 1 + multipart_etag_count += 1 + continue + + sig = _signature(row) + cur = db.execute( + "SELECT cnt FROM new_sigs WHERE size=? AND checksum_type=? AND checksum_value=?", + sig, + ) + if cur.fetchone(): + old_covered += 1 + else: + assert miss_writer is not None + miss_writer.writerow(row) + missing_count += 1 + + print(f" {old_total:,} OLD rows processed") + + finally: + db.close() + try: + os.unlink(db_path) + except OSError: + pass + + success = missing_count == 0 print("=== S3 Manifest Comparison Summary ===") - print(f"Source bucket: {old_bucket}") - print(f"Destination bucket: {new_bucket}") + print(f"Source bucket: {old_bucket}") + print(f"Destination bucket: {new_bucket}") print() - print(f"{old_bucket} objects total: {old_total}") - print(f"{old_bucket} objects confirmed in {new_bucket}: {old_covered}") - print(f"{old_bucket} objects confirmed missing: {missing_count}") - print(f"{old_bucket} objects unable to verify: {unverifiable_count}") - print(f"{new_bucket} objects with weak checksum: {len(new_weak_rows)}") + print(f"{old_bucket} objects total: {old_total:,}") + print(f"{old_bucket} objects confirmed in {new_bucket}: {old_covered:,}") + print(f"{old_bucket} objects confirmed missing: {missing_count:,}") + print(f"{old_bucket} objects unable to verify (total): {unverifiable_count:,}") + print(f" of which multipart ETags: {multipart_etag_count:,}") + print(f" of which no usable checksum: {unverifiable_count - multipart_etag_count:,}") + print(f"{new_bucket} objects with weak checksum: {weak_count:,}") print() print(f"Confirmed missing report: {args.missing_output}") print(f"Unverifiable report: {args.unverifiable_output}") - if new_weak_rows: + if weak_count: print(f"Weak checksum report: {args.weak_output}") print() if success: print("RESULT: PASS") - print(f"Every object in {old_bucket} is confirmed present in {new_bucket}.") + print(f"Every object in {old_bucket} with a stable checksum is confirmed present in {new_bucket}.") + if multipart_etag_count: + print( + f" Note: {multipart_etag_count:,} object(s) have multipart ETags and could not be verified by checksum." + ) else: print("RESULT: FAIL") if missing_count: - print(f" {missing_count} object(s) from {old_bucket} have no matching content signature in {new_bucket}.") + print(f" {missing_count:,} object(s) from {old_bucket} have no matching content signature in {new_bucket}.") if unverifiable_count: - print(f" {unverifiable_count} object(s) from {old_bucket} have no checksum and cannot be verified.") + print(f" {unverifiable_count:,} object(s) from {old_bucket} could not be verified.") + if multipart_etag_count: + print(f" {multipart_etag_count:,} have multipart ETags (ETag changes on re-upload; not a reliable match key).") + if unverifiable_count - multipart_etag_count: + print(f" {unverifiable_count - multipart_etag_count:,} have no usable checksum at all.") return 0 if success else 1 diff --git a/src/pds/pdc/s3_download.py b/src/pds/pdc/s3_download.py index c1a4f1b..c149c87 100644 --- a/src/pds/pdc/s3_download.py +++ b/src/pds/pdc/s3_download.py @@ -2,21 +2,47 @@ """Download files from an S3 bucket to a local directory. This script uses the boto3 library to download objects from a specified S3 -bucket that match a given prefix. The AWS profile, S3 bucket name, prefix, -and local destination directory are provided as command-line arguments or -through the AWS_PROFILE and AWS_BUCKET environment variables. +bucket that match a given prefix. Objects can optionally be filtered by their +last-modified timestamp. The AWS profile, S3 bucket name, prefix, and local +destination directory are provided as command-line arguments or through the +AWS_PROFILE and AWS_BUCKET environment variables. Usage: - python s3_download.py --source-prefix prefix/path/to/objects/ --local-dest-dir local/path/ + pdc-s3-download --source-prefix prefix/path/to/objects/ --local-dest-dir local/path/ + pdc-s3-download --source-prefix prefix/ --local-dest-dir local/ --start-datetime 2025-01-01 + pdc-s3-download --source-prefix prefix/ --local-dest-dir local/ \\ + --start-datetime 2025-01-01T00:00:00 --end-datetime 2025-06-01T00:00:00 + + All datetimes are interpreted as UTC. Accepted formats: YYYY-MM-DD or YYYY-MM-DDTHH:MM:SS. + + Time range behavior: + --start-datetime only → download objects modified from start to now + --end-datetime only → download objects modified up to end + both provided → download objects modified within [start, end] + neither provided → download all objects (no time filter) + (Optionally set AWS_PROFILE and AWS_BUCKET in your environment) """ import argparse import os +from datetime import datetime +from datetime import timezone import boto3 from botocore.exceptions import ClientError +def parse_datetime(value): + """Parse a datetime string in ISO 8601 format, returning an aware UTC datetime.""" + for fmt in ("%Y-%m-%dT%H:%M:%S", "%Y-%m-%d"): + try: + dt = datetime.strptime(value, fmt) + return dt.replace(tzinfo=timezone.utc) + except ValueError: + continue + raise argparse.ArgumentTypeError(f"Invalid datetime format: '{value}'. Use YYYY-MM-DD or YYYY-MM-DDTHH:MM:SS.") + + def main(): """Parse command-line arguments and download S3 objects to a local directory.""" # Get defaults from environment variables if available @@ -24,20 +50,34 @@ def main(): default_source_bucket = os.environ.get("AWS_BUCKET") parser = argparse.ArgumentParser( - description="Download files from an S3 bucket (filtered by a prefix) " "to a local directory." + description="Download files from an S3 bucket (filtered by a prefix) to a local directory." ) parser.add_argument( "--source-profile", default=default_source_profile, - help=("AWS profile name for the source bucket " "(defaults to AWS_PROFILE environment variable)"), + help=("AWS profile name for the source bucket (defaults to AWS_PROFILE environment variable)"), ) parser.add_argument( "--source-bucket", default=default_source_bucket, - help=("Name of the source S3 bucket " "(defaults to AWS_BUCKET environment variable)"), + help=("Name of the source S3 bucket (defaults to AWS_BUCKET environment variable)"), ) parser.add_argument("--source-prefix", required=True, help="Prefix in the source bucket to filter objects") parser.add_argument("--local-dest-dir", required=True, help="Local directory where files will be downloaded") + parser.add_argument( + "--start-datetime", + type=parse_datetime, + default=None, + metavar="DATETIME", + help="Only download objects last modified at or after this datetime (YYYY-MM-DD or YYYY-MM-DDTHH:MM:SS, UTC)", + ) + parser.add_argument( + "--end-datetime", + type=parse_datetime, + default=None, + metavar="DATETIME", + help="Only download objects last modified before or at this datetime (YYYY-MM-DD or YYYY-MM-DDTHH:MM:SS, UTC). Defaults to now if --start-datetime is provided.", + ) args = parser.parse_args() @@ -52,6 +92,11 @@ def main(): source_prefix = args.source_prefix local_dest_dir = args.local_dest_dir + start_dt = args.start_datetime + end_dt = args.end_datetime + if start_dt and not end_dt: + end_dt = datetime.now(tz=timezone.utc) + # Create a session for the source AWS profile and initialize S3 resource source_session = boto3.Session(profile_name=source_profile) source_s3 = source_session.resource("s3") @@ -64,13 +109,22 @@ def main(): # Loop through objects with the specified prefix in the source bucket for obj in source_bucket.objects.filter(Prefix=source_prefix): # Remove the source prefix from the object's key to construct a relative path - relative_path = obj.key[len(source_prefix) :] + relative_path = obj.key[len(source_prefix) :].lstrip("/") # Skip directory markers or keys that resolve to an empty relative path if not relative_path or obj.key.endswith("/"): print(f"Skipping directory marker or empty key: {obj.key}") continue + # Filter by last modified time range if specified + last_modified = obj.last_modified + if start_dt and last_modified < start_dt: + print(f"Skipping {obj.key} (last modified {last_modified} is before {start_dt})") + continue + if end_dt and last_modified > end_dt: + print(f"Skipping {obj.key} (last modified {last_modified} is after {end_dt})") + continue + # Construct the local file path local_file_path = os.path.join(local_dest_dir, relative_path) From ce6c2533599ea3acc6b66295c73e9f2729cafd24 Mon Sep 17 00:00:00 2001 From: Jordan Padams Date: Thu, 2 Apr 2026 14:19:48 -0700 Subject: [PATCH 3/3] Fix mypy/flake8 lint errors and upgrade to Python 3.12+ - Add missing docstrings across data_integrity modules and s3_download - Fix mypy: explicit boto3.Session kwargs instead of **dict unpack - Fix mypy: add unreachable raise after retry loops for missing return - Fix D212/D205 docstring formatting in build_s3_checksum_manifest - Fix D301: use r-string for module docstring with backslashes in s3_download - Fix B950: wrap long help string in s3_download - Drop Python 3.9 support; require >= 3.12; update tox envlist and classifiers Co-Authored-By: Claude Sonnet 4.6 --- setup.cfg | 6 ++- src/pds/pdc/data_integrity/__init__.py | 1 + .../build_s3_checksum_manifest.py | 54 ++++++++++++------- .../data_integrity/compare_s3_manifests.py | 16 ++++-- src/pds/pdc/s3_download.py | 8 ++- tox.ini | 4 +- 6 files changed, 59 insertions(+), 30 deletions(-) diff --git a/setup.cfg b/setup.cfg index 3274ac6..e2123f5 100644 --- a/setup.cfg +++ b/setup.cfg @@ -30,7 +30,8 @@ url = https://github.com/NASA-PDS/cloud-tools download_url = https://github.com/NASA-PDS/cloud-tools/releases/ classifiers = Programming Language :: Python :: 3 - Programming Language :: Python :: 3.9 + Programming Language :: Python :: 3.12 + Programming Language :: Python :: 3.13 License :: OSI Approved :: Apache Software License Operating System :: OS Independent @@ -55,11 +56,12 @@ namespace_packages = pds package_dir = = src packages = find_namespace: -python_requires = >= 3.9 +python_requires = >= 3.12 [options.extras_require] dev = black~=23.7.0 + detect-secrets~=1.5.0 flake8~=6.1.0 flake8-bugbear~=23.7.10 flake8-docstrings~=1.7.0 diff --git a/src/pds/pdc/data_integrity/__init__.py b/src/pds/pdc/data_integrity/__init__.py index e69de29..a667597 100644 --- a/src/pds/pdc/data_integrity/__init__.py +++ b/src/pds/pdc/data_integrity/__init__.py @@ -0,0 +1 @@ +"""Data integrity tools for cloud migrations.""" diff --git a/src/pds/pdc/data_integrity/build_s3_checksum_manifest.py b/src/pds/pdc/data_integrity/build_s3_checksum_manifest.py index 0803c2b..1275705 100644 --- a/src/pds/pdc/data_integrity/build_s3_checksum_manifest.py +++ b/src/pds/pdc/data_integrity/build_s3_checksum_manifest.py @@ -1,6 +1,5 @@ #!/usr/bin/env python3 -""" -Build a per-object checksum manifest for an S3 bucket. +"""Build a per-object checksum manifest for an S3 bucket. Checksum resolution order (first match wins): 1. Native S3 checksum from GetObjectAttributes (CRC64NVME preferred) @@ -19,14 +18,17 @@ --region us-west-2 --resume-from manifest.csv # resume an interrupted run (same file as --output) """ - from __future__ import annotations import argparse import csv import sys import time -from typing import Dict, Iterable, Optional, Set, Tuple +from typing import Dict +from typing import Iterable +from typing import Optional +from typing import Set +from typing import Tuple import boto3 from botocore.config import Config @@ -53,6 +55,7 @@ def parse_args() -> argparse.Namespace: + """Parse command-line arguments.""" parser = argparse.ArgumentParser() parser.add_argument("--bucket", required=True, help="S3 bucket name") parser.add_argument("--output", required=True, help="Output CSV path") @@ -67,16 +70,19 @@ def parse_args() -> argparse.Namespace: return parser.parse_args() -def build_session(profile: Optional[str], region: Optional[str]): - session_kwargs = {} +def build_session(profile: Optional[str], region: Optional[str]) -> boto3.Session: + """Create a boto3 Session with optional profile and region.""" + if profile and region: + return boto3.Session(profile_name=profile, region_name=region) if profile: - session_kwargs["profile_name"] = profile + return boto3.Session(profile_name=profile) if region: - session_kwargs["region_name"] = region - return boto3.Session(**session_kwargs) + return boto3.Session(region_name=region) + return boto3.Session() def load_existing_keys(csv_path: str) -> Set[str]: + """Load already-processed keys from an existing manifest CSV.""" keys: Set[str] = set() with open(csv_path, "r", newline="", encoding="utf-8") as f: reader = csv.DictReader(f) @@ -86,8 +92,8 @@ def load_existing_keys(csv_path: str) -> Set[str]: def choose_checksum(resp: Dict) -> Tuple[str, str, str]: - """ - Prefer CRC64NVME if present. Otherwise take the first available checksum. + """Prefer CRC64NVME if present, otherwise take first available checksum. + Returns: (algorithm, checksum_type, checksum_value) """ checksum = resp.get("Checksum", {}) or {} @@ -102,6 +108,7 @@ def choose_checksum(resp: Dict) -> Tuple[str, str, str]: def list_objects(s3_client, bucket: str, prefix: str) -> Iterable[Dict]: + """Paginate over all objects in a bucket matching the given prefix.""" paginator = s3_client.get_paginator("list_objects_v2") kwargs = {"Bucket": bucket} if prefix: @@ -113,6 +120,7 @@ def list_objects(s3_client, bucket: str, prefix: str) -> Iterable[Dict]: def get_object_attrs_with_retry(s3_client, bucket: str, key: str, retries: int = 5) -> Dict: + """Call GetObjectAttributes with exponential-backoff retry on throttle errors.""" delay = 1.0 for attempt in range(retries): try: @@ -128,9 +136,11 @@ def get_object_attrs_with_retry(s3_client, bucket: str, key: str, retries: int = delay *= 2 continue raise + raise RuntimeError(f"Exhausted {retries} retries for GetObjectAttributes on {key}") def head_object_with_retry(s3_client, bucket: str, key: str, retries: int = 5) -> Dict: + """Call HeadObject with exponential-backoff retry on throttle errors.""" delay = 1.0 for attempt in range(retries): try: @@ -142,6 +152,7 @@ def head_object_with_retry(s3_client, bucket: str, key: str, retries: int = 5) - delay *= 2 continue raise + raise RuntimeError(f"Exhausted {retries} retries for HeadObject on {key}") def parse_s3cmd_md5(attrs_value: str) -> str: @@ -159,6 +170,7 @@ def parse_s3cmd_md5(attrs_value: str) -> str: def main() -> int: + """Build a checksum manifest CSV for all objects in an S3 bucket.""" args = parse_args() session = build_session(args.profile, args.region) @@ -215,15 +227,17 @@ def main() -> int: checksum_type = "" checksum_value = f"ERROR:{err_code}" - writer.writerow([ - args.bucket, - key, - size, - algo, - checksum_type, - checksum_value, - etag, - ]) + writer.writerow( + [ + args.bucket, + key, + size, + algo, + checksum_type, + checksum_value, + etag, + ] + ) count += 1 if count % 1000 == 0: diff --git a/src/pds/pdc/data_integrity/compare_s3_manifests.py b/src/pds/pdc/data_integrity/compare_s3_manifests.py index 157be2e..ebf88a7 100644 --- a/src/pds/pdc/data_integrity/compare_s3_manifests.py +++ b/src/pds/pdc/data_integrity/compare_s3_manifests.py @@ -50,7 +50,6 @@ signature in NEW, or there are unverifiable rows beyond multipart ETags """ - from __future__ import annotations import argparse @@ -61,7 +60,10 @@ import re import sqlite3 import tempfile -from typing import Dict, Iterator, Optional, Tuple +from typing import Dict +from typing import Iterator +from typing import Optional +from typing import Tuple _MULTIPART_ETAG_RE = re.compile(r"^[0-9a-fA-F]+-\d+$") @@ -69,6 +71,7 @@ def parse_args() -> argparse.Namespace: + """Parse command-line arguments.""" parser = argparse.ArgumentParser( description="Compare two S3 checksum manifests to verify a bucket migration.", formatter_class=argparse.RawDescriptionHelpFormatter, @@ -227,6 +230,7 @@ def _build_new_index( def main() -> int: + """Compare two S3 checksum manifests and report missing or unverifiable objects.""" args = parse_args() with tempfile.NamedTemporaryFile(suffix=".db", delete=True) as tmp: @@ -340,11 +344,15 @@ def main() -> int: else: print("RESULT: FAIL") if missing_count: - print(f" {missing_count:,} object(s) from {old_bucket} have no matching content signature in {new_bucket}.") + print( + f" {missing_count:,} object(s) from {old_bucket} have no matching content signature in {new_bucket}." + ) if unverifiable_count: print(f" {unverifiable_count:,} object(s) from {old_bucket} could not be verified.") if multipart_etag_count: - print(f" {multipart_etag_count:,} have multipart ETags (ETag changes on re-upload; not a reliable match key).") + print( + f" {multipart_etag_count:,} have multipart ETags (ETag changes on re-upload; not a reliable match key)." + ) if unverifiable_count - multipart_etag_count: print(f" {unverifiable_count - multipart_etag_count:,} have no usable checksum at all.") diff --git a/src/pds/pdc/s3_download.py b/src/pds/pdc/s3_download.py index c149c87..9306a0a 100644 --- a/src/pds/pdc/s3_download.py +++ b/src/pds/pdc/s3_download.py @@ -1,5 +1,5 @@ #!/usr/bin/env python -"""Download files from an S3 bucket to a local directory. +r"""Download files from an S3 bucket to a local directory. This script uses the boto3 library to download objects from a specified S3 bucket that match a given prefix. Objects can optionally be filtered by their @@ -76,7 +76,11 @@ def main(): type=parse_datetime, default=None, metavar="DATETIME", - help="Only download objects last modified before or at this datetime (YYYY-MM-DD or YYYY-MM-DDTHH:MM:SS, UTC). Defaults to now if --start-datetime is provided.", + help=( + "Only download objects last modified before or at this datetime " + "(YYYY-MM-DD or YYYY-MM-DDTHH:MM:SS, UTC). " + "Defaults to now if --start-datetime is provided." + ), ) args = parser.parse_args() diff --git a/tox.ini b/tox.ini index a4ae16c..776458b 100644 --- a/tox.ini +++ b/tox.ini @@ -1,5 +1,5 @@ [tox] -envlist = py39, docs, lint +envlist = py312, docs, lint isolated_build = True [testenv] @@ -19,6 +19,6 @@ commands= skip_install = true [testenv:dev] -basepython = python3.9 +basepython = python3.12 usedevelop = True deps = .[dev]