Spaces:
Running
Running
File size: 2,612 Bytes
167c070 fc2fe9f 167c070 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 | """Cloudinary storage backend. Optional — only used when CLOUDINARY_* env vars are set."""
import io
import logging
import os
from typing import Optional
from .secrets import get_secret
logger = logging.getLogger("rga_auditor.cloudinary")
class CloudinaryStorage:
def __init__(self) -> None:
self._enabled = False
cloud_name = get_secret("CLOUDINARY_CLOUD_NAME") or os.getenv("CLOUDINARY_CLOUD_NAME")
api_key = get_secret("CLOUDINARY_API_KEY") or os.getenv("CLOUDINARY_API_KEY")
api_secret = get_secret("CLOUDINARY_API_SECRET") or os.getenv("CLOUDINARY_API_SECRET")
if not (cloud_name and api_key and api_secret):
logger.info("Cloudinary not configured — using local disk storage")
return
try:
import cloudinary
import cloudinary.uploader
cloudinary.config(cloud_name=cloud_name, api_key=api_key, api_secret=api_secret, secure=True)
self._uploader = cloudinary.uploader
self._enabled = True
logger.info("Cloudinary storage enabled (cloud=%s)", cloud_name)
except Exception as e:
logger.warning("Cloudinary init failed: %s", e)
@property
def is_enabled(self) -> bool:
return self._enabled
def upload(self, content: bytes, public_id: str, resource_type: str = "raw") -> dict:
if not self._enabled:
raise RuntimeError("Cloudinary not configured")
return self._uploader.upload(
io.BytesIO(content),
public_id=public_id,
resource_type=resource_type,
overwrite=True,
)
def get_url(self, public_id: str, resource_type: str = "raw") -> str:
if not self._enabled:
return ""
import cloudinary
return cloudinary.CloudinaryResource(public_id=public_id, resource_type=resource_type).build_url()
def get_signed_url(self, public_id: str, resource_type: str = "image", format: str = "pdf", version: str = None) -> str:
if not self._enabled:
return ""
from cloudinary.utils import cloudinary_url
opts = dict(
resource_type=resource_type,
type="upload",
format=format,
sign_url=True,
)
if version:
opts["version"] = version
url, _ = cloudinary_url(public_id, **opts)
return url
_storage: Optional[CloudinaryStorage] = None
def get_cloudinary() -> CloudinaryStorage:
global _storage
if _storage is None:
_storage = CloudinaryStorage()
return _storage
|