# [メタ情報] # 識別子: DAS棚田管理_実データ更新処理_exe # システム名: DAS棚田管理_実データ更新処理 # 技術種別: Misc # 機能名: Misc # 使用言語: Python ShellScript # 状態: 実行用 env終了 # [/メタ情報] 要約： このシステムは、Google Sheetsをインターフェースとしたファイル管理の自動化パイプラインです。`tanada_apply_button_daemon.py`は、Google Sheets上の手動ボタン（A10）によるAPPLY要求のみを監視し、深夜時間帯を除いて`tanada_apply.py`を起動します。`apply_runner.sh`は、SheetsのA8セルからファイル移動計画DBのパスを読み込み、それを固定名の適用DBにコピーし、`tanada_apply.py`を実行するためのシェルスクリプトです。コアロジックを担う`tanada_apply.py`は、適用DBに記録されたアクション（主にファイルの移動）をDRYRUNまたはAPPLYモードで実行します。このスクリプトは、「現段位がT0ではないファイルを新段位T0へ移動する」ポリシー違反を最優先でチェックし、違反があれば直ちに処理を停止する安全機構を備えています。処理結果はSheetsの指定セルにフィードバックされます。`tanada_make_apply_plan_db.py`は、シミュレーション結果DBから、`tanada_apply.py`が使用する固定名の適用計画DBを生成し、古いDBは世代管理してアーカイブします。`com.XXXXXX.tanada_apply_button_daemon.plist`は、デーモンをmacOSのLaunchAgentとして自動起動・常駐させ、必要な環境変数を設定します。このパイプラインは、手動トリガーと厳格な安全ポリシーにより、ファイル移動の誤操作を防ぎながらも自動化を実現しています。 ``` ################################################################################ ### 1) tanada_apply_button_daemon.py ################################################################################ #!/usr/bin/env python3 # -*- coding: utf-8 -*- from __future__ import annotations """ 【最重要：apply 起動制約（手動ボタンのみ）】 このデーモンは、Google Sheets の action_result シート上の 「A10に重ねた手動ボタン」が書き込む APPLY 要求セルだけを監視し、 そのセルが更新された場合に限り tanada_apply.py を起動する。 - 禁止：時間トリガー、定期実行、他セル(B2等)からの混線起動 - 許可：ユーザーが action_result シートの A10ボタンを手動で押した場合のみ 追加要件: - 「古すぎる plan_id を使った apply」を拒否して表示する（例：1時間以上前は中止） → 固定の apply_plan DB ではなく、Sheets に表示されている “最新 plan DB（A8）” を見て判定する - REQUESTED が古い（例：前日など）場合は「押していない残骸」とみなし、 CANCEL表示を出さずに REQUESTED を黙って自動クリアする。 - 実行結果が DONE / CANCEL / ERROR のいずれでも、最後に REQUESTED を必ずクリアする。 """ import os import sys import time import subprocess import sqlite3 from pathlib import Path from datetime import datetime, time as dtime from dotenv import load_dotenv # Google Client API Imports from google.oauth2.service_account import Credentials from googleapiclient.discovery import build # ====== 金庫（.env）の動的ロード ====== load_dotenv(Path.home() / "python_scripts" / ".env") # ====== apply専用（simulateと混線させない）====== SHEET_NAME = "action_result" APPLY_REQ = "B10" # A10ボタンが書く要求セル（例：REQUESTED:20260113_140050） APPLY_STATUS = "A11" # 状態表示（短文） PLAN_PATH_CELL = "A8" # simulate が出した最新 plan DB パス POLL_SEC = 5 # ===== 静音時間帯設定（深夜は実行しない）===== QUIET_START = dtime(0, 55) QUIET_END = dtime(2, 10) QUIET_SLEEP = 60 # ===== 実行する apply 本体 ===== PYTHON = "/usr/bin/python3" APPLY_SCRIPT = str(Path.home() / "python_scripts" / "tanada_apply.py") # ===== 古すぎる plan を拒否する閾値（秒）===== MAX_PLAN_AGE_SEC = 3600 # 1時間 # ===== 重複実行防止（永続）===== LAST_REQ_FILE = Path.home() / "das" / "meta" / "_last_request_action_result_B2.txt" def get_sheets_service(): cred_path = os.getenv("DAS_CRED_JSON") if not cred_path: raise ValueError("Environment variable DAS_CRED_JSON is not set.") p = Path(cred_path) if not p.is_absolute(): p = Path.home() / p creds = Credentials.from_service_account_file( str(p), scopes=["https://www.googleapis.com/auth/spreadsheets"] ) return build("sheets", "v4", credentials=creds) def get_sheet_values(service, spreadsheet_id, range_name): result = service.spreadsheets().values().get( spreadsheetId=spreadsheet_id, range=range_name ).execute() return result.get("values", []) def update_sheet_cell(service, spreadsheet_id, range_name, value): body = {"values": [[value]]} service.spreadsheets().values().update( spreadsheetId=spreadsheet_id, range=range_name, valueInputOption="USER_ENTERED", body=body ).execute() def check_quiet_time() -> bool: now = datetime.now().time() if QUIET_START <= now <= QUIET_END: return True return False def main(): spreadsheet_id = os.getenv("DAS_SHEET_ID") if not spreadsheet_id: print("[ERROR] DAS_SHEET_ID is not set in .env", file=sys.stderr) sys.exit(1) try: service = get_sheets_service() except Exception as e: print(f"[ERROR] Failed to initialize Sheets API: {e}", file=sys.stderr) sys.exit(1) LAST_REQ_FILE.parent.mkdir(parents=True, exist_ok=True) print(f"[INFO] Daemon started. Monitoring {SHEET_NAME}!{APPLY_REQ}...") while True: try: if check_quiet_time(): time.sleep(QUIET_SLEEP) continue range_to_read = f"{SHEET_NAME}!A1:B15" rows = get_sheet_values(service, spreadsheet_id, range_to_read) if not rows: time.sleep(POLL_SEC) continue val_b10 = "" val_a8 = "" if len(rows) > 9 and len(rows[9]) > 1: val_b10 = rows[9][1].strip() if len(rows) > 7 and len(rows[7]) > 0: val_a8 = rows[7][0].strip() if not val_b10.startswith("REQUESTED:"): time.sleep(POLL_SEC) continue req_time_str = val_b10.replace("REQUESTED:", "").strip() try: req_dt = datetime.strptime(req_time_str, "%Y%m%d_%H%M%S") except ValueError: print(f"[ERROR] Invalid format in B10: {val_b10}", file=sys.stderr) update_sheet_cell(service, spreadsheet_id, f"{SHEET_NAME}!{APPLY_STATUS}", "ERROR: Invalid Format") update_sheet_cell(service, spreadsheet_id, f"{SHEET_NAME}!{APPLY_REQ}", "") time.sleep(POLL_SEC) continue age_sec = (datetime.now() - req_dt).total_seconds() if age_sec > 86400: print(f"[INFO] Silent clear of old request: {val_b10} (age: {age_sec}s)") update_sheet_cell(service, spreadsheet_id, f"{SHEET_NAME}!{APPLY_REQ}", "") time.sleep(POLL_SEC) continue last_req = "" if LAST_REQ_FILE.exists(): last_req = LAST_REQ_FILE.read_text().strip() if val_b10 == last_req: time.sleep(POLL_SEC) continue print(f"[INFO] Processing request: {val_b10}") LAST_REQ_FILE.write_text(val_b10) if age_sec > MAX_PLAN_AGE_SEC: print(f"[ERROR] Request is too old. age={age_sec}s") update_sheet_cell(service, spreadsheet_id, f"{SHEET_NAME}!{APPLY_STATUS}", "CANCEL: Plan too old") update_sheet_cell(service, spreadsheet_id, f"{SHEET_NAME}!{APPLY_REQ}", "") continue runner_script = Path.home() / "python_scripts" / "apply_runner.sh" update_sheet_cell(service, spreadsheet_id, f"{SHEET_NAME}!{APPLY_STATUS}", "RUNNING") res = subprocess.run([str(runner_script)], capture_output=True, text=True) if res.returncode == 0: print("[INFO] Apply success") update_sheet_cell(service, spreadsheet_id, f"{SHEET_NAME}!{APPLY_STATUS}", "DONE") else: print(f"[ERROR] Apply failed: {res.stderr}", file=sys.stderr) update_sheet_cell(service, spreadsheet_id, f"{SHEET_NAME}!{APPLY_STATUS}", "ERROR: Run failed") update_sheet_cell(service, spreadsheet_id, f"{SHEET_NAME}!{APPLY_REQ}", "") except Exception as e: print(f"[ERROR] Daemon exception: {e}", file=sys.stderr) try: update_sheet_cell(service, spreadsheet_id, f"{SHEET_NAME}!{APPLY_STATUS}", "ERROR: Exception") update_sheet_cell(service, spreadsheet_id, f"{SHEET_NAME}!{APPLY_REQ}", "") except Exception: pass time.sleep(POLL_SEC) if __name__ == "__main__": main() ``` ``` ################################################################################ ### 2) apply_runner.sh ################################################################################ #!/bin/bash set -euo pipefail export PATH="/usr/bin:/bin:/usr/sbin:/sbin" LOCK_DIR="$HOME/das/locks" mkdir -p "$LOCK_DIR" LOCK_FILE="$LOCK_DIR/apply_pipeline.lock" # ---- 設定（固定名DB）---- APPLY_PLAN_DB="$HOME/das/meta/apply_plan__tanada_items_ALL.sqlite" TANADA_APPLY_PY="$HOME/python_scripts/tanada_apply.py" # Sheets 参照（.env から自動ロード） SHEET_NAME="${SHEET_NAME:-action_result}" PLAN_PATH_CELL="${PLAN_PATH_CELL:-A8}" APPLY_STATUS_CELL="${APPLY_STATUS_CELL:-A11}" APPLY_COMMENT_CELL="${APPLY_COMMENT_CELL:-A12}" # 二重起動だけ防止（mkdirロック） LOCK_DIR_PATH="${LOCK_DIR}/apply_pipeline.lockdir" if ! /bin/mkdir "${LOCK_DIR_PATH}" 2>/dev/null; then echo "[INFO] apply_runner: already running (lock=${LOCK_DIR_PATH})" >&2 exit 0 fi cleanup() { /bin/rmdir "${LOCK_DIR_PATH}" 2>/dev/null || true; } trap cleanup EXIT INT TERM # A8 から plan DB パスを取得（Pythonで読む） PLAN_DB=$(/usr/bin/python3 - <<'PY' import os import sys from pathlib import Path from dotenv import load_dotenv from google.oauth2.service_account import Credentials from googleapiclient.discovery import build load_dotenv(Path.home() / "python_scripts" / ".env") sid = os.environ.get("DAS_SHEET_ID", "").strip() cred = os.environ.get("DAS_CRED_JSON", "").strip() if not sid or not cred: print("", end="") sys.exit(0) p = Path(cred) if not p.is_absolute(): p = Path.home() / p try: creds = Credentials.from_service_account_file( str(p), scopes=["https://www.googleapis.com/auth/spreadsheets"] ) service = build("sheets", "v4", credentials=creds) result = service.spreadsheets().values().get( spreadsheetId=sid, range="action_result!A8" ).execute() values = result.get("values", []) if values and values[0]: print(values[0][0].strip(), end="") except Exception: print("", end="") PY ) if [ -z "${PLAN_DB}" ]; then echo "[ERROR] Could not resolve Plan DB Path from Sheet A8" >&2 exit 1 fi # 動的ユーザー名解決 CLEANED_PLAN_DB=$(echo "${PLAN_DB}" | perl -pe "s|^/Users/[^/]+|$HOME|") if [ ! -f "${CLEANED_PLAN_DB}" ]; then echo "[ERROR] Resolved DB path does not exist: ${CLEANED_PLAN_DB}" >&2 exit 1 fi mkdir -p "$(dirname "${APPLY_PLAN_DB}")" cp "${CLEANED_PLAN_DB}" "${APPLY_PLAN_DB}" # tanada_apply.py の実行 /usr/bin/python3 "${TANADA_APPLY_PY}" --db-path "${APPLY_PLAN_DB}" ``` ``` ################################################################################ ### 3) tanada_apply.py ################################################################################ #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ tanada_apply.py - plan-db を読み、items_all の action に従って DRYRUN / APPLY を行う。 - 方針: 「T0戻り禁止」を最優先で検査。違反があれば即終了。 """ from __future__ import annotations import argparse import os import sys import sqlite3 import shutil from dataclasses import dataclass from datetime import datetime from pathlib import Path from typing import Optional, List, Tuple from dotenv import load_dotenv # Google Client API Imports from google.oauth2.service_account import Credentials from googleapiclient.discovery import build # ====== 金庫（.env）の動的ロード ====== load_dotenv(Path.home() / "python_scripts" / ".env") @dataclass class ApplyItem: id: int src_path: str dst_path: str action: str status: str def get_sheets_service(): cred_path = os.getenv("DAS_CRED_JSON") if not cred_path: raise ValueError("Environment variable DAS_CRED_JSON is not set.") p = Path(cred_path) if not p.is_absolute(): p = Path.home() / p creds = Credentials.from_service_account_file( str(p), scopes=["https://www.googleapis.com/auth/spreadsheets"] ) return build("sheets", "v4", credentials=creds) def update_sheet_comment(service, spreadsheet_id, cell_range, text): try: body = {"values": [[text]]} service.spreadsheets().values().update( spreadsheetId=spreadsheet_id, range=cell_range, valueInputOption="USER_ENTERED", body=body ).execute() except Exception as e: print(f"[WARN] Failed to write sheet comment: {e}", file=sys.stderr) def validate_t0_reversal(src: str, dst: str) -> bool: """ T0戻り（逆流）禁止チェックのコアロジック。 T0ディレクトリ（アーカイブ等の最深部）から、より浅い作業フォルダへ ファイルが戻るような危険な移動操作を検出した場合、True（違反あり）を返す。 """ src_p = Path(src) dst_p = Path(dst) src_is_t0 = any(part.lower() in ("t0", "archive", "archived") for part in src_p.parts) dst_is_t0 = any(part.lower() in ("t0", "archive", "archived") for part in dst_p.parts) if src_is_t0 and not dst_is_t0: return True return False def resolve_path_dynamically(path_str: str) -> Path: """ パス内の /Users/XXXXXX/ などの環境固有ユーザー名を、 現在の実行環境のホームディレクトリ（Path.home()）に動的に解決する。 """ p = Path(path_str) parts = list(p.parts) if len(parts) > 2 and parts[1] == "Users": real_home = Path.home() relative_parts = parts[3:] return real_home.joinpath(*relative_parts) return p def main(): parser = argparse.ArgumentParser(description="DAS Apply Process") parser.add_argument("--db-path", type=str, required=True, help="Path to plan SQLite DB") parser.add_argument("--dryrun", action="store_true", help="Dry run mode") args = parser.parse_args() spreadsheet_id = os.getenv("DAS_SHEET_ID") status_cell = "action_result!A11" comment_cell = "action_result!A12" service = None if spreadsheet_id: try: service = get_sheets_service() except Exception as e: print(f"[WARN] Sheets API bypass: {e}", file=sys.stderr) db_path = resolve_path_dynamically(args.db_path) if not db_path.exists(): msg = f"Plan DB not found at: {db_path}" print(f"[ERROR] {msg}", file=sys.stderr) if service and spreadsheet_id: update_sheet_comment(service, spreadsheet_id, comment_cell, msg) sys.exit(1) conn = sqlite3.connect(db_path) cursor = conn.cursor() cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='items_all'") if not cursor.fetchone(): msg = "Table 'items_all' not found in Plan DB" print(f"[ERROR] {msg}", file=sys.stderr) if service and spreadsheet_id: update_sheet_comment(service, spreadsheet_id, comment_cell, msg) conn.close() sys.exit(1) cursor.execute("SELECT id, src_path, dst_path, action, status FROM items_all") rows = cursor.fetchall() items = [ApplyItem(r[0], r[1], r[2], r[3], r[4]) for r in rows] print(f"[INFO] Loaded {len(items)} items from plan DB.") # 1. 最優先バリデーション：T0戻り禁止の検証 reversal_violations = [] for item in items: if item.action in ("MOVE", "RENAME"): if validate_t0_reversal(item.src_path, item.dst_path): reversal_violations.append(item) if reversal_violations: violation_details = ", ".join([f"[ID:{i.id}] {i.src_path} -> {i.dst_path}" for i in reversal_violations]) msg = f"T0 Reversal Violation Detected! Process Aborted. Details: {violation_details}" print(f"[CRITICAL] {msg}", file=sys.stderr) if service and spreadsheet_id: update_sheet_comment(service, spreadsheet_id, comment_cell, msg) conn.close() sys.exit(2) # 2. 移動の実行 success_count = 0 fail_count = 0 ignored_count = 0 for item in items: if item.action in ("KEEP", "IGNORE", "NONE", ""): ignored_count += 1 continue src_p = resolve_path_dynamically(item.src_path) dst_p = resolve_path_dynamically(item.dst_path) # 実移動 (MOVE or RENAME) if item.action in ("MOVE", "RENAME"): if not src_p.exists(): print(f"[WARN] Source path does not exist: {src_p}") cursor.execute("UPDATE items_all SET status='ERROR: Src Missing' WHERE id=?", (item.id,)) fail_count += 1 continue if args.dryrun: print(f"[DRYRUN] Would move: {src_p} -> {dst_p}") cursor.execute("UPDATE items_all SET status='DRYRUN_OK' WHERE id=?", (item.id,)) success_count += 1 else: try: dst_p.parent.mkdir(parents=True, exist_ok=True) shutil.move(str(src_p), str(dst_p)) cursor.execute("UPDATE items_all SET status='APPLIED' WHERE id=?", (item.id,)) success_count += 1 except Exception as e: print(f"[ERROR] Failed to move {src_p} to {dst_p}: {e}", file=sys.stderr) cursor.execute("UPDATE items_all SET status='ERROR_MOVE_FAILED' WHERE id=?", (item.id,)) fail_count += 1 # 削除 (DELETE) elif item.action == "DELETE": if not src_p.exists(): cursor.execute("UPDATE items_all SET status='DELETE_BYPASSED: Src Missing' WHERE id=?", (item.id,)) ignored_count += 1 continue if args.dryrun: print(f"[DRYRUN] Would delete: {src_p}") cursor.execute("UPDATE items_all SET status='DRYRUN_OK' WHERE id=?", (item.id,)) success_count += 1 else: try: if src_p.is_dir(): shutil.rmtree(src_p) else: src_p.unlink() cursor.execute("UPDATE items_all SET status='DELETED' WHERE id=?", (item.id,)) success_count += 1 except Exception as e: print(f"[ERROR] Failed to delete {src_p}: {e}", file=sys.stderr) cursor.execute("UPDATE items_all SET status='ERROR_DELETE_FAILED' WHERE id=?", (item.id,)) fail_count += 1 conn.commit() conn.close() summary_msg = f"Apply Finished. Success: {success_count}, Fail: {fail_count}, Ignored: {ignored_count}" print(f"[INFO] {summary_msg}") if service and spreadsheet_id: update_sheet_comment(service, spreadsheet_id, comment_cell, summary_msg) if fail_count > 0: sys.exit(3) if __name__ == "__main__": main() ``` ``` ################################################################################ ### 4) tanada_make_apply_plan_db.py ################################################################################ #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ tanada_make_apply_plan_db.py - simulate結果DBをコピーして、実データ更新用DBとして確定する。 - アーカイブディレクトリへの履歴退避 """ from __future__ import annotations import argparse import shutil from dataclasses import dataclass from datetime import datetime from pathlib import Path from typing import List, Optional, Tuple # ====== パスの動的解決 ====== DEFAULT_META_DIR = Path.home() / "das" / "meta" APPLY_PLAN_FIXED_NAME = "apply_plan__tanada_items_ALL.sqlite" ARCHIVE_DIR_NAME = "_archive" def main(): parser = argparse.ArgumentParser(description="Create Apply Plan DB from latest simulation DB") parser.add_argument("--meta-dir", type=str, default=str(DEFAULT_META_DIR), help="Metadata directory") args = parser.parse_args() meta_dir = Path(args.meta_dir) if not meta_dir.exists(): print(f"[ERROR] Meta directory does not exist: {meta_dir}") return sim_dbs = list(meta_dir.glob("simulate_result__tanada_items_ALL_*.sqlite")) if not sim_dbs: print("[ERROR] No simulation DBs found in meta directory.") return sim_dbs.sort(key=lambda x: x.stat().st_mtime, reverse=True) latest_sim_db = sim_dbs[0] print(f"[INFO] Latest simulation DB found: {latest_sim_db.name}") archive_dir = meta_dir / ARCHIVE_DIR_NAME archive_dir.mkdir(exist_ok=True) apply_plan_db = meta_dir / APPLY_PLAN_FIXED_NAME if apply_plan_db.exists(): timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") archive_name = f"apply_plan__tanada_items_ALL_{timestamp}.sqlite" archive_path = archive_dir / archive_name print(f"[INFO] Archiving current apply plan: {apply_plan_db.name} -> {archive_path.name}") shutil.copy2(apply_plan_db, archive_path) print(f"[INFO] Deploying new apply plan: {latest_sim_db.name} -> {apply_plan_db.name}") shutil.copy2(latest_sim_db, apply_plan_db) print("[INFO] DB creation completed successfully.") if __name__ == "__main__": main() ``` ``` ################################################################################ ### 5) com.XXXXXX.tanada_apply_button_daemon.plist ################################################################################ Label com.XXXXXX.tanada_apply_button_daemon ProgramArguments /usr/bin/python3 /Users/XXXXXX/python_scripts/tanada_apply_button_daemon.py RunAtLoad KeepAlive StandardOutPath /Users/XXXXXX/das/logs/tanada_apply_button_daemon.out.log StandardErrorPath /Users/XXXXXX/das/logs/tanada_apply_button_daemon.err.log ```