"""Google 스프레드시트 쓰기 공통 계층 (서비스 계정). - 다른 모듈에서도 쓸 수 있게 `app/integrations/` 에 둔다. - 인증: 구글 클라우드 **서비스 계정** 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"] class GoogleSheetsWriter: """시트 1장을 통째로 다시 쓰는 용도의 얇은 래퍼.""" def __init__(self) -> None: self.enabled = False self.reason = "" self._service: Any = None self._client_email = "" 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 try: info = json.loads(raw) if raw else json.loads( open(path, encoding="utf-8").read() ) except (OSError, ValueError): self.reason = "서비스 계정 JSON 을 읽지 못했습니다." return 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 - 인증 실패 사유만 남긴다 self.reason = f"서비스 계정 인증 실패: {type(exc).__name__}" return self._client_email = str(info.get("client_email") or "") self.enabled = True @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], ) -> dict[str, Any]: """시트를 만들고(있으면 비우고) 표를 통째로 기록한다. 좌표는 모두 1-based(엑셀과 동일). 반환값에 시트 URL 을 담는다. """ 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)", )) requests.append(_fmt( header_row, first_col, header_row, last_col, {"horizontalAlignment": "CENTER", "verticalAlignment": "MIDDLE", "textFormat": {"bold": True}}, "userEnteredFormat(horizontalAlignment,verticalAlignment,textFormat)", )) 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 col, width in widths.items(): requests.append({ "updateDimensionProperties": { "range": { "sheetId": sheet_id, "dimension": "COLUMNS", "startIndex": col - 1, "endIndex": col, }, "properties": {"pixelSize": int(width) * 8}, "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