pharmacy-pos-qr-system/backend/services/kakao_client.py

"""
Kakao API Client for OAuth 2.0 Authentication
pharmacy-pos-qr-system용 경량 버전
"""
import os
import requests
import json
import logging
from datetime import datetime, timedelta
from typing import Dict, Any, Tuple
from urllib.parse import urlencode

logger = logging.getLogger(__name__)


class KakaoAPIClient:
    """카카오 API 클라이언트 (마일리지 적립용)"""

    def __init__(self):
        self.client_id = os.getenv('KAKAO_CLIENT_ID', '')
        self.client_secret = os.getenv('KAKAO_CLIENT_SECRET', '')
        self.redirect_uri = os.getenv(
            'KAKAO_REDIRECT_URI',
            'https://mile.0bin.in/claim/kakao/callback'
        )

        self.auth_base_url = 'https://kauth.kakao.com'
        self.api_base_url = 'https://kapi.kakao.com'

        self.session = requests.Session()
        self.session.headers.update({
            'Content-Type': 'application/x-www-form-urlencoded',
            'User-Agent': 'YouthPharmacy-Mileage/1.0'
        })

    def get_authorization_url(self, state: str = None) -> str:
        """카카오 OAuth 인증 URL 생성"""
        params = {
            'client_id': self.client_id,
            'redirect_uri': self.redirect_uri,
            'response_type': 'code',
            'scope': 'profile_nickname,profile_image,account_email'
        }

        if state:
            params['state'] = state

        return f"{self.auth_base_url}/oauth/authorize?{urlencode(params)}"

    def get_access_token(self, authorization_code: str) -> Tuple[bool, Dict[str, Any]]:
        """Authorization Code로 Access Token 요청"""
        url = f"{self.auth_base_url}/oauth/token"

        data = {
            'grant_type': 'authorization_code',
            'client_id': self.client_id,
            'client_secret': self.client_secret,
            'redirect_uri': self.redirect_uri,
            'code': authorization_code
        }

        try:
            headers = {'Content-Type': 'application/x-www-form-urlencoded'}
            response = self.session.post(url, data=data, headers=headers)

            logger.info(f"카카오 토큰 응답 상태: {response.status_code}")
            response.raise_for_status()

            token_data = response.json()

            if 'expires_in' in token_data:
                expires_at = datetime.now() + timedelta(seconds=token_data['expires_in'])
                token_data['expires_at'] = expires_at.isoformat()

            return True, token_data

        except requests.exceptions.RequestException as e:
            logger.error(f"카카오 토큰 요청 실패: {e}")

            error_details = {
                'error': 'token_request_failed',
                'error_description': f'Failed to get access token: {e}'
            }
            try:
                if hasattr(e, 'response') and e.response is not None:
                    kakao_error = e.response.json()
                    logger.error(f"카카오 API 오류: {kakao_error}")
                    error_details.update(kakao_error)
            except Exception:
                pass

            return False, error_details

        except json.JSONDecodeError as e:
            logger.error(f"카카오 응답 JSON 파싱 실패: {e}")
            return False, {
                'error': 'invalid_response',
                'error_description': f'Invalid JSON response: {e}'
            }

    def get_user_info(self, access_token: str) -> Tuple[bool, Dict[str, Any]]:
        """Access Token으로 사용자 정보 조회"""
        url = f"{self.api_base_url}/v2/user/me"

        headers = {
            'Authorization': f'Bearer {access_token}',
            'Content-Type': 'application/x-www-form-urlencoded'
        }

        try:
            response = requests.get(url, headers=headers)
            response.raise_for_status()

            user_data = response.json()
            normalized_user = self._normalize_user_data(user_data)

            return True, normalized_user

        except requests.exceptions.RequestException as e:
            logger.error(f"카카오 사용자 정보 조회 실패: {e}")
            return False, {
                'error': 'user_info_failed',
                'error_description': f'Failed to get user info: {e}'
            }

    def _normalize_user_data(self, kakao_user: Dict[str, Any]) -> Dict[str, Any]:
        """카카오 사용자 데이터를 정규화"""
        kakao_account = kakao_user.get('kakao_account', {})
        profile = kakao_account.get('profile', {})

        normalized = {
            'kakao_id': str(kakao_user.get('id')),
            'nickname': profile.get('nickname'),
            'profile_image': profile.get('profile_image_url'),
            'thumbnail_image': profile.get('thumbnail_image_url'),
            'email': kakao_account.get('email'),
            'is_email_verified': kakao_account.get('is_email_verified', False),
            'name': kakao_account.get('name'),
            'phone_number': kakao_account.get('phone_number'),
        }

        # None 값 제거
        normalized = {k: v for k, v in normalized.items() if v is not None}

        return normalized


# 싱글톤
_kakao_client = None


def get_kakao_client() -> KakaoAPIClient:
    """카카오 클라이언트 인스턴스 반환"""
    global _kakao_client
    if _kakao_client is None:
        _kakao_client = KakaoAPIClient()
    return _kakao_client