upload_and_move.py

#!/usr/bin/env python3
import argparse
import os
import sys
from pathlib import Path
from typing import Iterator, Optional

from botocore.config import Config
from botocore.exceptions import BotoCoreError, ClientError
from boto3.session import Session


DEFAULT_EXCLUDES = {
    ".git", ".hg", ".svn",
    "__pycache__", ".pytest_cache", ".mypy_cache", ".ruff_cache",
    ".venv", "venv", "node_modules",
    ".DS_Store",
}


def iter_files_recursively(root: Path, exclude_dirs: set[str]) -> Iterator[Path]:
    """
    Recursively iterates all files under `root`, excluding directories.
    Returns absolute file paths.
    """
    for p in root.rglob("*"):
        if p.is_dir():
            continue
        # Exclude by directory name at any depth
        if any(part in exclude_dirs for part in p.parts):
            continue
        yield p


def build_s3_client(
    region: Optional[str],
    endpoint_url: Optional[str],
    access_key: Optional[str],
    secret_key: Optional[str],
    session_token: Optional[str],
):
    session = Session(
        aws_access_key_id=access_key,
        aws_secret_access_key=secret_key,
        aws_session_token=session_token,
        region_name=region,
    )

    cfg = Config(
        retries={"max_attempts": 10, "mode": "standard"},
        connect_timeout=5,
        read_timeout=60,
    )

    return session.client("s3", endpoint_url=endpoint_url, config=cfg)


def normalize_prefix(prefix: str) -> str:
    if not prefix:
        return ""
    prefix = prefix.lstrip("/")
    if not prefix.endswith("/"):
        prefix += "/"
    return prefix


def s3_key_for_file(base_dir: Path, file_path: Path, prefix: str) -> str:
    rel = file_path.relative_to(base_dir).as_posix()
    return f"{prefix}{rel}"


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description=(
            "Recursively uploads files from a directory (resolved from the current working directory) to S3."
        )
    )

    parser.add_argument("--bucket", required=True, help="Destination bucket name (required).")
    parser.add_argument("--prefix", default="", help="Bucket prefix (folder). Example: uploads/")

    # Source directory is parameterized and resolved relative to the CWD
    parser.add_argument(
        "--source-dir",
        default=".",
        help="Source directory to scan (relative to where the command is executed). Default: '.'",
    )

    parser.add_argument(
        "--endpoint-url",
        default=None,
        help="S3 endpoint URL (e.g., http://localstack:4566). If omitted, boto3 uses AWS defaults.",
    )
    parser.add_argument("--region", default=None, help="AWS region. If omitted, boto3/env defaults are used.")
    parser.add_argument("--access-key", default=None, help="AWS_ACCESS_KEY_ID (optional; otherwise taken from env).")
    parser.add_argument("--secret-key", default=None, help="AWS_SECRET_ACCESS_KEY (optional; otherwise taken from env).")
    parser.add_argument("--session-token", default=None, help="AWS_SESSION_TOKEN (optional).")

    parser.add_argument(
        "--exclude-dir",
        action="append",
        default=[],
        help="Directory name to exclude. Repeatable. Example: --exclude-dir .git --exclude-dir node_modules",
    )

    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Do not upload; only print what would be uploaded.",
    )

    return parser.parse_args()


def resolve_source_dir(source_dir_arg: str) -> Path:
    """
    Resolves the source directory relative to the current working directory (CWD).
    If an absolute path is provided, it is used as-is.
    """
    p = Path(source_dir_arg)
    if p.is_absolute():
        return p.resolve()
    return (Path.cwd() / p).resolve()


def main() -> int:
    args = parse_args()

    cwd = Path.cwd().resolve()
    base_dir = resolve_source_dir(args.source_dir)

    if not base_dir.exists() or not base_dir.is_dir():
        print(f"[ERROR] Invalid --source-dir: {base_dir}", file=sys.stderr)
        return 2

    prefix = normalize_prefix(args.prefix)

    exclude_dirs = set(DEFAULT_EXCLUDES)
    exclude_dirs.update(args.exclude_dir)

    endpoint_url = args.endpoint_url or os.getenv("AWS_ENDPOINT_URL")

    s3 = build_s3_client(
        region=args.region or os.getenv("AWS_DEFAULT_REGION"),
        endpoint_url=endpoint_url,
        access_key=args.access_key or os.getenv("AWS_ACCESS_KEY_ID"),
        secret_key=args.secret_key or os.getenv("AWS_SECRET_ACCESS_KEY"),
        session_token=args.session_token or os.getenv("AWS_SESSION_TOKEN"),
    )

    files = list(iter_files_recursively(base_dir, exclude_dirs))

    if not files:
        print("[INFO] No files found to upload.")
        print(f"[INFO] CWD: {cwd}")
        print(f"[INFO] Source dir: {base_dir}")
        return 0

    print(f"[INFO] CWD: {cwd}")
    print(f"[INFO] Source dir: {base_dir}")
    print(f"[INFO] Bucket: {args.bucket}")
    print(f"[INFO] Prefix: {prefix or '(empty)'}")
    print(f"[INFO] Endpoint: {endpoint_url or '(AWS default)'}")
    print(f"[INFO] Files found: {len(files)}")

    if args.dry_run:
        for f in files:
            key = s3_key_for_file(base_dir, f, prefix)
            print(f"[DRY-RUN] Upload {f} -> s3://{args.bucket}/{key}")
        return 0

    uploaded = 0
    failed = 0

    for f in files:
        key = s3_key_for_file(base_dir, f, prefix)
        try:
            s3.upload_file(str(f), args.bucket, key)
            uploaded += 1
            print(f"[OK] {f.relative_to(base_dir)} -> s3://{args.bucket}/{key}")
        except (ClientError, BotoCoreError) as e:
            failed += 1
            print(f"[FAIL] {f.relative_to(base_dir)} -> s3://{args.bucket}/{key} | {e}", file=sys.stderr)

    print(f"[SUMMARY] uploaded={uploaded} failed={failed}")
    return 0 if failed == 0 else 1


if __name__ == "__main__":
    raise SystemExit(main())