# [メタ情報]
# 識別子: マイライブラリに基づく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

#!/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

FILE_ID = os.environ.get("LIB_JSON_FILE_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] LIB_JSON_FILE_ID が未設定です。", file=sys.stderr); sys.exit(2)
    creds_path = os.environ.get("GOOGLE_APPLICATION_CREDENTIALS")
    if not creds_path or not os.path.isfile(creds_path):
        print("[sync] 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

# -*- 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

# ======= 設定 =======
SA_JSON_PATH = "/Users/xxxxxxxxm1/keys/service.json"
SCOPES = ["https://www.googleapis.com/auth/spreadsheets.readonly"]

SPREADSHEET_ID = "17TMSsh8OF8bgcP0NEuM3X67oPfACVSmkGvy8w-DgrZY"
SHEET_NAME     = "シート1"
OUTPUT_PATH    = "/Users/xxxxxxxxm1/meta_gd_wp_data/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

# /Users/xxxxxxxxm1/python_scripts/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 が空行は除外
- 出力: /Users/xxxxxxxxm1/meta_gd_wp_data/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

# ======= 基本設定 =======
SHEET_ID = "<SHEET_ID>"
SHEET_TAB = "<SHEET_TAB>"
SERVICE_JSON = "/Users/xxxxxxxxm1/keys/service.json"
OUTPUT_PATH = "/Users/xxxxxxxxm1/meta_gd_wp_data/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 の
    /home/xxxxxxxx/xxxxxxxx.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

# ===== 設定 =====
HOME = Path.home()
META_DIR = Path(os.environ.get("META_DIR", str(HOME / "meta_gd_wp_data")))
REMOTE_DIR = "/home/xxxxxxxx/xxxxxxxx.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.xxxxxxxx.push-meta.plist
/Users/xxxxxxxxm1/Library/LaunchAgents/com.xxxxxxxx.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>
		<key>GOOGLE_APPLICATION_CREDENTIALS</key>
		<string>/Users/xxxxxxxxm1/keys/service.json</string>
		<key>LIB_JSON_FILE_ID</key>
		<string><あなたのID></string>
		<key>META_DIR</key>
		<string>/Users/xxxxxxxxm1/meta_gd_wp_data</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.xxxxxxxx.push-meta</string>
	<key>ProgramArguments</key>
	<array>
		<string>/bin/zsh</string>
		<string>-lc</string>
		<string>/usr/bin/python3 /Users/xxxxxxxxm1/python_scripts/upload_secure_json.py</string>
	</array>
	<key>RunAtLoad</key>
	<true/>
	<key>StandardErrorPath</key>
	<string>/Users/xxxxxxxxm1/Library/Logs/push_meta.err.log</string>
	<key>StandardOutPath</key>
	<string>/Users/xxxxxxxxm1/Library/Logs/push_meta.out.log</string>
	<key>StartInterval</key>
	<integer>30</integer>
	<key>WorkingDirectory</key>
	<string>/Users/xxxxxxxxm1</string>
</dict>
</plist>