# [メタ情報] # 識別子: vimeo更新システム_py利用_exe # システム名: 未分類 # 技術種別: Misc # 機能名: Misc # 使用言語: Python plist # 状態: 実行用 # [/メタ情報] 要約：Vimeo更新用のPython群。configは環境変数と列定義、トークン取得。commonはJST時刻と進捗文生成。gatewaysはgspreadラッパとF1のWP再生URL→Dropbox逆引き。workersはStep1(モード確認)、Step2(vid/urlで新規・置換・metadata更新)、Step3(題名・サムネ・字幕)。runnerがF2行を読みDropbox共有URLを直リンク化して実行しQに記録。kick_flagsはF2のB〜Eを走査しBがstep2_ready/step3_readyかつE=vimeoの行のみ順次runner起動し429考慮。LaunchAgentが20秒ごとにkick_flagsを起動。 PythonはM1 Mac miniに配置 GOOGLE_APPLICATION_CREDENTIALS = "" F2 GDOC_SPREADSHEET_ID = "" GDOC_SHEET_NAME = "" \ F1 GDOC_SPREADSHEET_ID = "" GDOC_SHEET_NAME = "" VIMEO_ACCESS_TOKEN = "" __init__.py 中身は空 config.py # -*- coding: utf-8 -*- """ config.py (Python 3.9 compatible) - 環境変数の読み込み（Sheets接続、Vimeo接続） - 列定義（A1表記） - 書き込み可能列の宣言 """ import os from dataclasses import dataclass from typing import Dict, Set # 対象列（A1） COL: Dict[str, str] = { "B": "B", # 状態 "C": "C", # 指示／パラメータ（vid格納も可） "E": "E", # 所在モード（vimeo/dynamic/youtube/none/skip） "Q": "Q", # 進捗サマリ＋メッセージ "AC": "AC", # Step1 時刻 "AD": "AD", # Step2 時刻 "AE": "AE", # Step3 時刻 } WRITABLE_COLS: Set[str] = {"B", "C", "Q", "AC", "AD", "AE"} E_ALLOWED_MODES: Set[str] = {"vimeo", "dynamic", "youtube", "none", "skip"} @dataclass class Settings: """Google Sheets 接続に必要な設定。""" spreadsheet_id: str sheet_name: str creds_path: str @classmethod def from_env(cls) -> "Settings": creds = os.environ.get("GOOGLE_APPLICATION_CREDENTIALS", "").strip() sid = os.environ.get("GDOC_SPREADSHEET_ID", "").strip() sname = os.environ.get("GDOC_SHEET_NAME", "").strip() missing = [] if not creds: missing.append("GOOGLE_APPLICATION_CREDENTIALS") if not sid: missing.append("GDOC_SPREADSHEET_ID") if not sname: missing.append("GDOC_SHEET_NAME") if missing: raise EnvironmentError("必須環境変数が未設定です: " + ", ".join(missing)) return cls(spreadsheet_id=sid, sheet_name=sname, creds_path=creds) def get_vimeo_token() -> str: """ Vimeoのパーソナルアクセストークン（Server-to-Server）を返す。 必須環境変数: VIMEO_ACCESS_TOKEN """ tok = os.environ.get("VIMEO_ACCESS_TOKEN", "").strip() if not tok: raise EnvironmentError("必須環境変数が未設定です: VIMEO_ACCESS_TOKEN") return tok __all__ = [ "Settings", "COL", "WRITABLE_COLS", "E_ALLOWED_MODES", "get_vimeo_token", ] common.py # vimeo_upd/common.py # Python 3.9 向けユーティリティ（最小限） from datetime import datetime, timezone, timedelta from typing import Optional def jst_now_str() -> str: """JSTの現在時刻を 'YYYY/MM/DD HH:MM:SS' で返す。""" JST = timezone(timedelta(hours=9)) return datetime.now(JST).strftime("%Y/%m/%d %H:%M:%S") def build_progress_line(step1_time: Optional[str], step2_time: Optional[str], step3_time: Optional[str]) -> str: """進捗サマリ1行を作る。空値は '--:--:--' 表示。""" def _fmt(v: Optional[str]) -> str: v = (v or "").strip() return v if v else "--:--:--" return ( f"Step1: 完了 ({_fmt(step1_time)}) " f"Step2: 完了 ({_fmt(step2_time)}) " f"Step3: 完了 ({_fmt(step3_time)})" ) gateways.py # -*- coding: utf-8 -*- """ gateways.py - Google Sheets へのアクセスラッパ - F2（運用シート）と F1（参照元シート）を安全に扱う - F2 の行読み書き（列記号ベース A,B,C,...） - F1 の「WP再生URL → DropboxリンクURL」逆引き 環境変数: GOOGLE_APPLICATION_CREDENTIALS ... サービスアカウントJSONのパス GDOC_SPREADSHEET_ID ... F2 スプレッドシートID GDOC_SHEET_NAME ... F2 ワークシート名（例: "シート1"） F1_SPREADSHEET_ID ... F1 スプレッドシートID F1_SHEET_NAME ... F1 ワークシート名（例: "sheet1"） F1_PLAY_URL_HEADER ... F1 の「WP再生URL」列の見出し文字列 F1_DROPBOX_URL_HEADER ... F1 の「DropboxリンクURL」列の見出し文字列 """ from __future__ import annotations import os import string from dataclasses import dataclass from typing import Dict, List, Optional import gspread from gspread.exceptions import WorksheetNotFound, APIError # ------------------------------------------------------------ # 設定 # ------------------------------------------------------------ @dataclass class Settings: google_application_credentials: str f2_spreadsheet_id: str f2_sheet_name: str f1_spreadsheet_id: str f1_sheet_name: str f1_play_url_header: str f1_dropbox_url_header: str @classmethod def from_env(cls) -> "Settings": creds = os.getenv("GOOGLE_APPLICATION_CREDENTIALS", "").strip() f2_id = os.getenv("GDOC_SPREADSHEET_ID", "").strip() f2_ws = os.getenv("GDOC_SHEET_NAME", "").strip() f1_id = os.getenv("F1_SPREADSHEET_ID", "").strip() f1_ws = os.getenv("F1_SHEET_NAME", "").strip() f1_play = os.getenv("F1_PLAY_URL_HEADER", "").strip() f1_drop = os.getenv("F1_DROPBOX_URL_HEADER", "").strip() missing = [] if not creds: missing.append("GOOGLE_APPLICATION_CREDENTIALS") if not f2_id: missing.append("GDOC_SPREADSHEET_ID") if not f2_ws: missing.append("GDOC_SHEET_NAME") if not f1_id: missing.append("F1_SPREADSHEET_ID") if not f1_ws: missing.append("F1_SHEET_NAME") if not f1_play: missing.append("F1_PLAY_URL_HEADER") if not f1_drop: missing.append("F1_DROPBOX_URL_HEADER") if missing: raise EnvironmentError("必須環境変数が未設定です: " + ", ".join(missing)) if not os.path.isfile(creds): raise FileNotFoundError(f"GOOGLE_APPLICATION_CREDENTIALS が存在しません: {creds}") return cls( google_application_credentials=creds, f2_spreadsheet_id=f2_id, f2_sheet_name=f2_ws, f1_spreadsheet_id=f1_id, f1_sheet_name=f1_ws, f1_play_url_header=f1_play, f1_dropbox_url_header=f1_drop, ) # ------------------------------------------------------------ # ユーティリティ # ------------------------------------------------------------ _COLS = list(string.ascii_uppercase) # 'A'..'Z' def col_letter_to_index(letter: str) -> int: """A->1, B->2 ... AA などは今回想定外（A..Z 前提）""" letter = (letter or "").strip().upper() if not letter or letter not in _COLS: raise ValueError(f"未知の列指定: {letter}") return _COLS.index(letter) + 1 def index_to_col_letter(idx: int) -> str: if 1 <= idx <= 26: return _COLS[idx - 1] raise ValueError(f"列インデックス範囲外: {idx}") # ------------------------------------------------------------ # シートクライアント # ------------------------------------------------------------ class SheetClient: def __init__(self, gc: gspread.Client, settings: Settings): self.gc = gc self.s = settings self.f2_spreadsheet_id = settings.f2_spreadsheet_id self.f2_sheet_name = settings.f2_sheet_name self.f1_spreadsheet_id = settings.f1_spreadsheet_id self.f1_sheet_name = settings.f1_sheet_name self.f1_play_url_header = settings.f1_play_url_header self.f1_dropbox_url_header = settings.f1_dropbox_url_header # ---------- 生成 ---------- @classmethod def from_env(cls) -> "SheetClient": s = Settings.from_env() gc = gspread.service_account(filename=s.google_application_credentials) return cls(gc, s) # ---------- 開く ---------- def open_spreadsheet(self, spreadsheet_id: Optional[str] = None): sid = spreadsheet_id or self.f2_spreadsheet_id return self.gc.open_by_key(sid) def open_worksheet(self, sheet_name: str, spreadsheet_id: Optional[str] = None): """ワークシート名の大小文字差・前後空白にもある程度耐性を持たせる""" ss = self.open_spreadsheet(spreadsheet_id) try: return ss.worksheet(sheet_name) except WorksheetNotFound: wanted = (sheet_name or "").strip().lower() try: for ws in ss.worksheets(): if ws.title.strip().lower() == wanted: return ws # 1枚しかない場合はそれを返す（運用簡便化） wss = ss.worksheets() if len(wss) == 1: return wss[0] except Exception: pass raise # ---------- 行読み ---------- def read_row_as_letters(self, row_index: int, sheet_name: Optional[str] = None, max_cols: int = 60) -> Dict[str, str]: """A..Z の列記号をキーにして返す（空セルは ""）""" ws = self.open_worksheet(sheet_name or self.f2_sheet_name, spreadsheet_id=self.f2_spreadsheet_id) rng = f"A{row_index}:{index_to_col_letter(min(max_cols, 26))}{row_index}" vals = ws.get(rng, value_render_option="FORMATTED_VALUE") row = (vals[0] if vals else []) out = {} for i in range(1, min(len(row), 26) + 1): out[index_to_col_letter(i)] = row[i - 1] # 足りない列は空文字で埋める for i in range(len(row) + 1, 26 + 1): out[index_to_col_letter(i)] = "" return out # ---------- 行更新 ---------- def update_row_by_letters(self, row_index: int, updates: Dict[str, str], sheet_name: Optional[str] = None): """updates は {'B':'step3_ready','C':'12345'} のように列記号で指定""" ws = self.open_worksheet(sheet_name or self.f2_sheet_name, spreadsheet_id=self.f2_spreadsheet_id) batch = [] for col, val in updates.items(): col = col.strip().upper() col_idx = col_letter_to_index(col) rng = f"{col}{row_index}:{col}{row_index}" batch.append({ "range": rng, "values": [[val]], }) if batch: ws.batch_update(batch, value_input_option="USER_ENTERED") # ---------- F1 逆引き ---------- def lookup_dropbox_url_from_f1(self, play_url: str) -> Optional[str]: """ F1（参照元）で「WP再生URL が play_url と一致する行」を探し、 その行の「DropboxリンクURL」を返す。 """ if not play_url: return None f1_ws = self.open_worksheet(self.f1_sheet_name, spreadsheet_id=self.f1_spreadsheet_id) # 見出し行（1行目）から対象列のインデックスを特定 header_vals = f1_ws.get("A1:Z1", value_render_option="FORMATTED_VALUE") headers = header_vals[0] if header_vals else [] # 見出し文字列を小文字化・前後空白トリムで比較 want_play = self.f1_play_url_header.strip().lower() want_drop = self.f1_dropbox_url_header.strip().lower() play_col_idx = None drop_col_idx = None for idx, h in enumerate(headers, start=1): hnorm = (h or "").strip().lower() if hnorm == want_play: play_col_idx = idx if hnorm == want_drop: drop_col_idx = idx if play_col_idx is None or drop_col_idx is None: raise RuntimeError( f"F1見出しが見つかりません: " f"WP再生URL='{self.f1_play_url_header}', DropboxリンクURL='{self.f1_dropbox_url_header}'" ) # 対象列だけを一括取得して走査（リクエスト節約） # 2行目以降を対象 last_row = 20000 # 十分大きく play_col_letter = index_to_col_letter(play_col_idx) drop_col_letter = index_to_col_letter(drop_col_idx) play_rng = f"{play_col_letter}2:{play_col_letter}{last_row}" drop_rng = f"{drop_col_letter}2:{drop_col_letter}{last_row}" play_vals = f1_ws.get(play_rng, value_render_option="FORMATTED_VALUE") drop_vals = f1_ws.get(drop_rng, value_render_option="FORMATTED_VALUE") # 文字列化して比較（完全一致） target = (play_url or "").strip() for i in range(len(play_vals)): pv = (play_vals[i][0] if play_vals[i] else "") if (pv or "").strip() == target: dv = (drop_vals[i][0] if i < len(drop_vals) and drop_vals[i] else "") return (dv or "").strip() or None return None workers.py # -*- coding: utf-8 -*- """ workers.py (Python 3.9) - Step2: Vimeo API 実装（安全モード） * C列に vid= があれば自動で replace（/videos/{id}/versions へ POST, file_name 必須） * vid= が無ければ upload（/me/videos へ POST, pull） * dynamic は明示指定時のみ（name/description/privacy.view） - Step3: サムネと字幕を任意で適用（thumb_url / vtt_url / lang / caption_name） - 失敗時は詳細エラーメッセージを返す（HTTPコードと先頭レスポンススニペット） - 環境変数: * VIMEO_ACCESS_TOKEN（必須） * VIMEO_DEBUG をセットするとHTTPリクエスト/レスポンス概要を標準出力に出す """ from typing import Dict, Any, Tuple, Optional import os import re import json import urllib.parse as up import requests from .config import get_vimeo_token # ========= ユーティリティ ========= def _mode_from_e(row: Dict[str, Any]) -> str: return str(row.get("E", "") or "").strip().lower() def _parse_kv_command(s: str) -> Tuple[str, Dict[str, str]]: """ "upload url=... name=..." のような文字列を (cmd, {key:value}) に分解。value内スペースは _ に置き換えて入力すると安全。 """ s = (s or "").strip() if not s: return "", {} parts = s.split() cmd = parts[0].lower() kv: Dict[str, str] = {} for p in parts[1:]: if "=" in p: k, v = p.split("=", 1) kv[k.strip()] = v.strip() return cmd, kv def _extract_vid_from_c(cval: str) -> Optional[str]: """ C列内の "vid=123456789" を抽出。 """ if not cval: return None m = re.search(r"\bvid=(\d+)\b", cval) return m.group(1) if m else None def _dbg_enabled() -> bool: return os.environ.get("VIMEO_DEBUG", "").strip() != "" def _short(s: Any, n: int = 300) -> str: try: t = str(s) except Exception: t = repr(s) return t[:n] def _hint_from_status(code: int) -> str: if code in (401, 403): return "（権限/トークンを確認：Server-to-Serverトークンか、upload/edit/create/public/private スコープ）" if code == 404: return "（vidが存在しないかアクセス不可）" if code == 400: return "（パラメータ不足/URL無効の可能性）" if code == 415: return "（Content-Type/アップロード方式の不一致の可能性）" if code == 422: return "（バリデーションエラー：URL到達不可・ファイル形式不正など）" return "" # ========= Vimeo API クライアント ========= class VimeoClient: def __init__(self, access_token: str): self.base = "https://api.vimeo.com" self.headers = { "Authorization": f"bearer {access_token}", "Accept": "application/vnd.vimeo.*+json;version=3.4", "Content-Type": "application/json", } self.timeout = 45 # pull だと少し長め # ---- 動画本体 ---- def upload_pull(self, link: str, name=None, description=None, privacy_view=None) -> Tuple[str, str]: payload: Dict[str, Any] = {"upload": {"approach": "pull", "link": link}} if name: payload["name"] = name if description: payload["description"] = description if privacy_view: payload["privacy"] = {"view": privacy_view} if _dbg_enabled(): print("[DEBUG] POST /me/videos", json.dumps(payload, ensure_ascii=False)) r = requests.post(f"{self.base}/me/videos", json=payload, headers=self.headers, timeout=self.timeout) if _dbg_enabled(): print("[DEBUG] RESP", r.status_code, _short(r.text)) if r.status_code not in (201, 202): raise RuntimeError(f"Vimeo upload failed: {r.status_code} {_short(r.text)} {_hint_from_status(r.status_code)}") try: data = r.json() except Exception: raise RuntimeError(f"Vimeo upload failed: invalid JSON {_short(r.text)}") uri = str(data.get("uri") or "") m = re.search(r"/videos/(\d+)", uri) if not m: raise RuntimeError(f"Unexpected response: uri={uri} {_short(data)}") vid = m.group(1) return vid, f"upload ok vid={vid}" def replace_pull(self, video_id: str, link: str) -> str: """ 置換は /videos/{id}/versions に POST（file_name 必須）で行う。 """ path_base = up.urlparse(link).path base = os.path.basename(path_base) or "video.mp4" payload = { "file_name": base, "upload": {"approach": "pull", "link": link} } if _dbg_enabled(): print("[DEBUG] POST /videos/{vid}/versions".format(vid=video_id), json.dumps(payload, ensure_ascii=False)) r = requests.post(f"{self.base}/videos/{video_id}/versions", json=payload, headers=self.headers, timeout=self.timeout) if _dbg_enabled(): print("[DEBUG] RESP", r.status_code, _short(r.text)) if r.status_code not in (201, 202): raise RuntimeError(f"Vimeo replace failed: {r.status_code} {_short(r.text)} {_hint_from_status(r.status_code)}") return f"replace ok vid={video_id}" def update_metadata(self, video_id: str, name=None, description=None, privacy_view=None) -> str: payload: Dict[str, Any] = {} if name: payload["name"] = name if description: payload["description"] = description if privacy_view: payload["privacy"] = {"view": privacy_view} if not payload: return "dynamic noop (no fields)" if _dbg_enabled(): print("[DEBUG] PATCH /videos/{vid}".format(vid=video_id), json.dumps(payload, ensure_ascii=False)) r = requests.patch(f"{self.base}/videos/{video_id}", json=payload, headers=self.headers, timeout=self.timeout) if _dbg_enabled(): print("[DEBUG] RESP", r.status_code, _short(r.text)) if r.status_code not in (200, 202): raise RuntimeError(f"Vimeo update failed: {r.status_code} {_short(r.text)} {_hint_from_status(r.status_code)}") return f"dynamic ok vid={video_id}" # ---- 追加アセット（任意） ---- def set_thumbnail_from_link(self, video_id: str, image_link: str) -> str: """ サムネイル適用。 """ payload = {"active": True, "link": image_link} if _dbg_enabled(): print("[DEBUG] POST /videos/{vid}/pictures".format(vid=video_id), json.dumps(payload, ensure_ascii=False)) r = requests.post(f"{self.base}/videos/{video_id}/pictures", json=payload, headers=self.headers, timeout=self.timeout) if _dbg_enabled(): print("[DEBUG] RESP", r.status_code, _short(r.text)) if r.status_code not in (200, 201): raise RuntimeError(f"Vimeo thumb failed: {r.status_code} {_short(r.text)} {_hint_from_status(r.status_code)}") return "thumb ok" def add_caption_from_link(self, video_id: str, vtt_link: str, language: str, name: Optional[str]) -> str: """ 字幕追加。 """ payload: Dict[str, Any] = {"type": "subtitles", "language": language, "link": vtt_link} if name: payload["name"] = name if _dbg_enabled(): print("[DEBUG] POST /videos/{vid}/texttracks".format(vid=video_id), json.dumps(payload, ensure_ascii=False)) r = requests.post(f"{self.base}/videos/{video_id}/texttracks", json=payload, headers=self.headers, timeout=self.timeout) if _dbg_enabled(): print("[DEBUG] RESP", r.status_code, _short(r.text)) if r.status_code not in (200, 201): raise RuntimeError(f"Vimeo texttrack failed: {r.status_code} {_short(r.text)} {_hint_from_status(r.status_code)}") return "vtt ok" # ========= Step実装 ========= def run_step1(row: Dict[str, Any], *, dry_run: bool = False) -> Tuple[str, str, Dict[str, str]]: mode = _mode_from_e(row) if not mode: return "error", "Step1: E列（所在モード）が未入力です。", {} return "step1_finished", f"Step1: モード='{mode}' 実行（dry={dry_run})", {} def run_step2(row: Dict[str, Any], *, dry_run: bool = False) -> Tuple[str, str, Dict[str, str]]: """ 安全モード: - dynamic は明示指定時のみ。 - C列に vid= が存在すれば replace（url= 必須）。無ければ upload（url= 必須）。 """ mode = _mode_from_e(row) if mode not in ("vimeo", "dynamic"): return "error", f"Step2: E列モードが vimeo/dynamic 以外: '{mode}'", {} cmd, kv = _parse_kv_command(str(row.get("C") or "")) token = get_vimeo_token() vc = VimeoClient(token) try: existing_vid = _extract_vid_from_c(str(row.get("C") or "")) # ---- dynamic は明示優先 ---- if cmd == "dynamic": vid = kv.get("vid") or existing_vid if not vid: return "error", "Step2 dynamic: vid= が見つかりません。", {} name = kv.get("name") description = kv.get("description") privacy_view = kv.get("privacy.view") if dry_run: return "step2_finished", f"(dry) dynamic vid={vid}", {} note = vc.update_metadata(vid, name=name, description=description, privacy_view=privacy_view) return "step2_finished", note, {"C": f"vid={vid}"} # ---- vid があれば replace、無ければ upload ---- if existing_vid: url = kv.get("url") if not url: return "error", "Step2 replace: url= が必要です。", {} if dry_run: return "step2_finished", f"(dry) replace vid={existing_vid} url={url}", {} note = vc.replace_pull(existing_vid, url) return "step2_finished", note, {"C": f"vid={existing_vid}"} else: url = kv.get("url") if not url: return "error", "Step2 upload: url= が必要です。", {} name = kv.get("name") description = kv.get("description") privacy_view = kv.get("privacy.view") if dry_run: return "step2_finished", f"(dry) upload url={url}", {} vid, note = vc.upload_pull(url, name=name, description=description, privacy_view=privacy_view) return "step2_finished", note, {"C": f"vid={vid}"} except requests.exceptions.RequestException as e: return "error", f"Step2 Vimeo API error: network {type(e).__name__}: {_short(e)}", {} except Exception as e: return "error", f"Step2 Vimeo API error: {type(e).__name__}: {_short(e)}", {} def run_step3(row: Dict[str, Any], *, dry_run: bool = False) -> Tuple[str, str, Dict[str, str]]: """ サムネ・字幕の適用（任意）。C列の同一行に以下のキーを置けます: - thumb_url=...jpg?dl=1 - vtt_url=...vtt?dl=1 - lang=ja（字幕の言語コード） - caption_name=日本語（字幕の表示名） いずれかが無ければ noop 扱い。 """ mode = _mode_from_e(row) if mode not in ("vimeo", "dynamic"): return "error", f"Step3: E列モードが vimeo/dynamic 以外: '{mode}'", {} cmd, kv = _parse_kv_command(str(row.get("C") or "")) # cmd自体は未使用でも良い vid = kv.get("vid") or _extract_vid_from_c(str(row.get("C") or "")) if not vid: return "error", "Step3: vid= が見つかりません（C列に vid=123... を含めてください）。", {} thumb_url = kv.get("thumb_url") vtt_url = kv.get("vtt_url") lang = kv.get("lang") or "ja" caption_name = kv.get("caption_name") # 何も指定が無ければ noop if not thumb_url and not vtt_url: return "step3_finished", "Step3 noop (thumb/vtt なし)", {"C": f"vid={vid}"} if dry_run: what = [] if thumb_url: what.append("thumb") if vtt_url: what.append("vtt") return "step3_finished", f"(dry) apply {','.join(what)} vid={vid}", {"C": f"vid={vid}"} token = get_vimeo_token() vc = VimeoClient(token) notes = [] try: if thumb_url: notes.append(vc.set_thumbnail_from_link(vid, thumb_url)) except requests.exceptions.RequestException as e: return "error", f"Step3 thumb error: network {type(e).__name__}: {_short(e)}", {} except Exception as e: return "error", f"Step3 thumb error: {type(e).__name__}: {_short(e)}", {} try: if vtt_url: notes.append(vc.add_caption_from_link(vid, vtt_url, lang, caption_name)) except requests.exceptions.RequestException as e: return "error", f"Step3 vtt error: network {type(e).__name__}: {_short(e)}", {} except Exception as e: return "error", f"Step3 vtt error: {type(e).__name__}: {_short(e)}", {} note = " & ".join(notes) if notes else "Step3 noop" return "step3_finished", note, {"C": f"vid={vid}"} runner.py # -*- coding: utf-8 -*- """ runner.py Vimeo 更新パイプライン起動スクリプト F2(シート1) 列割り当て: B: 状況 (step1_ready/step2_ready/step3_ready/FALSE/error) C: Vimeo動画ID (空=新規、数値=既存) D: 題名（Vimeoタイトルに反映） ← Step3グループ F: 初期画像ファイルURL ← Step3グループ H: WP再生URL (F1逆引きのキー) J: 字幕ファイルURL_vtt ← Step3グループ Q: 実行ログ ボタン運用: - 「全部を新規追加または更新」: B=step2_ready → 動画(新規/置換)実施後、Step3グループ(題名/サムネ/字幕)まで実施→B=FALSE - 「サムネ字幕を新規追加または更新」: B=step3_ready → Step3グループのみ実施→B=FALSE """ import sys import os import datetime import json import requests from urllib.parse import urlparse, urlencode, urlunparse, parse_qs from .gateways import SheetClient # --------------------------------------------- # 共通: Dropbox共有URLを直リンク(dl=1)へ # --------------------------------------------- def normalize_dropbox_link(url: str) -> str: if not url: return url try: pu = urlparse(url) host = pu.netloc q = parse_qs(pu.query) if "dropbox.com" in host: host = "dl.dropboxusercontent.com" # raw=1 は消して dl=1 を付与（なければ付与） if "raw" in q: q.pop("raw", None) if "dl" not in q: q["dl"] = ["1"] new_q = urlencode({k: v[0] for k, v in q.items()}) return urlunparse((pu.scheme, host, pu.path, "", new_q, "")) return url except Exception: return url def _trace_enabled() -> bool: return os.getenv("VIMEO_TRACE", "").strip() != "" def _debug_enabled() -> bool: return os.getenv("VIMEO_DEBUG", "").strip() != "" def tprint(*args): if _trace_enabled(): print("[TRACE]", *args) def dprint(*args): if _debug_enabled(): print("[DEBUG]", *args) # --------------------------------------------- # Vimeo API 共通 # --------------------------------------------- BASE = "https://api.vimeo.com" def _vimeo_headers(): token = os.getenv("VIMEO_ACCESS_TOKEN") if not token: raise RuntimeError("VIMEO_ACCESS_TOKEN が未設定です。") return { "Authorization": f"bearer {token}", "Accept": "application/vnd.vimeo.*+json;version=3.4", "Content-Type": "application/json", } # --------------------------------------------- # Step2: 動画 新規 or 置換 # --------------------------------------------- def upload_or_replace_video(video_id: str, dropbox_url: str) -> tuple[str, dict]: headers = _vimeo_headers() if video_id: # 置換 url = f"{BASE}/videos/{video_id}/versions" data = { "file_name": os.path.basename(urlparse(dropbox_url).path), "upload": {"approach": "pull", "link": dropbox_url}, } dprint("POST", url, json.dumps(data, ensure_ascii=False)) r = requests.post(url, headers=headers, json=data) else: # 新規 url = f"{BASE}/me/videos" data = {"upload": {"approach": "pull", "link": dropbox_url}} dprint("POST", url, json.dumps(data, ensure_ascii=False)) r = requests.post(url, headers=headers, json=data) dprint("RESP", r.status_code, r.text[:300]) if not r.ok: raise RuntimeError(f"Vimeo API エラー: {r.status_code} {r.text}") resp = r.json() new_vid = video_id or None # /videos//versions/ のときは を保持 if "uri" in resp: parts = resp["uri"].split("/") if "versions" in parts: try: idx = parts.index("videos") new_vid = parts[idx + 1] except Exception: pass else: new_vid = parts[-1] if not new_vid and "uri" in resp: new_vid = resp["uri"].split("/")[-1] dprint("Vimeo更新結果 vid=", new_vid) return new_vid, resp # --------------------------------------------- # Step3-A(グループ一部): タイトル更新（D列） # --------------------------------------------- def update_vimeo_title(video_id: str, title: str) -> dict: headers = _vimeo_headers() url = f"{BASE}/videos/{video_id}" data = {"name": title} dprint("PATCH", url, json.dumps(data, ensure_ascii=False)) r = requests.patch(url, headers=headers, json=data) dprint("RESP", r.status_code, r.text[:300]) if not r.ok: raise RuntimeError(f"Title update API エラー: {r.status_code} {r.text}") return r.json() # --------------------------------------------- # Step3-B(グループ一部): サムネイル画像の追加/更新（F列） # --------------------------------------------- def upload_thumbnail_to_vimeo(video_id: str, image_url: str) -> dict: headers = _vimeo_headers() url = f"{BASE}/videos/{video_id}/pictures" data = {"link": image_url, "active": True} dprint("POST", url, json.dumps(data, ensure_ascii=False)) r = requests.post(url, headers=headers, json=data) dprint("RESP", r.status_code, r.text[:300]) if not r.ok: raise RuntimeError(f"Thumbnail API エラー: {r.status_code} {r.text}") return r.json() # --------------------------------------------- # Step3-C(グループ一部): 字幕(VTT)の追加/更新（J列） # --------------------------------------------- def upload_subtitle_to_vimeo(video_id: str, vtt_url: str) -> dict: headers = _vimeo_headers() url = f"{BASE}/videos/{video_id}/texttracks" lang = os.getenv("VIMEO_SUB_LANG", "ja") name = os.getenv("VIMEO_SUB_NAME", "日本語") data = {"type": "subtitles", "language": lang, "name": name, "link": vtt_url} dprint("POST", url, json.dumps(data, ensure_ascii=False)) r = requests.post(url, headers=headers, json=data) dprint("RESP", r.status_code, r.text[:300]) if not r.ok: # 既存ありで 409 が出ることもあるため、必要に応じて拡張可 raise RuntimeError(f"Subtitle API エラー: {r.status_code} {r.text}") return r.json() # --------------------------------------------- # 実行本体 # --------------------------------------------- def run_once(row_idx: int): sc = SheetClient.from_env() row = sc.read_row_as_letters(row_idx) colB = (row.get("B") or "").strip() # 状況 colC = (row.get("C") or "").strip() # Vimeo ID colD = (row.get("D") or "").strip() # 題名（Vimeoタイトル） ← Step3グループ colF = (row.get("F") or "").strip() # 初期画像ファイルURL ← Step3グループ colH = (row.get("H") or "").strip() # WP再生URL colJ = (row.get("J") or "").strip() # 字幕ファイルURL_vtt ← Step3グループ tprint(f"B列 状況={colB}") tprint(f"C列 VimeoID={colC or '(空)'}") tprint(f"D列 題名={colD or '(空)'}") tprint(f"F列 初期画像URL={colF or '(空)'}") tprint(f"H列 再生URL={colH or '(空)'}") tprint(f"J列 字幕VTT={colJ or '(空)'}") ts = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") log_parts = [] # ----------------------------- # Step2: 動画の新規/置換 # ----------------------------- # step3_ready で起動された場合は動画本体は触らない do_video = (colB != "step3_ready") if do_video: if not colH: raise ValueError("H列にWP再生URLがありません。") # F1逆引き dropbox_raw = sc.lookup_dropbox_url_from_f1(colH) tprint("F1逆引き結果=", dropbox_raw or "(見つからず)") if not dropbox_raw: raise ValueError("F1で該当DropboxリンクURLが見つかりません。") # 直リンク化 direct_url = normalize_dropbox_link(dropbox_raw) tprint("最終的にVimeoへ渡すURL=", direct_url) # Vimeo置換/新規 new_vid, _ = upload_or_replace_video(colC, direct_url) colC = new_vid log_parts.append(("replace ok" if (row.get("C") or "").strip() else "upload ok") + f" vid={colC}") # Step3グループを同一ランで続行できるよう、変数上は step3_ready にする colB = "step3_ready" # ----------------------------- # Step3グループ: 題名 / サムネ / 字幕 # ----------------------------- if colB == "step3_ready": # (1) 題名（D列）→ Vimeoタイトル if colD: tprint(f"題名更新={colD}") _ = update_vimeo_title(colC, colD) log_parts.append("title ok") # (2) サムネ（F列） if colF: img_url = normalize_dropbox_link(colF) tprint("サムネ最終URL=", img_url) _ = upload_thumbnail_to_vimeo(colC, img_url) log_parts.append("thumb ok") # (3) 字幕（J列） if colJ: vtt_url = normalize_dropbox_link(colJ) tprint("字幕最終URL=", vtt_url) _ = upload_subtitle_to_vimeo(colC, vtt_url) log_parts.append("vtt ok") # グループ完了 → FALSE colB = "FALSE" # ----------------------------- # シート更新 # ----------------------------- qmsg = " | ".join(log_parts) + (f" ({ts})" if log_parts else "") updates = {"B": colB} if colC: updates["C"] = colC if qmsg.strip(): updates["Q"] = qmsg sc.update_row_by_letters(row_idx, updates) return (row.get("B") or "").strip(), updates # --------------------------------------------- # エントリポイント # --------------------------------------------- def main(): import argparse parser = argparse.ArgumentParser(description="Vimeo 更新 Runner") parser.add_argument("--row", type=int, required=True, help="対象行番号") args = parser.parse_args() before, after = run_once(args.row) print(f"[runner] {before} -> {after['B']}") return 0 if __name__ == "__main__": sys.exit(main()) kick_flags.py # -*- coding: utf-8 -*- """ kick_flags.py (省リクエスト・型安全・E列=vimeoフィルタ) - F2(シート1)の B〜E 列を一括取得し、 * B列が step2_ready / step3_ready * かつ E列が vimeo の行だけ vimeo_upd.runner を起動するキッカー。 - 429(API制限)が出たら待機して再試行。 - 各行の runner 実行間にスリープを入れてAPIにやさしく。 環境変数(任意): KICK_MAX_ROW ... 走査上限行 (既定: 2000) KICK_SLEEP_BETWEEN_RUNS_S ... 各行の実行間隔秒 (既定: 1) KICK_BATCH_RANGE ... 取得レンジ (例 "B2:E1200") を手動指定したい場合に使用 """ import os import time import subprocess from gspread.exceptions import APIError from vimeo_upd.gateways import SheetClient TARGET_STATES = {"step2_ready", "step3_ready"} def _cell_to_text(cell): """セル値を安全にテキストへ。None/真偽/数値も扱う""" if cell is None: return "" if isinstance(cell, str): return cell return str(cell) def _get_scan_range(max_row: int) -> str: # 任意で手動レンジを使いたい場合（デバッグ等） custom = os.getenv("KICK_BATCH_RANGE", "").strip() if custom: return custom return f"B2:E{max_row}" def _list_rows_to_run(ws, max_row: int): """B〜E列を一括で取得し、条件を満たす行番号のリストを返す""" rng = _get_scan_range(max_row) # TRUE/FALSE 等のフォーマット済み文字列で受け取る vals = ws.get(rng, value_render_option="FORMATTED_VALUE") # vals: [[B,C,D,E], [B,C,D,E], ...] を想定 to_run = [] start_row = 2 # B2:E2 が最初 for i, row in enumerate(vals, start=start_row): b = _cell_to_text(row[0] if len(row) > 0 else "").strip().lower() # 状況 e = _cell_to_text(row[3] if len(row) > 3 else "").strip().lower() # 所在（vimeo か） if b in TARGET_STATES and e == "vimeo": to_run.append(i) return to_run def main(): sc = SheetClient.from_env() max_row = int(os.getenv("KICK_MAX_ROW", "2000")) gap_s = float(os.getenv("KICK_SLEEP_BETWEEN_RUNS_S", "1")) while True: try: ws = sc.open_worksheet(sc.f2_sheet_name) rows = _list_rows_to_run(ws, max_row) if not rows: print("[kick] no rows to run") return print(f"[kick] rows to run: {rows}") for row_idx in rows: print(f"[kick] run runner for row={row_idx}") try: subprocess.run( ["python3", "-m", "vimeo_upd.runner", "--row", str(row_idx)], check=True, ) except subprocess.CalledProcessError as e: print(f"[kick] runner failed row={row_idx}: {e}") # 必要に応じて B=error に変更（任意） try: sc.update_row_by_letters(row_idx, {"B": "error"}) except Exception as ee: print(f"[kick] failed to mark error on row={row_idx}: {ee}") time.sleep(gap_s) return # 1回分のスキャンを終えて終了 except APIError as e: # 429 などで落ちた場合は待って再試行 print(f"[kick] APIError: {e}") time.sleep(60) except Exception as e: print(f"[kick] unexpected error: {e}") time.sleep(10) if __name__ == "__main__": main() /Users/xxxxxxxxm1/Library/LaunchAgents/com.xxxxxxxx.vimeo-kick.plist com.xxxxxxxx.vimeo-kick.plist Label com.xxxxxxxx.vimeo-kick ProgramArguments /usr/bin/python3 -m vimeo_upd.kick_flags StartInterval 20 RunAtLoad WorkingDirectory /Users/xxxxxxxxm1/python_scripts StandardOutPath /Users/xxxxxxxxm1/python_scripts/kick_flags.log StandardErrorPath /Users/xxxxxxxxm1/python_scripts/kick_flags.err KeepAlive EnvironmentVariables PATH /usr/local/bin:/opt/homebrew/bin:/usr/bin:/bin:/usr/sbin:/sbin:/Library/Developer/CommandLineTools/usr/bin GOOGLE_APPLICATION_CREDENTIALS /Users/xxxxxxxxm1/keys/service.json GDOC_SPREADSHEET_ID 17TMSsh8OF8bgcP0NEuM3X67oPfACVSmkGvy8w-DgrZY GDOC_SHEET_NAME シート1 F1_SPREADSHEET_ID <あなたのspreadsheetID> F1_SHEET_NAME sheet1 F1_PLAY_URL_HEADER WP再生URL F1_DROPBOX_URL_HEADER DropboxリンクURL VIMEO_ACCESS_TOKEN 9b47bf4c3db8a8e2e6da1a47bc0da106 VIMEO_SUB_LANG ja VIMEO_SUB_NAME 日本語