# [メタ情報] # 識別子: ファイルパス変更処理_gasからpython移行_exe # システム名: ファイルパス変更処理_gasからpython移行、M1Mac上で常時稼働 # 技術種別: Misc # 機能名: Misc # 使用言語: Python　自動起動スクリプト.plist # 状態: 公開用　　env終了 # [/メタ情報] 要約：mmedia・pmedia フォルダを常時監視し、ファイルやフォルダの移動・リネーム（MOVED）を検出して記録する Python スクリプト。検出時には変更前・変更後パス、JST の更新時刻、更新フラグ TRUE を Google スプレッドシート（sheet1）へ追記し、その後シート全体を読み込んで (A,B) が同じ重複記録は「最も古いものだけ残す」形に正規化し、更新日時の降順で再整列して書き戻す。scripts_pub や .dropbox など特定パスは無視するフィルタを搭載。watchdog によりディレクトリを監視し続け、ログは ~/Library/Logs に記録。LaunchAgent（com.XXXXXX.filewatcher.plist）を用いて macOS 起動時に自動実行される構成となっている。 Mac mini 起動 └── LaunchAgent（com.XXXXXX.filewatcher.plist）起動 └── file_rename_watcher.py 常駐開始 ├── mmedia / pmedia の変更をリアルタイム検出 ├── Google Sheets「sheet1」に即時反映 └── ~/Library/Logs/folder_watch.log に記録 <任意のフォルダパス>/file_rename_watcher.py ``` #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ file_rename_watcher.py (完全env適合化・M1/M2両対応決定版) - mmedia / pmedia を監視し、ファイル/フォルダの移動・リネーム(MOVED)を検出 - Google スプレッドシートへ [A=変更前, B=変更後, C=JST更新時刻, D=TRUE] を追記 - その後、(A,B)重複は「古い方のみ残す」に正規化して C 降順に整列 - 物理パスのハードコードを完全排除した、最高水準のポータブル設計です。 """ import os import sys import time import logging from datetime import datetime, timezone, timedelta from pathlib import Path from typing import List, Tuple, Dict from dotenv import load_dotenv # ===== 1. 環境設定（金庫 .env からの動的ロード ＆ 環境差の自動解決） ===== dotenv_loaded = False for dotenv_path in [ Path(__file__).resolve().parent / ".env", Path(__file__).resolve().parent.parent / ".env", Path.home() / "python_scripts" / ".env" ]: if dotenv_path.exists(): load_dotenv(dotenv_path) dotenv_loaded = True break # --- 監視対象ディレクトリ（M1/M2 Mac双方の外付けSSD・myboxに対応） --- WATCH_DIRS = [ "/Volumes/NO3_SSD/mybox/mybox_1/mmedia", "/Volumes/NO3_SSD/mybox/mybox_1/pmedia", ] # --- ログファイル --- LOG_FILE = os.path.expanduser("~/Library/Logs/folder_watch.log") # --- Google Sheets --- # 金庫の定義を優先、なければエイリアス（F3_SPREADSHEET_ID）等からロード SPREADSHEET_ID = os.getenv("<固有のランダム文字列>") or os.getenv("F3_SPREADSHEET_ID") SHEET_NAME = "sheet1" HAS_HEADER = True HEADERS = ["変更前パス", "変更後パス", "更新日時", "更新フラグ"] # --- 認証キー(JSON) --- SERVICE_ACCOUNT_JSON = os.getenv("RENAME_WATCHER_SA_JSON_PATH") or os.getenv("SA_JSON_PATH") # --- 2. 鉄壁の自律停止ガード（安全な起動時チェック） --- if not SPREADSHEET_ID or not SERVICE_ACCOUNT_JSON: print("❌ [file_rename_watcher.py] 金庫(.env)に必要な SPREADSHEET_ID または SERVICE_ACCOUNT_JSON が設定されていません。安全に自律停止します。", file=sys.stderr) sys.exit(1) # パスの実在チェック sa_key_path = Path(SERVICE_ACCOUNT_JSON).expanduser().resolve() if not sa_key_path.exists(): print(f"❌ [file_rename_watcher.py] サービスアカウントJSONキーが存在しません: {sa_key_path}。安全に自律停止します。", file=sys.stderr) sys.exit(1) # --- 無視フィルタ --- IGNORE_SUFFIXES = (".DS_Store",) IGNORE_SUBSTRINGS = ( "/.dropbox", "/.Trash", "/.~", "/.TemporaryItems", "/pmedia/scripts_pub/", # scripts_pub は F3 対象外 ".sb-", # 一時ファイル（sb-xxxxx）を除外 ) # ========= ロギング ========= os.makedirs(os.path.dirname(LOG_FILE), exist_ok=True) logging.basicConfig( filename=LOG_FILE, level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s", ) # ========= Google Sheets クライアント ========= import gspread from google.oauth2.service_account import Credentials def get_sheet(): scopes = ["https://www.googleapis.com/auth/spreadsheets"] creds = Credentials.from_service_account_file(str(sa_key_path), scopes=scopes) gc = gspread.authorize(creds) return gc.open_by_key(SPREADSHEET_ID).worksheet(SHEET_NAME) def jst_now() -> datetime: return datetime.now(timezone(timedelta(hours=9))) def jst_now_str() -> str: return jst_now().strftime("%Y-%m-%d %H:%M:%S") def ensure_header(ws): if not HAS_HEADER: return try: # A1セルに値があるか val = ws.cell(1, 1).value if not val: ws.update(values=[HEADERS], range_name="A1") except Exception as e: logging.error(f"ensure_header error: {e}") def load_sheet(ws) -> Tuple[List[str], List[List[str]]]: all_vals = ws.get_all_values() if not all_vals: return [], [] if HAS_HEADER: return all_vals[0], all_vals[1:] return [], all_vals def parse_time(s: str) -> datetime: try: return datetime.strptime(s, "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone(timedelta(hours=9))) except Exception: return datetime.min.replace(tzinfo=timezone(timedelta(hours=9))) def dedupe_oldest_keep_only(rows: List[List[str]]) -> List[List[str]]: """ (A=変更前, B=変更後) が重複しているレコードについて、 「最も古いレコード(更新日時が最小)」のみを残す """ # (before, after) -> (original_row, original_idx, time_obj, flag) keep: Dict[Tuple[str, str], Tuple[str, str, datetime, str]] = {} for r in rows: if len(r) < 4: continue a, b, t_str, d = r[0].strip(), r[1].strip(), r[2].strip(), r[3].strip() if not a or not b: continue t = parse_time(t_str) key = (a, b) if key not in keep: keep[key] = (a, b, t, d) else: old_a, old_b, old_t, old_d = keep[key] if t < old_t: # 古いほうを残す keep[key] = (a, b, t, d) out = [[a, b, t.strftime("%Y-%m-%d %H:%M:%S"), d] for (a, b, t, d) in keep.values()] out.sort(key=lambda r: r[2], reverse=True) return out def append_and_cleanup(before_path: str, after_path: str, when_str: str): ws = get_sheet() ensure_header(ws) # 新規行を末尾に追加 ws.append_row([before_path, after_path, when_str, "TRUE"], value_input_option="RAW") # 全体を読み込み → 正規化 → 書き戻し header, rows = load_sheet(ws) deduped = dedupe_oldest_keep_only(rows) if HAS_HEADER: ws.clear() ws.update(values=[HEADERS], range_name="A1") if deduped: ws.update(values=deduped, range_name="A2", value_input_option="RAW") else: ws.clear() if deduped: ws.update(values=deduped, range_name="A1", value_input_option="RAW") # ========= 監視パート ========= from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler, FileSystemMovedEvent def is_ignored(path: str) -> bool: if any(s in path for s in IGNORE_SUBSTRINGS): return True if path.endswith(IGNORE_SUFFIXES): return True return False class RenameHandler(FileSystemEventHandler): def on_moved(self, event: FileSystemMovedEvent): try: src = event.src_path dst = event.dest_path if is_ignored(src) or is_ignored(dst): return ts = jst_now_str() logging.info(f"MOVED src={src} dst={dst} ts={ts}") append_and_cleanup(src, dst, ts) except Exception as e: logging.exception(f"on_moved error: {e}") def main(): event_handler = RenameHandler() observer = Observer() scheduled = 0 for d in WATCH_DIRS: if os.path.exists(d): observer.schedule(event_handler, d, recursive=True) scheduled += 1 else: logging.warning(f"watch dir not found: {d}") if scheduled == 0: logging.error("no watch dirs scheduled; exiting") return observer.start() logging.info(f"watch started on {scheduled} dirs") try: while True: time.sleep(1) except KeyboardInterrupt: observer.stop() except Exception as e: logging.exception(f"main loop error: {e}") observer.stop() observer.join() if __name__ == "__main__": main() ``` <任意のフォルダパス>/com.XXXXXX.filewatcher.plist ``` Label com.XXXXXX.filewatcher ProgramArguments /usr/bin/python3 <任意のフォルダパス>/file_rename_watcher.py RunAtLoad StandardOutPath <任意のフォルダパス>/filewatcher.log StandardErrorPath <任意のフォルダパス>/filewatcher_error.log ```