GoogleドライブへアクセスするOAuth認証を自動化する方法がわからず困ってます。(Python)
解決したいこと
1.共有ドライブにOAuth認証によりユーザーアカウントでアクセス
2.スプレッドシートを取得→parquet型に変換
3.GCSへ保存
これらの処理を以下のmain()関数の中で書いており、CloudFunction+CloudSchedulerで毎日定期処理にかけたいと思っております。
ただこのままだと下のコード部分でブラウザに推移して手動でGoogleアカウントにログインする必要があります。
このログインを自動で実行できるように書き換えたいのですがわからず困ってます…
どなたか教えていただけると幸いです。
### ※※ブラウザで認証が必要※※
creds = flow.run_local_server(port=0)
### 実行結果
Please visit this URL to authorize this application:
https://accounts.google.com/o/oauth2/auth?response_type=code&client_id=132987612861-
4j24afrouontpeiv5ryy7sn64inhr.apps.googleusercontent.com&redirect_uri=
http%yyy%2Flocalhost%3yy6%2F&scope=httpsyyF%2Fwww.googleapis.com%2Fauth%2Fdrive.
readonly&state=XXXXXXXXXXXXXXXXXXXXXXXXXXX&access_type=offline
実行ごとに、readonly&state=XXXXXXXXXXXXXXXXXXXの部分が変わります。
該当する全体のソースコード
from __future__ import print_function
import io
import os
import key
import json
import os.path
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
from pprint import pprint
from webbrowser import Konqueror
from google.cloud import storage as gcs
from google.oauth2 import service_account
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.http import MediaIoBaseDownload, MediaIoBaseUpload, MediaFileUpload
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
SCOPES = ['https://www.googleapis.com/auth/drive.readonly']
def main(event, context):
"""Drive v3 API
共有Driveアクセス→Spreadsheet取得→parquet変換→GCSへuploadまで行う関数
"""
creds = None
file_id = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxxx' #共有ドライブ内の編集前データ
mime_type = 'text/csv'
# OAuth認証で共有ドライブへアクセス
if os.path.exists('token.json'):
creds = Credentials.from_authorized_user_file('token.json', SCOPES)
# 利用可能な(有効な)認証情報がない場合、ユーザーにログインを許可する
if not creds or not creds.valid:
if creds and creds.expired and creds.refresh_token:
creds.refresh(Request())
else:
flow = InstalledAppFlow.from_client_secrets_file(
'credentials.json', SCOPES)
### ※※ブラウザで認証が必要※※
creds = flow.run_local_server(port=0)##現状ではここで手動ログインが必要!
with open('token.json', 'w') as token:
token.write(creds.to_json())
try:
# 共有ドライブからスプレッドシートを取得
service = build('drive', 'v3', credentials=creds)
request = service.files().export_media(fileId=file_id, mimeType=mime_type)
fh = io.BytesIO()
downloader = MediaIoBaseDownload(fh, request)
done = False
print(io.StringIO(fh.getvalue().decode()))
while done is False:
status, done = downloader.next_chunk()
# "共有Drive/SpreadSheet"を読み込み→parquetに変換
df = pd.read_csv(io.StringIO(fh.getvalue().decode()))
table = pa.Table.from_pandas(df)
buf = pa.BufferOutputStream()
pq.write_table(table, buf,compression=None)
# GCSへ保存するためのservice_account
key_path = 'service_account_file.json'
service_account_info = json.load(open(key_path))
credentials = service_account.Credentials.from_service_account_info(service_account_info)
client = gcs.Client(
credentials=credentials,
project=credentials.project_id,
)
# 保存するGCSの情報
bucket_name = 'bucket-name'
blob_name = 'sample-folder/daily-data.parquet'#save_path
bucket = client.get_bucket(bucket_name)
blob = bucket.blob(blob_name)
# parquetデータをGCSへ保存
blob.upload_from_string(data=buf.getvalue().to_pybytes())
# ↓printが表示されれば、データが保存された
print("Blob '{}' created to '{}'!".format(blob_name, bucket_name))
except HttpError as error:
# TODO(developer) - Handle errors from drive API.
print(f'An error occurred: {error}')
自分で試したこと
seleniumを使ってブラウザを動かす方法を試してみましたが、ブラウザログインURLが毎回ちがうためうまく実装ができませんでした。←方法があるかもですが。。。
0