sami1220
@sami1220

Are you sure you want to delete the question?

If your question is resolved, you may close it.

Leaving a resolved question undeleted may help others!

We hope you find it useful!

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の部分が変わります。

上記コード部分の実行で遷移するブラウザ画面
スクリーンショット 2022-07-12 18.44.33.png

該当する全体のソースコード

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

No Answers yet.

Your answer might help someone💌