plantillas-proyectos/backend/core/storage_s3.py

"""
Cliente S3 (MinIO): bucket, objetos genéricos, presign, imports CSV.

Las claves de objeto deben generarse con ``core.s3_keys`` (p. ej. ``csv_import_key`` vía
``s3_key_for_csv_import``); no construir prefijos ``tenants/...`` aquí.
"""
import logging
from typing import Any, Dict, List, Optional

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

from core.config import settings
from core.s3_keys import csv_import_key

logger = logging.getLogger(__name__)


def _client():
    return boto3.client(
        "s3",
        endpoint_url=settings.S3_ENDPOINT_URL,
        aws_access_key_id=settings.S3_ACCESS_KEY,
        aws_secret_access_key=settings.S3_SECRET_KEY,
        region_name=settings.S3_REGION,
        use_ssl=settings.S3_USE_SSL,
        config=Config(
            signature_version="s3v4",
            s3={"addressing_style": "path"},
        ),
    )


def should_ensure_s3_bucket() -> bool:
    return settings.use_s3_object_storage


def ensure_s3_bucket() -> None:
    """Crea el bucket si no existe (idempotente)."""
    if not should_ensure_s3_bucket():
        return
    bucket = settings.S3_BUCKET
    client = _client()
    try:
        client.head_bucket(Bucket=bucket)
        logger.info("S3 bucket %s exists", bucket)
        return
    except ClientError as e:
        code = e.response.get("Error", {}).get("Code", "")
        if code not in ("404", "NoSuchBucket", "403"):
            logger.warning("head_bucket %s: %s", bucket, e)
    try:
        if settings.S3_REGION == "us-east-1":
            client.create_bucket(Bucket=bucket)
        else:
            client.create_bucket(
                Bucket=bucket,
                CreateBucketConfiguration={"LocationConstraint": settings.S3_REGION},
            )
        logger.info("S3 bucket %s created", bucket)
    except ClientError as e:
        logger.error("create_bucket %s failed: %s", bucket, e)
        raise


# Alias para código existente
def ensure_csv_import_bucket() -> None:
    ensure_s3_bucket()


def put_object_bytes(key: str, body: bytes, content_type: str = "application/octet-stream") -> None:
    _client().put_object(
        Bucket=settings.S3_BUCKET,
        Key=key,
        Body=body,
        ContentType=content_type,
    )


def put_csv_object(key: str, body: bytes, content_type: str = "text/csv") -> None:
    put_object_bytes(key, body, content_type=content_type)


def get_object_bytes(key: str) -> bytes:
    resp = _client().get_object(Bucket=settings.S3_BUCKET, Key=key)
    return resp["Body"].read()


def delete_object_if_exists(key: str) -> None:
    try:
        _client().delete_object(Bucket=settings.S3_BUCKET, Key=key)
    except ClientError as e:
        logger.warning("delete_object %s: %s", key, e)


def object_exists(key: str) -> bool:
    try:
        _client().head_object(Bucket=settings.S3_BUCKET, Key=key)
        return True
    except ClientError:
        return False


def presigned_get_url(key: str, expires_in: Optional[int] = None) -> str:
    sec = expires_in if expires_in is not None else settings.S3_PRESIGNED_EXPIRES_SECONDS
    return _client().generate_presigned_url(
        "get_object",
        Params={"Bucket": settings.S3_BUCKET, "Key": key},
        ExpiresIn=sec,
    )


def list_objects_tree(
    prefix: str,
    delimiter: str = "/",
    max_keys: int = 100,
    continuation_token: Optional[str] = None,
) -> Dict[str, Any]:
    """
    Lista objetos/prefijos como árbol virtual.

    Retorna:
      - ``prefixes``: subcarpetas (CommonPrefixes)
      - ``objects``: objetos directos bajo ``prefix``
      - ``next_continuation_token`` y ``is_truncated`` para paginación
    """
    params: Dict[str, Any] = {
        "Bucket": settings.S3_BUCKET,
        "Prefix": prefix,
        "Delimiter": delimiter,
        "MaxKeys": max(1, min(int(max_keys), 500)),
    }
    if continuation_token:
        params["ContinuationToken"] = continuation_token

    resp = _client().list_objects_v2(**params)
    common_prefixes: List[str] = [
        p.get("Prefix", "") for p in (resp.get("CommonPrefixes") or []) if p.get("Prefix")
    ]
    objects: List[Dict[str, Any]] = []
    for obj in resp.get("Contents") or []:
        key = obj.get("Key")
        if not key:
            continue
        if key == prefix:
            # Marcador de carpeta (objeto vacío con mismo nombre del prefijo).
            continue
        objects.append(
            {
                "key": key,
                "size": int(obj.get("Size", 0) or 0),
                "last_modified": obj.get("LastModified"),
                "etag": obj.get("ETag"),
                "storage_class": obj.get("StorageClass"),
            }
        )

    return {
        "prefixes": common_prefixes,
        "objects": objects,
        "next_continuation_token": resp.get("NextContinuationToken"),
        "is_truncated": bool(resp.get("IsTruncated")),
    }


def s3_key_for_csv_import(
    tenant_id,
    company_id: int,
    job_type: str,
    job_id: str,
) -> str:
    return csv_import_key(tenant_id, company_id, job_type, job_id)