# [メタ情報] # 識別子: Notebookのソースをローカルへ同期_exe # 補足: # [/メタ情報] 要約: 本システムは、Google Drive上の特定の図書館フォルダ(「gdoc_XXXXXX図書館」および「gdoc以外_XXXXXX図書館」)の内容を、ローカル外付けSSD(/Volumes/NO4_SSD/Notebookからソース同期)へ全自動・定期的に差分同期(エクスポート・保存)するプログラム群です。 macOSのバックグラウンド常駐機構(launchd)から直接 python3 バイナリを起動した際に発生する macOS セキュリティ(TCC)のアクセス拒否エラー([Errno 1] Operation not permitted)を回避するため、実行用 zsh ラッパースクリプト(run_gdrive_sync.sh)を仲介する最新構成を採用しています。 LaunchAgent(com.XXXXXX.gdrive.sync.plist)により、ログイン時に1時間間隔(StartInterval: 3600秒)で自動巡回し、Googleドキュメントは .txt 、スプレッドシートは .csv 、バイナリはそのままの形式で安全にローカルSSDへ上書き反映されます。 ================================================================================ ■ 1. 差分同期メインプログラム (gdrive_to_local_sync_exe.py) ================================================================================ 配置先: /Users/XXXXXX/python_scripts/gdrive_to_local_sync_exe.py ``` import os import sys import io import time import socket import datetime from pathlib import Path from dotenv import load_dotenv # 通信タイムアウトを180秒(3分)に延長(SSL/Socket Timeout対策) socket.setdefaulttimeout(180) # ========================================== # 環境変数のロードおよび起動時安全チェック (SOP-ST) # ========================================== env_path = Path("/Users/XXXXXX/python_scripts/.env") if env_path.exists(): load_dotenv(dotenv_path=env_path) CREDENTIALS_FILE = os.getenv("SA_JSON_PATH", "/Users/XXXXXX/keys/service.json") GDOC_FOLDER_ID = os.getenv("GDOC_XXXXXX_FOLDER_ID", "1vmQzwvSOmOay8HtDFxZb6Uhye6foVrdH") OTHER_FOLDER_ID = os.getenv("OTHER_XXXXXX_FOLDER_ID", "1xpNAIKTC5aF88KF7B7dgX9nPPJXNTh6R") LOCAL_SYNC_DIR = os.getenv("LOCAL_SYNC_DIR", "/Volumes/NO4_SSD/Notebookからソース同期") # 起動時ガバナンスチェック(鍵・パスの存在確認) missing_keys = [] if not CREDENTIALS_FILE or not os.path.exists(CREDENTIALS_FILE): missing_keys.append("SA_JSON_PATH (service.json)") if not GDOC_FOLDER_ID: missing_keys.append("GDOC_XXXXXX_FOLDER_ID") if not OTHER_FOLDER_ID: missing_keys.append("OTHER_XXXXXX_FOLDER_ID") if not LOCAL_SYNC_DIR: missing_keys.append("LOCAL_SYNC_DIR") if missing_keys: print(f"[ERROR] 必要な環境変数・鍵が設定されていません: {', '.join(missing_keys)}") print("[ERROR] 安全のため処理を自律停止します。.env の設定を確認してください。") sys.exit(1) # Google API 関連ライブラリのインポート try: import json from google.oauth2 import service_account from google.oauth2.credentials import Credentials from google_auth_oauthlib.flow import InstalledAppFlow from google.auth.transport.requests import Request from googleapiclient.discovery import build from googleapiclient.http import MediaIoBaseDownload from googleapiclient.errors import HttpError except ImportError as e: print(f"[ERROR] 必要なPythonライブラリが見つかりません: {e}") print("[HINT] pip install google-api-python-client google-auth-httplib2 google-auth-oauthlib python-dotenv を実行してください。") sys.exit(1) SCOPES = ['https://www.googleapis.com/auth/drive.readonly'] def get_drive_service(): """Google Drive API サービスの認証(サービスアカウント / OAuth 2.0 ハイブリッド)""" cred_file = CREDENTIALS_FILE if not Path(cred_file).exists(): cred_file = "credentials.json" if Path(cred_file).exists(): try: creds = service_account.Credentials.from_service_account_file( cred_file, scopes=SCOPES) return build('drive', 'v3', credentials=creds) except Exception as e: print(f"[WARNING] サービスアカウント認証に失敗しました: {e}。OAuth認証を試みます...") token_path = Path('/Users/XXXXXX/python_scripts/token.json') creds = None if token_path.exists(): creds = Credentials.from_authorized_user_file(str(token_path), 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_path, 'w') as token: token.write(creds.to_json()) return build('drive', 'v3', credentials=creds) def download_file_with_retry(request, dest_path, max_retries=5): """指数バックオフによる安全なファイル一括ダウンロード処理""" tmp_path = dest_path + ".tmp" for attempt in range(max_retries): try: fh = io.FileIO(tmp_path, 'wb') downloader = MediaIoBaseDownload(fh, request, chunksize=1024*1024) done = False while not done: status, done = downloader.next_chunk() fh.close() # 安全な原子置換(アトミック書き換え) if os.path.exists(dest_path): os.remove(dest_path) os.rename(tmp_path, dest_path) return True except Exception as e: if os.path.exists(tmp_path): try: os.remove(tmp_path) except: pass if attempt < max_retries - 1: wait_time = (2 ** attempt) + 1 time.sleep(wait_time) else: print(f" [ERROR] {max_retries} 回再試行しましたがダウンロードに失敗しました: {os.path.basename(dest_path)}") return False def sync_folder(service, folder_id, sync_target_dir): """指定されたGoogle Driveフォルダからファイルを同期""" os.makedirs(sync_target_dir, exist_ok=True) query = f"'{folder_id}' in parents and trashed = false" try: results = service.files().list( q=query, pageSize=1000, fields="nextPageToken, files(id, name, mimeType, modifiedTime)" ).execute() files = results.get('files', []) for file in files: file_id = file.get('id') file_name = file.get('name') mime_type = file.get('mimeType') modified_time_str = file.get('modifiedTime') safe_name = file_name.replace('/', '_') # Google Drive上の更新日時をエポックタイムへ変換 drive_mtime = datetime.datetime.fromisoformat(modified_time_str.replace('Z', '+00:00')).timestamp() # 拡張子の決定 if mime_type == 'application/vnd.google-apps.document': dest_path = os.path.join(sync_target_dir, f"{safe_name}.txt") elif mime_type == 'application/vnd.google-apps.spreadsheet': dest_path = os.path.join(sync_target_dir, f"{safe_name}.csv") else: dest_path = os.path.join(sync_target_dir, safe_name) # 差分判定(ローカルファイルが存在し、更新日時が新しければスキップ) if os.path.exists(dest_path): local_mtime = os.path.getmtime(dest_path) if local_mtime >= drive_mtime - 2: # 2秒の許容誤差 print(f" -> スキップ (変更なし): {os.path.basename(dest_path)}") continue print(f" -> 処理中 (更新保存): {file_name} ({mime_type})") try: if mime_type == 'application/vnd.google-apps.document': request = service.files().export_media(fileId=file_id, mimeType='text/plain') if download_file_with_retry(request, dest_path): os.utime(dest_path, (drive_mtime, drive_mtime)) elif mime_type == 'application/vnd.google-apps.spreadsheet': request = service.files().export_media(fileId=file_id, mimeType='text/csv') if download_file_with_retry(request, dest_path): os.utime(dest_path, (drive_mtime, drive_mtime)) else: request = service.files().get_media(fileId=file_id) if download_file_with_retry(request, dest_path): os.utime(dest_path, (drive_mtime, drive_mtime)) except Exception as e: print(f" [ERROR] ファイル処理スキップ ({file_name}): {e}") except Exception as e: print(f"[ERROR] フォルダID [{folder_id}] の取得に失敗しました: {e}") def main(): now_jst = datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=9))).strftime("%Y-%m-%d %H:%M:%S") print(f"=== Google Drive -> ローカル差分同期処理開始 (JST {now_jst}) ===") print(f"[*] 同期先ディレクトリ: {LOCAL_SYNC_DIR}") service = get_drive_service() print("\n--- 1. gdoc_XXXXXX図書館 の同期 ---") sync_folder(service, GDOC_FOLDER_ID, LOCAL_SYNC_DIR) print("\n--- 2. gdoc以外_XXXXXX図書館 の同期 ---") sync_folder(service, OTHER_FOLDER_ID, LOCAL_SYNC_DIR) print("\n=== 全同期処理が完了いたしました ===") if __name__ == "__main__": main() ``` ================================================================================ ■ 2. 実行用 Zsh ラッパースクリプト (run_gdrive_sync.sh) ================================================================================ 配置先: /Users/XXXXXX/python_scripts/run_gdrive_sync.sh (※chmod +x で実行権限を付与のこと) ``` #!/bin/zsh export PATH="/usr/bin:/bin:/usr/sbin:/sbin:/usr/local/bin" export LANG="ja_JP.UTF-8" # スクリプトディレクトリへ移動して同期プログラムを実行 cd /Users/XXXXXX/python_scripts /usr/bin/python3 /Users/XXXXXX/python_scripts/gdrive_to_local_sync_exe.py >> /Users/XXXXXX/python_scripts/logs/gdrive_sync.log 2>&1 ``` ================================================================================ ■ 3. LaunchAgent 設定ファイル (com.XXXXXX.gdrive.sync.plist) ================================================================================ 配置先: ~/Library/LaunchAgents/com.XXXXXX.gdrive.sync.plist ``` Label com.XXXXXX.gdrive.sync ProgramArguments /bin/zsh /Users/XXXXXX/python_scripts/run_gdrive_sync.sh RunAtLoad StartInterval 3600 WorkingDirectory /Users/XXXXXX/python_scripts StandardOutPath /Users/XXXXXX/python_scripts/logs/gdrive_sync.log StandardErrorPath /Users/XXXXXX/python_scripts/logs/gdrive_sync.log ``` /Users/XXXXXX/python_scripts/run_gdrive_sync.command ``` #!/bin/bash DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" /usr/bin/python3 "$DIR/gdrive_to_local_sync_exe.py" osascript -e 'display notification "Google Drive -> ローカル差分同期が完了いたしました。" with title "Gドライブ手動同期"' ```