feat: POS 실시간 판매 조회 웹 페이지 (Qt GUI 웹 버전)
This commit is contained in:
@@ -98,6 +98,89 @@ class KakaoAPIClient:
|
||||
'error_description': f'Invalid JSON response: {e}'
|
||||
}
|
||||
|
||||
def refresh_access_token(self, refresh_token: str) -> Tuple[bool, Dict[str, Any]]:
|
||||
"""Refresh Token으로 Access Token 갱신"""
|
||||
url = f"{self.auth_base_url}/oauth/token"
|
||||
|
||||
data = {
|
||||
'grant_type': 'refresh_token',
|
||||
'client_id': self.client_id,
|
||||
'refresh_token': refresh_token,
|
||||
}
|
||||
if self.client_secret:
|
||||
data['client_secret'] = self.client_secret
|
||||
|
||||
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_refresh_failed',
|
||||
'error_description': f'Failed to refresh 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
|
||||
|
||||
def get_user_info_with_refresh(
|
||||
self,
|
||||
access_token: str,
|
||||
refresh_token: str,
|
||||
token_expires_at: str = None
|
||||
) -> Tuple[bool, Dict[str, Any], Dict[str, Any]]:
|
||||
"""저장된 토큰으로 사용자 정보 조회 (만료 시 자동 갱신)
|
||||
|
||||
Returns:
|
||||
(성공여부, 사용자정보/에러, 갱신된 토큰 데이터 또는 빈 dict)
|
||||
"""
|
||||
new_token_data = {}
|
||||
|
||||
# 만료 확인: 5분 이내면 미리 갱신
|
||||
if token_expires_at:
|
||||
try:
|
||||
expires = datetime.fromisoformat(token_expires_at)
|
||||
if datetime.now() >= expires - timedelta(minutes=5):
|
||||
logger.info("Access token 만료 임박, 갱신 시도")
|
||||
success, refreshed = self.refresh_access_token(refresh_token)
|
||||
if success:
|
||||
access_token = refreshed['access_token']
|
||||
new_token_data = refreshed
|
||||
else:
|
||||
return False, refreshed, {}
|
||||
except (ValueError, TypeError) as e:
|
||||
logger.warning(f"token_expires_at 파싱 실패, 기존 토큰으로 시도: {e}")
|
||||
|
||||
# 사용자 정보 조회
|
||||
success, user_info = self.get_user_info(access_token)
|
||||
|
||||
if not success and refresh_token:
|
||||
# 실패 시 갱신 후 재시도
|
||||
logger.info("사용자 정보 조회 실패, 토큰 갱신 후 재시도")
|
||||
refresh_ok, refreshed = self.refresh_access_token(refresh_token)
|
||||
if refresh_ok:
|
||||
access_token = refreshed['access_token']
|
||||
new_token_data = refreshed
|
||||
success, user_info = self.get_user_info(access_token)
|
||||
|
||||
return success, user_info, new_token_data
|
||||
|
||||
def get_user_info(self, access_token: str) -> Tuple[bool, Dict[str, Any]]:
|
||||
"""Access Token으로 사용자 정보 조회"""
|
||||
url = f"{self.api_base_url}/v2/user/me"
|
||||
|
||||
Reference in New Issue
Block a user