# [メタ情報] # 識別子: videoembedjson生成送信videoctjson送信_python_exe # 補足: 実行用 env終了 # [/メタ情報] 要約: 本システムは、Googleスプレッドシート(F1_2:名前関係付け、F2:シート1)の動画データ更新をトリガーに、動画埋め込みコードの自動生成、JSONファイル作成、XServerへのアップロードを自動化するPythonスクリプト群です。 メインスクリプト`kick_flags.py`は、F1_2のI列またはF2 of B/U列にTRUEが検出されると起動。多重起動防止やエラー再試行機能を備え、以下の手順を実行します。まず`sync_video_embedcode.py`が、F1_2のルールに基づきF2シートのS列(埋め込みコード)を再生成し、U列(更新状況)をTRUEに設定後、F1_2のI列をFALSEに戻します。次に`generate_videoembed_json.py`でF2シートから動画情報を抽出し`videoembed.json`を生成。これを`xserver_uploader.py`がXServerへアップロードし、最後にF2のB/U列フラグをFALSEに戻します。 本システムはmacOSのLaunchAgentにより10秒間隔で自動実行されます。AppSheetで新規動画を追加する際、F2のU列初期値をTRUEに設定することで、この自動更新処理が確実に発火するよう運用されています。 M1 Mac mini === Python(kick_flags.py) === /Users/XXXXXX/python_scripts/kick_flags.py ```python #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ kick_flags.py (env化・M1 Mac本番稼働用) - 発火条件(どれか1つでも TRUE なら発火) (A) F1_2(名前関係付け)I列(更新フラグ) に TRUE が1件以上 (B) F2(シート1) B列(更新状況) or U列(埋め込みコード更新状況) に TRUE が1件以上 """ import warnings warnings.filterwarnings("ignore", category=UserWarning) import os import sys import time import subprocess import logging from contextlib import contextmanager from pathlib import Path from dotenv import load_dotenv import gspread from google.oauth2.service_account import Credentials # --- 環境変数 (.env) の動的ロード --- SCRIPT_DIR = Path(__file__).resolve().parent load_dotenv(SCRIPT_DIR / ".env") # ====== 環境変数からの読み込み ====== SA_JSON_PATH = os.getenv("GOOGLE_APPLICATION_CREDENTIALS") SPREADSHEET_F2_ID = os.getenv("F2_SPREADSHEET_ID") SHEET_NAME_F2 = os.getenv("F2_SHEET_NAME", "シート1") SPREADSHEET_F12_ID = os.getenv("F12_SPREADSHEET_ID") SHEET_NAME_F12 = os.getenv("F12_SHEET_NAME", "名前関係付け") PYTHON_BIN = os.getenv("PYTHON_BIN", "/opt/homebrew/bin/python3") if not SA_JSON_PATH or not SPREADSHEET_F2_ID or not SPREADSHEET_F12_ID: sys.exit("Error: Required environment variables are not set in .env") # 列定義 (0-based) F2_COL_B_UPDATED = 1 # B列: 更新状況 F2_COL_U_EMB_STATUS = 20 # U列: 埋め込みコード更新状況 F12_COL_I_UPDATE = 8 # I列: 更新フラグ # スコープ SC_RO = ["https://www.googleapis.com/auth/spreadsheets.readonly"] SC_RW = ["https://www.googleapis.com/auth/spreadsheets"] # スクリプトの相対パス (オリジナル名の維持) SYNC_SCRIPT = str(SCRIPT_DIR / "sync_video_embedcode.py") GEN_EMBED_SCRIPT = str(SCRIPT_DIR / "generate_videoembed_json.py") UPLOADER_SCRIPT = str(SCRIPT_DIR / "xserver_uploader.py") # ログ & ロック LOG_PATH = os.path.expanduser("~/Library/Logs/kick_flags.log") LOCK_PATH = "/tmp/kick_flags.lock" LOCK_TTL_SEC = 180 def log(msg: str) -> None: s = msg if msg.endswith("\n") else msg + "\n" sys.stdout.write(s) sys.stdout.flush() try: with open(LOG_PATH, "a", encoding="utf-8") as f: f.write(f"{time.strftime('%Y-%m-%d %H:%M:%S')} {s}") except Exception: pass def _open(ss_id: str, tab_name: str, scopes: list): creds = Credentials.from_service_account_file(SA_JSON_PATH, scopes=scopes) gc = gspread.authorize(creds) return gc.open_by_key(ss_id).worksheet(tab_name) def _normalize_bool(s: str) -> str: if s is None: return "" return str(s).strip().lower().replace(" ", "") TRUE_TOKENS = {"true", "1", "yes", "on", "y", "t", "✓", "✔"} def _rows_true(values: list, col_0based: int) -> list[int]: hits = [] if not values: return hits for i, row in enumerate(values[1:], start=2): # シートの行番号 (1-based) if len(row) <= col_0based: continue if _normalize_bool(row[col_0based]) in TRUE_TOKENS: hits.append(i) return hits def _detect_triggers() -> tuple[list[int], list[int]]: ws_f12 = _open(SPREADSHEET_F12_ID, SHEET_NAME_F12, SC_RO) ws_f2 = _open(SPREADSHEET_F2_ID, SHEET_NAME_F2, SC_RO) rows_f12 = ws_f12.get_all_values() rows_f2 = ws_f2.get_all_values() i_hits = _rows_true(rows_f12, F12_COL_I_UPDATE) b_hits = _rows_true(rows_f2, F2_COL_B_UPDATED) u_hits = _rows_true(rows_f2, F2_COL_U_EMB_STATUS) fu_hits = sorted(set(b_hits + u_hits)) return i_hits, fu_hits def _run(cmd: list[str], name: str, retries: int = 3) -> bool: delay = 3 for attempt in range(1, retries + 1): try: log(f"👟 子プロセス起動: {name} (試行 {attempt})") subprocess.run(cmd, check=True) return True except Exception as e: log(f"⚠️ {name} 失敗: {e}") if attempt == retries: return False time.sleep(delay) delay *= 2 return False def _reset_f2_flags(rows_1based: list[int]) -> int: if not rows_1based: return 0 ws = _open(SPREADSHEET_F2_ID, SHEET_NAME_F2, SC_RW) updates = [] for r in rows_1based: updates.append({"range": f"B{r}:B{r}", "values": [["FALSE"]]}) updates.append({"range": f"U{r}:U{r}", "values": [["FALSE"]]}) if updates: ws.batch_update(updates) return len(rows_1based) return 0 @contextmanager def _lock(path: str, ttl: int): now = time.time() if os.path.exists(path): try: st = os.stat(path) if now - st.st_mtime <= ttl: log("⚠️ 他のkick_flagsが実行中。スキップ。") yield False return except Exception: pass try: with open(path, "w") as f: f.write(str(os.getpid())) yield True finally: try: os.remove(path) except Exception: pass def main(): log("===== kick_flags.py: 起動(F12+F2スキャン) =====") with _lock(LOCK_PATH, LOCK_TTL_SEC) as acquired: if not acquired: return # 伝播遅延対策(2回読み) i_hits, fu_hits = _detect_triggers() if not (i_hits or fu_hits): time.sleep(1.5) i_hits, fu_hits = _detect_triggers() if not (i_hits or fu_hits): log("(フラグなし、終了)") return if i_hits: log(f"✅ F1_2.I TRUE 検出: rows={i_hits}") if fu_hits: log(f"✅ F2.B/U TRUE 検出: rows={fu_hits}") # 1) S列の再生成 sync_cmd = [PYTHON_BIN, SYNC_SCRIPT] if fu_hits: sync_cmd.append("--force") if not _run(sync_cmd, "sync_video_embedcode.py", retries=4): log("⛔ 同期失敗。処理を中断します。") return # 2) JSONファイルの生成 if not _run([PYTHON_BIN, GEN_EMBED_SCRIPT], "generate_videoembed_json.py", retries=3): log("⚠️ JSON生成失敗。") return # 3) Xserverへのアップロード # ★ env化・動的解決(金庫をロード) local_json_path = os.getenv("VIDEOEMBED_OUTPUT_PATH") or os.path.expanduser("~/meta_gd_wp_data/videoembed.json") if not _run([PYTHON_BIN, UPLOADER_SCRIPT, local_json_path], "xserver_uploader.py", retries=3): log("⚠️ Xserverへのアップロードに失敗。") return # 4) F2 の B/U を FALSE に戻す if fu_hits: restored = _reset_f2_flags(fu_hits) log(f"🧹 F2フラグ戻し: {restored} 行を FALSE に更新完了") log("🎉 全処理完了") if __name__ == "__main__": os.makedirs(os.path.dirname(LOG_PATH), exist_ok=True) main() ``` ================================================================================ === Python(sync_video_embedcode.py) === /Users/XXXXXX/python_scripts/sync_video_embedcode.py ```python #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ sync_video_embedcode.py (env化対応版) - F1_2(名前関係付け)の I列 に TRUE が1件でもあれば、 """ import warnings warnings.filterwarnings("ignore", category=UserWarning) import os import sys import json import argparse import time from pathlib import Path from dotenv import load_dotenv import gspread from google.oauth2.service_account import Credentials from typing import Dict, Tuple # --- 環境変数 (.env) の動的ロード --- SCRIPT_DIR = Path(__file__).resolve().parent load_dotenv(SCRIPT_DIR / ".env") # ====== 環境変数からの読み込み ====== SA_JSON_PATH = os.getenv("GOOGLE_APPLICATION_CREDENTIALS") SPREADSHEET_F2_ID = os.getenv("F2_SPREADSHEET_ID") SHEET_NAME_F2 = os.getenv("F2_SHEET_NAME", "シート1") SPREADSHEET_F12_ID = os.getenv("F12_SPREADSHEET_ID") SHEET_NAME_F12 = os.getenv("F12_SHEET_NAME", "名前関係付け") if not SA_JSON_PATH or not SPREADSHEET_F2_ID or not SPREADSHEET_F12_ID: sys.exit("Error: GOOGLE_APPLICATION_CREDENTIALS or spread sheet IDs are not set in env.") # スコープ SCOPES = ["https://www.googleapis.com/auth/spreadsheets"] # 列定義(0-based) F2_COL_E_LOCATION = 4 # E: 動画所在場所 F2_COL_S_EMBED = 18 # S: 埋め込みコード F2_COL_U_STATUS = 20 # U: 埋め込みコード更新状況(フラグ) F12_COL_A_LOCATION = 0 # A: 動画所在場所 F12_COL_F_TEMPLATE = 5 # F: 埋め込みテンプレート F12_COL_G_CONDITIONS = 6 # G: 対応条件(JSON) F12_COL_I_UPDATE_FLAG = 8 # I: 更新フラグ(TRUEで走る) def _auth(): creds = Credentials.from_service_account_file(SA_JSON_PATH, scopes=SCOPES) return gspread.authorize(creds) def _build_relation_map(rows_f12: list) -> Dict[str, Tuple[str, dict]]: relmap = {} for r in rows_f12[1:]: if len(r) < max(F12_COL_A_LOCATION, F12_COL_F_TEMPLATE, F12_COL_G_CONDITIONS) + 1: continue loc = str(r[F12_COL_A_LOCATION]).strip().lower() tmpl = str(r[F12_COL_F_TEMPLATE]).strip() cond_raw = str(r[F12_COL_G_CONDITIONS]).strip() if not loc or not tmpl: continue cond = {} if cond_raw: try: cond = json.loads(cond_raw) except Exception as e: print(f"⚠️ JSONパース失敗 ({loc}): {e}") relmap[loc] = (tmpl, cond) return relmap def _resolve_col_index_0based(col_name: str) -> int: s = col_name.strip().upper() if not s: return -1 num = 0 for char in s: num = num * 26 + (ord(char) - ord('A') + 1) return num - 1 def _regenerate_and_update(ws_f2, rows_f2: list, relmap: dict) -> bool: if len(rows_f2) <= 1: return False header = rows_f2[0] data_rows = rows_f2[1:] n_rows = len(rows_f2) updates = [] changed = False for idx, row in enumerate(data_rows, start=2): loc = str(row[F2_COL_E_LOCATION]).strip().lower() if F2_COL_E_LOCATION < len(row) else "" if loc not in relmap: continue tmpl, cond = relmap[loc] new_embed = tmpl for k, col_name in cond.items(): col_idx = _resolve_col_index_0based(col_name) col_val = "" if col_idx != -1 and col_idx < len(row): col_val = str(row[col_idx]).strip() new_embed = new_embed.replace(f"@@{k}@@", col_val) orig_embed = str(row[F2_COL_S_EMBED]).strip() if F2_COL_S_EMBED < len(row) else "" if new_embed != orig_embed: updates.append({"range": f"S{idx}:S{idx}", "values": [[new_embed]]}) updates.append({"range": f"U{idx}:U{idx}", "values": [["TRUE"]]}) changed = True if updates: ws_f2.batch_update(updates) print(f"F2: S/U を一括更新(行数={n_rows-1}、S変化あり={changed})") else: print("F2: 埋め込みコードの変更はありません。") return changed def main(): parser = argparse.ArgumentParser(description="Sync WordPress embed codes from F1_2 mapping.") parser.add_argument("--force", action="store_true", help="I=TRUE が無くても強制で全件再生成") args = parser.parse_args() print("=== sync_video_embedcode.py: start ===") gc = _auth() ws_f12 = gc.open_by_key(SPREADSHEET_F12_ID).worksheet(SHEET_NAME_F12) rows_f12 = ws_f12.get_all_values() hits = [ i for i, r in enumerate(rows_f12[1:], start=2) if len(r) > F12_COL_I_UPDATE_FLAG and (r[F12_COL_I_UPDATE_FLAG] or "").strip().upper() == "TRUE" ] if not args.force and not hits: print("I=TRUE が見つからないため、処理をスキップします。(--force で強制実行可)") print("=== sync_video_embedcode.py: done ===") return if not args.force: time.sleep(1.5) # AppSheet書き込み後の遅延対策 rows_f12 = ws_f12.get_all_values() relmap = _build_relation_map(rows_f12) ws_f2 = gc.open_by_key(SPREADSHEET_F2_ID).worksheet(SHEET_NAME_F2) rows_f2 = ws_f2.get_all_values() changed = _regenerate_and_update(ws_f2, rows_f2, relmap) if hits: data = [{"range": f"I{row}:I{row}", "values": [["FALSE"]]} for row in hits] ws_f12.batch_update(data) print(f"F1_2:I back to FALSE: {len(hits)}") print("=== sync_video_embedcode.py: done ===") if __name__ == "__main__": main() ``` ================================================================================ === Python(generate_videoembed_json.py) === /Users/XXXXXX/python_scripts/generate_videoembed_json.py ```python #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ generate_videoembed_json.py (env化対応版) - GAS版 videoembed.json と同一仕様・同一フォーマットで出力 """ import os import sys import json import gspread from google.oauth2.service_account import Credentials from pathlib import Path from dotenv import load_dotenv # --- 環境変数 (.env) の動的ロード --- SCRIPT_DIR = Path(__file__).resolve().parent load_dotenv(SCRIPT_DIR / ".env") # ====== 環境変数からの読み込み ====== SERVICE_JSON = os.getenv("GOOGLE_APPLICATION_CREDENTIALS") SHEET_ID = os.getenv("F2_SPREADSHEET_ID") SHEET_TAB = os.getenv("F2_SHEET_NAME", "シート1") # ★ env化・動的解決(金庫からロード、未指定時は本番用のファイル名へフォールバック) OUTPUT_PATH = os.getenv("VIDEOEMBED_OUTPUT_PATH") or os.path.expanduser("~/meta_gd_wp_data/videoembed.json") if not SERVICE_JSON or not SHEET_ID: sys.exit("Error: GOOGLE_APPLICATION_CREDENTIALS or F2_SPREADSHEET_ID is not set in .env") # ======= 列インデックス (0-based) ======= COL_A_PUBLISH = 0 # 公開/非公開 COL_C_VIDEOID = 2 # 動画ID COL_S_EMBED = 18 # WordPress埋め込みコード COL_W_VIDEO2 = 22 # 低画質動画URL (任意) # ======= 認証 ======= SCOPES = ["https://www.googleapis.com/auth/spreadsheets.readonly"] creds = Credentials.from_service_account_file(SERVICE_JSON, scopes=SCOPES) gc = gspread.authorize(creds) def _safe_get(row, idx): return row[idx] if idx < len(row) else "" def _to_int_or_str(v): s = str(v).strip() if not s: return None try: i = int(float(s)) return i except Exception: return s def _is_valid_embed(embed): if embed is None: return False e = str(embed).strip() if e in ("", "null", '""'): return False return True def generate_videoembed_json(): ws = gc.open_by_key(SHEET_ID).worksheet(SHEET_TAB) all_rows = ws.get_all_values() if not all_rows or len(all_rows) <= 1: print("[generate_videoembed_json] no data") with open(OUTPUT_PATH, "w", encoding="utf-8") as f: json.dump([], f, ensure_ascii=False, indent=2) return OUTPUT_PATH output = [] # 2行目からループ for row in all_rows[1:]: pub = _safe_get(row, COL_A_PUBLISH).strip() vid_raw = _safe_get(row, COL_C_VIDEOID).strip() embed = _safe_get(row, COL_S_EMBED).strip() video2 = _safe_get(row, COL_W_VIDEO2).strip() if pub == "公開" and vid_raw and _is_valid_embed(embed): vid = _to_int_or_str(vid_raw) if vid is None: continue item = { "videoid": vid, "embedCode": embed.strip() } if video2: item["video2"] = video2 output.append(item) # ソートして固定 output.sort(key=lambda x: str(x["videoid"])) # ★ 1. 新しく書き込む予定のJSONテキストをメモリ上に作成 new_json_str = json.dumps(output, ensure_ascii=False, indent=2, sort_keys=True) # ★ 2. 既存のファイルの中身を読み込む old_json_str = "" if os.path.exists(OUTPUT_PATH): with open(OUTPUT_PATH, "r", encoding="utf-8") as f: old_json_str = f.read() # ★ 3. 中身に変化があった時だけ「上書き保存」を実行する(原子的交換で安全に書き換え) if new_json_str != old_json_str: os.makedirs(os.path.dirname(OUTPUT_PATH), exist_ok=True) tmp_path = OUTPUT_PATH + ".tmp" with open(tmp_path, "w", encoding="utf-8") as f: f.write(new_json_str) os.replace(tmp_path, OUTPUT_PATH) print(f"[generate_videoembed_json] wrote: {OUTPUT_PATH} (items={len(output)})") else: print(f"[generate_videoembed_json] no changes. skipped.") return OUTPUT_PATH if __name__ == "__main__": generate_videoembed_json() ``` ================================================================================ === Python(xserver_uploader.py) === /Users/XXXXXX/python_scripts/xserver_uploader.py ```python #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ xserver_uploader.py (env化対応版) - JSON を Xserver へ原子的に安全反映 """ import os import sys import shlex import logging import subprocess import uuid from pathlib import Path from dotenv import load_dotenv # --- 環境変数 (.env) の動的ロード --- SCRIPT_DIR = Path(__file__).resolve().parent load_dotenv(SCRIPT_DIR / ".env") # ====== 環境変数からの読み込み ====== XS_USER = os.getenv("XS_USER", "XXXXXX") XS_HOST = os.getenv("XS_HOST", "XXXXXX.xsrv.jp") XS_SSH_PORT = int(os.getenv("XS_SSH_PORT", "10022")) SSH_KEY_PATH = os.getenv("SSH_KEY_PATH", os.path.expanduser("~/.ssh/id_xserver")) DEST_DIR = os.getenv("XSERVER_DEST_DIR", "/home/XXXXXX/XXXXXX.com/public_html/_secure/wp_json") def _run(cmd: list[str]) -> str: logging.info("RUN: %s", " ".join(shlex.quote(c) for c in cmd)) out = subprocess.check_output(cmd, stderr=subprocess.DEVNULL) return out.decode("utf-8", errors="replace").strip() def upload_secure_json(local_path: str): """JSON を Xserver へ原子的に反映(ホームへ一旦置いてから mv)""" src = Path(local_path) if not src.exists(): raise FileNotFoundError(f"local file not found: {src}") tmp_name = f"__upload_{uuid.uuid4().hex}.tmp" userhost = f"{XS_USER}@{XS_HOST}" remote_tmp_home = f"/home/{XS_USER}/{tmp_name}" # ① scp: まずホームへ scp_cmd = [ "scp", "-P", str(XS_SSH_PORT), "-o", "StrictHostKeyChecking=accept-new", "-o", "UserKnownHostsFile=~/.ssh/known_hosts", "-i", SSH_KEY_PATH, str(src), f"{userhost}:{remote_tmp_home}" ] _run(scp_cmd) # ② ssh: mv & chmod & mtime # _secure/wp_json は Xserver 上の WordPress から読み込む場所。 # ここへ原子的に差し替えるため、一時ファイルからリネーム(mv)で配置。 move_script = ( f"DEST={shlex.quote(DEST_DIR)}; " f"mkdir -p \"$DEST\"; " f"mv -f {shlex.quote(remote_tmp_home)} \"$DEST/{src.name}\"; " f"chmod 644 \"$DEST/{src.name}\"; " f"touch -m \"$DEST/{src.name}\"; " f"wc -c < \"$DEST/{src.name}\"; " ) out = _run([ "ssh", "-p", str(XS_SSH_PORT), "-i", SSH_KEY_PATH, "-o", "StrictHostKeyChecking=accept-new", "-o", "UserKnownHostsFile=~/.ssh/known_hosts", userhost, "bash", "-lc", move_script ]) logging.info("uploaded bytes: %s", out.strip()) print(f"Xserver 反映完了: {src.name} ({out.strip()} bytes)") if __name__ == "__main__": logging.basicConfig(level=logging.INFO) if len(sys.argv) > 1: upload_secure_json(sys.argv[1]) else: print("使用法: python3 xserver_uploader.py <ローカルJSONパス>") ``` ================================================================================ === plist(com.XXXXXX.kick-flags.plist) === /Users/XXXXXX/Library/LaunchAgents/com.XXXXXX.kick-flags.plist ```xml Disabled EnvironmentVariables PATH /opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin Label com.XXXXXX.kick-flags ProgramArguments /opt/homebrew/opt/python@3.11/bin/python3.11 /Users/XXXXXX/python_scripts/kick_flags.py RunAtLoad StandardErrorPath /Users/XXXXXX/Library/Logs/kick_flags.err StandardOutPath /Users/XXXXXX/Library/Logs/kick_flags.log StartInterval 10 ```