f8cd37ac8c
Includes: API (Express/TypeScript), Astro site, Python workers, document generators, FCC compliance tools, Canada CRTC formation, Ansible infrastructure, and deployment scripts. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
128 lines
4.3 KiB
Python
128 lines
4.3 KiB
Python
"""
|
|
MinIO (S3-compatible) client for document storage.
|
|
|
|
Uploads generated documents to MinIO and returns accessible URLs.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
from pathlib import Path
|
|
|
|
from minio import Minio
|
|
from minio.error import S3Error
|
|
|
|
LOG = logging.getLogger("document_gen.minio")
|
|
|
|
BUCKET = os.getenv("MINIO_BUCKET", "performancewest")
|
|
|
|
|
|
class MinioStorage:
|
|
"""S3-compatible document storage via MinIO."""
|
|
|
|
def __init__(self):
|
|
self.client = Minio(
|
|
endpoint=f"{os.getenv('MINIO_ENDPOINT', 'localhost')}:{os.getenv('MINIO_PORT', '9000')}",
|
|
access_key=os.getenv("MINIO_ACCESS_KEY", ""),
|
|
secret_key=os.getenv("MINIO_SECRET_KEY", ""),
|
|
secure=os.getenv("MINIO_SECURE", "false").lower() == "true",
|
|
)
|
|
self._ensure_bucket()
|
|
|
|
def _ensure_bucket(self):
|
|
"""Create the bucket if it doesn't exist."""
|
|
try:
|
|
if not self.client.bucket_exists(BUCKET):
|
|
self.client.make_bucket(BUCKET)
|
|
LOG.info("Created MinIO bucket: %s", BUCKET)
|
|
except S3Error as e:
|
|
LOG.error("MinIO bucket check failed: %s", e)
|
|
|
|
def upload(
|
|
self,
|
|
local_path: str | Path,
|
|
remote_path: str,
|
|
content_type: str = "application/octet-stream",
|
|
) -> str:
|
|
"""Upload a file to MinIO.
|
|
|
|
Args:
|
|
local_path: Local file path
|
|
remote_path: Object key in the bucket (e.g., "formations/PW-2026-XXXX/articles.pdf")
|
|
content_type: MIME type
|
|
|
|
Returns:
|
|
The object URL (internal MinIO URL)
|
|
"""
|
|
local_path = Path(local_path)
|
|
if not local_path.exists():
|
|
raise FileNotFoundError(f"File not found: {local_path}")
|
|
|
|
# Auto-detect content type
|
|
if content_type == "application/octet-stream":
|
|
suffix = local_path.suffix.lower()
|
|
content_types = {
|
|
".pdf": "application/pdf",
|
|
".docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
|
".doc": "application/msword",
|
|
".html": "text/html",
|
|
".txt": "text/plain",
|
|
".png": "image/png",
|
|
".jpg": "image/jpeg",
|
|
}
|
|
content_type = content_types.get(suffix, content_type)
|
|
|
|
try:
|
|
self.client.fput_object(
|
|
BUCKET,
|
|
remote_path,
|
|
str(local_path),
|
|
content_type=content_type,
|
|
)
|
|
LOG.info("Uploaded: %s → %s/%s", local_path.name, BUCKET, remote_path)
|
|
return f"{BUCKET}/{remote_path}"
|
|
except S3Error as e:
|
|
LOG.error("MinIO upload failed: %s", e)
|
|
raise
|
|
|
|
def download(self, remote_path: str, local_path: str | Path) -> Path:
|
|
"""Download a file from MinIO."""
|
|
local_path = Path(local_path)
|
|
local_path.parent.mkdir(parents=True, exist_ok=True)
|
|
try:
|
|
self.client.fget_object(BUCKET, remote_path, str(local_path))
|
|
LOG.info("Downloaded: %s/%s → %s", BUCKET, remote_path, local_path)
|
|
return local_path
|
|
except S3Error as e:
|
|
LOG.error("MinIO download failed: %s", e)
|
|
raise
|
|
|
|
def get_url(self, remote_path: str, expires_hours: int = 24) -> str:
|
|
"""Get a presigned URL for a file (for client download)."""
|
|
from datetime import timedelta
|
|
try:
|
|
url = self.client.presigned_get_object(
|
|
BUCKET, remote_path, expires=timedelta(hours=expires_hours),
|
|
)
|
|
return url
|
|
except S3Error as e:
|
|
LOG.error("MinIO presign failed: %s", e)
|
|
raise
|
|
|
|
def list_objects(self, prefix: str) -> list[str]:
|
|
"""List all objects under a prefix."""
|
|
try:
|
|
objects = self.client.list_objects(BUCKET, prefix=prefix, recursive=True)
|
|
return [obj.object_name for obj in objects]
|
|
except S3Error as e:
|
|
LOG.error("MinIO list failed: %s", e)
|
|
return []
|
|
|
|
def delete(self, remote_path: str):
|
|
"""Delete an object from MinIO."""
|
|
try:
|
|
self.client.remove_object(BUCKET, remote_path)
|
|
LOG.info("Deleted: %s/%s", BUCKET, remote_path)
|
|
except S3Error as e:
|
|
LOG.error("MinIO delete failed: %s", e)
|