from __future__ import annotations

import json
import os
import subprocess
from datetime import datetime


def list_objects(s3_uri: str, env: dict[str, str] | None = None) -> list[dict]:
    """List objects under an S3 prefix as ``[{name, size, last_modified_ms}]`` via the aws
    CLI. ``name`` is the bucket-relative key; ``last_modified_ms`` is epoch milliseconds.
    Mirror of ``common.gcp.list_objects`` for the GCS side. (Single page / up to 1000 keys —
    fine for the external-incremental discovery use; paginate if a source grows past that.)"""
    rest = s3_uri.removeprefix("s3://")
    bucket, _, prefix = rest.partition("/")
    completed = subprocess.run(
        ["aws", "s3api", "list-objects-v2", "--bucket", bucket, "--prefix", prefix, "--output", "json"],
        check=True,
        capture_output=True,
        text=True,
        env=env,
    )
    payload = json.loads(completed.stdout or "{}")
    return [
        {"name": item["Key"], "size": int(item.get("Size", 0)), "last_modified_ms": _epoch_ms(item.get("LastModified"))}
        for item in payload.get("Contents", [])
    ]


def _epoch_ms(value: str | None) -> int | None:
    if not value:
        return None
    text = value.replace("Z", "+00:00")
    try:
        parsed = datetime.fromisoformat(text)
    except ValueError:
        parsed = datetime.strptime(value, "%Y-%m-%dT%H:%M:%S%z")
    return int(parsed.timestamp() * 1000)


def aws_env(config) -> dict[str, str]:
    env = os.environ.copy()
    # Only override when the source config or environment actually provides a value, so a
    # missing profile/region falls back to the operator's ambient AWS config rather than a
    # baked-in default (the values live in .env.example, not the engine).
    if config.aws_profile:
        env["AWS_PROFILE"] = config.aws_profile
    if config.aws_region:
        env["AWS_REGION"] = config.aws_region
    return env