# [メタ情報]
# 識別子: マイライブラリに基づく3jsonのM1MacからXserverへの更新_exe
# システム名: 未分類
# 技術種別: Misc
# 機能名: Misc
# 使用言語: Python plist
# 状態: 公開用
# [/メタ情報]
# [/メタ情報]
要約：Google Driveのdropbox_wp_library.jsonをmd5で差分確認し、安全に取得・保存。F2（シート1）から公開データを読み、videoct.json（本番同構造・末尾改行）とvideoembed.json（公開・必須項目チェック、video2任意）を生成。upload_secure_json.pyは3 JSONをUTF-8検証しSHA-256差分比較のうえ一時名→原子的反映でXserverの/_secure/wp_jsonへ配置。必要時のみGDrive同期可。LaunchAgentが30秒毎に起動。

M1 Mac miniは、常時稼働。メタデータの加工と保存はM1Macが行い、アセットデータについては加工はM2Macなど各デバイスで行い保存はNAS(Dropboxとリンク)に行う。メタデータはM1MacからXserverへSCPで更新、アセットデータはNASからXserverへrsyncで更新。

　　　　　               GDrive 　　　M1Mac
　　　　　　　　　　　　gd_wp_data　meta_gd_wp_data
dropbox_wp_library.json 使う(GAS生成)　使う
videoct.json            使わない　　　　使う(PY生成)
videoembed.json　　　　　 使わない　　　 使う(PY生成)


┌──────────────────────────┐
│  Google Drive            │
│  （dropbox_wp_library） │
└──────────────┬───────────┘
               │
               │ (① 同期)
               ▼
┌──────────────────────────┐
│  meta_gd_wp_data (ローカル) │
│  ├── dropbox_wp_library.json  ← sync_dropbox_wp_library.py
│  ├── videoct.json             ← generate_videoct_json.py
│  └── videoembed.json          ← generate_videoembed_json.py
└──────────────┬───────────┘
               │
               │ (② 差分検出 + アップロード)
               ▼
┌──────────────────────────┐
│  Xserver _secure/wp_json │
│  ├── dropbox_wp_library.json
│  ├── videoct.json
│  └── videoembed.json
└──────────────┘

<任意のフォルダパス>/sync_dropbox_wp_library.py
sync_dropbox_wp_library.py

```
#!/usr/bin/env python3
import os, sys, io, json, tempfile
from google.oauth2 import service_account
from googleapiclient.discovery import build
from googleapiclient.http import MediaIoBaseDownload
from dotenv import load_dotenv

load_dotenv("<任意のフォルダパス>")

FILE_ID = os.environ.get("PUSH_META_LIB_JSON_ID")
META_DIR = os.environ.get("META_DIR", os.path.expanduser("~/meta_gd_wp_data"))
OUT_PATH = os.path.join(META_DIR, "dropbox_wp_library.json")

def main():
    if not FILE_ID:
        print("[sync] PUSH_META_LIB_JSON_ID が未設定です。", file=sys.stderr); sys.exit(2)
    creds_path = os.environ.get("PUSH_META_SA_JSON") or os.environ.get("<固有のランダム文字列>")
    if not creds_path or not os.path.isfile(creds_path):
        print("[sync] PUSH_META_SA_JSON (or GOOGLE_APPLICATION_CREDENTIALS) が不正です。", file=sys.stderr); sys.exit(2)

    os.makedirs(META_DIR, exist_ok=True)

    creds = service_account.Credentials.from_service_account_file(
        creds_path, scopes=["https://www.googleapis.com/auth/drive.readonly"]
    )
    svc = build("drive", "v3", credentials=creds, cache_discovery=False)

    meta = svc.files().get(fileId=FILE_ID, fields="id,name,modifiedTime,md5Checksum").execute()
    stamp_path = OUT_PATH + ".stamp.json"
    old = {}
    if os.path.exists(stamp_path):
        try:
            with open(stamp_path) as f: old = json.load(f)
        except Exception: pass
    if old.get("md5Checksum") == meta.get("md5Checksum"):
        print("[sync] 変更なし：", meta["modifiedTime"]); return

    req = svc.files().get_media(fileId=FILE_ID)
    buf = io.BytesIO()
    downloader = MediaIoBaseDownload(buf, req, chunksize=1024*1024)
    done = False
    while not done:
        status, done = downloader.next_chunk()
        if status: print(f"[sync] downloading {int(status.progress()*100)}%")
    data = buf.getvalue()

    try:
        json.loads(data.decode("utf-8"))
    except Exception as e:
        print("[sync] JSON decode error:", e, file=sys.stderr); sys.exit(1)

    fd, tmp = tempfile.mkstemp(prefix="libjson_", suffix=".json", dir=META_DIR)
    with os.fdopen(fd, "wb") as f: f.write(data)
    os.replace(tmp, OUT_PATH)
    with open(stamp_path, "w") as f:
        json.dump({"md5Checksum": meta.get("md5Checksum"),
                   "modifiedTime": meta.get("modifiedTime")}, f, ensure_ascii=False)

    print("[sync] 更新完了:", OUT_PATH, meta.get("modifiedTime"))

if __name__ == "__main__":
    main()
```

<任意のフォルダパス>/generate_videoct_json.py
generate_videoct_json.py

```
# -*- coding: utf-8 -*-
"""
generate_videoct_json.py（最終版）
F2(シート1)から公開行を抽出し、videoct.jsonと完全同一構造のpytest_videoct.jsonを生成。
"""

import os
import json
import gspread
from google.oauth2.service_account import Credentials
from dotenv import load_dotenv

load_dotenv("<任意のフォルダパス>")

# ======= 設定 =======
SA_JSON_PATH = os.getenv("PUSH_META_SA_JSON", "<任意のフォルダパス>/service.json")
SCOPES = ["https://www.googleapis.com/auth/spreadsheets.readonly"]

SPREADSHEET_ID = os.getenv("PUSH_META_SHEET_ID", "<固有のランダム文字列>")
SHEET_NAME     = "シート1"
OUTPUT_PATH    = "<任意のフォルダパス>/videoct.json"
# ====================


def _safe(row, idx):
    """行データから安全に値を取得（範囲外は空文字）"""
    return row[idx].strip() if idx < len(row) else ""


def generate_videoct_json():
    """F2からvideoct.jsonと完全同一構造のpytest_videoct.jsonを生成"""
    creds = Credentials.from_service_account_file(SA_JSON_PATH, scopes=SCOPES)
    gc = gspread.authorize(creds)
    ws = gc.open_by_key(SPREADSHEET_ID).worksheet(SHEET_NAME)
    rows = ws.get_all_values()

    if not rows:
        raise RuntimeError("F2 シートが空です。")

    output = []

    # 2行目以降を走査
    for r in rows[1:]:
        videoid = _safe(r, 2)
        if not videoid:
            continue  # C列=動画idが空ならスキップ

        item = {
            "videoid":       str(videoid),  # ★明示的に文字列化
            "image":         _safe(r, 5),   # F列 初期画像ファイルURL
            "video":         _safe(r, 7),   # H列 動画ファイルURL
            "video2":        _safe(r, 22),  # W列 低画質動画ファイルURL
            "subtitle":      _safe(r, 9),   # J列 字幕ファイルURL_vtt
            "subtitle_list": _safe(r, 11),  # L列 字幕一覧ファイルURL_txt
            "explain_line":  _safe(r, 13),  # N列 説明ファイルURL_txt
        }
        output.append(item)

    # JSON出力
    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:
        json.dump(output, f, ensure_ascii=False, indent=2)
        f.write("\n")  # ★末尾改行を追加（本番と完全一致させる）
    os.replace(tmp_path, OUTPUT_PATH)

    return OUTPUT_PATH


if __name__ == "__main__":
    path = generate_videoct_json()
    print(f"✅ 生成完了: {path}")
```

geberate_videoembed_json.pyは、
現在、c-geberate_videoembed_json.pyとなっている。

geberate_videoembed_json.py

# <任意のフォルダパス>/generate_videoembed_json.py
# -*- coding: utf-8 -*-
"""
完全版：GAS版 videoembed.json と同一仕様・同一フォーマットで出力
-----------------------------------------------------
- 元仕様:
    A列 (0): 公開/非公開      → "公開" の行のみ対象
    C列 (2): 動画ID           → videoid
    S列 (18): 埋め込みコード   → embedCode
    W列 (22): 低画質URL        → video2 (任意)
- その他の列は無視
- embedCode, videoid が空行は除外
- 出力: <任意のフォルダパス>/videoembed.json
- 構造: JSON配列（各要素 = {videoid, embedCode, (video2)})
- 整形: インデント2、改行あり、UTF-8 / ensure_ascii=False
"""

import os
import json
import gspread
from google.oauth2.service_account import Credentials
from dotenv import load_dotenv

load_dotenv("<任意のフォルダパス>")

# ======= 基本設定 =======
SHEET_ID = os.getenv("PUSH_META_SHEET_ID", "<固有のランダム文字列>")
SHEET_TAB = "シート1"
SERVICE_JSON = os.getenv("PUSH_META_SA_JSON", "<任意のフォルダパス>/service.json")
OUTPUT_PATH = "<任意のフォルダパス>/videoembed.json"

# ======= 列インデックス (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):
    """GAS同様、数値化可能ならint、そうでなければ文字列"""
    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):
    """GAS版の空・null・空文字チェック"""
    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 = []
    for row in all_rows[1:]:  # 0行目はヘッダ
        # 列取得（存在しない列は空文字で補う）
        publish = _safe_get(row, COL_A_PUBLISH).strip()
        videoid_raw = _safe_get(row, COL_C_VIDEOID).strip()
        embed = _safe_get(row, COL_S_EMBED)
        video2 = _safe_get(row, COL_W_VIDEO2).strip()

        # 公開判定
        if publish != "公開":
            continue
        # embedCode / videoid 必須チェック
        if not _is_valid_embed(embed) or videoid_raw == "":
            continue

        videoid_val = _to_int_or_str(videoid_raw)

        item = {
            "videoid": videoid_val,
            "embedCode": embed.strip(),
        }
        if video2:
            item["video2"] = video2
        output.append(item)

    # === JSON出力 ===
    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:
        json.dump(output, f, ensure_ascii=False, indent=2)
    os.replace(tmp_path, OUTPUT_PATH)

    print(f"[generate_videoembed_json] wrote: {OUTPUT_PATH} (items={len(output)})")
    return OUTPUT_PATH


# ======= メイン実行 =======
if __name__ == "__main__":
    generate_videoembed_json()



upload_secure_json.py

```
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
upload_secure_json.py  (diff-aware, atomic, locale-clean)

目的:
  - ローカル meta ディレクトリの JSON を Xserver の
    <任意のフォルダパス>/XXXXXX.com/public_html/_secure/wp_json
    に「変更があった場合のみ」原子的にアップロードする。
  - アップロード前の Google Drive 同期はデフォルトOFF（必要時のみ環境変数で有効化）。
  - ssh 実行時に LC_ALL=C LANG=C を指定し、ロケール警告を抑制・フィルタ。

前提:
  - ~/.ssh/config に Host エイリアス（既定: xsrv）を設定済み
  - scp/ssh は Host 名で接続（鍵やポート等は ~/.ssh/config に記述）

環境変数(任意):
  - META_DIR   : ローカルJSON置き場 (既定: ~/meta_gd_wp_data)
  - SSH_HOST   : 接続先Host名 (既定: xsrv)
  - SSH_OPTS   : 追加のssh/scpオプション (例: "-o SomeOpt=yes")
  - SYNC_GDRIVE: "true/1" で GDrive同期を一時有効化（既定: 無効）
"""

import os
import sys
import shlex
import json
import hashlib
import random
import string
import subprocess as sp
from pathlib import Path
from typing import Optional, Tuple
from dotenv import load_dotenv

load_dotenv("<任意のフォルダパス>")

# ===== 設定 =====
HOME = Path.home()
META_DIR = Path(os.environ.get("META_DIR", str(HOME / "meta_gd_wp_data")))
REMOTE_DIR = "<任意のフォルダパス>/XXXXXX.com/public_html/_secure/wp_json"
SSH_HOST = os.environ.get("SSH_HOST", "xsrv")
SSH_OPTS = shlex.split(os.environ.get("SSH_OPTS", "").strip())

TARGETS = [
    "videoembed.json",
    "videoct.json",
    "dropbox_wp_library.json",
]

HASH_CHUNK = 1024 * 1024  # 1MB

# ===== 実行ユーティリティ =====
def _run(cmd: list[str]) -> int:
    print("[exec]", " ".join(shlex.quote(c) for c in cmd))
    try:
        res = sp.run(cmd, check=True)
        return res.returncode
    except sp.CalledProcessError as e:
        print(f"[error] コマンド失敗: {e}", file=sys.stderr)
        return e.returncode

def _run_capture(cmd: list[str]) -> Tuple[int, str, str]:
    """stdout, stderr を受け取る版"""
    print("[exec]", " ".join(shlex.quote(c) for c in cmd))
    try:
        res = sp.run(cmd, check=True, stdout=sp.PIPE, stderr=sp.PIPE, text=True)
        return res.returncode, res.stdout.strip(), res.stderr.strip()
    except sp.CalledProcessError as e:
        return e.returncode, (e.stdout or "").strip(), (e.stderr or "").strip()

def _rand_tag(n=5) -> str:
    return "".join(random.choices(string.digits, k=n))

def _ssh(cmd_str: str) -> int:
    """ssh 経由でコマンド実行（ロケール警告抑制＋フィルタ）"""
    cmd = ["ssh", *SSH_OPTS, SSH_HOST, f"LC_ALL=C LANG=C {cmd_str}"]
    rc, out, err = _run_capture(cmd)
    # setlocale 警告だけ除去
    if err:
        err = "\n".join(line for line in err.splitlines() if "setlocale: LC_CTYPE" not in line)
        if err:
            sys.stderr.write(err + "\n")
    # stdout は不要（結果コードのみで十分）
    return rc

def _ssh_out(cmd_str: str) -> Tuple[int, str]:
    """ssh 経由で stdout を受け取る（ロケール警告抑制＋フィルタ）"""
    cmd = ["ssh", *SSH_OPTS, SSH_HOST, f"LC_ALL=C LANG=C {cmd_str}"]
    rc, out, err = _run_capture(cmd)
    if err:
        err = "\n".join(line for line in err.splitlines() if "setlocale: LC_CTYPE" not in line)
        if err:
            sys.stderr.write(err + "\n")
    return rc, out

def _scp(local: Path, remote_tmp: str) -> int:
    cmd = ["scp", *SSH_OPTS, str(local), f"{SSH_HOST}:{remote_tmp}"]
    return _run(cmd)

# ===== 便利関数 =====
def ensure_local_exists(p: Path) -> bool:
    if not p.exists():
        print(f"[warn] ローカルに存在しません: {p}", file=sys.stderr)
        return False
    if not p.is_file():
        print(f"[warn] ファイルではありません: {p}", file=sys.stderr)
        return False
    return True

def sha256_local(path: Path) -> str:
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(HASH_CHUNK), b""):
            h.update(chunk)
    return h.hexdigest()

def validate_json_utf8(path: Path) -> bool:
    try:
        text = path.read_text(encoding="utf-8")
        json.loads(text)
        return True
    except Exception as e:
        print(f"[error] JSON妥当性エラー: {path.name}: {e}", file=sys.stderr)
        return False

def sha256_remote(base_name: str) -> Optional[str]:
    """
    リモートの SHA-256 を取得。ファイルが無ければ None。
    利用可能なコマンドを順に試す: sha256sum, shasum -a 256, openssl dgst -sha256
    """
    remote_path = f"{REMOTE_DIR}/{base_name}"
    cmd = (
        "f=" + shlex.quote(remote_path) + " ; "
        "if [ ! -f \"$f\" ]; then echo '__NOFILE__'; exit 0; fi; "
        "(sha256sum \"$f\" 2>/dev/null || "
        " shasum -a 256 \"$f\" 2>/dev/null || "
        " openssl dgst -sha256 \"$f\" 2>/dev/null) | "
        "awk '{print $1}' | sed -e 's/^.*=\\s*//'"
    )
    rc, out = _ssh_out(cmd)
    if rc != 0:
        return None
    if out.strip() == "__NOFILE__":
        return None
    return out.strip() if out.strip() else None

# ===== アップロード（差分判定付き） =====
def upload_one(local_file: Path, base_name: str, tag: str) -> bool:
    """
    1ファイルを「変更があるときだけ」原子的にアップロード。
    - ローカルJSON妥当性チェック
    - リモートSHA-256と比較して同一なら SKIP
    - 異なる/存在しないときだけ scp -> mv -f -> chmod 600
    """
    if not validate_json_utf8(local_file):
        print(f"[skip] JSON不正のためスキップ: {base_name}")
        return False

    local_hash = sha256_local(local_file)
    remote_hash = sha256_remote(base_name)

    if remote_hash is not None and remote_hash == local_hash:
        print(f"[nochg] {base_name} 内容同一のためアップロード不要")
        return False

    remote_tmp = f"{REMOTE_DIR}/.__tmp_{tag}_{base_name}"
    remote_final = f"{REMOTE_DIR}/{base_name}"

    # 1) 一時名でアップロード
    if _scp(local_file, remote_tmp) != 0:
        print(f"[error] アップロード失敗: {base_name}", file=sys.stderr)
        return False

    # 2) サーバ側で原子的反映＋権限設定
    cmd = (
        f"mv -f {shlex.quote(remote_tmp)} {shlex.quote(remote_final)}"
        f" && chmod 600 {shlex.quote(remote_final)}"
    )
    if _ssh(cmd) != 0:
        print(f"[error] 反映( mv & chmod )失敗: {base_name}", file=sys.stderr)
        return False

    print(f"[write] {base_name} を更新（remote_hash: {remote_hash or 'none'} → {local_hash}）")
    return True

# ===== GDrive 同期（デフォルトOFF） =====
def maybe_sync_dropbox_wp_library():
    """
    通常は実行しない。必要になったときだけ
    SYNC_GDRIVE=true などで一時的に有効化。
    """
    flag = os.environ.get("SYNC_GDRIVE", "").lower()
    if flag in ("1", "true", "yes", "on"):
        py = "/usr/bin/python3"
        script = str(HOME / "python_scripts" / "sync_dropbox_wp_library.py")
        if Path(script).exists():
            print("[info] GDrive同期を実行します")
            _run([py, script])
        else:
            print(f"[info] 同期スクリプトが見つかりません: {script}（スキップ）")
    else:
        print("[info] GDrive同期は無効（SYNC_GDRIVE で有効化可）")

# ===== メイン =====
def main() -> int:
    # 0) 必要時のみ GDrive 同期（既定OFF）
    maybe_sync_dropbox_wp_library()

    # 1) ローカル存在チェック
    local_files: list[tuple[Path, str]] = []
    for name in TARGETS:
        p = META_DIR / name
        if ensure_local_exists(p):
            local_files.append((p, name))

    if not local_files:
        print("[info] 送るべきファイルがありません。")
        return 0

    # 2) 差分アップロード
    tag = _rand_tag()
    updated, skipped = 0, 0
    for p, name in local_files:
        print(f"[check] {name}")
        if upload_one(p, name, tag):
            updated += 1
        else:
            skipped += 1

    if updated:
        print(f"[done] 更新 {updated} 件 / スキップ {skipped} 件")
    else:
        print(f"[done] すべて変更なし（スキップ {skipped} 件）")
    return 0

if __name__ == "__main__":
    sys.exit(main())
```


com.XXXXXX.push-meta.plist
<任意のフォルダパス>/com.XXXXXX.push-meta.plist

```
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
	<key>EnvironmentVariables</key>
	<dict>
		<!-- 認証: Secrets are now managed via <任意のフォルダパス> -->
		<key>PUSH_META_SA_JSON</key>
		<string><任意のフォルダパス>/service.json</string>
		<key>PUSH_META_LIB_JSON_ID</key>
		<string></string>
		<key>META_DIR</key>
		<string><任意のフォルダパス>
		<key>PATH</key>
		<string>/usr/local/bin:/opt/homebrew/bin:/usr/bin:/bin:/usr/sbin:/sbin:/Library/Developer/CommandLineTools/usr/bin</string>
		<key>SSH_OPTS</key>
		<string></string>
		<key>SYNC_GDRIVE</key>
		<string>true</string>
	</dict>
	<key>Label</key>
	<string>com.XXXXXX.push-meta</string>
	<key>ProgramArguments</key>
	<array>
		<string>/bin/zsh</string>
		<string>-lc</string>
		<string>/usr/bin/python3 <任意のフォルダパス>/upload_secure_json.py</string>
	</array>
	<key>RunAtLoad</key>
	<true/>
	<key>StandardErrorPath</key>
	<string><任意のフォルダパス>/push_meta.err.log</string>
	<key>StandardOutPath</key>
	<string><任意のフォルダパス>/push_meta.out.log</string>
	<key>StartInterval</key>
	<integer>30</integer>
	<key>WorkingDirectory</key>
	<string><任意のフォルダパス>
</dict>
</plist>
```