Open1
python - google
antyuntyunGoogleDrive/ スプレッドシートを処理するクラスを書く時のメモ
setup
def __init__(self,
jsonf_path: str,
gdrive_service_account_credential: str,
):
logger.debug('init')
# 環境変数をクラス変数に格納
# サービスアカウントのスコープ指定
SCOPES = [
'https://www.googleapis.com/auth/drive',
'https://www.googleapis.com/auth/spreadsheets']
# 環境変数もしくは配置されているjsonファイルからサービスアカウントを認証
try:
if (jsonf_path):
logger.debug('try to get authentication by json-credential file...')
sa_creds = service_account.Credentials.from_service_account_file(jsonf_path)
else:
logger.debug('try to get authentication by env...')
tmp_path = 'tmp.json'
with open(tmp_path, mode='w') as f:
f.write(gdrive_service_account_credential)
sa_creds = service_account.Credentials.from_service_account_file(tmp_path)
os.remove(tmp_path)
except Exception as e:
logger.error('Fail to get authentication')
logger.error(e)
exit
logger.debug('Successs to get authentication')
scoped_creds = sa_creds.with_scopes(SCOPES)
self.drive_service = build('drive', 'v3', credentials=scoped_creds)
self.sheets = build('sheets', 'v4', credentials=scoped_creds)
self.gc = gspread.authorize(scoped_creds)
スプレッドシートをdfとして保存
def download_spreadsheet_as_df(self, spreadsheet_id: str, sheet_name: str) -> pd.DataFrame:
"""
GoogleDrive上のスプレッドシートをpandasのdataframeとして出力
左上詰めで1行目がヘッダとなっていることを前提とする
Parameters:
spreadsheet_id: ダウンロードするスプレッドシートファイルのid
sheet_name: ダウンロードするスプレッドシートのシート名
Returns:
pd.DataFrme
"""
response = self.sheets.spreadsheets().values().get(spreadsheetId=spreadsheet_id, range=sheet_name).execute()
logger.debug('response:')
logger.debug(pformat(response))
df = pd.DataFrame(response['values'][1:], columns=response['values'][0])
return df
ファイルの移動
def move_file(self, file_id: str, prev_parent_folder_id: str,
new_parent_folder_id: str) -> dict:
print(file_id)
print(prev_parent_folder_id)
print(new_parent_folder_id)
file = self.drive_service.files().update(fileId=file_id,
addParents=new_parent_folder_id,
removeParents=prev_parent_folder_id,
fields='id, parents',
supportsAllDrives=True).execute()
return file
dfをアップロード
スプレッドシートの場合は、mimeTypeは'application/vnd.google-apps.spreadsheet'を指定
def upload_df(self, df: pd.DataFrame, name: str, mimeType: str, parent_folder_id: str,
overwrite: bool = True) -> str:
"""
pandasのdatafrmeを指定したIDのフォルダ以下にアップロードする
Parameters:
df: アップロードするpd.DataFrame
name: アップロードした時のファイル名
mimeType: GoogleDrive上でのファイルタイプ
parent_folder_id: アップロード先の親フォルダ
overwrite: 同一名のファイルがあった時に上書きするかどうかのフラグ
Returns:
str: アップロードしたファイルのID
"""
# TODO: to_csvを挟まない処理に変更
tmp_csv_file_path = './tmp.csv'
df.to_csv(tmp_csv_file_path, index=False)
# PATCH処理になるかどうかの条件判定
patch_flag = False
# 指定フォルダ内に同一名ファイルが存在するか確認
if (self.exsits_in_folder(name, parent_folder_id)):
logger.debug('file already exists')
if (not overwrite):
logger.debug('do not overwrite.')
return ''
else:
# 上書きする場合はidを取得
logger.debug('will be overwritten')
exists_file_id = self.get_id_by_name_in_folder(name, parent_folder_id)
patch_flag = True
# TODO: 指定されたparent_folder_idがフォルダとして実在するかの確認
file_metadata = {
'name': name,
'parents': [parent_folder_id],
'mimeType': mimeType,
}
media = MediaFileUpload(
tmp_csv_file_path,
mimetype='text/csv',
resumable=True
)
# 上書きかどうかで利用API変更
if not patch_flag:
response = self.drive_service.files().create(
body=file_metadata, media_body=media, fields='id', supportsAllDrives=True
).execute()
else:
# update()だとファイルが上手く上書きされないので、patch処理ではなく 削除 -> ファイルアップロードに変更
# # https://developers.google.com/drive/api/v2/reference/files/update
# file_metadata = {
# 'name': name,
# }
# response = self.drive_service.files().update(
# fileId=exists_file_id,
# addParents=parent_folder_id,
# body=file_metadata,
# ).execute()
response = self.drive_service.files().delete(
fileId=exists_file_id, supportsAllDrives=True
).execute()
response = self.drive_service.files().create(
body=file_metadata, media_body=media, fields='id', supportsAllDrives=True
).execute()
os.remove(tmp_csv_file_path)
return response['id']
csvファイルをDLしdfとして読み込み
def download_csv_as_df(self, file_id: str) -> pd.DataFrame:
"""
GoogleDrive上のcsvをpandasのdataframeとしてDL
Parameters:
file_id: ダウンロードするcsvファイルのid
Returns:
pd.DataFrme
"""
# TODO: csv判定の追加
request = self.drive_service.files().get_media(fileId=file_id)
fh = io.BytesIO()
downloader = MediaIoBaseDownload(fh, request)
done = False
while done is False:
status, done = downloader.next_chunk()
return pd.read_csv(io.StringIO(fh.getvalue().decode()))
フォルダの作成
def create_folder(self, folder_name: str, parent_folder_id: str) -> str:
"""
GoogleDrive上に指定の名前でフォルダを作成し、作成されたフォルダのIDを返す
folder_name: 作成するフォルダ名
parend_folder_id: 作成するフォルダの親フォルダのID
"""
# 既にフォルダが存在する場合には,既存フォルダのidを返す
if (self.exsits_in_folder(folder_name, parent_folder_id)):
logger.debug('folder already exists')
return self.get_id_by_name_in_folder(folder_name, parent_folder_id)
# https://developers.google.com/drive/api/v3/reference/files/create
file_metadata = {
'name': folder_name,
'mimeType': 'application/vnd.google-apps.folder',
'parents': [parent_folder_id],
}
response = self.drive_service.files().create(body=file_metadata,
supportsAllDrives=True,
fields='id').execute()
return response['id']
ファイル名検索によるファイル有無判定
def exsits_in_folder(self, search_name: str, parent_folder_id: str) -> bool:
"""
GoogleDrive上の特定フォルダ内にファイル・フォルダが存在するかを名前検索
Parameters:
search_name: 検索名
parent_folder_id: 検索対象の親フォルダ
Returns:
bool
"""
check_result = False
# https://developers.google.com/resources/api-libraries/documentation/drive/v3/python/latest/drive_v3.files.html#list
# https://developers.google.com/drive/api/v3/reference/query-ref
condition_list = [
f'parents in "{parent_folder_id}"',
'trashed = false',
]
conditions = ' and '.join(condition_list)
response = self.drive_service.files().list(
supportsAllDrives=True,
includeItemsFromAllDrives=True,
q=conditions,
fields="nextPageToken, files(id, name, mimeType, fullFileExtension)").execute()
for file in response.get('files', []):
if file.get('name') == search_name:
check_result = True
break
return check_result
フォルダ内からファイル名で検索しidを取得
def get_id_by_name_in_folder(self, search_name: str, parent_folder_id: str) -> str:
"""
GoogleDrive上の特定フォルダ内に特定名のファイル・フォルダが存在するかを検索し、存在する場合idを返す。
存在しない場合は空文字を返す。
Parameters:
search_name: 検索名
parent_folder_id: 検索対象の親フォルダ
Returns:
str: 検索対象のID
"""
check_result = False
# https://developers.google.com/resources/api-libraries/documentation/drive/v3/python/latest/drive_v3.files.html#list
# https://developers.google.com/drive/api/v3/reference/query-ref
condition_list = [
f'parents in "{parent_folder_id}"',
'trashed = false',
]
conditions = ' and '.join(condition_list)
try:
response = self.drive_service.files().list(
supportsAllDrives=True,
includeItemsFromAllDrives=True,
q=conditions,
fields="nextPageToken, files(id, name, mimeType, fullFileExtension)").execute()
for file in response.get('files', []):
if file.get('name') == search_name:
check_result = file.get('id')
break
except BaseException:
# FileNotFoundのとき404を受け取りgoogleapiclient.errors.HttpErrorがraiseされる
logger.debug('NotFound')
return check_result