@source: 260313 heath-infer-step01

"""
@source: 260313 heath-infer-step01
@extracted: 2026-03-14
@description: AWS/GCP/Local 시크릿 관리 통합 추상화 계층. 환경에 따라 Provider 자동 선택.

사용법:
    secrets = get_secrets_manager()
    api_key = await secrets.get_secret('openai-api-key')

환경 변수:
    SECRETS_PROVIDER: 'aws' | 'gcp' | 'local' (기본: local)
    SECRETS_REGION: AWS/GCP 리전 (예: ap-northeast-2)
    SECRETS_PROJECT: GCP 프로젝트 ID
    SECRETS_ENCRYPTION_KEY: 로컬 암호화 키 (Fernet 키)
    SECRETS_LOCAL_PATH: 로컬 비밀 저장 경로 (기본: ~/.app-secrets)
    SECRETS_CACHE: 캐시 사용 여부 (기본: true)

의존성:
    pip install cryptography  (로컬 암호화)
    pip install boto3         (AWS)
    pip install google-cloud-secret-manager  (GCP)
"""
import base64
import json
import os
from abc import ABC, abstractmethod
from pathlib import Path
from typing import Any, Optional

# 암호화 라이브러리 (선택적)
try:
    from cryptography.fernet import Fernet
    from cryptography.hazmat.primitives import hashes
    from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
    CRYPTO_AVAILABLE = True
except ImportError:
    CRYPTO_AVAILABLE = False
    print("[WARN] cryptography 미설치 - 로컬 암호화 비활성화")


class SecretsProvider(ABC):
    """Secrets Manager 추상 인터페이스"""

    @abstractmethod
    async def get_secret(self, secret_id: str) -> Optional[str]:
        """비밀값 조회"""
        pass

    @abstractmethod
    async def set_secret(self, secret_id: str, value: str) -> bool:
        """비밀값 저장"""
        pass

    @abstractmethod
    async def delete_secret(self, secret_id: str) -> bool:
        """비밀값 삭제"""
        pass

    @abstractmethod
    async def list_secrets(self) -> list[str]:
        """비밀값 목록 조회"""
        pass


class LocalSecretsProvider(SecretsProvider):
    """
    로컬 Secrets Provider

    개발 환경용. 암호화된 JSON 파일에 비밀값을 저장합니다.
    """

    def __init__(self):
        # 비밀 저장 경로
        self._secrets_dir = Path(os.environ.get(
            'SECRETS_LOCAL_PATH',
            Path.home() / '.app-secrets'
        ))
        self._secrets_file = self._secrets_dir / 'secrets.enc'

        # 암호화 키
        self._fernet: Optional[Any] = None
        if CRYPTO_AVAILABLE:
            key = os.environ.get('SECRETS_ENCRYPTION_KEY')
            if key:
                try:
                    self._fernet = Fernet(key.encode())
                except Exception:
                    print("[WARN] 잘못된 암호화 키 - 평문 저장")
            else:
                print("[INFO] SECRETS_ENCRYPTION_KEY 미설정 - 평문 저장")

        # 디렉토리 생성
        self._secrets_dir.mkdir(parents=True, exist_ok=True)

    def _load_secrets(self) -> dict[str, str]:
        """저장된 비밀값 로드"""
        if not self._secrets_file.exists():
            return {}

        try:
            data = self._secrets_file.read_bytes()

            # 복호화
            if self._fernet:
                data = self._fernet.decrypt(data)

            return json.loads(data.decode('utf-8'))
        except Exception as e:
            print(f"[ERROR] 비밀값 로드 실패: {e}")
            return {}

    def _save_secrets(self, secrets: dict[str, str]) -> bool:
        """비밀값 저장"""
        try:
            data = json.dumps(secrets, ensure_ascii=False).encode('utf-8')

            # 암호화
            if self._fernet:
                data = self._fernet.encrypt(data)

            self._secrets_file.write_bytes(data)
            return True
        except Exception as e:
            print(f"[ERROR] 비밀값 저장 실패: {e}")
            return False

    async def get_secret(self, secret_id: str) -> Optional[str]:
        # 먼저 환경 변수 확인
        env_key = secret_id.upper().replace('-', '_')
        env_value = os.environ.get(env_key)
        if env_value:
            return env_value

        # 로컬 저장소 확인
        secrets = self._load_secrets()
        return secrets.get(secret_id)

    async def set_secret(self, secret_id: str, value: str) -> bool:
        secrets = self._load_secrets()
        secrets[secret_id] = value
        return self._save_secrets(secrets)

    async def delete_secret(self, secret_id: str) -> bool:
        secrets = self._load_secrets()
        if secret_id in secrets:
            del secrets[secret_id]
            return self._save_secrets(secrets)
        return False

    async def list_secrets(self) -> list[str]:
        secrets = self._load_secrets()
        return list(secrets.keys())


class AWSSecretsProvider(SecretsProvider):
    """
    AWS Secrets Manager Provider

    의존성: pip install boto3
    """

    def __init__(self):
        self._region = os.environ.get('SECRETS_REGION', 'ap-northeast-2')
        self._client = None

    def _get_client(self):
        if self._client is None:
            try:
                import boto3
                self._client = boto3.client(
                    'secretsmanager',
                    region_name=self._region
                )
            except ImportError:
                raise ImportError("boto3 미설치. pip install boto3")
        return self._client

    async def get_secret(self, secret_id: str) -> Optional[str]:
        try:
            client = self._get_client()
            response = client.get_secret_value(SecretId=secret_id)
            return response.get('SecretString')
        except Exception as e:
            print(f"[ERROR] AWS Secret 조회 실패: {secret_id} - {e}")
            return None

    async def set_secret(self, secret_id: str, value: str) -> bool:
        try:
            client = self._get_client()
            try:
                client.create_secret(Name=secret_id, SecretString=value)
            except client.exceptions.ResourceExistsException:
                client.update_secret(SecretId=secret_id, SecretString=value)
            return True
        except Exception as e:
            print(f"[ERROR] AWS Secret 저장 실패: {secret_id} - {e}")
            return False

    async def delete_secret(self, secret_id: str) -> bool:
        try:
            client = self._get_client()
            client.delete_secret(SecretId=secret_id, ForceDeleteWithoutRecovery=True)
            return True
        except Exception as e:
            print(f"[ERROR] AWS Secret 삭제 실패: {secret_id} - {e}")
            return False

    async def list_secrets(self) -> list[str]:
        try:
            client = self._get_client()
            response = client.list_secrets()
            return [s['Name'] for s in response.get('SecretList', [])]
        except Exception as e:
            print(f"[ERROR] AWS Secret 목록 조회 실패: {e}")
            return []


class GCPSecretsProvider(SecretsProvider):
    """
    GCP Secret Manager Provider

    의존성: pip install google-cloud-secret-manager
    """

    def __init__(self):
        self._project = os.environ.get('SECRETS_PROJECT')
        if not self._project:
            raise ValueError("SECRETS_PROJECT 환경 변수 필요")
        self._client = None

    def _get_client(self):
        if self._client is None:
            try:
                from google.cloud import secretmanager
                self._client = secretmanager.SecretManagerServiceClient()
            except ImportError:
                raise ImportError("google-cloud-secret-manager 미설치")
        return self._client

    async def get_secret(self, secret_id: str) -> Optional[str]:
        try:
            client = self._get_client()
            name = f"projects/{self._project}/secrets/{secret_id}/versions/latest"
            response = client.access_secret_version(request={"name": name})
            return response.payload.data.decode("utf-8")
        except Exception as e:
            print(f"[ERROR] GCP Secret 조회 실패: {secret_id} - {e}")
            return None

    async def set_secret(self, secret_id: str, value: str) -> bool:
        try:
            client = self._get_client()
            parent = f"projects/{self._project}"

            # 시크릿 생성 또는 버전 추가
            try:
                client.create_secret(
                    request={
                        "parent": parent,
                        "secret_id": secret_id,
                        "secret": {"replication": {"automatic": {}}},
                    }
                )
            except Exception:
                pass  # 이미 존재

            # 버전 추가
            secret_path = f"{parent}/secrets/{secret_id}"
            client.add_secret_version(
                request={
                    "parent": secret_path,
                    "payload": {"data": value.encode("utf-8")},
                }
            )
            return True
        except Exception as e:
            print(f"[ERROR] GCP Secret 저장 실패: {secret_id} - {e}")
            return False

    async def delete_secret(self, secret_id: str) -> bool:
        try:
            client = self._get_client()
            name = f"projects/{self._project}/secrets/{secret_id}"
            client.delete_secret(request={"name": name})
            return True
        except Exception as e:
            print(f"[ERROR] GCP Secret 삭제 실패: {secret_id} - {e}")
            return False

    async def list_secrets(self) -> list[str]:
        try:
            client = self._get_client()
            parent = f"projects/{self._project}"
            secrets = client.list_secrets(request={"parent": parent})
            return [s.name.split('/')[-1] for s in secrets]
        except Exception as e:
            print(f"[ERROR] GCP Secret 목록 조회 실패: {e}")
            return []


class SecretsManager:
    """
    Secrets Manager 통합 인터페이스

    환경에 따라 적절한 Provider를 선택합니다.
    """

    def __init__(self):
        provider_type = os.environ.get('SECRETS_PROVIDER', 'local').lower()

        if provider_type == 'aws':
            self._provider = AWSSecretsProvider()
        elif provider_type == 'gcp':
            self._provider = GCPSecretsProvider()
        elif provider_type == 'local':
            self._provider = LocalSecretsProvider()
        else:
            print(f"[WARN] 알 수 없는 Provider: {provider_type}, local 사용")
            self._provider = LocalSecretsProvider()

        self._cache: dict[str, str] = {}
        self._cache_enabled = os.environ.get('SECRETS_CACHE', 'true').lower() == 'true'

    async def get_secret(self, secret_id: str, use_cache: bool = True) -> Optional[str]:
        """
        비밀값 조회

        Args:
            secret_id: 비밀값 ID
            use_cache: 캐시 사용 여부

        Returns:
            비밀값 또는 None
        """
        # 캐시 확인
        if use_cache and self._cache_enabled and secret_id in self._cache:
            return self._cache[secret_id]

        value = await self._provider.get_secret(secret_id)

        # 캐시 저장
        if value and self._cache_enabled:
            self._cache[secret_id] = value

        return value

    async def set_secret(self, secret_id: str, value: str) -> bool:
        """비밀값 저장"""
        result = await self._provider.set_secret(secret_id, value)

        # 캐시 무효화
        if result and secret_id in self._cache:
            del self._cache[secret_id]

        return result

    async def delete_secret(self, secret_id: str) -> bool:
        """비밀값 삭제"""
        result = await self._provider.delete_secret(secret_id)

        # 캐시 무효화
        if secret_id in self._cache:
            del self._cache[secret_id]

        return result

    async def list_secrets(self) -> list[str]:
        """비밀값 목록 조회"""
        return await self._provider.list_secrets()

    def clear_cache(self):
        """캐시 초기화"""
        self._cache.clear()

    @property
    def provider_type(self) -> str:
        """현재 Provider 타입"""
        return type(self._provider).__name__


# 싱글톤
_manager: Optional[SecretsManager] = None


def get_secrets_manager() -> SecretsManager:
    """Secrets Manager 싱글톤 인스턴스"""
    global _manager
    if _manager is None:
        _manager = SecretsManager()
    return _manager


def generate_encryption_key() -> str:
    """
    로컬 암호화용 키 생성

    Returns:
        Base64 인코딩된 Fernet 키
    """
    if not CRYPTO_AVAILABLE:
        raise ImportError("cryptography 미설치")

    key = Fernet.generate_key()
    return key.decode('utf-8')