from flask import Flask, redirect, session, url_for, request
import google_auth_oauthlib.flow
import json
import os
import requests
from googleapiclient.discovery import build
import googleapiclient.errors
import google.oauth2.credentials
app = Flask('app')
# `FLASK_SECRET_KEY`는 세션에 사용됩니다. 임의의 문자열을 생성하여
# 시크릿으로 저장해야 합니다.
app.secret_key = os.environ.get('FLASK_SECRET_KEY') or os.urandom(24)
# `GOOGLE_APIS_OAUTH_SECRET`은 Google Cloud Credentials 패널에서
# 다운로드할 JSON 파일의 내용을 담고 있습니다. 다음 섹션 참조.
oauth_config = json.loads(os.environ['GOOGLE_OAUTH_SECRETS'])
# OAuth 플로우 구성 설정
oauth_flow = google_auth_oauthlib.flow.Flow.from_client_config(
oauth_config,
# scopes는 인증 후 사용자를 대신하여 접근할 API를 정의합니다
scopes=[
"https://www.googleapis.com/auth/userinfo.email",
"openid",
"https://www.googleapis.com/auth/userinfo.profile",
"https://www.googleapis.com/auth/spreadsheets.readonly"
]
)
# 로그인 페이지의 진입점입니다. `authorization_url`에 위치한 Google 로그인 서비스로
# 리다이렉트합니다. `redirect_uri`는 Google 로그인 서비스가 이 앱으로 다시
# 리다이렉트하는 데 사용할 URI입니다.
@app.route('/signin')
def signin():
# Replit 앱 내부에서는 http를 사용하지만 외부에서는 https로 접근하므로
# redirect_uri가 이와 일치해야 하기 때문에 URL을 http에서 https로 변환합니다
oauth_flow.redirect_uri = url_for('oauth2callback', _external=True).replace('http://', 'https://')
authorization_url, state = oauth_flow.authorization_url()
session['state'] = state
return redirect(authorization_url)
# Google 로그인 서비스가 다시 리다이렉트하는 엔드포인트입니다. Google Cloud의
# API 자격 증명 패널에서 "Authorized redirect URIs"에 추가해야 합니다.
# 액세스 토큰을 요청하기 위해 Google 엔드포인트를 호출하고 사용자 세션에 저장합니다.
# 이후 이 액세스 토큰을 사용하여 사용자를 대신해 API에 접근할 수 있습니다.
@app.route('/oauth2callback')
def oauth2callback():
if not session['state'] == request.args['state']:
return 'Invalid state parameter', 400
oauth_flow.fetch_token(authorization_response=request.url.replace('http:', 'https:'))
session['access_token'] = oauth_flow.credentials.token
return redirect("/")
# 유효한 액세스 토큰으로 userinfo API를 호출하여 사용자 정보를 가져옵니다.
# 이것은 액세스 토큰으로 사용자를 대신해 API에 접근하는 첫 번째 예시입니다.
def get_user_info(access_token):
response = requests.get("https://www.googleapis.com/oauth2/v3/userinfo", headers={
"Authorization": f"Bearer {access_token}"
})
if response.status_code == 200:
user_info = response.json()
return user_info
else:
print(f"Failed to fetch user info: {response.status_code} {response.text}")
return None
@app.route('/logout')
def logout():
session.clear()
return redirect('/')
# Google 스프레드시트 내의 모든 시트 가져오기
def get_sheets(sheets_api, spreadsheet_id) -> list[str]:
result = sheets_api.get(spreadsheetId=spreadsheet_id).execute()
return [sheet["properties"]["title"] for sheet in result["sheets"]]
# Google 스프레드시트 내 특정 시트의 데이터 가져오기
def get_sheet_data(sheets_api, spreadsheet_id, sheet_title) -> list[list[str]]:
result = (
sheets_api.values()
.get(spreadsheetId=spreadsheet_id, range=sheet_title)
.execute()
)
return result["values"]
# 스프레드시트 가져오기 폼 렌더링
@app.route("/import_spreadsheet_form")
def import_spreadsheet_form():
return """
<h3>Import Spreadsheet</h3>
<form action="/import_spreadsheet" method="POST">
<label>Spreadsheet ID</label>
<input type="text" name="spreadsheet_id">
<button type="submit">Import</button>
</form>
"""
@app.route("/import_spreadsheet", methods = ['POST'])
def import_spreadsheet():
if 'access_token' not in session:
return redirect('/signin')
spreadsheet_id = request.form["spreadsheet_id"]
credentials = google.oauth2.credentials.Credentials(token=session['access_token'])
service = build("sheets", "v4", credentials=credentials)
sheets_api = service.spreadsheets()
try:
sheets = get_sheets(sheets_api, spreadsheet_id)
data_by_sheets = {}
for sheet in sheets:
data = get_sheet_data(sheets_api, spreadsheet_id, sheet)
data_by_sheets[sheet] = data
except googleapiclient.errors.HttpError as e:
return f"upload failure"
dirpath = os.path.join("static", "uploads", spreadsheet_id)
filepath = os.path.join(dirpath, "data.json")
os.makedirs(dirpath, exist_ok=True)
with open(filepath, "w") as file:
json.dump(data_by_sheets, file)
return "upload success! Really!"
@app.route('/')
def welcome():
if "access_token" in session:
user_info = get_user_info(session["access_token"])
if user_info:
return f"""
Hello {user_info["given_name"]}!<br>
Your email address is {user_info["email"]}<br>
<a href="/signin">Sign In to Google</a><br>
<a href="/import_spreadsheet_form">Import a Sheet</a>
"""
return """
<h1>Welcome to Google Sheet Importer</h1>
<a href="/signin">Sign In to Google</a><br>
<a href="/import_spreadsheet_form">Import a Sheet</a>
"""
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000)
