king
21018a5332
구글 시트 요청에 pixelSize 를 8배로 넣어 A열이 80px 로 벌어져 있었다. WIDTHS_PX 를 픽셀 그대로 쓰고, xlsx 는 px/7 로 환산한다. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
315 lines
12 KiB
Python
315 lines
12 KiB
Python
"""Google 스프레드시트 쓰기 공통 계층.
|
|
|
|
인증 방식 두 가지를 지원한다(설정된 쪽을 자동 선택, 서비스 계정 우선).
|
|
|
|
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-api-python-client)가 없거나 설정이 비어 있으면
|
|
`enabled = False` 로 조용히 비활성화되고, 호출부는 건너뛴다.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from typing import Any
|
|
|
|
SCOPES = ["https://www.googleapis.com/auth/spreadsheets"]
|
|
TOKEN_URI = "https://oauth2.googleapis.com/token"
|
|
|
|
|
|
def _hex_color(value: str) -> dict[str, float]:
|
|
""""RRGGBB" → Sheets API 색상."""
|
|
v = (value or "").lstrip("#")
|
|
return {
|
|
"red": int(v[0:2], 16) / 255.0,
|
|
"green": int(v[2:4], 16) / 255.0,
|
|
"blue": int(v[4:6], 16) / 255.0,
|
|
}
|
|
|
|
|
|
class GoogleSheetsWriter:
|
|
"""시트 1장을 통째로 다시 쓰는 용도의 얇은 래퍼."""
|
|
|
|
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:
|
|
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 None
|
|
try:
|
|
from google.oauth2.service_account import Credentials
|
|
|
|
creds = Credentials.from_service_account_info(info, scopes=SCOPES)
|
|
except Exception as exc: # noqa: BLE001
|
|
self.reason = f"서비스 계정 인증 실패: {type(exc).__name__}"
|
|
return None
|
|
self._client_email = str(info.get("client_email") or "")
|
|
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:
|
|
"""공유 대상 확인용(비밀값 아님)."""
|
|
return self._client_email
|
|
|
|
# ────────────────────────────────────────────────
|
|
def _sheet_id(self, spreadsheet_id: str, title: str) -> int | None:
|
|
meta = self._service.spreadsheets().get(spreadsheetId=spreadsheet_id).execute()
|
|
for sh in meta.get("sheets", []):
|
|
props = sh.get("properties", {})
|
|
if props.get("title") == title:
|
|
return int(props.get("sheetId"))
|
|
return None
|
|
|
|
def _create_sheet(self, spreadsheet_id: str, title: str) -> int:
|
|
res = self._service.spreadsheets().batchUpdate(
|
|
spreadsheetId=spreadsheet_id,
|
|
body={"requests": [{"addSheet": {"properties": {"title": title}}}]},
|
|
).execute()
|
|
return int(res["replies"][0]["addSheet"]["properties"]["sheetId"])
|
|
|
|
def write_table(
|
|
self,
|
|
*,
|
|
spreadsheet_id: str,
|
|
title: str,
|
|
cells: dict[tuple[int, int], Any],
|
|
merges: list[tuple[int, int, int, int]],
|
|
header_row: int,
|
|
first_data_row: int,
|
|
last_row: int,
|
|
first_col: int,
|
|
last_col: int,
|
|
title_row: int,
|
|
widths: dict[int, int], # 열 번호(1-based) -> 픽셀
|
|
header_bg: str = "",
|
|
row_highlights: list[tuple[int, int, str]] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""시트를 만들고(있으면 비우고) 표를 통째로 기록한다.
|
|
|
|
좌표는 모두 1-based(엑셀과 동일). 반환값에 시트 URL 을 담는다.
|
|
header_bg / row_highlights 의 색은 "RRGGBB" 16진 문자열.
|
|
"""
|
|
if not self.enabled:
|
|
raise RuntimeError(self.reason or "Google Sheets 미설정")
|
|
|
|
sheet_id = self._sheet_id(spreadsheet_id, title)
|
|
created = sheet_id is None
|
|
if created:
|
|
sheet_id = self._create_sheet(spreadsheet_id, title)
|
|
|
|
rows_n = max([r for (r, _c) in cells] or [1])
|
|
cols_n = last_col
|
|
matrix: list[list[Any]] = [["" for _ in range(cols_n)] for _ in range(rows_n)]
|
|
for (r, c), value in cells.items():
|
|
matrix[r - 1][c - 1] = value
|
|
|
|
requests: list[dict[str, Any]] = [
|
|
# 기존 내용/서식/병합 초기화 → 같은 날짜로 다시 확정해도 깨끗하게 덮어쓴다
|
|
{"unmergeCells": {"range": {"sheetId": sheet_id}}},
|
|
{"updateCells": {"range": {"sheetId": sheet_id}, "fields": "*"}},
|
|
]
|
|
|
|
for (r1, c1, r2, c2) in merges:
|
|
requests.append({
|
|
"mergeCells": {
|
|
"mergeType": "MERGE_ALL",
|
|
"range": {
|
|
"sheetId": sheet_id,
|
|
"startRowIndex": r1 - 1, "endRowIndex": r2,
|
|
"startColumnIndex": c1 - 1, "endColumnIndex": c2,
|
|
},
|
|
}
|
|
})
|
|
|
|
def _fmt(r1: int, c1: int, r2: int, c2: int, fmt: dict[str, Any], fields: str) -> dict[str, Any]:
|
|
return {
|
|
"repeatCell": {
|
|
"range": {
|
|
"sheetId": sheet_id,
|
|
"startRowIndex": r1 - 1, "endRowIndex": r2,
|
|
"startColumnIndex": c1 - 1, "endColumnIndex": c2,
|
|
},
|
|
"cell": {"userEnteredFormat": fmt},
|
|
"fields": fields,
|
|
}
|
|
}
|
|
|
|
requests.append(_fmt(
|
|
title_row, first_col, title_row, last_col,
|
|
{"horizontalAlignment": "CENTER", "verticalAlignment": "MIDDLE",
|
|
"textFormat": {"bold": True, "fontSize": 16}},
|
|
"userEnteredFormat(horizontalAlignment,verticalAlignment,textFormat)",
|
|
))
|
|
header_fmt: dict[str, Any] = {
|
|
"horizontalAlignment": "CENTER", "verticalAlignment": "MIDDLE",
|
|
"textFormat": {"bold": True},
|
|
}
|
|
header_fields = "userEnteredFormat(horizontalAlignment,verticalAlignment,textFormat)"
|
|
if header_bg:
|
|
header_fmt["backgroundColor"] = _hex_color(header_bg)
|
|
header_fields = (
|
|
"userEnteredFormat(horizontalAlignment,verticalAlignment,"
|
|
"textFormat,backgroundColor)"
|
|
)
|
|
requests.append(_fmt(header_row, first_col, header_row, last_col,
|
|
header_fmt, header_fields))
|
|
if last_row >= first_data_row:
|
|
requests.append(_fmt(
|
|
first_data_row, first_col, last_row, last_col,
|
|
{"horizontalAlignment": "CENTER", "verticalAlignment": "MIDDLE",
|
|
"wrapStrategy": "WRAP"},
|
|
"userEnteredFormat(horizontalAlignment,verticalAlignment,wrapStrategy)",
|
|
))
|
|
# 제품명 칸만 왼쪽 정렬
|
|
requests.append(_fmt(
|
|
first_data_row, 9, last_row, 9,
|
|
{"horizontalAlignment": "LEFT"},
|
|
"userEnteredFormat(horizontalAlignment)",
|
|
))
|
|
border = {"style": "SOLID", "color": {"red": 0, "green": 0, "blue": 0}}
|
|
requests.append({
|
|
"updateBorders": {
|
|
"range": {
|
|
"sheetId": sheet_id,
|
|
"startRowIndex": header_row - 1, "endRowIndex": last_row,
|
|
"startColumnIndex": first_col - 1, "endColumnIndex": last_col,
|
|
},
|
|
"top": border, "bottom": border, "left": border, "right": border,
|
|
"innerHorizontal": border, "innerVertical": border,
|
|
}
|
|
})
|
|
|
|
# 특정 행 구간 배경색(예: 파렛트 출고 블록)
|
|
for (r1, r2, color) in (row_highlights or []):
|
|
requests.append(_fmt(
|
|
r1, first_col, r2, last_col,
|
|
{"backgroundColor": _hex_color(color)},
|
|
"userEnteredFormat(backgroundColor)",
|
|
))
|
|
|
|
for col, width in widths.items():
|
|
requests.append({
|
|
"updateDimensionProperties": {
|
|
"range": {
|
|
"sheetId": sheet_id, "dimension": "COLUMNS",
|
|
"startIndex": col - 1, "endIndex": col,
|
|
},
|
|
"properties": {"pixelSize": int(width)},
|
|
"fields": "pixelSize",
|
|
}
|
|
})
|
|
|
|
self._service.spreadsheets().batchUpdate(
|
|
spreadsheetId=spreadsheet_id, body={"requests": requests}
|
|
).execute()
|
|
|
|
self._service.spreadsheets().values().update(
|
|
spreadsheetId=spreadsheet_id,
|
|
range=f"'{title}'!A1",
|
|
valueInputOption="USER_ENTERED",
|
|
body={"values": matrix},
|
|
).execute()
|
|
|
|
return {
|
|
"created": created,
|
|
"title": title,
|
|
"url": f"https://docs.google.com/spreadsheets/d/{spreadsheet_id}/edit#gid={sheet_id}",
|
|
}
|
|
|
|
|
|
_writer: GoogleSheetsWriter | None = None
|
|
|
|
|
|
def get_writer() -> GoogleSheetsWriter:
|
|
"""프로세스 1개당 1회 초기화."""
|
|
global _writer # noqa: PLW0603
|
|
if _writer is None:
|
|
_writer = GoogleSheetsWriter()
|
|
return _writer
|