# Python으로 Money Forward 스타일 재무 PDF 리포트 만들기 - Part 3

## Google Sheets API 연동

### 개요

이번 파트에서는 Google Sheets API를 사용하여 예실관리 스프레드시트에서 데이터를 읽어오는 클라이언트를 구현합니다.

### 서비스 계정 인증

Google Sheets API를 사용하려면 서비스 계정이 필요합니다. 기존 프로젝트에서 사용하던 서비스 계정 키를 재사용합니다.

### sheets_client.py 구현

`src/sheets_client.py`:

```python
"""Google Sheets API 클라이언트"""
from typing import Any
from google.oauth2 import service_account
from googleapiclient.discovery import build

from config import SERVICE_ACCOUNT_KEY, SPREADSHEET_ID


class SheetsClient:
    """Google Sheets API 클라이언트 클래스"""

    SCOPES = ["https://www.googleapis.com/auth/spreadsheets.readonly"]

    def __init__(self):
        self._service = None

    @property
    def service(self):
        """Sheets API 서비스 객체 (지연 로딩)"""
        if self._service is None:
            credentials = service_account.Credentials.from_service_account_file(
                str(SERVICE_ACCOUNT_KEY),
                scopes=self.SCOPES
            )
            self._service = build("sheets", "v4", credentials=credentials)
        return self._service

    def get_sheet_data(
        self,
        range_name: str,
        spreadsheet_id: str = SPREADSHEET_ID
    ) -> list[list[Any]]:
        """시트 데이터 조회

        Args:
            range_name: 범위 (예: '予実出力!A1:AZ100')
            spreadsheet_id: 스프레드시트 ID

        Returns:
            2차원 배열 형태의 데이터
        """
        result = self.service.spreadsheets().values().get(
            spreadsheetId=spreadsheet_id,
            range=range_name
        ).execute()

        return result.get("values", [])

    def get_sheet_metadata(
        self,
        spreadsheet_id: str = SPREADSHEET_ID
    ) -> dict:
        """스프레드시트 메타데이터 조회"""
        return self.service.spreadsheets().get(
            spreadsheetId=spreadsheet_id
        ).execute()

    def get_all_sheets(
        self,
        spreadsheet_id: str = SPREADSHEET_ID
    ) -> list[dict]:
        """모든 시트 목록 조회"""
        metadata = self.get_sheet_metadata(spreadsheet_id)
        return [
            {
                "title": sheet["properties"]["title"],
                "sheet_id": sheet["properties"]["sheetId"],
                "index": sheet["properties"]["index"],
            }
            for sheet in metadata.get("sheets", [])
        ]


# 싱글톤 인스턴스
sheets_client = SheetsClient()
```

### 데이터 조회 함수

`src/data_fetcher.py`:

```python
"""데이터 조회 모듈"""
from typing import Optional
from sheets_client import sheets_client


def fetch_budget_data() -> list[list]:
    """予算入力 시트 데이터 조회"""
    return sheets_client.get_sheet_data("'予算入力'!A1:T50")


def fetch_actual_data() -> list[list]:
    """実績入力 시트 데이터 조회"""
    return sheets_client.get_sheet_data("'実績入力'!A1:T50")


def fetch_comparison_data() -> list[list]:
    """予実出力 시트 데이터 조회"""
    return sheets_client.get_sheet_data("'予実出力'!A1:AZ100")


def fetch_monthly_pl(month: Optional[str] = None) -> dict:
    """월별 손익 데이터 조회

    Args:
        month: 대상 월 (예: '2025-10'). None이면 전체 기간

    Returns:
        {
            'headers': ['항목', '2025/04月', ...],
            'rows': [['売上高', 628802, ...], ...]
        }
    """
    data = fetch_comparison_data()

    if not data:
        return {'headers': [], 'rows': []}

    # 헤더 행 찾기 (보통 4-5행)
    header_row_idx = None
    for i, row in enumerate(data[:10]):
        if row and '2025' in str(row):
            header_row_idx = i
            break

    if header_row_idx is None:
        return {'headers': [], 'rows': []}

    headers = data[header_row_idx]
    rows = data[header_row_idx + 1:]

    return {
        'headers': headers,
        'rows': rows
    }


def fetch_kpi_summary() -> dict:
    """KPI 요약 데이터 조회

    Returns:
        {
            'annual': {
                'revenue': 125947574,
                'gross_profit': 72841825,
                'operating_income': 51104333,
                'net_income': 51113902,
                'gross_margin': 57.84,
                'operating_margin': 40.58,
            },
            'monthly': {
                'month': '2025年10月',
                'revenue': 6100000,
                ...
            }
        }
    """
    data = fetch_comparison_data()

    # 실제 데이터 파싱 로직은 시트 구조에 따라 조정
    # 여기서는 기본 구조만 정의

    return {
        'annual': {
            'revenue': 0,
            'gross_profit': 0,
            'operating_income': 0,
            'net_income': 0,
            'gross_margin': 0.0,
            'operating_margin': 0.0,
        },
        'monthly': {
            'month': '',
            'revenue': 0,
            'gross_profit': 0,
            'operating_income': 0,
            'net_income': 0,
            'gross_margin': 0.0,
            'operating_margin': 0.0,
        }
    }
```

### 연결 테스트

`src/test_connection.py`:

```python
"""API 연결 테스트"""
from sheets_client import sheets_client
from data_fetcher import fetch_comparison_data, fetch_monthly_pl


def test_connection():
    """연결 테스트"""
    print("=== Google Sheets API 연결 테스트 ===\n")

    # 1. 시트 목록 조회
    print("1. 시트 목록:")
    sheets = sheets_client.get_all_sheets()
    for sheet in sheets:
        print(f"   - {sheet['title']} (ID: {sheet['sheet_id']})")

    # 2. 예실출력 데이터 조회
    print("\n2. 予実出力 데이터 샘플:")
    data = fetch_comparison_data()
    print(f"   총 {len(data)}개 행")

    if data:
        print(f"   첫 행: {data[0][:5]}...")

    # 3. 월별 P/L 데이터
    print("\n3. 월별 P/L 구조:")
    pl_data = fetch_monthly_pl()
    print(f"   헤더: {pl_data['headers'][:5]}...")
    print(f"   데이터 행 수: {len(pl_data['rows'])}")

    print("\n✓ 연결 테스트 완료!")


if __name__ == "__main__":
    test_connection()
```

### 실행 결과 예시

```bash
$ python src/test_connection.py

=== Google Sheets API 연결 테스트 ===

1. 시트 목록:
   - 予算入力 (ID: 0)
   - 実績入力 (ID: 1759121662)
   - 予実出力 (ID: 123456789)

2. 予実出力 데이터 샘플:
   총 45개 행
   첫 행: ['', '', '', '予算', '実績']...

3. 월별 P/L 구조:
   헤더: ['', '項目', '2025/04月', '2025/05月', '2025/06月']...
   데이터 행 수: 40

✓ 연결 테스트 완료!
```

### 에러 처리

실제 운영에서는 다양한 에러 상황을 처리해야 합니다:

```python
from googleapiclient.errors import HttpError


def safe_fetch(range_name: str) -> list[list]:
    """안전한 데이터 조회 (에러 처리 포함)"""
    try:
        return sheets_client.get_sheet_data(range_name)
    except HttpError as e:
        if e.resp.status == 404:
            print(f"시트를 찾을 수 없습니다: {range_name}")
        elif e.resp.status == 403:
            print("권한이 없습니다. 서비스 계정에 시트 접근 권한을 부여하세요.")
        else:
            print(f"API 에러: {e}")
        return []
    except FileNotFoundError:
        print("서비스 계정 키 파일을 찾을 수 없습니다.")
        return []
    except Exception as e:
        print(f"예상치 못한 에러: {e}")
        return []
```

### 캐싱 (선택사항)

API 호출을 줄이기 위한 간단한 캐싱:

```python
from functools import lru_cache
from datetime import datetime, timedelta


class CachedSheetsClient(SheetsClient):
    """캐싱 기능이 있는 Sheets 클라이언트"""

    def __init__(self, cache_ttl_seconds: int = 300):
        super().__init__()
        self._cache = {}
        self._cache_ttl = timedelta(seconds=cache_ttl_seconds)

    def get_sheet_data(self, range_name: str, **kwargs) -> list[list]:
        cache_key = f"{kwargs.get('spreadsheet_id', SPREADSHEET_ID)}:{range_name}"

        # 캐시 확인
        if cache_key in self._cache:
            cached_time, cached_data = self._cache[cache_key]
            if datetime.now() - cached_time < self._cache_ttl:
                return cached_data

        # API 호출
        data = super().get_sheet_data(range_name, **kwargs)

        # 캐시 저장
        self._cache[cache_key] = (datetime.now(), data)

        return data

    def clear_cache(self):
        """캐시 초기화"""
        self._cache.clear()
```

### 다음 단계

Part 4에서는 조회한 데이터를 처리하기 위한 데이터 모델(dataclass)을 설계합니다.

---

**시리즈 네비게이션**
- [x] Part 1: 프로젝트 소개
- [x] Part 2: 환경 설정
- [x] Part 3: Sheets API 연동 (현재)
- [ ] Part 4: 데이터 모델
- [ ] Part 5: PDF 기본 레이아웃
- [ ] Part 6: KPI 대시보드
- [ ] Part 7: 테이블 구현
- [ ] Part 8: 차트 생성
- [ ] Part 9: 복합 차트
- [ ] Part 10: 완성