mailova/app/storage.py

"""Image storage abstraction: local filesystem or S3-compatible object storage.

All system configuration lives in ``config.json`` in the project root::

    {
        "storage": {
            "backend": "local",
            "s3": {
                "endpoint": "https://s3.amazonaws.com",
                "bucket": "my-mailova-bucket",
                "region": "auto",
                "access_key": "AKIA...",
                "secret_key": "...",
                "public_url": "https://cdn.example.com/mailova"
            }
        }
    }

``storage.backend``: ``"local"`` (default) or ``"s3"``.

URL scheme
~~~~~~~~~~
Database stores **bare filenames** (``xxx.jpeg``) regardless of backend.
``resolve_url(key)`` converts a bare key to a full URL:
  - S3  → ``<public_url>/xxx.jpeg``
  - local → ``/uploads/xxx.jpeg``

Legacy records that still contain ``/uploads/xxx.jpeg`` are auto-stripped
by ``resolve_url``.
"""

from __future__ import annotations

import json
from pathlib import Path
from typing import Protocol

from app.config import UPLOAD_DIR

CONFIG_PATH = Path(__file__).resolve().parent.parent / "config.json"


class StorageBackend(Protocol):
    def put(self, key: str, data: bytes, content_type: str = "image/jpeg") -> str:
        """Store data and return the bare key."""
        ...

    def delete(self, key: str) -> None: ...

    def exists(self, key: str) -> bool: ...

    def resolve_url(self, key: str) -> str:
        """Convert a bare key (or legacy /uploads/xxx path) to a full URL."""
        ...


# ── Local Storage ──────────────────────────────────────────────────

class LocalStorage:
    def __init__(self) -> None:
        UPLOAD_DIR.mkdir(parents=True, exist_ok=True)

    def put(self, key: str, data: bytes, content_type: str = "") -> str:
        (UPLOAD_DIR / key).write_bytes(data)
        return key

    def delete(self, key: str) -> None:
        p = UPLOAD_DIR / key
        if p.exists():
            p.unlink()

    def exists(self, key: str) -> bool:
        return (UPLOAD_DIR / key).exists()

    def resolve_url(self, key: str) -> str:
        return f"/uploads/{_strip_prefix(key)}"


# ── S3 Storage ─────────────────────────────────────────────────────

class S3Storage:
    def __init__(
        self,
        endpoint_url: str,
        bucket: str,
        access_key: str,
        secret_key: str,
        region: str = "auto",
        public_url: str = "",
    ) -> None:
        import boto3

        self.bucket = bucket
        self.public_url = public_url.rstrip("/")
        self._client = boto3.client(
            "s3",
            endpoint_url=endpoint_url or None,
            aws_access_key_id=access_key,
            aws_secret_access_key=secret_key,
            region_name=region or "auto",
        )

    def put(self, key: str, data: bytes, content_type: str = "image/jpeg") -> str:
        self._client.put_object(
            Bucket=self.bucket, Key=key, Body=data, ContentType=content_type,
            CacheControl="public, max-age=2592000, immutable",
        )
        return key

    def delete(self, key: str) -> None:
        self._client.delete_object(Bucket=self.bucket, Key=key)

    def exists(self, key: str) -> bool:
        try:
            self._client.head_object(Bucket=self.bucket, Key=key)
            return True
        except Exception:
            return False

    def resolve_url(self, key: str) -> str:
        k = _strip_prefix(key)
        if self.public_url:
            return f"{self.public_url}/{k}"
        # private bucket: generate presigned URL (7 days expiry)
        return self._client.generate_presigned_url(
            "get_object",
            Params={"Bucket": self.bucket, "Key": k},
            ExpiresIn=86400,
        )


# ── Helpers ────────────────────────────────────────────────────────

def _strip_prefix(key: str) -> str:
    """Strip ``/uploads/`` prefix from legacy DB values."""
    if key.startswith("/uploads/"):
        return key[len("/uploads/"):]
    return key


# ── Config loading ─────────────────────────────────────────────────

def load_config() -> dict:
    """Load config.json. Returns empty dict if missing."""
    if not CONFIG_PATH.exists():
        return {}
    try:
        return json.loads(CONFIG_PATH.read_text(encoding="utf-8"))
    except Exception:
        return {}


# ── Singleton accessor ─────────────────────────────────────────────

_backend: StorageBackend | None = None


def get_storage() -> StorageBackend:
    """Get the active storage backend based on config.json."""
    global _backend
    if _backend is not None:
        return _backend

    cfg = load_config()
    backend = cfg.get("storage", {}).get("backend", "local")

    if backend == "s3":
        s3 = cfg.get("storage", {}).get("s3", {})
        if not s3.get("bucket"):
            # fallback to local if S3 is misconfigured
            _backend = LocalStorage()
        else:
            _backend = S3Storage(
                endpoint_url=s3.get("endpoint", ""),
                bucket=s3["bucket"],
                access_key=s3.get("access_key", ""),
                secret_key=s3.get("secret_key", ""),
                region=s3.get("region", "auto"),
                public_url=s3.get("public_url", ""),
            )
    else:
        _backend = LocalStorage()
    return _backend


def resolve_url(key: str | None) -> str:
    """Convenience: resolve a possibly-None image key to a URL (or empty)."""
    if not key:
        return ""
    return get_storage().resolve_url(key)


def reset_storage() -> None:
    """Force re-detection on next get_storage() call."""
    global _backend
    _backend = None