# [メタ情報] # 識別子: scripts_exeフォルダからgdocを生成_exe # 補足: env終了 # [/メタ情報] 要約： このシステムは、ローカルの特定のテキストファイルをGoogleドキュメントに自動同期するためのものです。`sync_exe_docs.py`は、指定されたローカルフォルダ内の`_exe.txt`ファイルを更新日時順に収集し、設定された複数のGoogleドキュメントに分割して書き込みます。各ドキュメントには目次とファイルの内容が含まれ、Google Docs APIを用いて認証・更新されます。 `watcher_exe.py`は、このローカルフォルダのファイル変更（作成、変更、移動）をリアルタイムで監視し、`_exe.txt`ファイルの変更が検知されると、一定のクールダウン期間を経て`sync_exe_docs.py`の同期処理を自動的にトリガーします。 最後に、`com.XXXXXX.exe-docsupdater.plist`はmacOSのLaunchAgent設定ファイルであり、ユーザーログイン時に`watcher_exe.py`をバックグラウンドで自動起動・常駐させ、この監視・同期プロセスを継続的に実行します。これにより、ローカルのファイルが更新されるたびに、対応するGoogleドキュメントが自動的に最新の状態に保たれるよう設計されています。 /Users/XXXXXX/python_scripts/sync_exe_docs.py ``` #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ sync_exe_docs.py (SOP・完全env適合化ポータブル仕様版) - 特定のフォルダ内の _exe.txt を更新日時降順で収集 - Google Docs API 経由で10分割した Google ドキュメント ID群へ自動同期 - 機密情報（ドキュメントID・パス）をコード内に1文字も残さず、金庫（.env）から動的ロード """ import os import os.path import sys 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__)) # ===== 1. 環境設定（金庫 .env からの動的ロード ＆ 環境差の自動解決） ===== # 自身のフォルダ、親フォルダ、またはホーム直下の python_scripts の .env を動的に探索 dotenv_loaded = False for dotenv_path in [ os.path.join(SCRIPT_DIR, ".env"), os.path.join(os.path.dirname(SCRIPT_DIR), ".env"), os.path.join(os.path.expanduser("~"), "python_scripts", ".env") ]: if os.path.exists(dotenv_path): load_dotenv(dotenv_path) dotenv_loaded = True break # (A) 監視・同期元のフォルダパスを動的取得 SOURCE_FOLDER_PATH = os.getenv("SCRIPTS_EXE_SRC") if not SOURCE_FOLDER_PATH: print("❌ エラー: 環境変数 SCRIPTS_EXE_SRC が .env に設定されていません。") sys.exit(1) # 末尾のスラッシュを保証 if not SOURCE_FOLDER_PATH.endswith('/'): SOURCE_FOLDER_PATH += '/' CLIENT_SECRET_FILE = os.path.join(SCRIPT_DIR, 'credentials.json') TOKEN_FILE = os.path.join(SCRIPT_DIR, 'token.json') # (B) 書き込み先のGoogleドキュメントIDリストを金庫から動的バインド TARGET_DOC_IDS = [] missing_ids = [] for idx in range(1, 11): env_key = f"SYNC_EXE_DOC_ID_{idx}" doc_id = os.getenv(env_key) if not doc_id: missing_ids.append(env_key) else: TARGET_DOC_IDS.append(doc_id) if missing_ids: print("❌ エラー: 金庫 (.env) に以下の環境変数が設定されていません。") for key in missing_ids: print(f" - {key}") sys.exit(1) # 1つのドキュメントに含めるファイル数 FILES_PER_DOC = 15 # 対象とするサフィックスのリスト（_exe.txt） TARGET_SUFFIXES = ['_exe.txt'] # 除外ファイル EXCLUDE_FILES = ['空_exe.txt', '空_wip.txt', '.DS_Store'] EXCLUDE_DIRS = ['.git', '__pycache__', '.idea', 'venv', 'node_modules'] SCOPES = ['https://www.googleapis.com/auth/documents'] # ============================================ def get_sorted_files_list(root_folder): """ ディレクトリを探索し、対象ファイル（*_exe.txt）の情報をリスト化して返す（更新日時降順） """ collected_files = [] skipped_files = 0 print(f"📂 探索開始: {root_folder}") if not os.path.exists(root_folder): print(f"⚠️ 警告: フォルダ {root_folder} が存在しません。") return collected_files, skipped_files for root, dirs, files in os.walk(root_folder): # 除外ディレクトリ dirs[:] = [d for d in dirs if d not in EXCLUDE_DIRS] for file in files: norm_name = unicodedata.normalize('NFC', file) if norm_name in EXCLUDE_FILES: skipped_files += 1 continue if any(norm_name.endswith(suffix) for suffix in TARGET_SUFFIXES): full_path = os.path.join(root, file) try: mtime = os.path.getmtime(full_path) collected_files.append({ 'path': full_path, 'name': norm_name, 'mtime': mtime }) except Exception as e: print(f"❌ タイムスタンプ取得失敗 ({norm_name}): {e}") # 更新日時の降順（最新のファイルが先頭）でソート collected_files.sort(key=lambda x: x['mtime'], reverse=True) return collected_files, skipped_files 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("🌍 ブラウザを開いてログインしてください...") if not os.path.exists(CLIENT_SECRET_FILE): raise FileNotFoundError(f"エラー: {CLIENT_SECRET_FILE} が見つかりません。") 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 update_google_doc(target_doc_id, text_content): """ 指定されたIDのドキュメントを更新する """ short_id = target_doc_id[:5] + "..." + target_doc_id[-5:] if len(target_doc_id) > 10 else target_doc_id try: creds = get_credentials() service = build('docs', 'v1', credentials=creds) # ドキュメントの末尾インデックス（現在の長さ）を取得 doc = service.documents().get(documentId=target_doc_id).execute() end_index = doc.get('body').get('content')[-1].get('endIndex') - 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 format_text_for_chunk(files_chunk, toc_start_num): """ 分割されたファイルを、目次＋本文の形に整形する """ toc_lines = [] body_lines = [] toc_lines.append("========================================================================") toc_lines.append("【目次】") for idx, f in enumerate(files_chunk): serial_num = toc_start_num + idx base_name = os.path.splitext(f['name'])[0] toc_lines.append(f"{serial_num}. {base_name}") body_lines.append("========================================================================") body_lines.append(f"INDEX: No.{serial_num:03d}") body_lines.append(f"FILENAME: {f['name']}") body_lines.append("------------------------------------------------------------------------") try: with open(f['path'], 'r', encoding='utf-8') as file_obj: content = file_obj.read() body_lines.append(content) except Exception as e: body_lines.append(f"❌ ファイルの読み込みに失敗しました: {e}") body_lines.append("\n") toc_lines.append("========================================================================\n") full_text = "\n".join(toc_lines) + "\n".join(body_lines) return full_text 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}〜{min(end_idx, total_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} (空ファイル) の更新 ---") 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() ``` /Users/XXXXXX/python_scripts/watcher_exe.py ``` #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ watcher_exe.py (監視用常駐デーモン) - 指定の _exe.txt フォルダを常時監視 - 変更検知後、連射ガードとクールダウン5秒を経て自動製本 (sync_exe_docs.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_exe_docs class ChangeHandler(FileSystemEventHandler): """変更を検知するクラス""" def __init__(self): self.last_modified = time.time() # 確実にエディタの保存（ディスク書き込み）が完了するよう、待機を5秒に延長 self.cooldown_seconds = 5 self.file_last_seen = {} def process_event(self, src_path): """イベントの共通処理""" filename = os.path.basename(src_path) # 隠しファイルやスクリプト自身は無視 if filename.startswith('.') or filename in ['watcher_exe.py', 'sync_exe_docs.py']: return normalized_name = unicodedata.normalize('NFC', filename) if normalized_name in sync_exe_docs.EXCLUDE_FILES: return # 対象のサフィックス（_exe.txt）を持っている場合のみトリガー if any(normalized_name.endswith(suffix) for suffix in sync_exe_docs.TARGET_SUFFIXES): current_time = time.time() last_seen = self.file_last_seen.get(src_path, 0) # 連射ガード（10秒） if current_time - last_seen < 10: return self.file_last_seen[src_path] = current_time print(f"👀 検知: {normalized_name} が作成・更新されました。") self.last_modified = current_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_exe_docs) sync_exe_docs.main_sync() except Exception as e: print(f"❌ 重大エラー: {e}") print("="*30 + "\n") print(f"監視中... 対象: {sync_exe_docs.SOURCE_FOLDER_PATH}") if __name__ == "__main__": target_path = sync_exe_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_exe監視デーモンが起動しました") print(f"監視フォルダ: {target_path}") print("--------------------------------------------------") last_triggered = 0 try: while True: time.sleep(0.5) # 最後に変更を検知してから指定秒数（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() ``` /Users/XXXXXX/Library/LaunchAgents/com.XXXXXX.exe-docsupdater.plist ``` Label com.XXXXXX.exe-docsupdater ProgramArguments /usr/bin/python3 -u /Users/XXXXXX/python_scripts/watcher_exe.py WorkingDirectory /Users/XXXXXX/python_scripts RunAtLoad StandardOutPath /tmp/exe-docsupdater.out StandardErrorPath /tmp/exe-docsupdater.err ```