feat(sheets): 사용자 OAuth 리프레시 토큰 인증 지원
조직 정책(iam.disableServiceAccountKeyCreation)으로 서비스 계정 키를 만들 수 없는 환경 대응. 서비스 계정이 설정돼 있으면 그쪽을 먼저 쓰고, 없으면 GOOGLE_SHEETS_OAUTH_REFRESH_TOKEN + client id/secret 으로 인증한다. (client id/secret 미지정 시 로그인용 GOOGLE_CLIENT_ID/SECRET 재사용) scripts/google_sheets_authorize.py: 데스크톱 앱 OAuth 클라이언트로 1회 동의해 refresh token 을 발급받는 스크립트. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
@@ -1,11 +1,21 @@
|
||||
"""Google 스프레드시트 쓰기 공통 계층 (서비스 계정).
|
||||
"""Google 스프레드시트 쓰기 공통 계층.
|
||||
|
||||
- 다른 모듈에서도 쓸 수 있게 `app/integrations/` 에 둔다.
|
||||
- 인증: 구글 클라우드 **서비스 계정** JSON.
|
||||
인증 방식 두 가지를 지원한다(설정된 쪽을 자동 선택, 서비스 계정 우선).
|
||||
|
||||
1) 사용자 OAuth 리프레시 토큰 — 조직 정책으로 서비스 계정 **키 발급이 막힌** 경우
|
||||
· `GOOGLE_SHEETS_OAUTH_REFRESH_TOKEN` 1회 동의로 받은 refresh token
|
||||
· `GOOGLE_SHEETS_OAUTH_CLIENT_ID` / `..._SECRET`
|
||||
(없으면 로그인용 `GOOGLE_CLIENT_ID` / `GOOGLE_CLIENT_SECRET` 사용)
|
||||
토큰 발급: `python scripts/google_sheets_authorize.py`
|
||||
이 방식은 토큰을 발급한 **사용자 권한**으로 동작하므로, 그 사용자가 이미
|
||||
편집할 수 있는 문서면 별도 공유가 필요 없다.
|
||||
|
||||
2) 서비스 계정 JSON
|
||||
· `GOOGLE_SHEETS_CREDENTIALS` 서비스 계정 JSON 파일 경로
|
||||
· `GOOGLE_SHEETS_CREDENTIALS_JSON` JSON 본문(파일 대신 환경변수로 넣을 때)
|
||||
대상 스프레드시트를 서비스 계정 이메일(client_email)에 **편집자로 공유**해야 한다.
|
||||
- 비밀값(키)은 로그·화면에 절대 출력하지 않는다.
|
||||
· `GOOGLE_SHEETS_CREDENTIALS_JSON` JSON 본문(파일 대신 환경변수로)
|
||||
대상 문서를 서비스 계정 이메일(client_email)에 **편집자로 공유**해야 한다.
|
||||
|
||||
- 비밀값(키/토큰)은 로그·화면에 절대 출력하지 않는다.
|
||||
- 라이브러리(google-api-python-client)가 없거나 설정이 비어 있으면
|
||||
`enabled = False` 로 조용히 비활성화되고, 호출부는 건너뛴다.
|
||||
"""
|
||||
@@ -17,6 +27,7 @@ import os
|
||||
from typing import Any
|
||||
|
||||
SCOPES = ["https://www.googleapis.com/auth/spreadsheets"]
|
||||
TOKEN_URI = "https://oauth2.googleapis.com/token"
|
||||
|
||||
|
||||
class GoogleSheetsWriter:
|
||||
@@ -25,39 +36,91 @@ class GoogleSheetsWriter:
|
||||
def __init__(self) -> None:
|
||||
self.enabled = False
|
||||
self.reason = ""
|
||||
self.auth_mode = "" # "service_account" | "oauth"
|
||||
self._service: Any = None
|
||||
self._client_email = ""
|
||||
|
||||
try:
|
||||
from googleapiclient.discovery import build
|
||||
except ImportError:
|
||||
self.reason = "google-api-python-client 미설치"
|
||||
return
|
||||
|
||||
creds = self._service_account_creds()
|
||||
if creds is None:
|
||||
creds = self._oauth_creds()
|
||||
if creds is None:
|
||||
if not self.reason:
|
||||
self.reason = (
|
||||
"GOOGLE_SHEETS_OAUTH_REFRESH_TOKEN 또는 "
|
||||
"GOOGLE_SHEETS_CREDENTIALS(_JSON) 미설정"
|
||||
)
|
||||
return
|
||||
|
||||
try:
|
||||
self._service = build("sheets", "v4", credentials=creds, cache_discovery=False)
|
||||
except Exception as exc: # noqa: BLE001 - 사유만 남긴다(비밀값 미출력)
|
||||
self.reason = f"Sheets 클라이언트 생성 실패: {type(exc).__name__}"
|
||||
return
|
||||
self.enabled = True
|
||||
|
||||
# ── 인증 ─────────────────────────────────────────
|
||||
def _service_account_creds(self) -> Any:
|
||||
raw = (os.getenv("GOOGLE_SHEETS_CREDENTIALS_JSON") or "").strip()
|
||||
path = (os.getenv("GOOGLE_SHEETS_CREDENTIALS") or "").strip()
|
||||
if not raw and not path:
|
||||
self.reason = "GOOGLE_SHEETS_CREDENTIALS(_JSON) 미설정"
|
||||
return
|
||||
|
||||
return None
|
||||
try:
|
||||
info = json.loads(raw) if raw else json.loads(
|
||||
open(path, encoding="utf-8").read()
|
||||
)
|
||||
except (OSError, ValueError):
|
||||
self.reason = "서비스 계정 JSON 을 읽지 못했습니다."
|
||||
return
|
||||
|
||||
return None
|
||||
try:
|
||||
from google.oauth2.service_account import Credentials
|
||||
from googleapiclient.discovery import build
|
||||
except ImportError:
|
||||
self.reason = "google-api-python-client 미설치"
|
||||
return
|
||||
|
||||
try:
|
||||
creds = Credentials.from_service_account_info(info, scopes=SCOPES)
|
||||
self._service = build("sheets", "v4", credentials=creds, cache_discovery=False)
|
||||
except Exception as exc: # noqa: BLE001 - 인증 실패 사유만 남긴다
|
||||
except Exception as exc: # noqa: BLE001
|
||||
self.reason = f"서비스 계정 인증 실패: {type(exc).__name__}"
|
||||
return
|
||||
|
||||
return None
|
||||
self._client_email = str(info.get("client_email") or "")
|
||||
self.enabled = True
|
||||
self.auth_mode = "service_account"
|
||||
return creds
|
||||
|
||||
def _oauth_creds(self) -> Any:
|
||||
refresh_token = (os.getenv("GOOGLE_SHEETS_OAUTH_REFRESH_TOKEN") or "").strip()
|
||||
if not refresh_token:
|
||||
return None
|
||||
client_id = (
|
||||
os.getenv("GOOGLE_SHEETS_OAUTH_CLIENT_ID")
|
||||
or os.getenv("GOOGLE_CLIENT_ID")
|
||||
or ""
|
||||
).strip()
|
||||
client_secret = (
|
||||
os.getenv("GOOGLE_SHEETS_OAUTH_CLIENT_SECRET")
|
||||
or os.getenv("GOOGLE_CLIENT_SECRET")
|
||||
or ""
|
||||
).strip()
|
||||
if not client_id or not client_secret:
|
||||
self.reason = "GOOGLE_SHEETS_OAUTH_CLIENT_ID/SECRET 미설정"
|
||||
return None
|
||||
try:
|
||||
from google.oauth2.credentials import Credentials
|
||||
|
||||
creds = Credentials(
|
||||
token=None,
|
||||
refresh_token=refresh_token,
|
||||
token_uri=TOKEN_URI,
|
||||
client_id=client_id,
|
||||
client_secret=client_secret,
|
||||
scopes=SCOPES,
|
||||
)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
self.reason = f"OAuth 자격 생성 실패: {type(exc).__name__}"
|
||||
return None
|
||||
self.auth_mode = "oauth"
|
||||
return creds
|
||||
|
||||
@property
|
||||
def client_email(self) -> str:
|
||||
|
||||
Reference in New Issue
Block a user