#!/usr/bin/env python3
"""
R2 smoke test (S3-compatible).

This script intentionally NEVER prints secret values. It is safe to run in CI/CD logs.

Required env vars:
- R2_ENDPOINT_URL
- R2_BUCKET
- R2_ACCESS_KEY_ID (or R2_AccessID)
- R2_SECRET_ACCESS_KEY (or R2_Secret_Access_Key)
"""

from __future__ import annotations

import os
from typing import Optional

import boto3
from botocore.config import Config as BotoConfig
from botocore.exceptions import ClientError


def _env(name: str) -> str:
    return os.environ.get(name, "").strip()


def _env_required(name: str) -> str:
    val = _env(name)
    if not val:
        raise RuntimeError(f"Missing env var: {name}")
    return val


def main() -> None:
    endpoint = _env_required("R2_ENDPOINT_URL")
    bucket = _env_required("R2_BUCKET")

    access_key = _env("R2_ACCESS_KEY_ID") or _env("R2_AccessID")
    secret_key = _env("R2_SECRET_ACCESS_KEY") or _env("R2_Secret_Access_Key")
    if not access_key:
        raise RuntimeError("Missing env var: R2_ACCESS_KEY_ID (or R2_AccessID)")
    if not secret_key:
        raise RuntimeError("Missing env var: R2_SECRET_ACCESS_KEY (or R2_Secret_Access_Key)")

    prefix = _env("R2_PREFIX")

    print("=== R2 SMOKE TEST ===")
    print("Endpoint:", endpoint)
    print("Bucket:", bucket)
    if prefix:
        print("Prefix:", prefix)
    print("Access key present:", True)
    print("Secret key present:", True)

    s3 = boto3.client(
        "s3",
        endpoint_url=endpoint,
        aws_access_key_id=access_key,
        aws_secret_access_key=secret_key,
        region_name="auto",
        config=BotoConfig(signature_version="s3v4"),
    )

    # 1) List buckets (verifies auth + endpoint)
    resp = s3.list_buckets()
    buckets = [b["Name"] for b in resp.get("Buckets", [])]
    print("\nBuckets:")
    for b in buckets:
        print(" -", b)

    # 2) Check target bucket exists (R2 can still allow access even if list is limited)
    if bucket not in buckets:
        print("\n(note) Target bucket not present in list_buckets output; continuing anyway (some policies hide it).")

    # 3) List first few objects (optional)
    list_kwargs = {"Bucket": bucket, "MaxKeys": 10}
    if prefix:
        list_kwargs["Prefix"] = prefix.strip("/") + "/"

    try:
        resp2 = s3.list_objects_v2(**list_kwargs)
        keys = [o["Key"] for o in resp2.get("Contents", [])]
        print("\nSample objects:")
        if not keys:
            print(" (none)")
        else:
            for k in keys:
                print(" -", k)
    except ClientError as e:
        print("\nlist_objects_v2 failed (bucket/policy?):", e.response.get("Error", {}).get("Code"))

    print("\n✅ R2 smoke test complete")


if __name__ == "__main__":
    main()