# [メタ情報] # 識別子: scripts_pubフォルダからgドキュメントを生成_exe # システム名: 公開スクリプト（コード）用 # 技術種別: Misc # 機能名: 260205 # 使用言語: [] # 状態: 公開用 # [/メタ情報] 要約： APIキーエラー発生時は、まず`token.json`ファイルを削除し、`sync_docs.py`スクリプトを再実行してブラウザでGoogleアカウントに再認証します。認証が完了し「更新完了」が表示されたら、`launchctl unload`と`launchctl load`コマンドを順に実行し、バックグラウンドのファイル監視プロセスを再起動して新しい認証トークンを読み込ませます。 このシステムは、ローカルディレクトリ内のスクリプトやテキストファイルをGoogleドキュメントに自動で同期するものです。`sync_docs.py`が指定されたソースフォルダのファイルを収集し、更新日時順に整理して目次付きでGoogleドキュメントへ書き込みます。この際、ファイルは複数のGoogleドキュメントに分割して保存されます。`watcher.py`は、このソースフォルダのファイル変更を常時監視し、変更を検知すると`sync_docs.py`を自動的に実行して同期処理をトリガーします。`com.XXXXXX.docsupdater.plist`は、macOSのLaunchAgentとして`watcher.py`をシステム起動時に自動起動・常駐させ、ファイル更新の自動同期を可能にしています。 ------------------------ APIキーエラーになった時 復旧手順： 1. 古いトークンを削除して再実行する Bash cd <任意のフォルダパス> rm token.json python3 sync_docs.py 2. ブラウザでの再認証 「🌍 ブラウザを開いてログインしてください...」と表示され、URLが出力されるかブラウザが立ち上がります。Googleアカウントでログインし、アクセスを許可してください。 ターミナルにDoc #1〜#4の「✅ 更新完了」が最後まで表示されれば成功です。 3. バックグラウンド監視プロセスの再起動 新しいトークンをシステムに読み込ませるため、以下の2行を順番に実行します。 Bash launchctl unload ~/Library/LaunchAgents/com.XXXXXX.docsupdater.plist launchctl load ~/Library/LaunchAgents/com.XXXXXX.docsupdater.plist <任意のフォルダパス>/sync_docs.py ``` import os import os.path import urllib.parse import unicodedata import socket from googleapiclient.discovery import build from google.auth.transport.requests import Request from google.oauth2.credentials import Credentials from google_auth_oauthlib.flow import InstalledAppFlow from dotenv import load_dotenv # デフォルトのタイムアウト設定（10分） socket.setdefaulttimeout(600) SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) # ================= 設定エリア ================= load_dotenv("<任意のフォルダパス>") # 監視・同期元の公開スクリプトフォルダ SOURCE_FOLDER_PATH = "<任意のフォルダパス>" BASE_URL = "https://XXXXXX.com/script_list/?pubtxt=" CLIENT_SECRET_FILE = os.path.join(SCRIPT_DIR, 'credentials.json') TOKEN_FILE = os.path.join(SCRIPT_DIR, 'token.json') # 【設定】書き込み先のGoogleドキュメントIDリスト（masking_info.txtに準拠した20枠完全対応） TARGET_DOC_IDS = [ os.environ.get("SYNC_DOC_ID_1"), os.environ.get("SYNC_DOC_ID_2"), os.environ.get("SYNC_DOC_ID_3"), os.environ.get("SYNC_DOC_ID_4"), os.environ.get("SYNC_DOC_ID_5"), os.environ.get("SYNC_DOC_ID_6"), os.environ.get("SYNC_DOC_ID_7"), os.environ.get("SYNC_DOC_ID_8"), os.environ.get("SYNC_DOC_ID_9"), os.environ.get("SYNC_DOC_ID_10"), os.environ.get("SYNC_DOC_ID_11"), os.environ.get("SYNC_DOC_ID_12"), os.environ.get("SYNC_DOC_ID_13"), os.environ.get("SYNC_DOC_ID_14"), os.environ.get("SYNC_DOC_ID_15"), os.environ.get("SYNC_DOC_ID_16"), os.environ.get("SYNC_DOC_ID_17"), os.environ.get("SYNC_DOC_ID_18"), os.environ.get("SYNC_DOC_ID_19"), os.environ.get("SYNC_DOC_ID_20"), ] # 1つのドキュメントに含めるファイル数 FILES_PER_DOC = 15 TARGET_EXTENSIONS = [ '.txt', '.py', '.log', '.sh', '.md', '.json', '.yaml', '.html', '.css', '.js', '.php' ] EXCLUDE_FILES = [ 'key_patterns.txt', '.DS_Store', 'credentials.json', 'token.json', 'sync_docs.py', 'watcher.py' ] EXCLUDE_DIRS = ['.git', '__pycache__', '.idea', 'venv', 'node_modules'] SCOPES = ['https://www.googleapis.com/auth/documents'] # ============================================ def get_credentials(): creds = None if os.path.exists(TOKEN_FILE): creds = Credentials.from_authorized_user_file(TOKEN_FILE, SCOPES) if not creds or not creds.valid: if creds and creds.expired and creds.refresh_token: creds.refresh(Request()) else: print("🌍 ブラウザを開いてログインしてください...") flow = InstalledAppFlow.from_client_secrets_file(CLIENT_SECRET_FILE, SCOPES) creds = flow.run_local_server(port=0) with open(TOKEN_FILE, 'w') as token: token.write(creds.to_json()) return creds def get_sorted_files_list(folder_path): file_list = [] skipped_count = 0 for root, dirs, files in os.walk(folder_path): dirs[:] = [d for d in dirs if d not in EXCLUDE_DIRS] for file in files: if file in EXCLUDE_FILES: skipped_count += 1 continue ext = os.path.splitext(file)[1].lower() if ext in TARGET_EXTENSIONS: full_path = os.path.join(root, file) mtime = os.path.getmtime(full_path) file_list.append((full_path, mtime)) else: skipped_count += 1 file_list.sort(key=lambda x: x[1], reverse=True) return [f[0] for f in file_list], skipped_count def format_text_for_chunk(file_paths, toc_start_num): toc_lines = [] body_parts = [] toc_lines.append("=" * 80) toc_lines.append("【目次】") for i, path in enumerate(file_paths): num = toc_start_num + i rel_path = os.path.relpath(path, SOURCE_FOLDER_PATH) filename = os.path.basename(path) encoded_filename = urllib.parse.quote(filename) web_url = f"{BASE_URL}{encoded_filename}" toc_lines.append(f"{num}. {rel_path}") body_parts.append("=" * 80) body_parts.append(f"INDEX: No.{num:03d}") body_parts.append(f"FILENAME: {rel_path}") body_parts.append(f"URL: {web_url}") body_parts.append("-" * 80) try: with open(path, 'r', encoding='utf-8') as f: body_parts.append(f.read()) except Exception as e: body_parts.append(f"[ファイル読み込みエラー: {e}]") body_parts.append("") toc_lines.append("=" * 80) full_text = "\n".join(toc_lines) + "\n\n\n" + "\n\n".join(body_parts) return unicodedata.normalize('NFC', full_text) def update_google_doc(target_doc_id, text_content): if not target_doc_id: return short_id = target_doc_id[-5:] print(f"Google Docs APIに接続中... (ID: ...{short_id})") if not text_content: text_content = "(このドキュメントに割り当てられるファイルはありません)" try: creds = get_credentials() service = build('docs', 'v1', credentials=creds) doc = service.documents().get( documentId=target_doc_id, fields="body(content(endIndex))" ).execute() content = doc.get('body').get('content') end_index = content[-1]['endIndex'] - 1 if content else 1 requests = [] if end_index > 1: requests.append({'deleteContentRange': {'range': {'startIndex': 1, 'endIndex': end_index}}}) requests.append({'insertText': {'location': {'index': 1}, 'text': text_content}}) service.documents().batchUpdate(documentId=target_doc_id, body={'requests': requests}).execute() print(f"✅ 更新完了 (ID: ...{short_id})") except Exception as e: print(f"❌ APIエラー (ID: ...{short_id}): {e}") def main_sync(): all_files, skipped = get_sorted_files_list(SOURCE_FOLDER_PATH) total_files = len(all_files) print("-" * 30) print(f"読み込み成功: {total_files} ファイル") print(f"除外スキップ: {skipped} ファイル") for i, doc_id in enumerate(TARGET_DOC_IDS): start_idx = i * FILES_PER_DOC end_idx = start_idx + FILES_PER_DOC chunk_files = all_files[start_idx:end_idx] toc_start_num = start_idx + 1 current_range = f"{start_idx + 1}-{start_idx + len(chunk_files)}" if chunk_files else "空" if chunk_files: print(f"\n--- Doc #{i+1} (Files {current_range}) の更新 ---") chunk_text = format_text_for_chunk(chunk_files, toc_start_num) update_google_doc(doc_id, chunk_text) else: print(f"\n--- Doc #{i+1} (Files {current_range}) の更新 ---") update_google_doc(doc_id, "(このドキュメントに割り当てられるファイルはありません)") if total_files > len(TARGET_DOC_IDS) * FILES_PER_DOC: limit = len(TARGET_DOC_IDS) * FILES_PER_DOC print(f"\n⚠️ 注意: ファイル数が上限({limit})を超えています。残りのファイルは同期されません。") if __name__ == "__main__": main_sync() ``` <任意のフォルダパス>/watcher.py ``` import time import sys import os import unicodedata from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler sys.path.append(os.path.dirname(os.path.abspath(__file__))) import sync_docs class ChangeHandler(FileSystemEventHandler): def __init__(self): super().__init__() self.last_modified = 0 self.cooldown_seconds = 2 def process_event(self, filepath): norm_path = unicodedata.normalize('NFC', filepath) filename = os.path.basename(norm_path) if filename in sync_docs.EXCLUDE_FILES: return ext = os.path.splitext(filename)[1].lower() if ext not in sync_docs.TARGET_EXTENSIONS: return print(f"👀 検知: {filename} が変更されました") self.last_modified = time.time() def on_modified(self, event): if not event.is_directory: self.process_event(event.src_path) def on_created(self, event): if not event.is_directory: self.process_event(event.src_path) def on_deleted(self, event): if not event.is_directory: self.process_event(event.src_path) def on_moved(self, event): if not event.is_directory: self.process_event(event.dest_path) def run_sync_process(): print("\n" + "="*30) print("🔄 Googleドキュメント同期プロセスを開始...") try: import importlib importlib.reload(sync_docs) sync_docs.main_sync() except Exception as e: print(f"❌ 重大エラー: {e}") print("="*30 + "\n") print(f"監視中... 対象: {sync_docs.SOURCE_FOLDER_PATH}") if __name__ == "__main__": target_path = sync_docs.SOURCE_FOLDER_PATH if not os.path.exists(target_path): print(f"❌ エラー: 監視対象のフォルダが見つかりません: {target_path}") sys.exit(1) event_handler = ChangeHandler() observer = Observer() observer.schedule(event_handler, target_path, recursive=True) observer.start() print(f"👁️ scripts_pub監視デーモンが起動しました") print(f"監視フォルダ: {target_path}") print("<固有のランダム文字列>") last_triggered = 0 try: while True: time.sleep(0.5) if event_handler.last_modified > last_triggered: if time.time() - event_handler.last_modified > event_handler.cooldown_seconds: run_sync_process() last_triggered = time.time() except KeyboardInterrupt: observer.stop() print("\n🛑 終了します") observer.join() ``` <任意のフォルダパス>/com.XXXXXX.docsupdater.plist ``` Label com.XXXXXX.docsupdater ProgramArguments /usr/bin/python3 -u <任意のフォルダパス>/watcher.py WorkingDirectory <任意のフォルダパス> RunAtLoad StandardOutPath /tmp/docsupdater.out StandardErrorPath /tmp/docsupdater.err ```