79 lines
2.6 KiB
Python
79 lines
2.6 KiB
Python
"""Configuration management for OBS R2 Uploader."""
|
|
|
|
import os
|
|
from typing import Optional
|
|
from pathlib import Path
|
|
from dotenv import load_dotenv
|
|
|
|
|
|
class Config:
|
|
"""Configuration manager for R2 upload settings."""
|
|
|
|
def __init__(self, env_path: Optional[str] = None):
|
|
"""Initialize configuration from environment variables.
|
|
|
|
Args:
|
|
env_path: Optional path to .env file. If None, searches in current and parent dirs.
|
|
"""
|
|
if env_path:
|
|
load_dotenv(env_path)
|
|
else:
|
|
# Search for .env in current directory and parent directories
|
|
current_dir = Path.cwd()
|
|
for parent in [current_dir] + list(current_dir.parents):
|
|
env_file = parent / ".env"
|
|
if env_file.exists():
|
|
load_dotenv(env_file)
|
|
break
|
|
|
|
# Required settings
|
|
self.account_id = os.getenv("R2_ACCOUNT_ID")
|
|
self.access_key_id = os.getenv("R2_ACCESS_KEY_ID")
|
|
self.secret_access_key = os.getenv("R2_SECRET_ACCESS_KEY")
|
|
self.bucket_name = os.getenv("R2_BUCKET_NAME", "obs-videos")
|
|
|
|
# Construct endpoint URL
|
|
if self.account_id:
|
|
self.endpoint = os.getenv(
|
|
"R2_ENDPOINT",
|
|
f"https://{self.account_id}.r2.cloudflarestorage.com"
|
|
)
|
|
else:
|
|
self.endpoint = os.getenv("R2_ENDPOINT")
|
|
|
|
# Public domain for sharing
|
|
self.public_domain = os.getenv("PUBLIC_DOMAIN", "videos.jeffemmett.com")
|
|
|
|
# Optional settings
|
|
self.obs_recording_dir = os.getenv("OBS_RECORDING_DIR")
|
|
self.auto_delete = os.getenv("AUTO_DELETE_AFTER_UPLOAD", "false").lower() == "true"
|
|
|
|
def validate(self) -> tuple[bool, list[str]]:
|
|
"""Validate that all required configuration is present.
|
|
|
|
Returns:
|
|
Tuple of (is_valid, list of missing settings)
|
|
"""
|
|
required = {
|
|
"R2_ACCOUNT_ID": self.account_id,
|
|
"R2_ACCESS_KEY_ID": self.access_key_id,
|
|
"R2_SECRET_ACCESS_KEY": self.secret_access_key,
|
|
"R2_ENDPOINT": self.endpoint,
|
|
}
|
|
|
|
missing = [key for key, value in required.items() if not value]
|
|
return (len(missing) == 0, missing)
|
|
|
|
def get_public_url(self, filename: str) -> str:
|
|
"""Generate public URL for uploaded file.
|
|
|
|
Args:
|
|
filename: Name of the uploaded file
|
|
|
|
Returns:
|
|
Full public URL
|
|
"""
|
|
# Remove protocol if present in public_domain
|
|
domain = self.public_domain.replace("https://", "").replace("http://", "")
|
|
return f"https://{domain}/{filename}"
|